ive harnessed the harness
Something went wrong. Try again.
11 kB · 273 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274"""Tests for evidence resolver, workspace info, and maintenance origination SDK functions."""from __future__ import annotations
import unittest
from klbr_runtime.context import ( Bridge, ExecutionScope, assess_maintenance, current, originate_maintenance, resolve_evidence, workspace_info,)from klbr.maintenance import load_maintenance_evidence
class MockWire: def __init__(self, responses: dict | None = None): self.responses = responses or {} self.sent = [] self.bridge: Bridge | None = None
async def send(self, msg): self.sent.append(msg) req_method = msg.get("request", {}).get("method") if req_method in self.responses and self.bridge is not None: self.bridge.reply({ "kind": "host_reply", "id": msg["id"], "execution_id": msg["execution_id"], "reply": {"kind": "ok", "value": self.responses[req_method]}, })
class EvidenceResolverTests(unittest.IsolatedAsyncioTestCase): """Verify evidence resolver and workspace info SDK functions."""
async def test_resolve_evidence_parses_refs(self): expected_result = { "resolved": [ { "ref": "event:42", "kind": "event", "event": {"id": 42, "kind": "tool.result", "payload": {"error": "TypeError"}, "created_ms": 1000}, "workspace": {"path": "/measured/workspace", "session_id": "test-session"}, "revision": {"id": "rev-abc", "epoch": 1}, } ], "unavailable": [ {"ref": "event:999", "reason": "not_found"} ], }
wire = MockWire({"evidence.resolve": expected_result}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope)
try: result = await resolve_evidence(["event:42", "event:999"])
self.assertEqual(len(result["resolved"]), 1) self.assertEqual(result["resolved"][0]["ref"], "event:42") self.assertEqual(result["resolved"][0]["workspace"]["path"], "/measured/workspace") self.assertEqual(len(result["unavailable"]), 1) self.assertEqual(result["unavailable"][0]["ref"], "event:999") self.assertEqual(result["unavailable"][0]["reason"], "not_found") finally: current.reset(token)
async def test_workspace_info_returns_measured_path_and_session(self): expected_result = { "path": "/measured/workspace/project", "session_id": "session-xyz", "active_revision": "rev-123", "epoch": 2, }
wire = MockWire({"workspace.info": expected_result}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope)
try: result = await workspace_info()
self.assertEqual(result["path"], "/measured/workspace/project") self.assertEqual(result["session_id"], "session-xyz") self.assertEqual(result["active_revision"], "rev-123") self.assertEqual(result["epoch"], 2) finally: current.reset(token)
async def test_load_maintenance_evidence_builds_packet_with_missing(self): wire_data = { "resolved": [ { "ref": "event:10", "kind": "event", "event": {"id": 10, "kind": "scratchpad", "payload": {"error": "IndexError: out of bounds"}}, } ], "unavailable": [ {"ref": "source:missing.py", "reason": "file not found in workspace or release"} ], }
wire = MockWire({"evidence.resolve": wire_data}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope)
try: evidence = await load_maintenance_evidence(["event:10", "source:missing.py"])
self.assertEqual(evidence.observation_id, "event:10") self.assertEqual(evidence.error_trace, "IndexError: out of bounds") self.assertIn("source:missing.py", evidence.missing_evidence) self.assertEqual(evidence.evidence_refs, ("event:10", "source:missing.py")) finally: current.reset(token)
async def test_originate_maintenance_requests_runnable_work(self): expected_reply = { "item_id": "work-maint-123", "work_item_id": "work-maint-123", "seq": 45, }
wire = MockWire({"maintenance.originate": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope)
try: res = await originate_maintenance( purpose="repair", reason="failure observed in check", evidence_refs=["event:10"], impact="high", ) self.assertEqual(res["item_id"], "work-maint-123") self.assertEqual(res["seq"], 45) self.assertEqual(wire.sent[0]["request"]["method"], "maintenance.originate") self.assertEqual(wire.sent[0]["request"]["purpose"], "repair") finally: current.reset(token)
async def test_assess_maintenance_requests_host_recording(self): expected_reply = { "status": "recorded", "work_item_id": "work-assess-1", "seq": 46, }
wire = MockWire({"maintenance.assess": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope)
try: res = await assess_maintenance( work_item_id="work-assess-1", outcome="needs-evidence", reason="missing file", missing_refs=["source:foo.py"], revisit_condition="awaiting file", candidate_revision="rev-1", check_receipts=["check-1"], ) self.assertEqual(res["status"], "recorded") self.assertEqual(res["work_item_id"], "work-assess-1") self.assertEqual(wire.sent[0]["request"]["method"], "maintenance.assess") self.assertEqual(wire.sent[0]["request"]["outcome"], "needs-evidence") self.assertEqual(wire.sent[0]["request"]["missing_refs"], ["source:foo.py"]) finally: current.reset(token)
class AssessMaintenanceBoundaryTests(unittest.IsolatedAsyncioTestCase): """The SDK validates the closed assessment outcome set before forwarding."""
def _scope_and_wire(self, expected_reply: dict) -> tuple: wire = MockWire({"maintenance.assess": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) return scope, wire, token
async def test_invalid_outcome_fails_locally_and_is_not_forwarded(self): scope, wire, token = self._scope_and_wire({"status": "recorded"}) try: with self.assertRaises(ValueError) as ctx: await assess_maintenance( work_item_id="work-assess-1", outcome="actviated", reason="fixture typo", ) self.assertIn("is not a disposition for a work item", str(ctx.exception)) self.assertEqual(wire.sent, [], "an invalid outcome must never reach the host") finally: current.reset(token)
async def test_missing_required_variant_fields_fail_locally(self): scope, wire, token = self._scope_and_wire({"status": "recorded"}) try: with self.assertRaises(ValueError): await assess_maintenance( work_item_id="work-assess-1", outcome="no-change", ) with self.assertRaises(ValueError): await assess_maintenance( work_item_id="work-assess-1", outcome="needs-evidence", ) with self.assertRaises(ValueError): await assess_maintenance( work_item_id="work-assess-1", outcome="activated", check_receipts=["event:10"], ) self.assertEqual(wire.sent, []) finally: current.reset(token)
async def test_valid_assessments_forward_the_closed_fields(self): expected_reply = {"status": "recorded", "work_item_id": "work-assess-1", "seq": 46}
wire = MockWire({"maintenance.assess": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope)
try: res = await assess_maintenance( work_item_id="work-assess-1", outcome="needs-evidence", reason="missing file", missing_refs=["source:foo.py"], revisit_condition="awaiting file", revisit_deadline_ms=50_000, ) self.assertEqual(res["status"], "recorded") self.assertEqual(res["work_item_id"], "work-assess-1") self.assertEqual(wire.sent[0]["request"]["method"], "maintenance.assess") self.assertEqual(wire.sent[0]["request"]["outcome"], "needs-evidence") self.assertEqual(wire.sent[0]["request"]["missing_refs"], ["source:foo.py"]) self.assertEqual(wire.sent[0]["request"]["revisit_deadline_ms"], 50_000)
res = await assess_maintenance( work_item_id="work-assess-1", outcome="deferred", reason="checkpoint retained", trigger_ms=60_000, ) self.assertEqual(res["status"], "recorded") self.assertEqual(wire.sent[-1]["request"]["outcome"], "deferred") self.assertEqual(wire.sent[-1]["request"]["trigger_ms"], 60_000) finally: current.reset(token)
if __name__ == "__main__": unittest.main()