diff options
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_send.py')
| -rw-r--r-- | .ai/scripts/tests/test_cross_agent_send.py | 210 |
1 files changed, 0 insertions, 210 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_send.py b/.ai/scripts/tests/test_cross_agent_send.py deleted file mode 100644 index f716e95..0000000 --- a/.ai/scripts/tests/test_cross_agent_send.py +++ /dev/null @@ -1,210 +0,0 @@ -"""Tests for cross-agent-send. - -Subprocess-based: treat the script as a black-box CLI and assert on its -exit codes, stdout, and the files it produces. -""" - -from __future__ import annotations - -import os -import subprocess -import textwrap -from pathlib import Path - -import pytest - -SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-send" - - -def _make_message(tmp_path: Path, conv_id: str = "test-conv", seq: int = 1, msg_type: str = "request", - proto_version: str = "5") -> Path: - msg = tmp_path / "msg.org" - msg.write_text(textwrap.dedent(f"""\ - #+TITLE: Test message - #+CONVERSATION_ID: {conv_id} - #+MESSAGE_TYPE: {msg_type} - #+SEQUENCE: {seq} - #+TIMESTAMP: 2026-04-27T05:00:00-05:00 - #+PROTOCOL_VERSION: {proto_version} - - Body. - """)) - return msg - - -def _run(args: list[str], env: dict | None = None, cwd: Path | None = None) -> subprocess.CompletedProcess: - return subprocess.run( - [str(SCRIPT), *args], - capture_output=True, - text=True, - env=env, - cwd=cwd, - ) - - -@pytest.fixture -def isolated_env(tmp_path, monkeypatch): - """Redirect HOME so peers.toml, HALT, marker files are scoped to the test.""" - fake_home = tmp_path / "home" - fake_home.mkdir() - monkeypatch.setenv("HOME", str(fake_home)) - # Pre-create projects/ so derive_sender_project has somewhere to look. - (fake_home / "projects" / "homelab").mkdir(parents=True) - return fake_home - - -def test_send_help(isolated_env): - """--help works without side effects.""" - result = _run(["--help"], env={**os.environ, "HOME": str(isolated_env)}) - assert result.returncode == 0 - assert "Send a cross-agent message" in result.stdout - - -def test_send_missing_message_file(isolated_env): - """Nonexistent message file returns general error.""" - import socket - machine = socket.gethostname().split(".")[0] - result = _run( - [f"{machine}.homelab", str(isolated_env / "nonexistent.org")], - env={**os.environ, "HOME": str(isolated_env)}, - ) - assert result.returncode == 1 - assert "not found" in result.stderr.lower() - - -def test_send_invalid_destination_format(isolated_env, tmp_path): - """Destination without . returns dest-not-found exit code.""" - msg = _make_message(tmp_path) - result = _run( - ["bogus", str(msg)], - env={**os.environ, "HOME": str(isolated_env)}, - ) - assert result.returncode == 2 - assert "<machine>.<project>" in result.stderr or "destination" in result.stderr.lower() - - -def test_send_dest_not_in_peers(isolated_env, tmp_path): - """Cross-machine destination with no peers.toml entry exits 2.""" - msg = _make_message(tmp_path) - result = _run( - ["unknownmachine.homelab", str(msg)], - env={**os.environ, "HOME": str(isolated_env)}, - ) - assert result.returncode == 2 - assert "not found in peers" in result.stderr - - -def test_send_frontmatter_missing_required(isolated_env, tmp_path): - """Message missing required fields exits 4.""" - bad = tmp_path / "bad.org" - bad.write_text("#+TITLE: nope\n\nBody.\n") - import socket - machine = socket.gethostname().split(".")[0] - result = _run( - [f"{machine}.homelab", str(bad)], - env={**os.environ, "HOME": str(isolated_env)}, - ) - assert result.returncode == 4 - assert "missing required fields" in result.stderr - - -def test_send_invalid_message_type(isolated_env, tmp_path): - """Unknown MESSAGE_TYPE exits 4.""" - msg = _make_message(tmp_path, msg_type="frobnicate") - import socket - machine = socket.gethostname().split(".")[0] - result = _run( - [f"{machine}.homelab", str(msg)], - env={**os.environ, "HOME": str(isolated_env)}, - ) - assert result.returncode == 4 - assert "MESSAGE_TYPE" in result.stderr - - -def test_send_halt_blocks(isolated_env, tmp_path): - """When HALT exists, send refuses with exit 5.""" - halt = isolated_env / ".config" / "cross-agent-comms" / "HALT" - halt.parent.mkdir(parents=True) - halt.write_text("test halt\n") - msg = _make_message(tmp_path) - import socket - machine = socket.gethostname().split(".")[0] - result = _run( - [f"{machine}.homelab", str(msg)], - env={**os.environ, "HOME": str(isolated_env)}, - ) - assert result.returncode == 5 - assert "halt active" in result.stderr.lower() - - -def test_send_same_machine_no_sign_delivers(isolated_env, tmp_path): - """Same-machine delivery with --no-sign produces a canonically named file.""" - msg = _make_message(tmp_path, conv_id="my-conv") - import socket - machine = socket.gethostname().split(".")[0] - # Sender is derived from CWD walking up to ~/projects/<name>/ - cwd = isolated_env / "projects" / "homelab" - result = _run( - [f"{machine}.homelab", str(msg), "--no-sign"], - env={**os.environ, "HOME": str(isolated_env)}, - cwd=cwd, - ) - assert result.returncode == 0, f"stderr={result.stderr}" - inbox = isolated_env / "projects" / "homelab" / "inbox" / "from-agents" - files = list(inbox.glob("*-from-homelab-my-conv.org")) - assert len(files) == 1 - # No sig file with --no-sign. - assert not list(inbox.glob("*.asc")) - # Canonical filename pattern. - assert files[0].name.startswith("2026") and files[0].name.endswith("-from-homelab-my-conv.org") - - -def test_send_same_machine_signed_writes_asc(isolated_env, tmp_path): - """Signed delivery writes both .org and .asc.""" - msg = _make_message(tmp_path, conv_id="signed-conv") - import socket - machine = socket.gethostname().split(".")[0] - cwd = isolated_env / "projects" / "homelab" - # Use the real GPG keyring (not isolating GPG — Craig's existing keys are fine for tests). - real_env = {**os.environ, "HOME": str(isolated_env), "GNUPGHOME": str(Path.home() / ".gnupg")} - result = _run( - [f"{machine}.homelab", str(msg)], - env=real_env, - cwd=cwd, - ) - if result.returncode != 0: - pytest.skip(f"GPG signing unavailable in this environment: {result.stderr}") - inbox = isolated_env / "projects" / "homelab" / "inbox" / "from-agents" - org_files = list(inbox.glob("*-from-homelab-signed-conv.org")) - asc_files = list(inbox.glob("*-from-homelab-signed-conv.org.asc")) - assert len(org_files) == 1 - assert len(asc_files) == 1 - - -def test_send_filename_ignores_input_basename(isolated_env, tmp_path): - """User's input filename is ignored; canonical filename is generated.""" - weird = tmp_path / "weird-user-name.org" - weird.write_text(textwrap.dedent("""\ - #+TITLE: Title - #+CONVERSATION_ID: ignored-input - #+MESSAGE_TYPE: request - #+SEQUENCE: 1 - #+TIMESTAMP: 2026-04-27T05:00:00-05:00 - #+PROTOCOL_VERSION: 5 - - Body. - """)) - import socket - machine = socket.gethostname().split(".")[0] - cwd = isolated_env / "projects" / "homelab" - result = _run( - [f"{machine}.homelab", str(weird), "--no-sign"], - env={**os.environ, "HOME": str(isolated_env)}, - cwd=cwd, - ) - assert result.returncode == 0 - inbox = isolated_env / "projects" / "homelab" / "inbox" / "from-agents" - # No file named after the user's input. - assert not (inbox / "weird-user-name.org").exists() - # Canonical naming used. - assert list(inbox.glob("*-from-homelab-ignored-input.org")) |
