diff options
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_recv.py')
| -rw-r--r-- | .ai/scripts/tests/test_cross_agent_recv.py | 176 |
1 files changed, 0 insertions, 176 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_recv.py b/.ai/scripts/tests/test_cross_agent_recv.py deleted file mode 100644 index 27c53a5..0000000 --- a/.ai/scripts/tests/test_cross_agent_recv.py +++ /dev/null @@ -1,176 +0,0 @@ -"""Tests for cross-agent-recv.""" - -from __future__ import annotations - -import json -import os -import subprocess -from pathlib import Path - -import pytest - -SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-recv" - - -def _make_message(path: Path, *, conv_id: str = "test-conv", seq: int = 1, msg_type: str = "request", - proto_version: str = "5", title: str = "Test", requires_tools: str | None = None, - body: str = "Body.\n") -> Path: - fm_lines = [ - f"#+TITLE: {title}", - f"#+CONVERSATION_ID: {conv_id}", - f"#+MESSAGE_TYPE: {msg_type}", - f"#+SEQUENCE: {seq}", - "#+TIMESTAMP: 2026-04-27T05:00:00-05:00", - f"#+PROTOCOL_VERSION: {proto_version}", - ] - if requires_tools: - fm_lines.append(f"#+REQUIRES_TOOLS: {requires_tools}") - path.write_text("\n".join(fm_lines) + "\n\n" + body) - return path - - -def _run(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): - fake_home = tmp_path / "home" - fake_home.mkdir() - monkeypatch.setenv("HOME", str(fake_home)) - return fake_home - - -def test_recv_help(isolated_env): - result = _run(["--help"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 0 - assert "Receive and decide" in result.stdout - - -def test_recv_missing_file_rejects(isolated_env, tmp_path): - result = _run([str(tmp_path / "nope.org")], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 3 # reject - - -def test_recv_malformed_frontmatter_rejects(isolated_env, tmp_path): - bad = tmp_path / "bad.org" - bad.write_text("not org-mode at all\n") - result = _run([str(bad), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 3 - assert "decision: reject" in result.stdout - - -def test_recv_missing_required_field_rejects(isolated_env, tmp_path): - msg = tmp_path / "msg.org" - # Missing PROTOCOL_VERSION among others. - msg.write_text("#+TITLE: x\n#+CONVERSATION_ID: c\n\nBody.\n") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 3 - assert "missing required" in result.stdout - - -def test_recv_protocol_version_mismatch_query(isolated_env, tmp_path): - msg = _make_message(tmp_path / "msg.org", proto_version="4") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 2 # query - assert "PROTOCOL_VERSION mismatch" in result.stdout - - -def test_recv_invalid_message_type_rejects(isolated_env, tmp_path): - msg = _make_message(tmp_path / "msg.org", msg_type="banana") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 3 - assert "invalid MESSAGE_TYPE" in result.stdout - - -def test_recv_missing_signature_rejects(isolated_env, tmp_path): - """When verify is on, a missing .asc sibling rejects.""" - msg = _make_message(tmp_path / "msg.org") - # No .asc sidecar. - result = _run([str(msg)], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 3 - assert "signature file missing" in result.stdout - - -def test_recv_valid_processes(isolated_env, tmp_path): - """A valid message with --no-verify and no dedup match → process.""" - msg = _make_message(tmp_path / "msg.org") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 0 # process - assert "decision: process" in result.stdout - assert "sha256:" in result.stdout - - -def test_recv_dedup_against_identical_existing(isolated_env, tmp_path): - """Same content + same SEQUENCE in same dir → dedup.""" - inbox = tmp_path / "inbox" - inbox.mkdir() - first = _make_message(inbox / "20260427T100000Z-from-x-c.org", conv_id="c", seq=5) - # Second message with same content — name differs (canonical-style would have different timestamp). - second = _make_message(inbox / "20260427T100100Z-from-x-c.org", conv_id="c", seq=5) - # Bodies must be byte-identical for hash equality. - second.write_bytes(first.read_bytes()) - result = _run([str(second), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 1 # dedup - assert "decision: dedup" in result.stdout - - -def test_recv_collision_with_different_content_processes(isolated_env, tmp_path): - """Same SEQUENCE + same CONVERSATION_ID but different content → process both.""" - inbox = tmp_path / "inbox" - inbox.mkdir() - _make_message(inbox / "20260427T100000Z-from-x-c.org", conv_id="c", seq=5, body="First body.\n") - second = _make_message(inbox / "20260427T100100Z-from-x-c.org", conv_id="c", seq=5, body="Different body.\n") - result = _run([str(second), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 0 # process - assert "decision: process" in result.stdout - - -def test_recv_requires_tools_missing_query(isolated_env, tmp_path): - """REQUIRES_TOOLS naming a definitely-missing binary → query.""" - msg = _make_message(tmp_path / "msg.org", requires_tools="definitely-not-installed-xyzzy-9000") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 2 # query - assert "required tools unavailable" in result.stdout - - -def test_recv_requires_tools_present_processes(isolated_env, tmp_path): - """REQUIRES_TOOLS naming a real binary → process.""" - msg = _make_message(tmp_path / "msg.org", requires_tools="ls,cat") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 0 - assert "decision: process" in result.stdout - - -def test_recv_json_output(isolated_env, tmp_path): - msg = _make_message(tmp_path / "msg.org") - result = _run([str(msg), "--no-verify", "--json"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 0 - payload = json.loads(result.stdout) - assert payload["decision"] == "process" - assert payload["message_type"] == "request" - assert payload["conversation_id"] == "test-conv" - - -def test_recv_halt_blocks(isolated_env, tmp_path): - halt = isolated_env / ".config" / "cross-agent-comms" / "HALT" - halt.parent.mkdir(parents=True) - halt.write_text("halted\n") - msg = _make_message(tmp_path / "msg.org") - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 5 - assert "halt active" in result.stderr.lower() - - -def test_recv_halt_leaves_message_in_place(isolated_env, tmp_path): - """Per spec: under HALT, recv must NOT move/dedup/reject — leave file in place.""" - halt = isolated_env / ".config" / "cross-agent-comms" / "HALT" - halt.parent.mkdir(parents=True) - halt.write_text("halted\n") - msg = _make_message(tmp_path / "msg.org") - pre_content = msg.read_text() - result = _run([str(msg), "--no-verify"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 5 - # File still exists with same content. - assert msg.exists() - assert msg.read_text() == pre_content |