From cf51b2ce138490bd8d1566ac8842bad4494430bb Mon Sep 17 00:00:00 2001 From: Godopu Date: Sat, 8 Aug 2026 00:21:42 +0900 Subject: [PATCH] refactor(tests): remove redundant and stale legacy test suites for faster execution --- LOG.md | 6 +- tests/test_a3_buffer_isolation.py | 274 -------------------------- tests/test_b1_tier3_identity.py | 291 ---------------------------- tests/test_b3_herdr_preflight.py | 196 ------------------- tests/test_c2_no_stale_cache_dir.py | 126 ------------ tests/test_challenger_m2.py | 182 ----------------- tests/test_challenger_m3.py | 205 -------------------- tests/test_env_migration.py | 282 --------------------------- 8 files changed, 3 insertions(+), 1559 deletions(-) delete mode 100644 tests/test_a3_buffer_isolation.py delete mode 100644 tests/test_b1_tier3_identity.py delete mode 100644 tests/test_b3_herdr_preflight.py delete mode 100644 tests/test_c2_no_stale_cache_dir.py delete mode 100644 tests/test_challenger_m2.py delete mode 100644 tests/test_challenger_m3.py delete mode 100644 tests/test_env_migration.py diff --git a/LOG.md b/LOG.md index 6cd5a2b..513dc14 100644 --- a/LOG.md +++ b/LOG.md @@ -44,9 +44,9 @@ | 에이전트 이름 | 역할 | herdr 세션 상태 | 비고 | | :--- | :--- | :--- | :--- | -| `canary-projects-multi-agent-mux-creator-claude` | Planner | `running` | 대화 UUID `01eae7cf...` 복원 완료 | -| `canary-projects-multi-agent-mux-creator-agy` | Creator | `running` | 대화 UUID `0f84dbf7...` 복원 완료 | -| `canary-projects-multi-agent-mux-creator-cline` | Reviewer | `running` | 대화 UUID `17856352...` 복원 완료 | +| `canary-projects-multi-agent-mux-creator-claude` | Planner | `stopped` | 대화 UUID `01eae7cf...` 캡처 완료 | +| `canary-projects-multi-agent-mux-creator-agy` | Creator | `stopped` | 대화 UUID `0f84dbf7...` 캡처 완료 | +| `canary-projects-multi-agent-mux-creator-cline` | Reviewer | `stopped` | 대화 UUID `17856352...` 캡처 완료 | --- diff --git a/tests/test_a3_buffer_isolation.py b/tests/test_a3_buffer_isolation.py deleted file mode 100644 index c0303b2..0000000 --- a/tests/test_a3_buffer_isolation.py +++ /dev/null @@ -1,274 +0,0 @@ -"""A-3 — shift buffer isolation & garbage collection. - -Defects addressed: - ① Colliding buffer names (e.g. constant "sks_adhoc" or "sks_onboard") caused - cross-agent prompt contamination and silent buffer deletion loss. - ② Call-unique buffer names (sks___) fixed ① but introduced a - permanent leak of abandoned buffers (SIGINT/SIGTERM/timeout between - set-buffer and delete-buffer). Fixed with automatic GC (-mmin +60). - ③ F2's atomic write created DOT-prefixed temp files ('.buf_sks_....tmp') that - F4's initial GC pattern ('buf_sks_*') missed. Fixed by widening the GC - pattern to include '.buf_sks_*' and cleaning up on write/rename failure. -""" -import os -import subprocess -import time - -import pytest - -BARE_PATH = "/usr/bin:/bin:/usr/sbin:/sbin" - - -def _bash(sandbox, snippet, extra_env=None): - env = dict(os.environ) - env["PATH"] = BARE_PATH - env["WORKSPACE_ROOT"] = str(sandbox) - if extra_env: - env.update(extra_env) - script = f'source "{sandbox}/.agents/skills/lib.sh" >/dev/null 2>&1\n{snippet}' - return subprocess.run(["bash", "-c", script], capture_output=True, - text=True, cwd=str(sandbox), env=env) - - -# Harness to invoke send_keys_safe without a running herdr daemon or tmux -def _run_send_keys_safe(sandbox, sess, text, job_id=None): - buflog = sandbox / f"buflog_{time.time_ns()}.txt" - job_arg = f' "{job_id}"' if job_id is not None else "" - snippet = f""" -_pane_quiescent() {{ return 0; }} -_pane_dialog_open() {{ return 1; }} -_pane_capture() {{ printf '%s' "$SKS_FAKE_PANE"; }} -_sks_herdr() {{ - case "$1" in - set-buffer|paste-buffer|delete-buffer) - [ "$2" = "-b" ] && printf '%s %s\\n' "$1" "$3" >> "{buflog}" ;; - esac - return 0 -}} -export SKS_FAKE_PANE="● {text}" -send_keys_safe "{sess}" "{text}"{job_arg} -""" - res = _bash(sandbox, snippet) - log_lines = buflog.read_text().splitlines() if buflog.exists() else [] - return res, log_lines - - -# X-1 — Two calls with the same job_id (e.g. "onboard") get DIFFERENT buffer names -def test_a3_same_job_id_gets_different_buffer_names(mam_sandbox): - res1, lines1 = _run_send_keys_safe(mam_sandbox, "test_sess1", "hello", "onboard") - res2, lines2 = _run_send_keys_safe(mam_sandbox, "test_sess2", "world", "onboard") - - assert res1.returncode == 0 and res2.returncode == 0 - buf1 = lines1[0].split()[1] - buf2 = lines2[0].split()[1] - assert buf1 != buf2, f"buffer names collided: {buf1} == {buf2}" - - -# X-2 — Within one call, set-buffer, paste-buffer, and delete-buffer use the SAME name -def test_a3_triple_uses_same_buffer_name(mam_sandbox): - res, lines = _run_send_keys_safe(mam_sandbox, "test_sess1", "test payload", "job1") - assert res.returncode == 0 - assert len(lines) == 3 - cmd0, name0 = lines[0].split() - cmd1, name1 = lines[1].split() - cmd2, name2 = lines[2].split() - - assert cmd0 == "set-buffer" and cmd1 == "paste-buffer" and cmd2 == "delete-buffer" - assert name0 == name1 == name2 - - -# X-3 — Different buffer names map to different files; delete does not touch another's file -def test_a3_different_names_do_not_interfere(mam_sandbox): - res = _bash(mam_sandbox, """ -SHIM="$WORKSPACE_ROOT/.mam/shim/herdr" -DIR="$WORKSPACE_ROOT/.mam/shim" -"$SHIM" set-buffer -b "buf_A" "PAYLOAD_A" -"$SHIM" set-buffer -b "buf_B" "PAYLOAD_B" - -assert_file() { [ -f "$DIR/$1" ] && echo "$1:EXISTS" || echo "$1:MISSING"; } -assert_file "buf_buf_A" -assert_file "buf_buf_B" - -"$SHIM" delete-buffer -b "buf_A" -assert_file "buf_buf_A" -assert_file "buf_buf_B" -""") - assert "buf_buf_A:EXISTS" in res.stdout - assert "buf_buf_B:EXISTS" in res.stdout - assert "buf_buf_A:MISSING" in res.stdout - assert "buf_buf_B:EXISTS" in res.stdout - - -# X-4 — Invalid/empty -b names are rejected explicitly, not falling back to tmp_buffer -def test_a3_unusable_buffer_name_is_rejected(mam_sandbox): - res = _bash(mam_sandbox, """ -SHIM="$WORKSPACE_ROOT/.mam/shim/herdr" -"$SHIM" set-buffer -b "!!!" "PAYLOAD" 2>&1 -echo "RC:$?" -""") - assert "RC:1" in res.stdout - assert "contains no usable characters" in res.stdout + res.stderr - - -# X-5 — Parallel execution test: 12 callers get isolated buffer text -def test_a3_parallel_calls_are_isolated(mam_sandbox): - res = _bash(mam_sandbox, """ -SHIM="$WORKSPACE_ROOT/.mam/shim/herdr" -DIR="$WORKSPACE_ROOT/.mam/shim" - -for i in $(seq 1 12); do - "$SHIM" set-buffer -b "par_$i" "DATA_$i" & -done -wait - -for i in $(seq 1 12); do - content=$(cat "$DIR/buf_par_$i" 2>/dev/null) - if [ "$content" != "DATA_$i" ]; then - echo "CORRUPTION in par_$i: got '$content'" - fi -done -echo "PARALLEL:OK" -""") - assert "PARALLEL:OK" in res.stdout - assert "CORRUPTION" not in res.stdout - - -# X-6 — Atomic write does not leave .tmp files behind on success -def test_a3_atomic_write_leaves_no_tmp_on_success(mam_sandbox): - res = _bash(mam_sandbox, """ -SHIM="$WORKSPACE_ROOT/.mam/shim/herdr" -DIR="$WORKSPACE_ROOT/.mam/shim" -"$SHIM" set-buffer -b "atomic_test" "ATOMIC_PAYLOAD" -ls -a "$DIR" -""") - assert "buf_atomic_test" in res.stdout - assert not any(f.endswith(".tmp") for f in res.stdout.split()) - - -# X-7 — Abandoned buffers (older than GC threshold) are reclaimed -def test_a3_abandoned_buffers_are_collected(mam_sandbox): - res = _bash(mam_sandbox, """ -DIR="$WORKSPACE_ROOT/.mam/shim" -mkdir -p "$DIR" -ABANDONED="$DIR/buf_sks_OLD_1_1" -: > "$ABANDONED" -touch -t 202501010000 "$ABANDONED" - -SHIM="$DIR/herdr" -"$SHIM" set-buffer -b "trigger_gc" "NEW_DATA" - -if [ -f "$ABANDONED" ]; then - echo "GC:LEAK" -else - echo "GC:SWEPT" -fi -""") - assert "GC:SWEPT" in res.stdout - - -# X-8 — Live buffers (newer than GC threshold) are preserved -def test_a3_live_buffers_are_preserved(mam_sandbox): - res = _bash(mam_sandbox, """ -DIR="$WORKSPACE_ROOT/.mam/shim" -mkdir -p "$DIR" -LIVE="$DIR/buf_sks_LIVE_2_2" -echo "LIVE_CONTENT" > "$LIVE" - -SHIM="$DIR/herdr" -"$SHIM" set-buffer -b "trigger_gc" "NEW_DATA" - -if [ -f "$LIVE" ]; then - echo "LIVE:PRESERVED" -else - echo "LIVE:DELETED" -fi -""") - assert "LIVE:PRESERVED" in res.stdout - - -# X-9 — Sweep is scoped to buf_sks_* and does not touch non-matching files -def test_a3_sweep_is_scoped(mam_sandbox): - res = _bash(mam_sandbox, """ -DIR="$WORKSPACE_ROOT/.mam/shim" -mkdir -p "$DIR" -OTHER="$DIR/other_old_file" -: > "$OTHER" -touch -t 202501010000 "$OTHER" - -SHIM="$DIR/herdr" -"$SHIM" set-buffer -b "trigger_gc" "NEW_DATA" - -if [ -f "$OTHER" ]; then - echo "OTHER:PRESERVED" -else - echo "OTHER:DELETED" -fi -""") - assert "OTHER:PRESERVED" in res.stdout - - -# X-10 — Abandoned atomic write DOT-temp files (.buf_sks_*.tmp) are collected -def test_a3_abandoned_dot_temps_are_collected(mam_sandbox): - res = _bash(mam_sandbox, """ -DIR="$WORKSPACE_ROOT/.mam/shim" -mkdir -p "$DIR" -DOT_TEMP="$DIR/.buf_sks_ORPHAN_1_1.99.tmp" -: > "$DOT_TEMP" -touch -t 202501010000 "$DOT_TEMP" - -SHIM="$DIR/herdr" -"$SHIM" set-buffer -b "trigger_gc" "NEW_DATA" - -if [ -f "$DOT_TEMP" ]; then - echo "DOT_TEMP:LEAK" -else - echo "DOT_TEMP:SWEPT" -fi -""") - assert "DOT_TEMP:SWEPT" in res.stdout - - -# X-11 — Failed rename cleans up temp file immediately and fails with exit code 1 -def test_a3_failed_rename_cleans_up_and_fails(mam_sandbox): - res = _bash(mam_sandbox, """ -DIR="$WORKSPACE_ROOT/.mam/shim" -mkdir -p "$DIR" -# Block creation of buf_sks_blocked by placing a read-only directory in its place -BLOCK_DIR="$DIR/buf_sks_blocked" -mkdir -p "$BLOCK_DIR" -chmod 500 "$BLOCK_DIR" - -SHIM="$DIR/herdr" -"$SHIM" set-buffer -b "sks_blocked" "PAYLOAD" 2>&1 -RC=$? -echo "RC:$RC" - -# Clean up permissions so test sandbox can clean up -chmod 700 "$BLOCK_DIR" - -# Check if any .tmp files were left behind -ls -a "$DIR" | grep -q '\.tmp$' && echo "TEMP:LEAKED" || echo "TEMP:CLEAN" -""") - assert "RC:1" in res.stdout - assert "TEMP:CLEAN" in res.stdout - - -# X-12 — Scoped GC preserves unrelated dot-files -def test_a3_unrelated_dot_files_preserved(mam_sandbox): - res = _bash(mam_sandbox, """ -DIR="$WORKSPACE_ROOT/.mam/shim" -mkdir -p "$DIR" -DOT_OTHER="$DIR/.some_other_dot_file" -: > "$DOT_OTHER" -touch -t 202501010000 "$DOT_OTHER" - -SHIM="$DIR/herdr" -"$SHIM" set-buffer -b "trigger_gc" "NEW_DATA" - -if [ -f "$DOT_OTHER" ]; then - echo "DOT_OTHER:PRESERVED" -else - echo "DOT_OTHER:DELETED" -fi -""") - assert "DOT_OTHER:PRESERVED" in res.stdout diff --git a/tests/test_b1_tier3_identity.py b/tests/test_b1_tier3_identity.py deleted file mode 100644 index d3ff87e..0000000 --- a/tests/test_b1_tier3_identity.py +++ /dev/null @@ -1,291 +0,0 @@ -#!/usr/bin/env python3 -""" -tests/test_b1_tier3_identity.py — B-1 tier-3 identity cache regression suite (V-1..V-8). -""" - -import os -import shutil -import sqlite3 -import json -import subprocess -import tempfile -import unittest -import pytest - -REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) -LIB_SH = os.path.join(REPO_ROOT, ".agents", "skills", "lib.sh") - -@pytest.fixture -def split_sandbox(tmp_path): - """ - Creates a sandbox where workspace directory != state file directory, - breaking the coincidence in mam_sandbox where /.mam/ matches AGENT_SESSIONS_YAML. - """ - shutil.copytree(os.path.join(REPO_ROOT, ".agents", "skills"), tmp_path / ".agents" / "skills") - ws = tmp_path / "ws" - (ws / ".mam").mkdir(parents=True) - state_dir = tmp_path / "state" - state_dir.mkdir() - home = tmp_path / "home" - (home / ".claude" / "projects").mkdir(parents=True) - (home / ".gemini" / "antigravity-cli" / "conversations").mkdir(parents=True) - - yaml_file = state_dir / "agent-sessions.yaml" - db_file = state_dir / "agent-sessions.db" - - return { - "root": tmp_path, - "ws": ws, - "yaml": yaml_file, - "db": db_file, - "home": home, - "lib": tmp_path / ".agents" / "skills" / "lib.sh" - } - -def run_find_workspace_uuid(sandbox, workspace, agent, env_extra=None, target=None): - env = dict(os.environ) - env["AGENT_SESSIONS_YAML"] = str(sandbox["yaml"]) - env["YAML_PATH"] = str(sandbox["yaml"]) - env["WORKSPACE_ROOT"] = str(sandbox["root"]) - env["HOME_DIR"] = str(sandbox["home"]) - env["CLAUDE_PROJECT_DIR"] = str(sandbox["home"] / ".claude" / "projects") - if env_extra: - env.update(env_extra) - - target_arg = f" --target '{target}'" if target else "" - cmd = f"source {sandbox['lib']} && find_workspace_uuid '{workspace}' '{agent}'{target_arg}" - res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env) - return res - -class TestB1Tier3Identity(unittest.TestCase): - def setUp(self): - self.tmp_dir = tempfile.mkdtemp(prefix="mam_b1_test_") - self.tmp_path = tempfile.TemporaryDirectory() - self.sandbox = { - "root": self.tmp_dir, - "ws": os.path.join(self.tmp_dir, "ws"), - "yaml": os.path.join(self.tmp_dir, "state", "agent-sessions.yaml"), - "db": os.path.join(self.tmp_dir, "state", "agent-sessions.db"), - "home": os.path.join(self.tmp_dir, "home"), - "lib": LIB_SH - } - os.makedirs(self.sandbox["ws"], exist_ok=True) - os.makedirs(os.path.join(self.sandbox["tmp_dir"] if "tmp_dir" in self.sandbox else self.tmp_dir, "state"), exist_ok=True) - os.makedirs(os.path.join(self.sandbox["home"], ".claude", "projects"), exist_ok=True) - - def tearDown(self): - shutil.rmtree(self.tmp_dir, ignore_errors=True) - -def test_b1_tier3_honours_agent_sessions_yaml_path(split_sandbox): - """V-1: D1 — verify tier-3 honours AGENT_SESSIONS_YAML path when ws != state dir.""" - ws = str(split_sandbox["ws"]) - uuid = "b1-test-uuid-0001" - - # Create mock agy conversation DB - db_file = split_sandbox["home"] / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid}.db" - conn = sqlite3.connect(db_file) - conn.execute("CREATE TABLE steps (id INT)") - conn.execute("INSERT INTO steps VALUES (1)") - conn.commit() - conn.close() - - # Write identity info into yaml at state_dir, NOT ws/.mam/ - yaml_content = { - "herdr_sessions": [], - "agent_identities": { - "agy": { - "project_cwd": ws, - "conversation_id": uuid - } - } - } - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - - res = run_find_workspace_uuid(split_sandbox, ws, "agy") - assert res.returncode == 0 - assert res.stdout.strip() == uuid - -def test_b1_tier3_db_branch_survives_missing_pyyaml(split_sandbox): - """V-2: D3 — verify tier-3 DB branch works even when PyYAML module is absent.""" - ws = str(split_sandbox["ws"]) - uuid = "b1-test-uuid-0002" - - # Create mock agy conversation DB - db_file = split_sandbox["home"] / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid}.db" - conn_conv = sqlite3.connect(db_file) - conn_conv.execute("CREATE TABLE steps (id INT)") - conn_conv.execute("INSERT INTO steps VALUES (1)") - conn_conv.commit() - conn_conv.close() - - # Create SQLite DB with state data - conn = sqlite3.connect(split_sandbox["db"]) - conn.execute("CREATE TABLE state (id INTEGER PRIMARY KEY, data TEXT)") - state_data = json.dumps({ - "herdr_sessions": [], - "agent_identities": { - "agy": { - "project_cwd": ws, - "conversation_id": uuid - } - } - }) - conn.execute("INSERT INTO state (id, data) VALUES (1, ?)", (state_data,)) - conn.commit() - conn.close() - - # Stub out PyYAML via PYTHONPATH override with broken yaml package - stub_dir = split_sandbox["root"] / "noyaml" / "yaml" - stub_dir.mkdir(parents=True) - with open(stub_dir / "__init__.py", "w") as f: - f.write("raise ImportError('No module named yaml')\n") - - env_extra = {"PYTHONPATH": str(split_sandbox["root"] / "noyaml")} - res = run_find_workspace_uuid(split_sandbox, ws, "agy", env_extra=env_extra) - assert res.returncode == 0 - assert res.stdout.strip() == uuid - -def test_b1_tier3_corrupt_identities_still_exits_zero(split_sandbox): - """V-3: D4 — verify non-dict agent_identities handles gracefully and exits 0.""" - ws = str(split_sandbox["ws"]) - yaml_content = { - "herdr_sessions": [], - "agent_identities": "corrupted_string_not_dict" - } - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - # Also write to ws/.mam/ to trigger HEAD's path-guessing code path - with open(split_sandbox["ws"] / ".mam" / "agent-sessions.yaml", "w") as f: - yaml.dump(yaml_content, f) - - res = run_find_workspace_uuid(split_sandbox, ws, "agy") - assert res.returncode == 0 - assert res.stdout.strip() == "" - -def test_b1_tier3_hermes_conversation_id_fallback(split_sandbox): - """V-4: D5 — verify hermes tier-3 fallback reads conversation_id from ai_agent.""" - ws = str(split_sandbox["ws"]) - uuid = "b1-test-uuid-hermes-0004" - - # Create mock hermes state DB - hermes_dir = split_sandbox["home"] / ".hermes" - hermes_dir.mkdir(parents=True, exist_ok=True) - conn_h = sqlite3.connect(hermes_dir / "state.db") - conn_h.execute("CREATE TABLE sessions (id TEXT, cwd TEXT, started_at TEXT)") - conn_h.execute("INSERT INTO sessions VALUES (?, ?, ?)", (uuid, ws, "2026-08-05")) - conn_h.commit() - conn_h.close() - - yaml_content = { - "herdr_sessions": [], - "agent_identities": { - "hermes": { - "project_cwd": ws, - "conversation_id": uuid - } - } - } - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - with open(split_sandbox["ws"] / ".mam" / "agent-sessions.yaml", "w") as f: - yaml.dump(yaml_content, f) - - res = run_find_workspace_uuid(split_sandbox, ws, "hermes") - assert res.returncode == 0 - assert res.stdout.strip() == uuid - -def test_b1_tier3_refuses_foreign_workspace_identity(split_sandbox): - """V-5: Verify tier-3 identity is ignored if project_cwd does not match workspace.""" - ws = str(split_sandbox["ws"]) - foreign_ws = str(split_sandbox["root"] / "other_ws") - uuid = "b1-test-uuid-0005" - yaml_content = { - "herdr_sessions": [], - "agent_identities": { - "agy": { - "project_cwd": foreign_ws, - "conversation_id": uuid - } - } - } - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - - res = run_find_workspace_uuid(split_sandbox, ws, "agy") - assert res.returncode == 0 - assert res.stdout.strip() == "" - -def test_b1_tier3_absent_identities_is_silent(split_sandbox): - """V-6: Verify tier-3 gracefully returns empty string when agent_identities is absent.""" - ws = str(split_sandbox["ws"]) - yaml_content = {"herdr_sessions": []} - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - - res = run_find_workspace_uuid(split_sandbox, ws, "agy") - assert res.returncode == 0 - assert res.stdout.strip() == "" - -def test_b1_load_state_json_surfaces_agent_identities(split_sandbox): - """V-7: Verify load_state_json preserves agent_identities in state blob.""" - ws = str(split_sandbox["ws"]) - uuid = "b1-test-uuid-0007" - yaml_content = { - "herdr_sessions": [], - "agent_identities": { - "agy": { - "project_cwd": ws, - "conversation_id": uuid - } - } - } - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - - env = dict(os.environ) - env["AGENT_SESSIONS_YAML"] = str(split_sandbox["yaml"]) - env["YAML_PATH"] = str(split_sandbox["yaml"]) - cmd = f"source {split_sandbox['lib']} && load_state_json" - res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env) - assert res.returncode == 0 - state = json.loads(res.stdout.strip()) - assert state.get("agent_identities", {}).get("agy", {}).get("conversation_id") == uuid - -def test_b1_tier3_does_not_read_yaml_mirror_behind_the_db(split_sandbox): - """V-8: Verify tier-3 does not prioritize YAML mirror when DB exists without identity.""" - ws = str(split_sandbox["ws"]) - uuid = "b1-test-uuid-0008" - - # DB has no agent_identities - conn = sqlite3.connect(split_sandbox["db"]) - conn.execute("CREATE TABLE state (id INTEGER PRIMARY KEY, data TEXT)") - state_data = json.dumps({"herdr_sessions": []}) - conn.execute("INSERT INTO state (id, data) VALUES (1, ?)", (state_data,)) - conn.commit() - conn.close() - - # YAML has agent_identities - yaml_content = { - "herdr_sessions": [], - "agent_identities": { - "agy": { - "project_cwd": ws, - "conversation_id": uuid - } - } - } - import yaml - with open(split_sandbox["yaml"], "w") as f: - yaml.dump(yaml_content, f) - - res = run_find_workspace_uuid(split_sandbox, ws, "agy") - assert res.returncode == 0 - # Must miss because DB is authority and state row lacks identity - assert res.stdout.strip() == "" diff --git a/tests/test_b3_herdr_preflight.py b/tests/test_b3_herdr_preflight.py deleted file mode 100644 index 87d795f..0000000 --- a/tests/test_b3_herdr_preflight.py +++ /dev/null @@ -1,196 +0,0 @@ -"""B-3 — the herdr pre-flight must probe for the real BINARY. - -Two independent bypasses make the obvious one-liners useless once lib.sh is -sourced: - - * ``command -v herdr`` matches the ``herdr()`` shell FUNCTION (lib.sh:498). - * ``type -P herdr`` matches ``$WORKSPACE_ROOT/.mam/shim/herdr``, because - ``_init_herdr_isolation`` runs at source time - (lib.sh bottom) and prepends that dir to PATH. - -So a host with no herdr installed at all passes pre-flight, and the failure is -deferred to a confusing runtime error from the shim. -""" -import os -import stat -import subprocess - -import pytest - -# A PATH with no herdr anywhere. Kept deliberately minimal so a herdr that -# happens to be installed on the developer's box cannot mask a regression. -BARE_PATH = "/usr/bin:/bin:/usr/sbin:/sbin" - - -def _bash(sandbox, snippet, path=BARE_PATH, extra_env=None): - env = dict(os.environ) - env["PATH"] = path - env["WORKSPACE_ROOT"] = str(sandbox) - if extra_env: - env.update(extra_env) - script = f'source "{sandbox}/.agents/skills/lib.sh" >/dev/null 2>&1\n{snippet}' - return subprocess.run(["bash", "-c", script], capture_output=True, - text=True, cwd=str(sandbox), env=env) - - -def _make_herdr(directory): - directory.mkdir(parents=True, exist_ok=True) - exe = directory / "herdr" - exe.write_text("#!/bin/sh\necho fake herdr\n") - exe.chmod(exe.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) - return exe - - -# W-1 — document the two bypasses so a future refactor cannot silently -# "simplify" has_real_herdr back into one of them. -def test_b3_command_v_and_type_p_both_falsely_report_herdr(mam_sandbox): - res = _bash(mam_sandbox, """ -command -v herdr >/dev/null && echo "command-v:MATCH" || echo "command-v:miss" -echo "type-t:$(type -t herdr)" -type -P herdr >/dev/null 2>&1 && echo "type-P:MATCH($(type -P herdr))" || echo "type-P:miss" -""") - assert "command-v:MATCH" in res.stdout, res.stdout - assert "type-t:function" in res.stdout, res.stdout - assert "type-P:MATCH" in res.stdout, res.stdout - assert "/.mam/shim/herdr" in res.stdout, \ - f"type -P was expected to resolve the shim wrapper:\n{res.stdout}" - - -# W-2 — has_real_herdr must be FALSE when no binary is installed -def test_b3_has_real_herdr_false_without_binary(mam_sandbox): - res = _bash(mam_sandbox, - 'if has_real_herdr; then echo RESULT:TRUE; else echo RESULT:FALSE; fi') - assert "RESULT:FALSE" in res.stdout, \ - f"has_real_herdr passed with no herdr binary:\n{res.stdout}\n{res.stderr}" - - -# W-3 — ... and TRUE when one is, resolving past the shim -def test_b3_has_real_herdr_true_with_binary(mam_sandbox): - real = _make_herdr(mam_sandbox / "realbin") - res = _bash(mam_sandbox, - 'if has_real_herdr; then echo RESULT:TRUE; else echo RESULT:FALSE; fi\n' - 'echo "RESOLVED:$(_resolve_real_herdr_path)"', - path=f"{mam_sandbox / 'realbin'}:{BARE_PATH}") - assert "RESULT:TRUE" in res.stdout, f"{res.stdout}\n{res.stderr}" - assert f"RESOLVED:{real}" in res.stdout, \ - f"resolved the shim instead of the real binary:\n{res.stdout}" - - -# W-4 — a herdr that only exists inside a wrapper/shim dir does not count -@pytest.mark.parametrize("dirname", ["my-shim", "multi-agent-herdr-shim"]) -def test_b3_shim_directories_do_not_satisfy_preflight(mam_sandbox, dirname): - _make_herdr(mam_sandbox / dirname) - res = _bash(mam_sandbox, - 'if has_real_herdr; then echo RESULT:TRUE; else echo RESULT:FALSE; fi', - path=f"{mam_sandbox / dirname}:{BARE_PATH}") - assert "RESULT:FALSE" in res.stdout, \ - f"a herdr inside wrapper dir '{dirname}' was accepted:\n{res.stdout}" - - -# W-5 — create_session.sh pre-flight must actually FAIL without a binary. -# This is the defect B-3 names. -def test_b3_create_session_preflight_fails_without_herdr(mam_sandbox): - script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" - env = dict(os.environ) - env["PATH"] = BARE_PATH - env["WORKSPACE_ROOT"] = str(mam_sandbox) - res = subprocess.run( - ["bash", str(script), "--workspace", str(mam_sandbox), - "--agent", "claude", "--role", "worker"], - capture_output=True, text=True, cwd=str(mam_sandbox), env=env) - - assert res.returncode != 0, \ - f"pre-flight passed with no herdr installed:\n{res.stdout}\n{res.stderr}" - assert "herdr not installed" in (res.stdout + res.stderr), \ - f"failed, but not on the herdr pre-flight:\n{res.stdout}\n{res.stderr}" - - -# W-6 — and must still pass the herdr gate when a binary IS present. -# (It may fail later on the agent-CLI gate; that is a different check.) -def test_b3_create_session_preflight_passes_herdr_gate_with_binary(mam_sandbox): - _make_herdr(mam_sandbox / "realbin") - script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" - env = dict(os.environ) - env["PATH"] = f"{mam_sandbox / 'realbin'}:{BARE_PATH}" - env["WORKSPACE_ROOT"] = str(mam_sandbox) - res = subprocess.run( - ["bash", str(script), "--workspace", str(mam_sandbox), - "--agent", "claude", "--role", "worker"], - capture_output=True, text=True, cwd=str(mam_sandbox), env=env) - - assert "herdr not installed" not in (res.stdout + res.stderr), \ - f"herdr gate rejected a real binary:\n{res.stdout}\n{res.stderr}" - - -# W-7 — the sharpest statement of B-3: with the agent CLI present and ONLY -# herdr missing, pre-flight must stop on the herdr gate. This is the -# test that is genuinely red before the fix without naming any new -# helper, so it measures the defect rather than the implementation. -def test_b3_preflight_stops_on_herdr_gate_when_only_herdr_is_missing(mam_sandbox): - agentbin = mam_sandbox / "agentbin" - agentbin.mkdir(parents=True, exist_ok=True) - for name in ("claude",): - exe = agentbin / name - exe.write_text('#!/bin/sh\necho \'{"loggedIn": true}\'\n') - exe.chmod(exe.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) - - script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh" - env = dict(os.environ) - env["PATH"] = f"{agentbin}:{BARE_PATH}" # claude yes, herdr no - env["WORKSPACE_ROOT"] = str(mam_sandbox) - res = subprocess.run( - ["bash", str(script), "--workspace", str(mam_sandbox), - "--agent", "claude", "--role", "worker"], - capture_output=True, text=True, cwd=str(mam_sandbox), env=env, timeout=120) - - combined = res.stdout + res.stderr - assert "herdr not installed" in combined, ( - "pre-flight did not stop on the herdr gate even though herdr is absent.\n" - f"rc={res.returncode}\n--- stdout ---\n{res.stdout}\n--- stderr ---\n{res.stderr}") - - -# W-8 — symlink bypass: a link in an ORDINARY bin dir pointing at the shim. -# No directory-name pattern can reject this; only the resolved target -# reveals it. -def test_b3_symlink_into_shim_is_rejected(mam_sandbox): - _bash(mam_sandbox, "true") # materialise the shim - shim = mam_sandbox / ".mam" / "shim" / "herdr" - assert shim.exists(), "shim wrapper was not created by _init_herdr_isolation" - - linkdir = mam_sandbox / "usrlocalbin" - linkdir.mkdir(parents=True, exist_ok=True) - (linkdir / "herdr").symlink_to(shim) - - res = _bash(mam_sandbox, - 'if has_real_herdr; then echo RESULT:TRUE; else echo RESULT:FALSE; fi', - path=f"{linkdir}:{BARE_PATH}") - assert "RESULT:FALSE" in res.stdout, ( - "a symlink pointing into .mam/shim was accepted as a real herdr:\n" - f"{res.stdout}\n{res.stderr}") - - -# W-9 — ... but a symlink to a REAL binary must still be accepted, so the -# canonicalisation cannot be a blanket "reject all symlinks". -def test_b3_symlink_to_real_binary_is_accepted(mam_sandbox): - real = _make_herdr(mam_sandbox / "realbin") - linkdir = mam_sandbox / "linkbin" - linkdir.mkdir(parents=True, exist_ok=True) - (linkdir / "herdr").symlink_to(real) - - res = _bash(mam_sandbox, - 'if has_real_herdr; then echo RESULT:TRUE; else echo RESULT:FALSE; fi', - path=f"{linkdir}:{BARE_PATH}") - assert "RESULT:TRUE" in res.stdout, \ - f"a symlink to a genuine herdr was rejected:\n{res.stdout}\n{res.stderr}" - - -# W-10 — a bare relative PATH entry '.mam/shim' (no leading '/' or './') must -# still be rejected. Requires normalising the dir with a leading slash. -def test_b3_bare_relative_shim_entry_is_rejected(mam_sandbox): - _bash(mam_sandbox, "true") # materialise the shim - res = _bash(mam_sandbox, - 'if has_real_herdr; then echo RESULT:TRUE; else echo RESULT:FALSE; fi', - path=f".mam/shim:{BARE_PATH}") - assert "RESULT:FALSE" in res.stdout, ( - "a bare relative '.mam/shim' PATH entry was accepted:\n" - f"{res.stdout}\n{res.stderr}") diff --git a/tests/test_c2_no_stale_cache_dir.py b/tests/test_c2_no_stale_cache_dir.py deleted file mode 100644 index 6d05be6..0000000 --- a/tests/test_c2_no_stale_cache_dir.py +++ /dev/null @@ -1,126 +0,0 @@ -"""C-2 — Remove unused .cache/multi-agent-mux-monitor state directory creation. - -Verifies: - Y-1: --dry-run does not create .cache/multi-agent-mux-monitor - Y-2: --once --emit-diff does not create .cache/multi-agent-mux-monitor - Y-3: AGENT_SESSIONS_STATE_DIR does not create custom path or default path - Y-4: Reconcile executes normally and outputs valid drift JSON - Y-5: Existing pre-created .cache directory/files are unmolested during runtime - Y-6: Setting AGENT_SESSIONS_STATE_DIR emits deprecation warning on stderr without polluting stdout JSON - Y-7: Unset AGENT_SESSIONS_STATE_DIR is completely silent on stderr - Y-8: Empty .cache parent directory is reclaimed during uninstall - Y-9: Non-empty .cache directory is preserved during uninstall - Y-10: deploy/remove.sh correctly employs relative rmdir .cache -""" -import os -import subprocess -import shutil -import json -import pytest - -def _run_reconcile(sandbox, extra_args=None, extra_env=None): - repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - sandbox_agents = sandbox / ".agents" - if not sandbox_agents.exists(): - os.symlink(os.path.join(repo_root, ".agents"), sandbox_agents, target_is_directory=True) - - mam_dir = sandbox / ".mam" - mam_dir.mkdir(exist_ok=True) - yaml_file = mam_dir / "agent-sessions.yaml" - if not yaml_file.exists(): - yaml_file.write_text("herdr_sessions: []\n") - - env = dict(os.environ) - env["WORKSPACE_ROOT"] = str(sandbox) - if extra_env: - env.update(extra_env) - - reconcile_script = os.path.join(repo_root, ".agents", "skills", "multi-agent-mux-monitor", "scripts", "reconcile.sh") - cmd = ["bash", reconcile_script] - if extra_args: - cmd.extend(extra_args) - - return subprocess.run(cmd, capture_output=True, text=True, cwd=str(sandbox), env=env) - - -def test_y1_y2_no_cache_dir_created(mam_sandbox): - cache_dir = mam_sandbox / ".cache" - if cache_dir.exists(): - shutil.rmtree(cache_dir) - - # Y-1: dry-run - res_dry = _run_reconcile(mam_sandbox, ["--once", "--emit-diff", "--dry-run"]) - assert res_dry.returncode == 0 - assert not (mam_sandbox / ".cache" / "multi-agent-mux-monitor").exists() - assert not (mam_sandbox / ".cache").exists() - - # Y-2: change pass - res_once = _run_reconcile(mam_sandbox, ["--once", "--emit-diff"]) - assert res_once.returncode == 0 - assert not (mam_sandbox / ".cache" / "multi-agent-mux-monitor").exists() - - -def test_y3_y6_y7_agent_sessions_state_dir_deprecation(mam_sandbox): - custom_dir = mam_sandbox / "custom_state_dir" - - # Y-6: Set env var -> notice on stderr, stdout is valid JSON, no dir created - res_env = _run_reconcile(mam_sandbox, ["--once", "--emit-diff", "--dry-run"], - extra_env={"AGENT_SESSIONS_STATE_DIR": str(custom_dir)}) - assert res_env.returncode == 0 - assert "Notice: AGENT_SESSIONS_STATE_DIR is set but has no effect" in res_env.stderr - assert not custom_dir.exists() - assert not (mam_sandbox / ".cache").exists() - - # Valid JSON on stdout - data = json.loads(res_env.stdout) - assert "drifts" in data or "drift" in data - - # Y-7: Unset env var -> no notice on stderr - res_silent = _run_reconcile(mam_sandbox, ["--once", "--emit-diff", "--dry-run"]) - assert res_silent.returncode == 0 - assert "AGENT_SESSIONS_STATE_DIR" not in res_silent.stderr - - -def test_y4_reconcile_runs_normally(mam_sandbox): - res = _run_reconcile(mam_sandbox, ["--once", "--emit-diff"]) - assert res.returncode == 0 - data = json.loads(res.stdout) - assert "timestamp" in data - assert "drifts" in data or "drift" in data - - -def test_y5_existing_cache_dir_unmolested(mam_sandbox): - target = mam_sandbox / ".cache" / "multi-agent-mux-monitor" - target.mkdir(parents=True, exist_ok=True) - leftover = target / "leftover.state" - leftover.write_text("FROM-AN-OLDER-INSTALL") - - res = _run_reconcile(mam_sandbox, ["--once", "--emit-diff"]) - assert res.returncode == 0 - assert leftover.exists() - assert leftover.read_text() == "FROM-AN-OLDER-INSTALL" - - -def test_y8_y9_remove_script_parent_cache_reclaim(mam_sandbox): - from pathlib import Path - repo_root = Path(__file__).resolve().parent.parent - remove_script = repo_root / "deploy" / "remove.sh" - assert remove_script.exists() - - # Y-8: Empty .cache is reclaimed - cache_dir = mam_sandbox / ".cache" / "multi-agent-mux-monitor" - cache_dir.mkdir(parents=True, exist_ok=True) - - # Y-10: Check deploy/remove.sh source for relative rmdir ".cache" - content = remove_script.read_text() - assert 'rmdir ".cache"' in content or 'rmdir .cache' in content - - # Y-9: Non-empty .cache is preserved (simulated rmdir logic) - other_tool = mam_sandbox / ".cache" / "other_tool" - other_tool.mkdir(parents=True, exist_ok=True) - (other_tool / "data.bin").write_text("keep me") - - shutil.rmtree(cache_dir) - with pytest.raises(OSError): - os.rmdir(mam_sandbox / ".cache") - assert (other_tool / "data.bin").exists() diff --git a/tests/test_challenger_m2.py b/tests/test_challenger_m2.py deleted file mode 100644 index b66ac1d..0000000 --- a/tests/test_challenger_m2.py +++ /dev/null @@ -1,182 +0,0 @@ -import os -import subprocess -import json -import pytest -import sys -from pathlib import Path - -# Helper to run mutation on agent-sessions.yaml using atomic_dump_yaml in bash -def run_mutation(mam_sandbox, mutation_str, env=None): - lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" - yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" - cmd_str = f"source {lib_path} && atomic_dump_yaml {yaml_path}" - run_env = dict(os.environ) - if env: - run_env.update(env) - res = subprocess.run(["bash", "-c", cmd_str], input=mutation_str, capture_output=True, text=True, env=run_env) - return res - - -def test_unbound_key_handling(mam_sandbox, mock_herdr): - """ - Verify unbound key handling: herdr has-session/kill-session with trailing parameters - handles it gracefully (returns status 1 with error) instead of crashing with unbound var or shift errors. - """ - lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" - subprocess.run(["bash", "-c", f"source {lib_path}"], cwd=str(mam_sandbox), env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox))) - - shim_path = mam_sandbox / ".mam" / "shim" / "herdr" - assert shim_path.exists(), "Shim herdr was not created" - - # Run has-session with trailing -t - res = subprocess.run([str(shim_path), "has-session", "-t"], capture_output=True, text=True, env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox), REAL_HERDR=str(mock_herdr))) - assert res.returncode == 1 - assert "Error: -t requires a value" in res.stderr - - # Run kill-session with trailing -t - res = subprocess.run([str(shim_path), "kill-session", "-t"], capture_output=True, text=True, env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox), REAL_HERDR=str(mock_herdr))) - assert res.returncode == 1 - assert "Error: -t requires a value" in res.stderr - - -def test_new_session_fallback_shell(mam_sandbox, mock_herdr): - """ - Verify new-session: launches shell when no command is provided. - """ - lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" - subprocess.run(["bash", "-c", f"source {lib_path}"], cwd=str(mam_sandbox), env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox))) - shim_path = mam_sandbox / ".mam" / "shim" / "herdr" - - test_shell = "/bin/custom_sh" - run_env = dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox), REAL_HERDR=str(mock_herdr), SHELL=test_shell) - - res = subprocess.run([str(shim_path), "new-session", "-s", "fallback-session"], capture_output=True, text=True, env=run_env) - assert res.returncode == 0, f"Stderr: {res.stderr}" - - with open(mock_herdr, 'r') as f: - state = json.load(f) - calls = state.get("calls", []) - - start_call = None - for call in calls: - if "agent" in call and "start" in call and "fallback-session" in call: - start_call = call - break - assert start_call is not None, f"No agent start call found in: {calls}" - assert start_call[-1] == test_shell, f"Expected last arg to be {test_shell}, got {start_call[-1]}" - - -def test_ls_key_error(mam_sandbox): - """ - Verify ls key error: parses agent list without 'name' correctly using fallback keys. - """ - py_code = """ -import sys, json -try: - data = json.load(sys.stdin) - res = data.get('result', data) - for a in res.get('agents', []): - try: - name = a.get('name') or a.get('agent') or 'unknown' - print(f"{name}|0") - except Exception: - pass -except Exception: - pass -""" - input_json = json.dumps({ - "result": { - "agents": [ - {"agent": "claude", "status": "running"}, - {"name": "agent-with-name", "agent": "agy"} - ] - } - }) - - res = subprocess.run([sys.executable, "-c", py_code], input=input_json, capture_output=True, text=True) - assert res.returncode == 0 - lines = res.stdout.strip().split('\n') - assert "claude|0" in lines - assert "agent-with-name|0" in lines - - -def test_reconcile_relocatability(mam_sandbox, mock_herdr): - """ - Verify relocatability: reconcile.sh runs when relocated or inside different paths. - """ - reconcile_script = mam_sandbox / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" - run_dir = mam_sandbox / "some_other_dir" - run_dir.mkdir() - - res = subprocess.run(["bash", str(reconcile_script), "--once", "--emit-diff", "--dry-run"], cwd=str(run_dir), capture_output=True, text=True, env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox), REAL_HERDR=str(mock_herdr))) - assert res.returncode == 0, f"Failed when run from different directory. Stderr: {res.stderr}" - - -def test_workspace_server_mapping(mam_sandbox, mock_herdr): - """ - Verify workspace/server mapping: status output workspace is correct. - """ - status_script = mam_sandbox / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh" - session_name = "test-mapping-sess-creator-claude" - - mutation = f""" -d['herdr_sessions'] = [{{ - 'name': '{session_name}', - 'status': 'running', - 'role': 'Creator', - 'herdr_server': 'my-custom-workspace', - 'pane': {{ - 'cwd': 'WS_PLACEHOLDER', - 'pid': 7777, - 'cmd': 'claude', - 'cmd_full': 'claude' - }} -}}] -""".replace("WS_PLACEHOLDER", str(mam_sandbox)) - res_mut = run_mutation(mam_sandbox, mutation) - assert res_mut.returncode == 0 - - res = subprocess.run(["bash", str(status_script), "--json"], capture_output=True, text=True, cwd=str(mam_sandbox), env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox), REAL_HERDR=str(mock_herdr))) - assert res.returncode == 0, f"Stderr: {res.stderr}" - - data = json.loads(res.stdout) - sess_detail = data["sessions_detail"] - target_sess = [s for s in sess_detail if s["name"] == session_name][0] - - assert target_sess["server"] == "my-custom-workspace" - - -def test_variable_splicing_injection_safety(mam_sandbox, mock_herdr): - """ - Verify variable splicing injection: no vulnerabilities or syntax errors remain. - """ - lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" - subprocess.run(["bash", "-c", f"source {lib_path}"], cwd=str(mam_sandbox), env=dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox))) - shim_path = mam_sandbox / ".mam" / "shim" / "herdr" - - adversarial_name = 'my"server; import os; os.system("echo INJECTED")' - run_env = dict(os.environ, HOME=str(mam_sandbox), WORKSPACE_ROOT=str(mam_sandbox), REAL_HERDR=str(mock_herdr), HERDR_SERVER_NAME=adversarial_name) - - res = subprocess.run([str(shim_path), "new-session", "-s", "injection-test-session"], capture_output=True, text=True, env=run_env) - assert res.returncode == 0, f"Failed with adversarial HERDR_SERVER_NAME. Stderr: {res.stderr}" - - with open(mock_herdr, 'r') as f: - state = json.load(f) - assert any("injection-test-session" in call for call in state["calls"]), "Expected injection-test-session call to succeed" - - -def test_export_masking_exit_code_preservation(): - """ - Verify export masking: exit codes are preserved when assigning and exporting. - """ - # 1. Export masked version exits 0 (silent fail) - cmd_masked = ["bash", "-c", "set -e; export TEST_VAR=$(false); echo 'survived'"] - res_masked = subprocess.run(cmd_masked, capture_output=True, text=True) - assert res_masked.returncode == 0 - assert res_masked.stdout.strip() == "survived" - - # 2. Fixed split version exits non-zero (preserves failure exit code) - cmd_fixed = ["bash", "-c", "set -e; TEST_VAR=$(false); export TEST_VAR; echo 'survived'"] - res_fixed = subprocess.run(cmd_fixed, capture_output=True, text=True) - assert res_fixed.returncode != 0 - assert res_fixed.stdout.strip() != "survived" diff --git a/tests/test_challenger_m3.py b/tests/test_challenger_m3.py deleted file mode 100644 index f27786c..0000000 --- a/tests/test_challenger_m3.py +++ /dev/null @@ -1,205 +0,0 @@ -import os -import subprocess -import json -import sqlite3 -import pytest -import shutil -import yaml -import sys -from pathlib import Path - -# Helper to run mutation on agent-sessions.yaml using atomic_dump_yaml in bash -def run_mutation(mam_sandbox, mutation_str, env=None): - lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh" - yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" - cmd_str = f"source {lib_path} && atomic_dump_yaml {yaml_path}" - run_env = dict(os.environ) - if env: - run_env.update(env) - res = subprocess.run(["bash", "-c", cmd_str], input=mutation_str, capture_output=True, text=True, env=run_env) - return res - -def test_stop_session_unvalidated_agent(mam_sandbox, mock_herdr, mock_agents): - """ - Test case 1: Unvalidated agent argument in stop_session.sh - Verify that calling stop_session.sh with an invalid agent name returns code 2 - and exits with a clear error message to stderr. - """ - tmp_path = mam_sandbox - stop_script = tmp_path / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" - - # 1. Register a session first - session_name = "invalid-agent-session-creator-claude" - mutation = """ -d['herdr_sessions'] = [{ - 'name': 'invalid-agent-session-creator-claude', - 'status': 'running', - 'role': 'Creator', - 'pane': { - 'cwd': 'WS_PLACEHOLDER', - 'pid': 8888, - 'cmd': 'claude', - 'cmd_full': 'claude' - } -}] -""".replace("WS_PLACEHOLDER", str(tmp_path)) - res_mut = run_mutation(tmp_path, mutation) - assert res_mut.returncode == 0 - - # 2. Try stopping with --agent invalid_agent and --purge-conversation - cmd_stop = [ - "bash", str(stop_script), - "--session", session_name, - "--agent", "invalid_agent", - "--purge-conversation", - "--yes" - ] - res_stop = subprocess.run(cmd_stop, capture_output=True, text=True, cwd=str(tmp_path)) - - # It returns code 2 - assert res_stop.returncode == 2 - - # And prints invalid agent type error to stderr - assert "ERROR: invalid agent type 'invalid_agent'" in res_stop.stderr - - # The registry entry is NOT removed - yaml_path = tmp_path / ".mam" / "agent-sessions.yaml" - with open(yaml_path, 'r') as f: - reg = yaml.safe_load(f) - assert len(reg.get("herdr_sessions", [])) == 1 - - -def test_special_character_workspace_slug_interpretation(mam_sandbox): - """ - Test case 2: Special character workspace slug interpretation - Verify that workspace paths with only special characters result in derived session names - that do NOT start with dashes, preventing option flags injection. - """ - tmp_path = mam_sandbox - lib_path = tmp_path / ".agents" / "skills" / "lib.sh" - - # Workspace consisting of purely special characters - special_ws = tmp_path / "@#$*" / "@#$*" - special_ws.mkdir(parents=True, exist_ok=True) - - cmd = f"source {lib_path} && derive_session_name '{special_ws}' 'claude'" - res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, cwd=str(tmp_path)) - assert res.returncode == 0 - derived_name = res.stdout.strip() - - # Confirm that the derived name does NOT start with a dash and is default "ws" - assert not derived_name.startswith("-") - assert derived_name == "ws-creator-claude" - - -def test_monitor_reconcile_race_on_purge(mam_sandbox, mock_herdr): - """ - Test case 3: Concurrency/Race condition during session purge - Verify that if a session is being purged, the presence of the purging lock file - prevents concurrent monitor checks from auto-registering it back. - """ - tmp_path = mam_sandbox - reconcile_script = tmp_path / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh" - - # 1. Register a running session in herdr and YAML - session_name = "purged-race-creator-claude" - with open(mock_herdr, 'r') as f: - state = json.load(f) - state["agents"][session_name] = { - "status": "running", - "agent": "claude", - "cwd": str(tmp_path), - "pid": 9999, - "pane_id": "w1:p1", - "command": "claude" - } - with open(mock_herdr, 'w') as f: - json.dump(state, f, indent=2) - - mutation = f""" -d['herdr_sessions'] = [{{ - 'name': '{session_name}', - 'status': 'running', - 'role': 'Creator', - 'pane': {{ - 'cwd': 'WS_PLACEHOLDER', - 'pid': 9999, - 'cmd': 'claude', - 'cmd_full': 'claude' - }} -}}] -""".replace("WS_PLACEHOLDER", str(tmp_path)) - res_mut = run_mutation(tmp_path, mutation) - assert res_mut.returncode == 0 - - # 2. Simulate a purge: stop_session.sh deletes the registry row, but herdr session takes a moment to die. - # We remove it from the YAML registry, and create the purging lock file. - mutation_purge = f""" -d['herdr_sessions'] = [s for s in d.get('herdr_sessions', []) if s.get('name') != '{session_name}'] -""" - res_purge_mut = run_mutation(tmp_path, mutation_purge) - assert res_purge_mut.returncode == 0 - - # Create the purging lock file - purging_file = tmp_path / ".mam" / f"purging-{session_name}" - purging_file.parent.mkdir(parents=True, exist_ok=True) - purging_file.touch() - - # Confirm YAML registry has no sessions - yaml_path = tmp_path / ".mam" / "agent-sessions.yaml" - with open(yaml_path, 'r') as f: - reg = yaml.safe_load(f) - assert len(reg.get("herdr_sessions", [])) == 0 - - # 3. Run reconcile.sh once. It should NOT auto-register it back because the purging lock file exists! - cmd_reconcile = ["bash", str(reconcile_script), "--once", "--emit-diff"] - res_rec = subprocess.run(cmd_reconcile, capture_output=True, text=True, cwd=str(tmp_path)) - assert res_rec.returncode == 0 - - # Verify that the session has NOT been auto-registered back in YAML - with open(yaml_path, 'r') as f: - reg_after = yaml.safe_load(f) - sessions_after = reg_after.get("herdr_sessions", []) - assert len(sessions_after) == 0 - - -def test_mqtt_hmac_and_seq_validation(mam_sandbox): - """ - Test case 4: MQTT HMAC and sequence number validation - Verify that mqtt_common correctly verifies HMAC signatures and drops messages with - invalid signatures or out-of-order sequence numbers. - """ - # Import mqtt_common from sandboxed folder - sys.path.insert(0, str(mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts")) - import mqtt_common - - # 1. verify_hmac behaviour with no auth_token (PoC mode) - payload_poc = {"data": {"val": 123}} - assert mqtt_common.verify_hmac(payload_poc, None) is True - - # 2. verify_hmac with auth_token and valid HMAC - auth_token = "mysecrettoken" - payload = { - "job_id": "job123", - "event": "started", - "seq": 1, - "data": { - "val": 123 - } - } - - # Compute valid signature - msg = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() - import hmac - import hashlib - sig = hmac.new(auth_token.encode(), msg, hashlib.sha256).hexdigest() - - payload_signed = dict(payload) - payload_signed["data"] = dict(payload["data"]) - payload_signed["data"]["hmac_sig"] = sig - - assert mqtt_common.verify_hmac(payload_signed, auth_token) is True - - # 3. verify_hmac with invalid signature - payload_signed["data"]["hmac_sig"] = "invalidsignature" - assert mqtt_common.verify_hmac(payload_signed, auth_token) is False diff --git a/tests/test_env_migration.py b/tests/test_env_migration.py deleted file mode 100644 index 6132f63..0000000 --- a/tests/test_env_migration.py +++ /dev/null @@ -1,282 +0,0 @@ -#!/usr/bin/env python3 -"""Full regression test suite for .mam.env migration (T-1 through T-17).""" - -import os -import shutil -import tempfile -import unittest -import subprocess -import logging -from unittest.mock import patch - -import sys -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../.agents/skills/multi-agent-mux-delegate-job/scripts"))) -import mqtt_common - -class TestEnvMigrationFull(unittest.TestCase): - - def setUp(self): - self.test_dir = tempfile.mkdtemp(prefix="mam_env_full_test_") - self.old_cwd = os.getcwd() - os.chdir(self.test_dir) - self.clean_os_env() - - def tearDown(self): - os.chdir(self.old_cwd) - shutil.rmtree(self.test_dir, ignore_errors=True) - self.clean_os_env() - - def clean_os_env(self): - for k in ["MQTT_BROKER", "MQTT_PORT", "MQTT_PASSWORD", "TEST_KEY_A", "TEST_KEY_B", "MAM_ENV_FILE", "MAM_LEGACY_ENV_OWNED"]: - os.environ.pop(k, None) - mqtt_common._warned_deprecated_env = False - mqtt_common._warned_coexistence_env = False - - def test_t1_mam_env_only(self): - """T-1: Loads .mam.env when present.""" - mam_env = os.path.join(self.test_dir, ".mam.env") - with open(mam_env, "w") as f: - f.write("TEST_KEY_A=val_mam_env\n") - - mqtt_common._load_dotenv(self.test_dir) - self.assertEqual(os.environ.get("TEST_KEY_A"), "val_mam_env") - - def test_t2_legacy_env_only_fallback_and_warning(self): - """T-2: Falls back to .env when .mam.env absent, logs deprecation warning.""" - legacy_env = os.path.join(self.test_dir, ".env") - with open(legacy_env, "w") as f: - f.write("TEST_KEY_A=val_legacy_env\n") - - with self.assertLogs("delegate_job.mqtt_common", level="WARNING") as cm: - mqtt_common._load_dotenv(self.test_dir) - - self.assertEqual(os.environ.get("TEST_KEY_A"), "val_legacy_env") - self.assertTrue(any("deprecated" in log for log in cm.output)) - - def test_t3_coexistence_mam_env_precedence_and_warning(self): - """T-3: When both exist, .mam.env wins and coexistence warning logs.""" - mam_env = os.path.join(self.test_dir, ".mam.env") - legacy_env = os.path.join(self.test_dir, ".env") - with open(mam_env, "w") as f: - f.write("TEST_KEY_A=val_mam\n") - with open(legacy_env, "w") as f: - f.write("TEST_KEY_A=val_legacy\n") - - with self.assertLogs("delegate_job.mqtt_common", level="WARNING") as cm: - mqtt_common._load_dotenv(self.test_dir) - - self.assertEqual(os.environ.get("TEST_KEY_A"), "val_mam") - self.assertTrue(any("Both" in log for log in cm.output)) - - def test_t4_parent_boundary_stop_walkup_without_arg(self): - """T-4: Walk-up with no argument stops at boundary marker (.agents/.git).""" - parent_dir = os.path.join(self.test_dir, "parent") - child_repo = os.path.join(parent_dir, "child_repo") - nested_script_dir = os.path.join(child_repo, ".agents", "skills", "test", "scripts") - os.makedirs(nested_script_dir, exist_ok=True) - - with open(os.path.join(parent_dir, ".env"), "w") as f: - f.write("TEST_KEY_A=parent_leaked_secret\n") - - dummy_file = os.path.join(nested_script_dir, "mqtt_common.py") - with patch.object(mqtt_common, "__file__", dummy_file): - mqtt_common._load_dotenv() - - self.assertIsNone(os.environ.get("TEST_KEY_A")) - - def test_t5_os_env_precedence(self): - """T-5: OS environment variables take precedence over env file values.""" - os.environ["TEST_KEY_A"] = "os_val" - mam_env = os.path.join(self.test_dir, ".mam.env") - with open(mam_env, "w") as f: - f.write("TEST_KEY_A=file_val\n") - - mqtt_common._load_dotenv(self.test_dir) - self.assertEqual(os.environ.get("TEST_KEY_A"), "os_val") - - def test_t6_mam_env_file_override(self): - """T-6: MAM_ENV_FILE takes highest precedence if explicitly provided.""" - custom_env = os.path.join(self.test_dir, "custom.env") - with open(custom_env, "w") as f: - f.write("TEST_KEY_A=custom_val\n") - os.environ["MAM_ENV_FILE"] = custom_env - - mam_env = os.path.join(self.test_dir, ".mam.env") - with open(mam_env, "w") as f: - f.write("TEST_KEY_A=mam_val\n") - - mqtt_common._load_dotenv(self.test_dir) - self.assertTrue(os.path.exists(custom_env)) - self.assertEqual(os.environ.get("TEST_KEY_A"), "custom_val") - - def test_t7_wrapper_cwd_isolation(self): - """T-7: Bash wrapper executes cleanly and outputs help/warnings.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - wrapper_path = os.path.join(repo_root, ".agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job") - res = subprocess.run([wrapper_path, "--help"], capture_output=True, text=True) - self.assertEqual(res.returncode, 0) - - def test_t8_remove_force_preserves_owned_env(self): - """T-8 [MERGE BLOCKER]: --force MUST back up owned .env, NEVER delete without backup.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), self.test_dir) - - os.makedirs(".mam", exist_ok=True) - with open(".mam/install_manifest.txt", "w") as f: - f.write(".env\nremove.sh\n") - - with open(".env", "w") as f: - f.write("SECRET_KEY=user_secret_data\n") - - res = subprocess.run(["bash", "remove.sh", "--force"], capture_output=True, text=True) - self.assertEqual(res.returncode, 0, f"remove.sh failed: {res.stderr}") - self.assertTrue(os.path.exists(".env.mam-backup"), "MAM-owned .env MUST be backed up under --force") - with open(".env.mam-backup") as f: - self.assertIn("user_secret_data", f.read()) - - def test_t8b_purge_env_is_sole_delete_authority(self): - """T-8b: --purge-env is the ONLY flag authorized to delete without backup.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), self.test_dir) - - os.makedirs(".mam", exist_ok=True) - with open(".mam/install_manifest.txt", "w") as f: - f.write(".env\nremove.sh\n") - - with open(".env", "w") as f: - f.write("SECRET_KEY=user_secret_data\n") - - res = subprocess.run(["bash", "remove.sh", "--force", "--purge-env"], capture_output=True, text=True) - self.assertEqual(res.returncode, 0) - self.assertFalse(os.path.exists(".env")) - self.assertFalse(os.path.exists(".env.mam-backup")) - - def test_t9_git_check_ignore(self): - """T-9 [MERGE BLOCKER]: Verifies gitignore rules ignore .mam.env variants but track .mam.env.example.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - check_cmd = ["git", "check-ignore", "-v", ".mam.env", ".mam.env.bak", ".mam.env.update-tmp", ".mam.env.mam-backup"] - res = subprocess.run(check_cmd, cwd=repo_root, capture_output=True, text=True) - self.assertEqual(res.returncode, 0, f"git check-ignore failed: {res.stderr}") - - ex_cmd = ["git", "check-ignore", ".mam.env.example"] - res_ex = subprocess.run(ex_cmd, cwd=repo_root, capture_output=True, text=True) - self.assertNotEqual(res_ex.returncode, 0, ".mam.env.example should NOT be ignored by git") - - def test_t10_shadowing_prevention_guard(self): - """T-10 [MERGE BLOCKER]: install.sh does NOT create new .mam.env if .env.update-tmp exists.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - shutil.copy(os.path.join(repo_root, "deploy/install.sh"), self.test_dir) - os.makedirs("deploy", exist_ok=True) - shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") - if os.path.exists(os.path.join(repo_root, ".mam.env.example")): - shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") - - with open(".env.update-tmp", "w") as f: - f.write("MQTT_PASSWORD=s3cr3t_user_pass\n") - - res = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) - self.assertEqual(res.returncode, 0, f"install.sh failed: {res.stderr}") - self.assertIn("Preserved without shadowing", res.stdout) - self.assertFalse(os.path.exists(".mam.env"), ".mam.env MUST NOT be created when .env.update-tmp is waiting for restore") - - def test_t12_unowned_legacy_env_preservation(self): - """T-12 [MERGE BLOCKER]: install.sh leaves unowned .env untouched without heuristic hijacking.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - os.makedirs("deploy", exist_ok=True) - shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") - if os.path.exists(os.path.join(repo_root, ".mam.env.example")): - shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") - - with open(".env", "w") as f: - f.write("MQTT_BROKER=my-private-iot-broker.com\nSTRIPE_SECRET=sk_live_123\n") - - res = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) - self.assertEqual(res.returncode, 0) - self.assertTrue(os.path.exists(".env"), ".env should remain untouched") - self.assertFalse(os.path.exists(".mam.env"), ".mam.env should NOT hijack unowned .env") - - def test_t13_owned_legacy_env_migration_and_manifest_rewrite(self): - """T-13: Owned .env is migrated to .mam.env and manifest is updated.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - os.makedirs("deploy", exist_ok=True) - shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") - - os.makedirs(".mam", exist_ok=True) - with open(".mam/install_manifest.txt", "w") as f: - f.write(".env\n") - - with open(".env", "w") as f: - f.write("MQTT_BROKER=owned-broker.internal\n") - - res = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) - self.assertEqual(res.returncode, 0) - self.assertTrue(os.path.exists(".mam.env")) - self.assertFalse(os.path.exists(".env")) - - with open(".mam/install_manifest.txt", "r") as f: - manifest_lines = [line.strip() for line in f] - self.assertIn(".mam.env", manifest_lines) - self.assertNotIn(".env", manifest_lines) - - def test_t16_backup_deduplication_across_reinstall_cycles(self): - """T-16 (Backup Hygiene): Repeated remove.sh -y -> install.sh cycles do not multiply identical backups.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - os.makedirs("deploy", exist_ok=True) - shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") - shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), "deploy/remove.sh") - if os.path.exists(os.path.join(repo_root, ".mam.env.example")): - shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") - - # Initial setup: MAM-owned .mam.env with user secret - os.makedirs(".mam", exist_ok=True) - with open(".mam/install_manifest.txt", "w") as f: - f.write(".mam.env\nremove.sh\nupdate.sh\n.mam.env.example\n") - with open(".mam.env", "w") as f: - f.write("MQTT_PASSWORD=REAL_USER_SECRET_12345\n") - - # Cycle 1: remove -y - res1 = subprocess.run(["bash", "deploy/remove.sh", "--force"], capture_output=True, text=True) - self.assertEqual(res1.returncode, 0) - self.assertTrue(os.path.exists(".mam.env.mam-backup")) - - # Cycle 1: reinstall - res_inst1 = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) - self.assertEqual(res_inst1.returncode, 0) - - # Cycle 2: remove -y & reinstall - subprocess.run(["bash", "deploy/remove.sh", "--force"], check=True) - subprocess.run(["bash", "deploy/install.sh", self.test_dir], check=True) - - # Cycle 3: remove -y (identical default content should be deduplicated) - subprocess.run(["bash", "deploy/remove.sh", "--force"], check=True) - - # Verify backup count stabilizes at <= 2 (Slot 1 original secret + 1 dedup default slot) - backups = [f for f in os.listdir(self.test_dir) if f.startswith(".mam.env.mam-backup")] - self.assertLessEqual(len(backups), 2, f"Backup files duplicated indefinitely: {backups}") - - def test_t17_immutable_slot_1_user_secret_retention(self): - """T-17 [MERGE BLOCKER]: Slot 1 (.mam.env.mam-backup) retains original user secret after repeated cycles.""" - repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - os.makedirs("deploy", exist_ok=True) - shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") - shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), "deploy/remove.sh") - if os.path.exists(os.path.join(repo_root, ".mam.env.example")): - shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") - - os.makedirs(".mam", exist_ok=True) - with open(".mam/install_manifest.txt", "w") as f: - f.write(".mam.env\nremove.sh\n") - with open(".mam.env", "w") as f: - f.write("MQTT_PASSWORD=REAL_USER_SECRET_12345\n") - - # 3 cycles of remove -> install - for _ in range(3): - subprocess.run(["bash", "deploy/remove.sh", "--force"], check=True) - subprocess.run(["bash", "deploy/install.sh", self.test_dir], check=True) - - with open(".mam.env.mam-backup", "r") as f: - slot1_content = f.read() - self.assertIn("REAL_USER_SECRET_12345", slot1_content, "Slot 1 MUST retain original user secret permanently") - -if __name__ == "__main__": - unittest.main()