multi-agent-mux/tests/test_issue72_installer.py

#!/usr/bin/env python3
"""Issue #72 — the installer side of the multi-agent-mux-update skill.

The update skill runs the new install.sh in place over an existing install
(no remove.sh, .mam kept). These pin what that in-place run must do:
install_state keeps one line per key, no updater is shipped any more and an
old one is cleaned up, and framework files the new version no longer ships
are deleted unless the user modified them.
"""
from __future__ import annotations

import os
import shutil
import subprocess
import tempfile

import pytest

REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))

# Extra framework-owned files only the "old" source ships.
RETIRED_SKILL = ".agents/skills/multi-agent-mux-retired72"
RETIRED_UNMODIFIED = [RETIRED_SKILL + "/SKILL.md", RETIRED_SKILL + "/scripts/old.sh"]
RETIRED_SHARED_DIR = ".agents/skills/multi-agent-mux-retired72b"
RETIRED_SHARED = RETIRED_SHARED_DIR + "/SKILL.md"
RETIRED_MODIFIED = ".agents/hooks/retired72_hook.sh"


def _copy_tree(dest):
    shutil.copytree(REPO_ROOT, dest,
                    ignore=shutil.ignore_patterns(".git", ".venv", ".mam", ".mam_deploy",
                                                  "__pycache__", ".pytest_cache"),
                    symlinks=True)


@pytest.fixture(scope="module")
def sources():
    """(new, old): the tree as is, and the same tree plus files it no longer ships."""
    tmp = tempfile.mkdtemp(prefix="mam_issue72_src_")
    new, old = os.path.join(tmp, "new"), os.path.join(tmp, "old")
    _copy_tree(new)
    _copy_tree(old)
    for rel in RETIRED_UNMODIFIED + [RETIRED_SHARED, RETIRED_MODIFIED]:
        path = os.path.join(old, rel)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w") as f:
            f.write("# shipped by the old version only: %s\n" % rel)
    try:
        yield new, old
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


@pytest.fixture
def ws():
    tmp = tempfile.mkdtemp(prefix="mam_issue72_ws_")
    try:
        yield os.path.join(tmp, "workspace")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def _install(src, ws):
    env = dict(os.environ)
    env.update({"MAM_REPO_URL": src, "MAM_SKIP_VENV": "1"})
    res = subprocess.run(["bash", os.path.join(src, "deploy", "install.sh"), ws],
                         env=env, capture_output=True, text=True, timeout=600)
    assert res.returncode == 0, res.stdout + res.stderr
    return res


def _remove(ws):
    res = subprocess.run(["bash", os.path.join(ws, ".mam_deploy", "remove.sh"), "--force"],
                         cwd=ws, capture_output=True, text=True, timeout=600)
    assert res.returncode == 0, res.stdout + res.stderr
    return res


def _read(ws, rel):
    with open(os.path.join(ws, rel)) as f:
        return f.read()


def _manifest(ws):
    return _read(ws, ".mam/install_manifest.txt").splitlines()


def _hashed_paths(ws):
    return [line.split(None, 1)[1] for line in _read(ws, ".mam/asset_hashes.txt").splitlines()
            if line.strip()]


def test_in_place_reinstall_keeps_one_line_per_install_state_key(sources, ws):
    new, _ = sources
    _install(new, ws)
    with open(os.path.join(ws, ".mam", "install_state"), "w") as f:
        f.write("first=a\ngitignore_created=1\nlast=b\n")

    _install(new, ws)

    assert _read(ws, ".mam/install_state").splitlines() == [
        "first=a", "gitignore_created=1", "last=b"]
    # remove.sh must read the key as 1: the .gitignore it created goes away.
    _remove(ws)
    assert not os.path.exists(os.path.join(ws, ".gitignore"))


def test_fresh_install_ships_no_updater(sources, ws):
    new, _ = sources
    res = _install(new, ws)
    assert not os.path.exists(os.path.join(ws, ".mam_deploy", "update.sh"))
    manifest = _manifest(ws)
    assert ".mam_deploy/update.sh" not in manifest and "update.sh" not in manifest
    assert "bash .mam_deploy/update.sh" not in res.stdout
    assert "multi-agent-mux-update/scripts/update.sh" in res.stdout


