Adam Moussa
0842ff778b
Live github (issue-comment poster) and claude_code (file-drop) transports, plus GithubIntake (labeled issue -> coordinator.start_task, de-duped). run-team _build_transport now wires github/claude_code live (was SystemExit) + adds the intake-github subcommand. claude_code drop-path also neutralizes backslash (defense-in-depth).
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Unit tests for agent_team.transport.claude_code_live (§3.3.1, §7.1 P4, D10).
|
|
|
|
The live file-drop wiring is the production backing for the §3.3.1 injected
|
|
Claude-Code ``delivery`` seam. These tests prove the contract entirely with
|
|
mocks (no network, no SDK, and the real filesystem is exercised only through
|
|
``tmp_path`` or injected fakes): the sink writes a prompt drop and returns the
|
|
session id, the ``channel_ref`` round-trips the ``question_id`` through a real
|
|
``ClaudeCodeAdapter``, a dropped answer is read back into a ``parse_answer``
|
|
payload that maps to the original question, and an unaddressable / unwritable
|
|
drop fails loudly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from agent_team.transport.base import QuestionSet
|
|
from agent_team.transport.claude_code_adapter import (
|
|
VIA,
|
|
ClaudeCodeAdapter,
|
|
ClaudeCodeDeliveryError,
|
|
build_channel_ref,
|
|
render_prompt,
|
|
)
|
|
from agent_team.transport.claude_code_live import (
|
|
ANSWER_SUFFIX,
|
|
PROMPT_SUFFIX,
|
|
build_claude_code_delivery,
|
|
build_file_drop_reader,
|
|
build_file_drop_writer,
|
|
build_live_claude_code_transport,
|
|
read_answer_payload,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Test doubles / helpers #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class _RecordingWriter:
|
|
"""A fake prompt-writer recording the (path, prompt) it was handed."""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[Path, str]] = []
|
|
|
|
def __call__(self, path: Path, prompt: str) -> None:
|
|
self.calls.append((path, prompt))
|
|
|
|
|
|
def _question_set(**overrides: Any) -> QuestionSet:
|
|
defaults: dict[str, Any] = {
|
|
"thread_id": "t1",
|
|
"question_id": "q1",
|
|
"turn": 0,
|
|
"questions": ["Proceed with the dependency bump?"],
|
|
}
|
|
defaults.update(overrides)
|
|
return QuestionSet(**defaults)
|
|
|
|
|
|
def _prompt_for(question_id: str) -> str:
|
|
return render_prompt(
|
|
question_id=question_id,
|
|
turn=0,
|
|
question_set=_question_set(question_id=question_id),
|
|
deadline="2026-06-18T00:00:00Z",
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Clean import (no optional SDK, no network) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_module_imports_cleanly() -> None:
|
|
"""The module reloads without any optional dependency or network."""
|
|
module = importlib.reload(
|
|
importlib.import_module("agent_team.transport.claude_code_live")
|
|
)
|
|
assert hasattr(module, "build_claude_code_delivery")
|
|
assert hasattr(module, "build_live_claude_code_transport")
|
|
assert hasattr(module, "read_answer_payload")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# delivery sink: addressing + session id #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_delivery_derives_session_id_from_embedded_marker() -> None:
|
|
"""The session id is the question_id embedded in the prompt marker."""
|
|
writer = _RecordingWriter()
|
|
delivery = build_claude_code_delivery("/drop", writer=writer)
|
|
|
|
session_id = delivery(session_hint="", prompt=_prompt_for("qEmbed"))
|
|
|
|
assert session_id == "qEmbed"
|
|
|
|
|
|
def test_delivery_falls_back_to_session_hint_without_marker() -> None:
|
|
"""A markerless prompt is addressed by the session_hint."""
|
|
writer = _RecordingWriter()
|
|
delivery = build_claude_code_delivery("/drop", writer=writer)
|
|
|
|
session_id = delivery(session_hint="mac-sess-9", prompt="bare prompt, no marker")
|
|
|
|
assert session_id == "mac-sess-9"
|
|
|
|
|
|
def test_delivery_unaddressable_prompt_raises() -> None:
|
|
"""No marker and no session_hint cannot be addressed: fail loudly."""
|
|
delivery = build_claude_code_delivery("/drop", writer=_RecordingWriter())
|
|
|
|
with pytest.raises(ClaudeCodeDeliveryError):
|
|
delivery(session_hint="", prompt="bare prompt, no marker")
|
|
|
|
|
|
def test_delivery_writes_prompt_to_drop_path() -> None:
|
|
"""The sink hands the writer a path under the drop dir with PROMPT_SUFFIX."""
|
|
writer = _RecordingWriter()
|
|
delivery = build_claude_code_delivery("/drop", writer=writer)
|
|
|
|
delivery(session_hint="", prompt=_prompt_for("qWrite"))
|
|
|
|
assert len(writer.calls) == 1
|
|
path, prompt = writer.calls[0]
|
|
assert path == Path("/drop") / f"qWrite{PROMPT_SUFFIX}"
|
|
assert "qWrite" in prompt
|
|
|
|
|
|
def test_delivery_sanitizes_separators_in_session_id() -> None:
|
|
"""A session_hint with path separators cannot escape the drop directory."""
|
|
writer = _RecordingWriter()
|
|
delivery = build_claude_code_delivery("/drop", writer=writer)
|
|
|
|
delivery(session_hint="../../etc/passwd", prompt="no marker here")
|
|
|
|
path, _prompt = writer.calls[0]
|
|
# The drop must stay inside the drop directory: separators are flattened so
|
|
# the file is a single component under /drop, not a traversal out of it.
|
|
assert path.parent == Path("/drop")
|
|
assert path.name == f"_.._etc_passwd{PROMPT_SUFFIX}"
|
|
assert path == Path("/drop") / path.name
|
|
|
|
|
|
def test_delivery_wraps_writer_oserror() -> None:
|
|
"""A writer OSError surfaces as ClaudeCodeDeliveryError (failed post)."""
|
|
|
|
def _boom(path: Path, prompt: str) -> None:
|
|
raise OSError("disk full")
|
|
|
|
delivery = build_claude_code_delivery("/drop", writer=_boom)
|
|
|
|
with pytest.raises(ClaudeCodeDeliveryError, match="failed to write"):
|
|
delivery(session_hint="", prompt=_prompt_for("qBoom"))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Wired through a real adapter: post -> channel_ref #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_post_question_round_trips_question_id_as_channel_ref() -> None:
|
|
"""Through a real adapter, the drop's session id becomes the channel_ref."""
|
|
writer = _RecordingWriter()
|
|
transport = ClaudeCodeAdapter(
|
|
build_claude_code_delivery("/drop", writer=writer),
|
|
)
|
|
|
|
channel_ref = transport.post_question(
|
|
thread_id="t1",
|
|
question_id="q1",
|
|
turn=0,
|
|
question_set=_question_set(),
|
|
deadline="2026-06-18T00:00:00Z",
|
|
)
|
|
|
|
assert channel_ref == build_channel_ref("q1", "q1")
|
|
assert writer.calls, "a prompt drop must have been written"
|
|
|
|
|
|
def test_convenience_factory_returns_adapter() -> None:
|
|
"""``build_live_claude_code_transport`` yields a ClaudeCodeAdapter."""
|
|
transport = build_live_claude_code_transport("/drop", session_hint="mac-1")
|
|
assert isinstance(transport, ClaudeCodeAdapter)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Inbound: read dropped answer -> parse_answer payload #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_read_answer_payload_maps_back_to_question_id() -> None:
|
|
"""A dropped answer reads into a payload that parse_answer maps correctly."""
|
|
channel_ref = build_channel_ref("q1", "q1")
|
|
|
|
def _reader(path: Path) -> str | None:
|
|
assert path == Path("/drop") / f"q1{ANSWER_SUFFIX}"
|
|
return "ship it"
|
|
|
|
payload = read_answer_payload(channel_ref, "/drop", reader=_reader)
|
|
|
|
assert payload == {"channel_ref": channel_ref, "answer": "ship it"}
|
|
# The payload must feed parse_answer and recover the original question_id.
|
|
adapter = ClaudeCodeAdapter()
|
|
assert adapter.parse_answer(payload) == ("q1", "ship it", VIA)
|
|
|
|
|
|
def test_read_answer_payload_none_when_no_answer_dropped() -> None:
|
|
"""No dropped answer yet returns None so reconcile can poll idempotently."""
|
|
|
|
def _reader(path: Path) -> str | None:
|
|
return None
|
|
|
|
payload = read_answer_payload(
|
|
build_channel_ref("q1", "q1"), "/drop", reader=_reader
|
|
)
|
|
|
|
assert payload is None
|
|
|
|
|
|
def test_read_answer_payload_rejects_foreign_channel_ref() -> None:
|
|
"""A non-Claude-Code channel_ref is rejected rather than mis-read."""
|
|
with pytest.raises(ValueError):
|
|
read_answer_payload("1718000000.001100", "/drop", reader=lambda p: None)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Full round-trip over the real filesystem (tmp_path, no network) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_filesystem_writer_and_reader_round_trip(tmp_path: Path) -> None:
|
|
"""The default filesystem writer/reader round-trip a prompt and answer."""
|
|
drop = tmp_path / "claude-drop"
|
|
|
|
# Post a question with the live filesystem-backed sink.
|
|
transport = build_live_claude_code_transport(drop)
|
|
channel_ref = transport.post_question(
|
|
thread_id="t1",
|
|
question_id="qFS",
|
|
turn=0,
|
|
question_set=_question_set(question_id="qFS"),
|
|
deadline="2026-06-18T00:00:00Z",
|
|
)
|
|
assert channel_ref == build_channel_ref("qFS", "qFS")
|
|
|
|
prompt_file = drop / f"qFS{PROMPT_SUFFIX}"
|
|
assert prompt_file.exists()
|
|
assert "qFS" in prompt_file.read_text(encoding="utf-8")
|
|
|
|
# Before the harness answers, the reader yields None.
|
|
assert read_answer_payload(channel_ref, drop) is None
|
|
|
|
# The harness drops an answer file; the reader picks it up.
|
|
(drop / f"qFS{ANSWER_SUFFIX}").write_text("done", encoding="utf-8")
|
|
payload = read_answer_payload(channel_ref, drop)
|
|
assert payload == {"channel_ref": channel_ref, "answer": "done"}
|
|
assert transport.parse_answer(payload) == ("qFS", "done", VIA)
|
|
|
|
|
|
def test_build_file_drop_writer_creates_dir(tmp_path: Path) -> None:
|
|
"""The filesystem writer creates a missing drop directory on first write."""
|
|
drop = tmp_path / "nested" / "drop"
|
|
writer = build_file_drop_writer(drop)
|
|
|
|
writer(drop / f"qX{PROMPT_SUFFIX}", "hello")
|
|
|
|
assert (drop / f"qX{PROMPT_SUFFIX}").read_text(encoding="utf-8") == "hello"
|
|
|
|
|
|
def test_build_file_drop_reader_missing_file_returns_none(tmp_path: Path) -> None:
|
|
"""The filesystem reader returns None for an absent answer file."""
|
|
reader = build_file_drop_reader(tmp_path)
|
|
assert reader(tmp_path / "absent.answer.txt") is None
|