from __future__ import annotations import importlib.util import json import subprocess import sys import tempfile import unittest from pathlib import Path from typing import cast ROOT = Path(__file__).resolve().parents[1] MODULE_PATH = ROOT / "scripts" / "kanban-hygiene-advisory.py" def load_module(): spec = importlib.util.spec_from_file_location("kanban_hygiene_advisory", MODULE_PATH) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def task(task_id: str, title: str, status: str = "ready", **extra): row = { "id": task_id, "title": title, "status": status, "assignee": "engineer", "created_at": 1_780_000_000, "updated_at": 1_780_000_100, } row.update(extra) return row class KanbanHygieneAdvisoryTests(unittest.TestCase): def setUp(self): self.mod = load_module() def advisory(self, tasks, now=1_780_003_600): return self.mod.advisory( tasks, board="npu-maximization", now=now, input_metadata={}, include_evidence=False, ) def test_output_contract_and_authority_flags_are_all_false(self): output = self.advisory([ task("t_spec", "spec: Kanban/task hygiene classifier", body_excerpt="Define dry-run labels and next gate.") ]) self.assertEqual(output["schema"], "kanban_hygiene_advisory_v1") self.assertTrue(output["dry_run"]) self.assertEqual(output["counts"]["tasks"], 1) self.assertTrue(output["npu_proof"]["required_for_npu_claims"]) self.assertFalse(output["npu_proof"]["attempted"]) self.assertTrue(output["authority"]) self.assertTrue(all(value is False for value in output["authority"].values())) def test_required_labels_and_kanban_lane_gate(self): output = self.advisory([ task("t1", "spec: Kanban/task hygiene classifier", body_excerpt="Read board summaries and suggest review-needed next gate labels.") ]) item = output["items"][0] for key in ["task_type", "project", "lane", "blocker", "staleness", "duplicate", "review_needed", "next_gate"]: self.assertIn(key, item) self.assertEqual(item["task_type"]["value"], "spec") self.assertEqual(item["project"]["value"], "npu-maximization") self.assertEqual(item["lane"]["value"], "kanban_hygiene") self.assertEqual(item["next_gate"]["value"], "ready_for_implementation") def test_lifecycle_chain_is_not_duplicate_even_with_same_normalized_title(self): rows = [ task("t_spec", "spec: Kanban hygiene advisory", children=["t_impl"]), task("t_impl", "implement: Kanban hygiene advisory", parents=["t_spec"], children=["t_review"]), task("t_review", "review: Kanban hygiene advisory", parents=["t_impl"]), ] output = self.advisory(rows) self.assertEqual(output["counts"]["duplicates"], 0) self.assertTrue(all(not item["duplicate"]["is_duplicate"] for item in output["items"])) def test_duplicate_same_type_lane_and_normalized_title_is_flagged(self): rows = [ task("t_a", "implement: dry-run Kanban hygiene advisory", body_excerpt="Kanban board summaries"), task("t_b", "implement: dry run kanban hygiene advisory", body_excerpt="Kanban board summaries"), ] output = self.advisory(rows) self.assertEqual(output["counts"]["duplicates"], 1) dupes = [item for item in output["items"] if item["duplicate"]["is_duplicate"]] self.assertEqual(len(dupes), 1) self.assertEqual(dupes[0]["next_gate"]["value"], "dedupe_review") def test_staleness_is_deterministic_with_now(self): output = self.advisory([ task("t_run", "implement: NPU service", status="running", updated_at=1_780_000_000, heartbeat_at=1_780_000_000) ], now=1_780_007_201) item = output["items"][0] self.assertEqual(item["staleness"]["value"], "stale_lock") self.assertEqual(item["next_gate"]["value"], "investigate_stale_lock") self.assertEqual(output["counts"]["stale"], 1) def test_review_required_marker_sets_ready_for_review(self): output = self.advisory([ task( "t_impl", "implement: dry-run Kanban hygiene advisory", status="blocked", body_excerpt="review-required: code change needs review", changed_files=["scripts/kanban-hygiene-advisory.py"], tests_run=8, ) ]) item = output["items"][0] self.assertTrue(item["review_needed"]["value"]) self.assertEqual(item["review_needed"]["kind"], "code_change") self.assertEqual(item["next_gate"]["value"], "ready_for_review") def test_missing_parent_waits_without_marking_blocked(self): output = self.advisory([ task("t_child", "implement: context gate", status="todo", parents=["t_parent"], body_excerpt="RAG context gate") ]) item = output["items"][0] self.assertEqual(item["blocker"]["value"], "missing_parent") self.assertFalse(item["blocker"]["blocked"]) self.assertEqual(item["next_gate"]["value"], "wait_for_parents") def test_npu_claim_without_busy_delta_routes_to_proof_gate(self): for excerpt in [ "NPU classifier returned HTTP 200 but missing busy delta evidence", "NPU reranker reported npu_busy_delta_us=0", "NPU reranker reported npu_busy_delta_us=-5", "NPU reranker reported npu_busy_delta_us=-0.1", ]: with self.subTest(excerpt=excerpt): output = self.advisory([task("t_npu", "test: NPU classifier smoke", body_excerpt=excerpt)]) item = output["items"][0] self.assertTrue(item["review_needed"]["value"]) self.assertEqual(item["review_needed"]["kind"], "npu_proof_gate") self.assertEqual(item["next_gate"]["value"], "needs_npu_proof") def test_npu_proof_gate_dominates_review_required_marker(self): for excerpt in [ "review-required: NPU reranker reported npu_busy_delta_us=0 after smoke", "review-required: NPU classifier returned HTTP 200 but missing busy delta evidence", ]: with self.subTest(excerpt=excerpt): output = self.advisory([ task( "t_npu_review", "implement: NPU classifier smoke", status="blocked", body_excerpt=excerpt, changed_files=["scripts/npu-classifier.py"], tests_run=1, ) ]) item = output["items"][0] self.assertTrue(item["review_needed"]["value"]) self.assertEqual(item["review_needed"]["kind"], "npu_proof_gate") self.assertEqual(item["next_gate"]["value"], "needs_npu_proof") def test_cli_accepts_jsonl_auto_format_and_invalid_schema_exits_nonzero(self): good_rows = [ json.dumps(task("t1", "docs: service map update", body_excerpt="runbook README")), json.dumps(task("t2", "ops: utilization digest", body_excerpt="health metrics digest")), ] with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as handle: handle.write("\n".join(good_rows)) good_path = handle.name try: result = subprocess.run( [sys.executable, str(MODULE_PATH), "--input", good_path, "--board", "npu-maximization", "--now", "1780003600"], capture_output=True, text=True, check=False, ) finally: Path(good_path).unlink(missing_ok=True) self.assertEqual(result.returncode, 0, result.stderr) parsed = json.loads(result.stdout) self.assertEqual(parsed["counts"]["tasks"], 2) bad = subprocess.run( [sys.executable, str(MODULE_PATH)], input=json.dumps({"tasks": [{"id": "missing-fields"}]}), capture_output=True, text=True, check=False, ) self.assertNotEqual(bad.returncode, 0) self.assertIn("missing required fields", bad.stderr) if __name__ == "__main__": unittest.main()