aboutsummaryrefslogtreecommitdiff
path: root/.ai/scripts/tests/test_cross_agent_send.py
diff options
context:
space:
mode:
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_send.py')
-rw-r--r--.ai/scripts/tests/test_cross_agent_send.py210
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"))