"""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()