tegwick
6d2ccc7760
Some checks failed
tamq-ci / test (push) Failing after 5s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
import json
|
|
|
|
from tamq.capture import render_jsonl, render_markdown
|
|
from tamq.cli import main
|
|
from tamq.store import Store
|
|
|
|
|
|
def test_capture_renderers_expose_protocol_but_not_duplicate_message_bodies(tmp_path):
|
|
store = Store(tmp_path / "queue.sqlite3")
|
|
message_id = store.add(
|
|
"source", "target", "Please review.\nReply accepted.", provenance="worker_output"
|
|
)
|
|
store.record_protocol_event(
|
|
"delivery.attempted",
|
|
repo="target",
|
|
peer_repo="source",
|
|
message_id=message_id,
|
|
provenance="worker_output",
|
|
delivery_mode="trigger",
|
|
outcome="attempted",
|
|
)
|
|
rows = store.protocol_events(repo="source")
|
|
|
|
markdown = render_markdown(rows, scope="repository `source`")
|
|
assert "# TAMQ protocol capture" in markdown
|
|
assert "Accepted messages: 1" in markdown
|
|
assert "## Review cues" in markdown
|
|
assert markdown.count("Please review.") == 1
|
|
assert "does not contain unrelated pane output" in markdown
|
|
|
|
records = [json.loads(line) for line in render_jsonl(rows).splitlines()]
|
|
assert records[0]["message_body"] == "Please review.\nReply accepted."
|
|
assert "message_body" not in records[1]
|
|
store.close()
|
|
|
|
|
|
def test_capture_cli_filters_and_writes_review_file(tmp_path, monkeypatch, capsys):
|
|
database = tmp_path / "queue.sqlite3"
|
|
monkeypatch.setattr("tamq.cli.db_path", lambda: database)
|
|
store = Store(database)
|
|
store.add("one", "two", "hello", provenance="operator_input")
|
|
store.add("three", "four", "ignore", provenance="worker_output")
|
|
store.close()
|
|
output = tmp_path / "review" / "protocol.md"
|
|
|
|
assert main(["capture", "--repo", "one", "--output", str(output)]) == 0
|
|
assert "Wrote 1 protocol event" in capsys.readouterr().out
|
|
report = output.read_text()
|
|
assert "hello" in report
|
|
assert "ignore" not in report
|
|
|
|
|
|
def test_capture_cli_rejects_non_positive_limit(tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setattr("tamq.cli.db_path", lambda: tmp_path / "queue.sqlite3")
|
|
assert main(["capture", "--limit", "0"]) == 2
|
|
assert "capture limit must be a positive integer" in capsys.readouterr().err
|