This repository has no description
Something went wrong. Try again.
6.5 kB · 163 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164"""Unit tests for capability checkups, observation processing, and G07/G09/G13/G16/G22."""from __future__ import annotations
import tempfileimport timeimport unittestfrom pathlib import Path
from klbr import checksfrom 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()