256 lines
7.7 KiB
Python
256 lines
7.7 KiB
Python
"""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
|