multi-agent-mux/tests/test_version_consistency.py

# ==============================================================================
# test_version_consistency.py — 3-Way Version Lockstep & Immutability Test Suite
# ==============================================================================
import os
import re
import subprocess
from pathlib import Path
import yaml

REPO_ROOT = Path(__file__).resolve().parent.parent
LIB_SH = REPO_ROOT / ".agents" / "skills" / "lib.sh"
VERSIONS_MD = REPO_ROOT / "VERSIONS.md"
SKILLS_DIR = REPO_ROOT / ".agents" / "skills"
EAGER_PY = SKILLS_DIR / "lib_py" / "loop" / "eager.py"

SKILL_NAMES = [
    "multi-agent-mux-create",
    "multi-agent-mux-stop",
    "multi-agent-mux-resume",
    "multi-agent-mux-status",
    "multi-agent-mux-monitor",
    "multi-agent-mux-delegate-job",
    "multi-agent-mux-loop",
    "multi-agent-mux-orc-onboard",
]


def _get_lib_sh_mam_version():
    cmd = f'source "{LIB_SH}" && echo "$MAM_VERSION"'
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, cwd=str(REPO_ROOT))
    assert res.returncode == 0, f"Failed to source lib.sh: {res.stderr}"
    return res.stdout.strip()


def _get_versions_md_versions():
    text = VERSIONS_MD.read_text(encoding="utf-8")
    
    # 1. Current release framework version
    m_curr = re.search(r"-\s*\*\*프레임워크 버전\*\*:\s*`v(\d+\.\d+\.\d+)`", text)
    assert m_curr, "Failed to parse Current Release framework version in VERSIONS.md"
    current_framework_ver = m_curr.group(1)

    # 2. Skill matrix table versions
    table_matches = re.findall(r"\|\s*\*\*`([^`]+)`\*\*\s*\|\s*`(\d+\.\d+\.\d+)`\s*\|", text)
    matrix_dict = dict(table_matches)
    
    return current_framework_ver, matrix_dict


def _get_eager_py_framework_version():
    """Read FRAMEWORK_VERSION statically — importing lib_py.loop.eager would
    pull the whole framework onto sys.path just to read one constant."""
    text = EAGER_PY.read_text(encoding="utf-8")
    m = re.search(r'^FRAMEWORK_VERSION\s*=\s*"(\d+\.\d+\.\d+)"', text, re.M)
    assert m, f"Failed to parse FRAMEWORK_VERSION in {EAGER_PY}"
    return m.group(1)


def _get_skill_md_frontmatter_version(skill_name: str):
    skill_file = SKILLS_DIR / skill_name / "SKILL.md"
    assert skill_file.is_file(), f"Skill file not found: {skill_file}"
    content = skill_file.read_text(encoding="utf-8")
    
    # Strict frontmatter parsing
    fm_match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
    assert fm_match, f"No YAML frontmatter found in {skill_file}"
    fm_data = yaml.safe_load(fm_match.group(1))
    assert isinstance(fm_data, dict) and "version" in fm_data, f"No version in frontmatter of {skill_file}"
    return str(fm_data["version"])


def test_three_way_version_lockstep():
    """T-4: lib.sh(MAM_VERSION) == VERSIONS.md(Current + Matrix 8 items) == 8x SKILL.md frontmatters."""
    # 1. lib.sh runtime constant
    lib_ver = _get_lib_sh_mam_version()
    assert lib_ver != "", "MAM_VERSION in lib.sh is empty"

    # 2. VERSIONS.md
    vmd_curr, vmd_matrix = _get_versions_md_versions()
    assert lib_ver == vmd_curr, f"lib.sh MAM_VERSION ({lib_ver}) != VERSIONS.md current ({vmd_curr})"

    for skill in SKILL_NAMES:
        assert skill in vmd_matrix, f"Skill {skill} missing from VERSIONS.md matrix"
        assert vmd_matrix[skill] == lib_ver, f"VERSIONS.md matrix for {skill} ({vmd_matrix[skill]}) != lib.sh ({lib_ver})"

    # 3. 8x SKILL.md files
    for skill in SKILL_NAMES:
        fm_ver = _get_skill_md_frontmatter_version(skill)
        assert fm_ver == lib_ver, f"{skill}/SKILL.md version ({fm_ver}) != lib.sh ({lib_ver})"

    # 4. Runtime constant in lib_py/loop/eager.py.
    eager_ver = _get_eager_py_framework_version()
    assert eager_ver == lib_ver, (
        f"eager.py FRAMEWORK_VERSION ({eager_ver}) != lib.sh MAM_VERSION ({lib_ver})"
    )


def test_runtime_version_gate_rejects_a_partially_bumped_tree():
    """A tree where only one of the two runtime constants was bumped must not run.

    lib.sh exports MAM_VERSION into the environment and eager_load() compares it
    against FRAMEWORK_VERSION, so a half-updated deployment fails closed at loop
    startup rather than running two versions of the framework against each other.
    This guards the mechanism; the values themselves are pinned by the lockstep
    test above.
    """
    import importlib.util

    spec = importlib.util.spec_from_file_location("_eager_under_test", EAGER_PY)
    eager = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(eager)

    assert eager.FRAMEWORK_VERSION == _get_lib_sh_mam_version()

    env_before = os.environ.get("MAM_VERSION")
    os.environ["MAM_VERSION"] = "0.0.0-stale"
    try:
        try:
            eager.verify_framework_version()
        except RuntimeError as exc:
            assert "MAM_VERSION mismatch" in str(exc)
        else:
            raise AssertionError("a stale MAM_VERSION was accepted by the runtime gate")
    finally:
        if env_before is None:
            os.environ.pop("MAM_VERSION", None)
        else:
            os.environ["MAM_VERSION"] = env_before


def test_mam_version_is_not_env_overridable():
    """T-0 (Constraint 1): MAM_VERSION cannot be spoofed by caller environment."""
    cmd = f'source "{LIB_SH}" && echo "$MAM_VERSION"'
    env = os.environ.copy()
    env["MAM_VERSION"] = "9.9.9"
    res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env, cwd=str(REPO_ROOT))
    assert res.returncode == 0
    assert res.stdout.strip() != "9.9.9", "MAM_VERSION was improperly overwritten by caller environment"
    assert res.stdout.strip() == _get_lib_sh_mam_version()