aboutsummaryrefslogtreecommitdiff
path: root/.ai/scripts/tests/test_cross_agent_halt.py
diff options
context:
space:
mode:
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_halt.py')
-rw-r--r--.ai/scripts/tests/test_cross_agent_halt.py204
1 files changed, 0 insertions, 204 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_halt.py b/.ai/scripts/tests/test_cross_agent_halt.py
deleted file mode 100644
index f8bf0b3..0000000
--- a/.ai/scripts/tests/test_cross_agent_halt.py
+++ /dev/null
@@ -1,204 +0,0 @@
-"""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