210 lines
8.9 KiB
Python
210 lines
8.9 KiB
Python
"""Tests for scripts/assemble_telegram_visible_direct_claim_capture_receipt.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import assemble_telegram_visible_direct_claim_capture_receipt as receipt # noqa: E402
|
|
|
|
STRUCTURED_FIXTURE_READBACKS = {
|
|
"DC-01": "\nDB readback: Proposal: f004bbb2-ac9a-481f-b7b8-74319373ba6a; Status: applied; Applied at: 2026-07-05.",
|
|
"DC-02": "\nDB readback: Proposal: a64df080-8502-42e2-98f4-9bbdecb8da73; Status: approved; Applied at: none.",
|
|
"DC-03": "\nDB readback: Proposal: a64df080-8502-42e2-98f4-9bbdecb8da73; Status: approved; Applied at: none.",
|
|
"DC-04": (
|
|
"\nDB readback: Proposal: e987b07c-9c23-45ba-b95c-d6d04c6536c7; Status: pending_review; "
|
|
"Applied at: none; Source_ref: telegram-resolved-attachment:doc_996e2e86cd94."
|
|
),
|
|
"DC-05": (
|
|
"\nDB readback: claims: 1837; sources: 4145; claim_edges: 4916; "
|
|
"claim_evidence: 4670; kb_proposals: 26."
|
|
),
|
|
"DC-06": "\nDB readback: Proposal: f004bbb2-ac9a-481f-b7b8-74319373ba6a; Status: applied; Applied at: 2026-07-05.",
|
|
}
|
|
|
|
KNOWN_GOOD_REPLIES = {
|
|
"DC-01": (
|
|
"Mostly still proposals. Approved is not the same as applied. Canonical public.* changes require applied_at "
|
|
"and row-level postflight proof; I cannot claim the DB changed without that readback. Next proof-changing "
|
|
"follow-up: validate or rebuild a strict payload for the approved proposal before review and authorization."
|
|
),
|
|
"DC-02": (
|
|
"No, not canonical. Helmer is approved and staged, but its public.sources, public.claims, evidence, edges, "
|
|
"and reasoning tool rows are not applied. Explicit operator authorization comes only after strict payload "
|
|
"validation. Next proof-changing follow-up: validate or rebuild the payload, review it, then authorize apply and "
|
|
"postflight."
|
|
),
|
|
"DC-03": (
|
|
"No. Fresh readback: the decision-matrix schema is absent; reviewer status is not a decision-matrix vote. "
|
|
"matrix_voters, proposal_votes, and proposal_decisions do not exist, so canonical public.* state must be "
|
|
"checked separately. Next proof-changing follow-up: validate the strict proposal payload before using the "
|
|
"reviewer/admin path."
|
|
),
|
|
"DC-04": (
|
|
"Not just pointer mismatch. Telegram file refs, document evaluations, proposal source_ref keys, raw files, "
|
|
"and canonical public.sources rows are distinct layers; pending_review also reflects reviewer state. The "
|
|
"pointer claim does not prove the whole cause. Next proof-changing follow-up: run a row-link audit and draft the "
|
|
"guarded source/evidence apply contract."
|
|
),
|
|
"DC-05": (
|
|
"Staging yes, canonical KB change not safe to demo from chat. Approved is not the same as applied. A staging "
|
|
"write leaves a proposal UUID and row-level readback artifact; canonical public.claim_edges mutation requires "
|
|
"explicit operator authorization and retained postflight proof. Next proof-changing follow-up: stage a separately "
|
|
"reviewed strict add_edge canary with two existing canonical claim IDs."
|
|
),
|
|
"DC-06": (
|
|
"No. SOUL.md is a runtime/rendered artifact, not canonical Postgres and not the source of truth. Canonical "
|
|
"identity requires new or updated public.* row IDs plus renderer and postflight proof; direct edits do not "
|
|
"change canonical identity. Next proof-changing follow-up: audit the identity rows and stage the missing proposal."
|
|
),
|
|
}
|
|
|
|
|
|
def load_packet() -> dict:
|
|
return json.loads(receipt.DEFAULT_PACKET.read_text(encoding="utf-8"))
|
|
|
|
|
|
def retained_visible_replies() -> dict[str, str]:
|
|
data = json.loads(
|
|
(
|
|
REPO_ROOT
|
|
/ "docs/reports/leo-working-state-20260709/telegram-visible-direct-claim-capture-results-current.json"
|
|
).read_text(encoding="utf-8")
|
|
)
|
|
return {item["prompt_id"]: item["reply"] for item in data["results"]}
|
|
|
|
|
|
def good_capture(packet: dict) -> dict:
|
|
messages = []
|
|
for index, item in enumerate(packet["exact_messages"], start=1):
|
|
messages.append(
|
|
{
|
|
"prompt_id": item["prompt_id"],
|
|
"sent_text": item["message"],
|
|
"reply": KNOWN_GOOD_REPLIES[item["prompt_id"]] + STRUCTURED_FIXTURE_READBACKS[item["prompt_id"]],
|
|
"sent_at_utc": f"2026-07-10T03:0{index}:00Z",
|
|
"reply_at_utc": f"2026-07-10T03:0{index}:30Z",
|
|
"telegram_message_id": f"sent-{index}",
|
|
"reply_message_id": f"reply-{index}",
|
|
}
|
|
)
|
|
return {
|
|
"operator_authorization_text": packet["explicit_operator_authorization_text"],
|
|
"capture_source": {
|
|
"platform": "telegram",
|
|
"chat_label": "Leo",
|
|
"chat_id": "-5146042086",
|
|
"captured_at_utc": "2026-07-10T03:08:00Z",
|
|
"captured_by": "codex",
|
|
"final_screenshot": "outputs/telegram-visible-direct-claim-suite-20260710/final.png",
|
|
"final_accessibility": "outputs/telegram-visible-direct-claim-suite-20260710/final.txt",
|
|
},
|
|
"db_before_after": {
|
|
"before": {"public.claims": 1837},
|
|
"after": {"public.claims": 1837},
|
|
"db_counts_changed": False,
|
|
},
|
|
"service_readback": {
|
|
"before": {"ActiveState": "active", "SubState": "running"},
|
|
"after": {"ActiveState": "active", "SubState": "running"},
|
|
},
|
|
"messages": messages,
|
|
}
|
|
|
|
|
|
def test_receipt_without_capture_emits_awaiting_capture_template():
|
|
packet = load_packet()
|
|
|
|
data = receipt.build_receipt(packet, None, source_capture_path=None)
|
|
|
|
assert data["status"] == "awaiting_capture"
|
|
assert data["receipt_ready"] is False
|
|
assert data["capture_template"]["messages"][0]["prompt_id"] == "DC-01"
|
|
assert data["checks"]["missing_prompt_ids"] == ["DC-01", "DC-02", "DC-03", "DC-04", "DC-05", "DC-06"]
|
|
|
|
|
|
def test_good_capture_scores_and_receipt_passes():
|
|
packet = load_packet()
|
|
|
|
data = receipt.build_receipt(packet, good_capture(packet), source_capture_path=Path("capture.json"))
|
|
|
|
assert data["status"] == "pass"
|
|
assert data["receipt_ready"] is True
|
|
assert data["score"]["pass"] is True
|
|
assert data["score"]["passes"] == data["score"]["expected_prompt_count"] == 6
|
|
assert len(data["results"]) == 6
|
|
assert data["results"][0]["evidence_tier"] == "telegram_visible_capture"
|
|
|
|
|
|
def test_retained_visible_capture_exposes_dc05_apply_path_issue():
|
|
packet = load_packet()
|
|
capture = good_capture(packet)
|
|
replies = retained_visible_replies()
|
|
capture["messages"][4]["reply"] = replies["DC-05"] + STRUCTURED_FIXTURE_READBACKS["DC-05"]
|
|
|
|
data = receipt.build_receipt(packet, capture, source_capture_path=Path("capture.json"))
|
|
|
|
assert data["status"] == "fail"
|
|
assert data["score"]["failures"][0]["prompt_id"] == "DC-05"
|
|
assert data["score"]["failures"][0]["response_issue_detected"] is True
|
|
|
|
|
|
def test_capture_with_message_text_mismatch_fails():
|
|
packet = load_packet()
|
|
capture = good_capture(packet)
|
|
capture["messages"][0]["sent_text"] = "wrong"
|
|
|
|
data = receipt.build_receipt(packet, capture, source_capture_path=Path("capture.json"))
|
|
|
|
assert data["status"] == "fail"
|
|
assert data["receipt_ready"] is False
|
|
assert data["checks"]["sent_text_exact"] is False
|
|
assert data["checks"]["text_mismatch_prompt_ids"] == ["DC-01"]
|
|
|
|
|
|
def test_capture_with_empty_reply_fails():
|
|
packet = load_packet()
|
|
capture = good_capture(packet)
|
|
capture["messages"][1]["reply"] = ""
|
|
|
|
data = receipt.build_receipt(packet, capture, source_capture_path=Path("capture.json"))
|
|
|
|
assert data["status"] == "fail"
|
|
assert data["receipt_ready"] is False
|
|
assert data["checks"]["replies_nonempty"] is False
|
|
assert data["checks"]["empty_reply_prompt_ids"] == ["DC-02"]
|
|
|
|
|
|
def test_main_writes_awaiting_capture_artifacts(tmp_path: Path):
|
|
out = tmp_path / "receipt.json"
|
|
md = tmp_path / "receipt.md"
|
|
results = tmp_path / "results.json"
|
|
|
|
code = receipt.main(["--out", str(out), "--markdown-out", str(md), "--results-out", str(results)])
|
|
|
|
assert code == 0
|
|
assert json.loads(out.read_text(encoding="utf-8"))["status"] == "awaiting_capture"
|
|
assert "Telegram-Visible Direct-Claim Capture Receipt" in md.read_text(encoding="utf-8")
|
|
assert not results.exists()
|
|
|
|
|
|
def test_receipt_records_cli_override_paths(tmp_path: Path):
|
|
packet_path = tmp_path / "packet.json"
|
|
results = tmp_path / "results.json"
|
|
packet = load_packet()
|
|
|
|
packet_path.write_text(json.dumps(packet), encoding="utf-8")
|
|
data = receipt.build_receipt(
|
|
packet,
|
|
good_capture(packet),
|
|
source_capture_path=tmp_path / "capture.json",
|
|
packet_path=packet_path,
|
|
results_out=results,
|
|
)
|
|
|
|
assert data["packet_path"] == str(packet_path)
|
|
assert data["results_out"] == str(results)
|