diff options
| author | Craig Jennings <c@cjennings.net> | 2026-05-06 21:59:52 -0500 |
|---|---|---|
| committer | Craig Jennings <c@cjennings.net> | 2026-05-06 21:59:52 -0500 |
| commit | d81b23ad6b6e437dfe3c338a00a4be39bc555146 (patch) | |
| tree | 2d4b0d7890fd1fc70d81282b81fed2808c28a106 /.ai/scripts/tests/test_cross_agent_halt.py | |
| parent | 201377f57430ef28d02e703a2191434bbee55c75 (diff) | |
| download | rulesets-d81b23ad6b6e437dfe3c338a00a4be39bc555146.tar.gz rulesets-d81b23ad6b6e437dfe3c338a00a4be39bc555146.zip | |
chore(ai): initialize project notes and Claude tooling surfaces
Replace the seed notes.org with project-specific context (layout, install modes, task tracker location, recent inflection point). Bring in the synced template surfaces (protocols, workflows, scripts, references, retrospectives, someday-maybe) as tracked content for this content/documentation project.
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_halt.py')
| -rw-r--r-- | .ai/scripts/tests/test_cross_agent_halt.py | 204 |
1 files changed, 204 insertions, 0 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_halt.py b/.ai/scripts/tests/test_cross_agent_halt.py new file mode 100644 index 0000000..f8bf0b3 --- /dev/null +++ b/.ai/scripts/tests/test_cross_agent_halt.py @@ -0,0 +1,204 @@ +"""Tests for cross-agent-halt and cross-agent-resume (TDD).""" + +from __future__ import annotations + +import os +import subprocess +import textwrap +from pathlib import Path + +import pytest + +HALT_SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-halt" +RESUME_SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-resume" + + +def _run(script: Path, args: list[str], env: dict | None = None) -> subprocess.CompletedProcess: + return subprocess.run([str(script), *args], capture_output=True, text=True, env=env) + + +@pytest.fixture +def isolated_env(tmp_path, monkeypatch): + """Isolated HOME + a fake systemctl that records calls without acting.""" + fake_home = tmp_path / "home" + fake_home.mkdir() + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + # Fake systemctl: no-op, exit 0. + fake_systemctl = fake_bin / "systemctl" + fake_systemctl.write_text("#!/usr/bin/env bash\nexit 0\n") + fake_systemctl.chmod(0o755) + # Fake ssh: succeed only for known-good host. + fake_ssh = fake_bin / "ssh" + fake_ssh.write_text(textwrap.dedent("""\ + #!/usr/bin/env bash + # Find the destination arg (skip flags). + target="" + for arg in "$@"; do + case "$arg" in + -*|*=*) ;; + *@*|localhost|*.local|*.invalid) target="$arg"; break ;; + *) target="$arg"; break ;; + esac + done + case "$target" in + *invalid*|*unreachable*) exit 255 ;; + *) exit 0 ;; + esac + """)) + fake_ssh.chmod(0o755) + + monkeypatch.setenv("HOME", str(fake_home)) + # Prepend our fake bin so systemctl + ssh are intercepted, but keep real /bin etc. + monkeypatch.setenv("PATH", f"{fake_bin}:{os.environ.get('PATH', '')}") + return fake_home + + +# ---- cross-agent-halt ---- + + +def test_halt_help(isolated_env): + result = _run(HALT_SCRIPT, ["--help"], env={**os.environ, "HOME": str(isolated_env), + "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + assert "halt" in result.stdout.lower() + + +def test_halt_creates_halt_file(isolated_env): + halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT" + assert not halt_file.exists() + result = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), + "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + assert halt_file.exists() + + +def test_halt_with_reason_writes_body(isolated_env): + result = _run(HALT_SCRIPT, ["pausing for incident review"], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT" + assert halt_file.exists() + assert "pausing for incident review" in halt_file.read_text() + + +def test_halt_idempotent(isolated_env): + """Running halt twice doesn't error.""" + halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT" + r1 = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert r1.returncode == 0 + assert halt_file.exists() + r2 = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert r2.returncode == 0 + assert halt_file.exists() + + +def test_halt_does_not_pkill(isolated_env): + """Per design: halt does NOT call pkill. Verify by checking no pkill process gets launched.""" + # Replace pkill in PATH with something that fails loudly so we'd see if halt invoked it. + fake_bin = isolated_env.parent / "bin" + pkill = fake_bin / "pkill" + pkill.write_text("#!/usr/bin/env bash\necho 'PKILL CALLED' >&2\nexit 99\n") + pkill.chmod(0o755) + result = _run(HALT_SCRIPT, [], env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + assert "PKILL CALLED" not in result.stderr + + +def test_halt_tailnet_reports_per_peer(isolated_env): + """--tailnet iterates peers.toml and reports per-peer status.""" + cfg = isolated_env / ".config" / "cross-agent-comms" + cfg.mkdir(parents=True) + (cfg / "peers.toml").write_text(textwrap.dedent("""\ + [peers.velox] + host = "velox" + ssh_user = "cjennings" + + [peers.bogus] + host = "definitely-unreachable.invalid" + ssh_user = "cjennings" + """)) + result = _run(HALT_SCRIPT, ["--tailnet"], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + # Partial halt → exit 1. + assert result.returncode == 1 + assert "velox" in result.stdout + assert "bogus" in result.stdout + # ✓ marker for velox, ✗ for bogus. + assert "✓" in result.stdout + assert "✗" in result.stdout + assert "PARTIAL" in result.stdout or "partial" in result.stdout.lower() + + +def test_halt_tailnet_all_reachable_exits_zero(isolated_env): + cfg = isolated_env / ".config" / "cross-agent-comms" + cfg.mkdir(parents=True) + (cfg / "peers.toml").write_text(textwrap.dedent("""\ + [peers.velox] + host = "velox" + ssh_user = "cjennings" + """)) + result = _run(HALT_SCRIPT, ["--tailnet"], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + assert "velox" in result.stdout + + +# ---- cross-agent-resume ---- + + +def test_resume_help(isolated_env): + result = _run(RESUME_SCRIPT, ["--help"], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + assert "resume" in result.stdout.lower() + + +def test_resume_removes_halt_file(isolated_env): + halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT" + halt_file.parent.mkdir(parents=True) + halt_file.write_text("halted") + assert halt_file.exists() + result = _run(RESUME_SCRIPT, [], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + assert not halt_file.exists() + + +def test_resume_when_no_halt_active_succeeds(isolated_env): + """No HALT to clear is not an error.""" + result = _run(RESUME_SCRIPT, [], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + + +def test_resume_prints_per_session_instructions(isolated_env): + """Resume must surface that polling does NOT auto-resume.""" + halt_file = isolated_env / ".config" / "cross-agent-comms" / "HALT" + halt_file.parent.mkdir(parents=True) + halt_file.write_text("halted") + result = _run(RESUME_SCRIPT, [], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 0 + out = result.stdout.lower() + assert "polling" in out + assert "auto" in out or "explicit" in out or "session" in out + + +def test_resume_tailnet_partial_failure_exit_1(isolated_env): + cfg = isolated_env / ".config" / "cross-agent-comms" + cfg.mkdir(parents=True) + (cfg / "peers.toml").write_text(textwrap.dedent("""\ + [peers.velox] + host = "velox" + + [peers.bogus] + host = "unreachable-host.invalid" + """)) + halt_file = cfg / "HALT" + halt_file.write_text("halted") + result = _run(RESUME_SCRIPT, ["--tailnet"], + env={**os.environ, "HOME": str(isolated_env), "PATH": os.environ["PATH"]}) + assert result.returncode == 1 + assert "velox" in result.stdout + assert "bogus" in result.stdout |
