"""Unit tests for capability checkups, observation processing, and G07/G09/G13/G16/G22."""
from __future__ import annotations
import tempfile
import time
import unittest
from pathlib import Path
from klbr import checks
from klbr.checks import (
CheckResult,
FailureGroup,
Observation,
check_api_consistency,
check_guidance_delivery,
check_resource_access,
compute_action_identity,
group_failure_observations,
)
class CapabilityCheckTests(unittest.TestCase):
def test_typed_check_variants(self):
now = time.time()
h = CheckResult.healthy("test_cap", "scope1", "it works", now + 100.0)
self.assertEqual(h.status, "healthy")
self.assertEqual(h.effective_status(now), "healthy")
self.assertEqual(h.effective_status(now + 200.0), "stale")
d = CheckResult.degraded("test_cap", "scope1", "slow", impact="latency doubled")
self.assertEqual(d.status, "degraded")
self.assertEqual(d.effective_status(now), "degraded")
u = CheckResult.unavailable("test_cap", "scope1", "no socket", dependency="daemon_sock")
self.assertEqual(u.status, "unavailable")
unk = CheckResult.unknown("test_cap", "scope1", reason="timed out", missing_evidence=("trace",))
self.assertEqual(unk.status, "unknown")
self.assertEqual(unk.missing_evidence, ("trace",))
def test_g09_observation_grouping_and_deduplication(self):
now = time.time()
obs = [
Observation(
seq=1, session_id="s1", operation_id="op_diff",
component_revision="rev-1", signature="TypeError: invalid arg",
workspace="/ws/main", payload={"err": "TypeError"}
),
Observation(
seq=2, session_id="s1", operation_id="op_diff",
component_revision="rev-1", signature="TypeError: invalid arg",
workspace="/ws/main", payload={"err": "TypeError"}
),
Observation(
seq=3, session_id="s2", operation_id="op_diff",
component_revision="rev-1", signature="TypeError: invalid arg",
workspace="/ws/main", payload={"err": "TypeError"}
),
Observation(
seq=4, session_id="s1", operation_id="op_push",
component_revision="rev-1", signature="NetworkError: timeout",
workspace="/ws/main", payload={"err": "NetworkError"}
),
]
groups = group_failure_observations(obs)
self.assertEqual(len(groups), 2)
diff_group = next(g for g in groups if g.operation == "op_diff")
self.assertEqual(diff_group.count, 3)
self.assertEqual(diff_group.trail, (1, 2, 3))
self.assertTrue(diff_group.action_identity.startswith("act-"))
# Idempotency of action identity: same inputs produce identical action identity
act_id_2 = compute_action_identity("op_diff", "rev-1", "TypeError: invalid arg", "/ws/main")
self.assertEqual(diff_group.action_identity, act_id_2)
def test_api_consistency_check(self):
import math
# math has sqrt(x), sin(x), pow(x, y)
res = check_api_consistency(math, {"sqrt": ["x"], "pow": ["x", "y"]})
self.assertEqual(res.status, "healthy")
# Missing function
res_missing = check_api_consistency(math, {"nonexistent_function": []})
self.assertEqual(res_missing.status, "unavailable")
# Missing parameter in documentation vs actual
class DummyMod:
def greet(self, name):
pass
res_deg = check_api_consistency(DummyMod(), {"greet": ["name", "punctuation"]})
self.assertEqual(res_deg.status, "degraded")
def test_g07_guidance_delivery_check(self):
receipt_ok = {
"guidance_status": "delivered_intact",
"included_spans": [
{"source_id": "pref:commit_summary", "intact": True}
],
"omitted_items": []
}
res = check_guidance_delivery(receipt_ok, ["pref:commit_summary"])
self.assertEqual(res.status, "healthy")
# Required directive dropped / truncated
receipt_dropped = {
"guidance_status": "delivered_intact",
"included_spans": [],
"omitted_items": []
}
res_dropped = check_guidance_delivery(receipt_dropped, ["pref:commit_summary"])
self.assertEqual(res_dropped.status, "degraded")
self.assertIn("missing", res_dropped.evidence)
# Unresolved budget
receipt_budget = {
"guidance_status": {"kind": "unresolved_budget", "needed": 500},
"included_spans": [],
"omitted_items": []
}
res_budget = check_guidance_delivery(receipt_budget, ["pref:commit_summary"])
self.assertEqual(res_budget.status, "degraded")
def test_g22_resource_access_check(self):
with tempfile.TemporaryDirectory() as tempdir:
src = Path(tempdir) / "src"
draft = Path(tempdir) / "draft"
src.mkdir()
draft.mkdir()
res = check_resource_access(src, draft)
self.assertEqual(res.status, "healthy")
# Missing path
res_unavail = check_resource_access(src, Path(tempdir) / "nonexistent")
self.assertEqual(res_unavail.status, "unavailable")
def test_failure_identity_shared_fixture(self):
import json
fixture_path = Path(__file__).resolve().parents[3] / "tests/runtime-fixtures/failure-identity.json"
if not fixture_path.exists():
self.skipTest(f"fixture not found at {fixture_path}")
fixture = json.loads(fixture_path.read_text())
for case in fixture["test_cases"]:
actual = compute_action_identity(
case["operation"],
case["component_revision"],
case["signature"],
case["workspace"],
)
self.assertEqual(actual, case["expected_identity"], f"Mismatch for: {case['description']}")
for pair in fixture["collision_pairs"]:
left = compute_action_identity(**pair["left"])
right = compute_action_identity(**pair["right"])
if pair.get("must_differ", False):
self.assertNotEqual(left, right, f"Collision in pair: {pair['description']}")
if __name__ == "__main__":
unittest.main()