tmpl/multi-agent-mux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(tests): remove redundant and stale legacy test suites for faster execution

Browse Source
This commit is contained in:
Godopu
2026-08-08 00:21:42 +09:00
parent 399242dac5
commit cf51b2ce13
8 changed files with 3 additions and 1559 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_a3_buffer_isolation.py
-274
View File
@@ -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_<job>_<pid>_<rand>) 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
tests/test_b1_tier3_identity.py
-291
View File
@@ -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 <ws>/.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() == ""
tests/test_b3_herdr_preflight.py
-196
View File
@@ -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}")
tests/test_c2_no_stale_cache_dir.py
-126
View File
@@ -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()
tests/test_challenger_m2.py
-182
View File
@@ -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"
tests/test_challenger_m3.py
-205
View File
@@ -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
tests/test_env_migration.py
-282
View File
@@ -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()