def test_in_place_reinstall_deletes_an_old_mam_deploy_updater(sources, ws):
    new, _ = sources
    _install(new, ws)
    with open(os.path.join(ws, ".mam_deploy", "update.sh"), "w") as f:
        f.write("#!/usr/bin/env bash\n# old updater\n")
    with open(os.path.join(ws, ".mam", "install_manifest.txt"), "a") as f:
        f.write(".mam_deploy/update.sh\n")

    _install(new, ws)

    assert not os.path.exists(os.path.join(ws, ".mam_deploy", "update.sh"))
    assert ".mam_deploy/update.sh" not in _manifest(ws)


def test_legacy_root_updater_goes_only_when_the_manifest_lists_it(sources, ws):
    new, _ = sources
    _install(new, ws)
    root_updater = os.path.join(ws, "update.sh")

    # Listed: the installer put it there (very old layout).
    with open(root_updater, "w") as f:
        f.write("# legacy updater\n")
    with open(os.path.join(ws, ".mam", "install_manifest.txt"), "a") as f:
        f.write("update.sh\n")
    _install(new, ws)
    assert not os.path.exists(root_updater)
    assert not os.path.exists(os.path.join(ws, ".mam_deploy", "update.sh"))
    manifest = _manifest(ws)
    assert "update.sh" not in manifest and ".mam_deploy/update.sh" not in manifest

    # Not listed: the user's own file stays untouched.
    with open(root_updater, "w") as f:
        f.write("# user's own script\n")
    _install(new, ws)
    assert _read(ws, "update.sh") == "# user's own script\n"


def test_files_the_new_version_no_longer_ships(sources, ws):
    new, old = sources
    _install(old, ws)
    for rel in RETIRED_UNMODIFIED + [RETIRED_SHARED, RETIRED_MODIFIED]:
        assert rel in _manifest(ws) and rel in _hashed_paths(ws), rel
    # The user edits one retired file and adds their own file next to another.
    with open(os.path.join(ws, RETIRED_MODIFIED), "a") as f:
        f.write("# my change\n")
    with open(os.path.join(ws, RETIRED_SHARED_DIR, "notes.txt"), "w") as f:
        f.write("mine\n")
    # A base/ copy of a retired file must go with it.
    base_copy = os.path.join(ws, ".mam", "base", RETIRED_UNMODIFIED[0])
    os.makedirs(os.path.dirname(base_copy))
    shutil.copy(os.path.join(ws, RETIRED_UNMODIFIED[0]), base_copy)

    res = _install(new, ws)

    for rel in RETIRED_UNMODIFIED + [RETIRED_SHARED]:
        assert not os.path.exists(os.path.join(ws, rel)), rel
    assert not os.path.exists(os.path.join(ws, RETIRED_SKILL)), "emptied directory left behind"
    assert _read(ws, RETIRED_SHARED_DIR + "/notes.txt") == "mine\n"
    assert "# my change" in _read(ws, RETIRED_MODIFIED)
    assert "Local modification detected" in res.stderr and RETIRED_MODIFIED in res.stderr
    for rel in RETIRED_UNMODIFIED:
        assert rel not in res.stderr, rel
    assert not os.path.exists(base_copy)
    manifest, hashed = _manifest(ws), _hashed_paths(ws)
    for rel in RETIRED_UNMODIFIED + [RETIRED_SHARED, RETIRED_MODIFIED]:
        assert rel not in manifest and rel not in hashed, rel
    # Still-shipped framework files are untouched by the cleanup.
    assert ".agents/skills/lib.sh" in manifest and ".agents/skills/lib.sh" in hashed

    # The kept file is the user's now: uninstalling leaves it in place.
    _remove(ws)
    assert "# my change" in _read(ws, RETIRED_MODIFIED)


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0,
                    reason="root reads a mode-000 file")
def test_local_source_update_backup_is_not_staged(sources, ws):
    new, _ = sources
    backup = os.path.join(new, ".mam", "update-backup")
    os.makedirs(backup)
    unreadable = os.path.join(backup, "unreadable")
    with open(unreadable, "w") as f:
        f.write("x")
    os.chmod(unreadable, 0)
    try:
        _install(new, ws)
    finally:
        os.chmod(unreadable, 0o600)
        shutil.rmtree(os.path.join(new, ".mam"))