File size: 3,135 Bytes
f109bf8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """Smoke: only selected agents receive WorkItems and run."""
from __future__ import annotations
import unittest
from execution.models import Evidence, EvidenceBundle
from execution.dispatcher import (
ExecutionRequest,
build_work_items,
dispatch,
stub_handler,
)
SELECTED = [
"LegacyCobol",
"Database",
"ReverseEngineering",
"Provenance",
"Risk",
"Response",
]
class TestSparseDispatch(unittest.TestCase):
def setUp(self) -> None:
self.bundle = EvidenceBundle(
sources=(
Evidence(
evidence_id="e1",
source_id="s1",
source_type="cobol_source",
document_id="pay.cbl",
location="lines:100-200",
content_hash="abc",
content="COMPUTE BALANCE ...",
),
),
)
self.req = ExecutionRequest(
request_id="req-cobol-1",
task=(
"Explain how this COBOL payment program calculates "
"the account balance and determine whether its Java "
"replacement preserves the behavior."
),
evidence_bundle=self.bundle,
required_outputs=("business_rules", "equivalence_result"),
permissions=frozenset(
{"read_source", "read_copybooks", "read_schema", "read_evidence"}
),
risk_policy="escalate_on_high",
)
self.handlers = {aid: stub_handler(aid) for aid in SELECTED}
# Quant must never run even if a handler exists.
self.handlers["Quant"] = stub_handler("Quant")
def test_work_items_only_for_selected(self) -> None:
items = build_work_items(SELECTED, self.req)
self.assertEqual([i.agent_id for i in items], SELECTED)
self.assertTrue(all(i.request_id == "req-cobol-1" for i in items))
self.assertNotIn("Quant", [i.agent_id for i in items])
def test_dispatch_invokes_only_selected(self) -> None:
invoked: list[str] = []
def wrap(aid: str):
inner = stub_handler(aid)
def _h(item):
invoked.append(aid)
return inner(item)
return _h
handlers = {aid: wrap(aid) for aid in self.handlers}
results = dispatch(SELECTED, self.req, handlers)
self.assertEqual([r.agent_id for r in results], SELECTED)
self.assertEqual(invoked, SELECTED)
self.assertNotIn("Quant", invoked)
self.assertTrue(all(r.status == "ok" for r in results))
def test_empty_selection_runs_nobody(self) -> None:
results = dispatch([], self.req, self.handlers)
self.assertEqual(results, [])
def test_evidence_bundle_not_bare_text(self) -> None:
items = build_work_items(["LegacyCobol"], self.req)
src = items[0].evidence_bundle.sources[0]
self.assertEqual(src.evidence_id, "e1")
self.assertEqual(src.content_hash, "abc")
self.assertIsNotNone(src.source_id)
if __name__ == "__main__":
unittest.main()
|