feat(deploy): implement lib_ownership, registry merge, and deployment test suite

2026-08-07 23:48:49 +09:00
parent cc11a02784
commit 399242dac5
11 changed files with 939 additions and 172 deletions
@@ -0,0 +1,255 @@
+"""Deploy key-level registry merge tests for .agents/hooks.json.
+
+Tests for Rev.2 key-level 3-way merge rules (R-1 to R-10).
+"""
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+
+import pytest
+
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+
+
+@pytest.fixture
+def ws():
+    """A pristine source copy plus empty target directory."""
+    tmp = tempfile.mkdtemp(prefix="mam_deploy_reg_")
+    src, tgt = os.path.join(tmp, "src"), os.path.join(tmp, "tgt")
+    shutil.copytree(REPO_ROOT, src,
+                    ignore=shutil.ignore_patterns(".git", ".venv", ".mam",
+                                                  ".mam_deploy", "__pycache__"),
+                    symlinks=True)
+    os.makedirs(tgt)
+    try:
+        yield src, tgt
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def _install(src, tgt, **extra):
+    env = dict(os.environ)
+    env.update({"MAM_REPO_URL": src, "MAM_SKIP_VENV": "1",
+                "MAM_SKIP_GITIGNORE": "1"})
+    env.update(extra)
+    return subprocess.run(["bash", os.path.join(src, "deploy", "install.sh"), tgt],
+                          env=env, capture_output=True, text=True)
+
+
+def _install_mam(src, tgt):
+    return subprocess.run(["bash", os.path.join(src, "deploy", "install_mam.sh"),
+                          "--target", tgt], capture_output=True, text=True)
+
+
+def _read_json(path):
+    with open(path, encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _write_json(path, data):
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2)
+
+
+def _add_user_hook(tgt, name="custom-user-hook"):
+    p = os.path.join(tgt, ".agents", "hooks.json")
+    d = _read_json(p)
+    d[name] = {"PreToolUse": [{"matcher": "user_action", "hooks": [{"type": "command", "command": "echo user"}]}]}
+    _write_json(p, d)
+    return name
+
+
+def _ship_new_hook(src, name="mam-loop-timeout-guard"):
+    script = os.path.join(src, ".agents", "hooks", "loop_timeout_guard.sh")
+    with open(script, "w") as f:
+        f.write('#!/usr/bin/env bash\necho \'{"decision":"allow"}\'\n')
+    os.chmod(script, 0o755)
+    p = os.path.join(src, ".agents", "hooks.json")
+    d = _read_json(p)
+    d[name] = {"PreToolUse": [{"matcher": "file_change", "hooks": [
+        {"type": "command", "command": "./hooks/loop_timeout_guard.sh", "timeout": 10}]}]}
+    _write_json(p, d)
+    return name
+
+
+def test_r1_user_hook_does_not_block_new_framework_hook(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    user_hook = _add_user_hook(tgt)
+    new_hook = _ship_new_hook(src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    tgt_data = _read_json(os.path.join(tgt, ".agents", "hooks.json"))
+    assert user_hook in tgt_data, "User's custom hook was lost"
+    assert new_hook in tgt_data, "New framework hook was not delivered"
+
+
+def test_r2_upstream_fix_to_an_existing_hook_lands(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    _add_user_hook(tgt)
+
+    p_src = os.path.join(src, ".agents", "hooks.json")
+    d_src = _read_json(p_src)
+    key = list(d_src.keys())[0]
+    d_src[key]["updated_by_upstream"] = True
+    _write_json(p_src, d_src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    tgt_data = _read_json(os.path.join(tgt, ".agents", "hooks.json"))
+    assert tgt_data.get(key, {}).get("updated_by_upstream") is True
+
+
+def test_r3_user_modified_hook_is_preserved_and_reported(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    p_tgt = os.path.join(tgt, ".agents", "hooks.json")
+    d_tgt = _read_json(p_tgt)
+    key = list(d_tgt.keys())[0]
+    d_tgt[key]["user_custom_setting"] = 123
+    _write_json(p_tgt, d_tgt)
+
+    p_src = os.path.join(src, ".agents", "hooks.json")
+    d_src = _read_json(p_src)
+    d_src[key]["upstream_competing_setting"] = 456
+    _write_json(p_src, d_src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    tgt_data = _read_json(p_tgt)
+    assert tgt_data[key].get("user_custom_setting") == 123, "User edit was overwritten"
+    assert "kept yours" in res.stderr or "kept your" in res.stderr or "Registry merged" in res.stderr
+
+
+def test_r4_retired_upstream_hook_removed_if_unmodified(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    p_src = os.path.join(src, ".agents", "hooks.json")
+    d_src = _read_json(p_src)
+    retired_key = list(d_src.keys())[0]
+    del d_src[retired_key]
+    _write_json(p_src, d_src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    tgt_data = _read_json(os.path.join(tgt, ".agents", "hooks.json"))
+    assert retired_key not in tgt_data, "Retired unmodified hook was not removed"
+
+
+def test_r4b_retired_upstream_hook_kept_if_user_modified(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    p_tgt = os.path.join(tgt, ".agents", "hooks.json")
+    d_tgt = _read_json(p_tgt)
+    retired_key = list(d_tgt.keys())[0]
+    d_tgt[retired_key]["modified_by_user"] = True
+    _write_json(p_tgt, d_tgt)
+
+    p_src = os.path.join(src, ".agents", "hooks.json")
+    d_src = _read_json(p_src)
+    del d_src[retired_key]
+    _write_json(p_src, d_src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    tgt_data = _read_json(p_tgt)
+    assert retired_key in tgt_data, "User-modified retired hook should be kept"
+
+
+def test_r5_merge_preserves_backup_and_stderr_notification(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    _add_user_hook(tgt)
+    _ship_new_hook(src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+    assert "Registry merged" in res.stderr or "added" in res.stderr
+
+    backup_dir = os.path.join(tgt, ".mam", "skill-backups")
+    assert os.path.exists(backup_dir) and len(os.listdir(backup_dir)) > 0
+
+
+def test_r6_missing_base_falls_back_to_additive_merge(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    shutil.rmtree(os.path.join(tgt, ".mam", "base"), ignore_errors=True)
+
+    user_hook = _add_user_hook(tgt)
+    new_hook = _ship_new_hook(src)
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    tgt_data = _read_json(os.path.join(tgt, ".agents", "hooks.json"))
+    assert user_hook in tgt_data
+    assert new_hook in tgt_data
+
+
+def test_r7_corrupt_registry_preserved_and_install_continues(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    p_tgt = os.path.join(tgt, ".agents", "hooks.json")
+    with open(p_tgt, "w") as f:
+        f.write("INVALID JSON {{{")
+
+    res = _install(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    with open(p_tgt) as f:
+        assert "INVALID JSON" in f.read()
+
+
+def test_r8_merge_is_idempotent(ws):
+    src, tgt = ws
+    assert _install(src, tgt).returncode == 0
+
+    _add_user_hook(tgt)
+    _ship_new_hook(src)
+
+    res1 = _install(src, tgt)
+    assert res1.returncode == 0
+    t1_content = open(os.path.join(tgt, ".agents", "hooks.json")).read()
+
+    res2 = _install(src, tgt)
+    assert res2.returncode == 0
+    t2_content = open(os.path.join(tgt, ".agents", "hooks.json")).read()
+
+    assert t1_content == t2_content
+
+
+def test_r9_install_mam_records_hash_db_and_base(ws):
+    src, tgt = ws
+    res = _install_mam(src, tgt)
+    assert res.returncode == 0, res.stderr
+
+    assert os.path.exists(os.path.join(tgt, ".mam", "asset_hashes.txt"))
+    assert os.path.exists(os.path.join(tgt, ".mam", "base", ".agents", "hooks.json"))
+
+
+def test_r10_missing_ownership_rules_abort_rather_than_degrade(ws):
+    src, tgt = ws
+    lib = os.path.join(src, "deploy", "lib_ownership.sh")
+    assert os.path.exists(lib), "deploy/lib_ownership.sh is not shipped"
+    os.remove(lib)
+
+    res = _install(src, tgt)
+    assert res.returncode != 0
+    assert "lib_ownership.sh" in res.stderr