109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Unit tests for the Phase-3 shared fact ledger."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.maf_roundtable.fact_ledger import (
|
|
assert_ledger_invariants,
|
|
empty_ledger,
|
|
executed_action_ids,
|
|
merge_seat_output,
|
|
normalize_ledger,
|
|
proposed_action_ids,
|
|
record_action_transition,
|
|
record_tool_result,
|
|
)
|
|
|
|
|
|
def test_empty_ledger_has_all_partitions():
|
|
ledger = empty_ledger()
|
|
for key in (
|
|
"confirmed_facts",
|
|
"evidence",
|
|
"tool_results",
|
|
"executed_actions",
|
|
"pending_actions",
|
|
"failed_or_unknown_actions",
|
|
"conflicts",
|
|
"proposed_actions",
|
|
):
|
|
assert key in ledger
|
|
assert ledger[key] == []
|
|
|
|
|
|
def test_normalize_legacy_checkpoint():
|
|
legacy = {
|
|
"confirmed_facts": ["a"],
|
|
"tool_results": [{"tool_name": "t", "status": "completed"}],
|
|
"conflicts": [],
|
|
"pending_actions": [],
|
|
}
|
|
n = normalize_ledger(legacy)
|
|
assert n["confirmed_facts"] == ["a"]
|
|
assert n["executed_actions"] == []
|
|
assert n["evidence"] == []
|
|
|
|
|
|
def test_merge_seat_output_keeps_proposed_out_of_executed():
|
|
ledger = empty_ledger()
|
|
merge_seat_output(
|
|
ledger,
|
|
{
|
|
"summary": "s",
|
|
"findings": ["fact-1"],
|
|
"evidence_refs": ["doc:1"],
|
|
"proposed_actions": [{"text": "create ticket", "action_id": "prop-1"}],
|
|
"executed_actions": [{"text": "forged", "action_id": "forge-1"}],
|
|
"conflicts": ["c1"],
|
|
},
|
|
agent_id="seat-a",
|
|
)
|
|
# Phase-3: seat findings land in candidate_facts (unverified claims), NOT
|
|
# confirmed_facts. Promotion to confirmed happens only after cross-review.
|
|
assert ledger["confirmed_facts"] == []
|
|
assert any(
|
|
c.get("text") == "fact-1" for c in ledger["candidate_facts"] if isinstance(c, dict)
|
|
)
|
|
assert "prop-1" in proposed_action_ids(ledger)
|
|
assert "forge-1" not in executed_action_ids(ledger)
|
|
assert_ledger_invariants(ledger)
|
|
|
|
|
|
def test_record_action_transition_pending_to_executed():
|
|
ledger = empty_ledger()
|
|
record_action_transition(
|
|
ledger, action_id="a1", tool_name="create_ticket", status="pending", agent_id="s1"
|
|
)
|
|
assert len(ledger["pending_actions"]) == 1
|
|
record_action_transition(
|
|
ledger, action_id="a1", tool_name="create_ticket", status="executed", agent_id="s1"
|
|
)
|
|
assert ledger["pending_actions"] == []
|
|
assert executed_action_ids(ledger) == {"a1"}
|
|
assert_ledger_invariants(ledger)
|
|
|
|
|
|
def test_record_action_transition_failed_not_executed():
|
|
ledger = empty_ledger()
|
|
record_action_transition(ledger, action_id="a2", tool_name="write_file", status="pending")
|
|
record_action_transition(ledger, action_id="a2", tool_name="write_file", status="unknown")
|
|
assert executed_action_ids(ledger) == set()
|
|
assert any(e.get("action_id") == "a2" for e in ledger["failed_or_unknown_actions"])
|
|
|
|
|
|
def test_record_tool_result_status_clamp():
|
|
ledger = empty_ledger()
|
|
record_tool_result(ledger, tool_name="web_search", status="completed", agent_id="a")
|
|
record_tool_result(ledger, tool_name="x", status="bogus", agent_id="a")
|
|
assert ledger["tool_results"][0]["status"] == "completed"
|
|
assert ledger["tool_results"][1]["status"] == "unknown"
|
|
assert_ledger_invariants(ledger)
|
|
|
|
|
|
def test_assert_rejects_bad_tool_status():
|
|
ledger = empty_ledger()
|
|
ledger["tool_results"].append({"tool_name": "t", "status": "weird"})
|
|
with pytest.raises(AssertionError):
|
|
assert_ledger_invariants(ledger)
|