import hashlib import json import subprocess import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] VALIDATOR = ROOT / "scripts" / "validate-capability-baseline.py" VERSION = "2026.08.08.1" def dump(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() class CapabilityBaselineValidatorTests(unittest.TestCase): def make_root(self) -> tuple[tempfile.TemporaryDirectory[str], Path]: temporary = tempfile.TemporaryDirectory() root = Path(temporary.name) inventory = { "schema_version": 1, "product": "crank-community", "flows": [ { "id": "api-operation-test-run", "type": "api", "requirements": ["FR-2", "FR-46"], "user_outcome": "An administrator can execute an Operation test run.", "owner": "runtime-community", "status": "implemented", "capabilities": ["tools"], "evidence": {"automated": ["docs/capability-baseline/results.json"], "manual": ["docs/capability-baseline/manual-checklist.md"]}, } ], } required = { "baseline_version": VERSION, "required_flow_ids": ["api-operation-test-run"], "surface_groups": [{"id": "operations", "flow_ids": ["api-operation-test-run"]}], } taxonomy = { "baseline_version": VERSION, "implementation_statuses": ["implemented", "planned", "gap", "blocked"], "execution_verdicts": ["pass", "fail", "blocked", "skipped", "flaky", "not_run"], "evidence_modes": ["automated", "manual_only"], "full_pass": {"implementation_status": "implemented", "execution_verdict": "pass", "evidence_mode": "automated"}, } checklist = f"# Community UI baseline checklist\n\nbaseline_version: {VERSION}\n\n## UI-01 Operations\n\n- flow_id: api-operation-test-run\n- states: happy, error, recovery\n- verdict: not_run\n- reason: API-only fixture\n" results = { "baseline_version": VERSION, "source_revision": "0123456789abcdef0123456789abcdef01234567", "environment_class": "community-test", "runs": [{ "id": "run-api-operation-test", "command_id": "python-tooling-tests", "flow_ids": ["api-operation-test-run"], "execution_verdict": "pass", "evidence_mode": "automated", "accepted": True, "source_report_sha256": "a" * 64, "collector": "capability-baseline-collector-v1", "source_revision": "0123456789abcdef0123456789abcdef01234567", "environment_class": "community-test", }], "manual_results": [{ "check_id": "UI-01", "evidence_mode": "manual_only", "execution_verdict": "not_run", "flow_ids": ["api-operation-test-run"], "next_evidence": "Automate this fixture state.", }], "defects": [], } paths = { "inventory": root / "docs/capability-inventory.json", "required_surfaces": root / "docs/capability-baseline/required-surfaces.json", "taxonomy": root / "docs/capability-baseline/outcome-taxonomy.json", "checklist": root / "docs/capability-baseline/manual-checklist.md", "results": root / "docs/capability-baseline/results.json", } dump(paths["inventory"], inventory) dump(paths["required_surfaces"], required) dump(paths["taxonomy"], taxonomy) paths["checklist"].write_text(checklist, encoding="utf-8") dump(paths["results"], results) schema = { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://crank.local/schemas/capability-baseline.schema.json", "title": "Crank Community Capability Baseline Manifest", "type": "object", "additionalProperties": False, "required": ["schema_version", "baseline_version", "artifacts"], "properties": { "schema_version": {"const": 1}, "baseline_version": {"type": "string", "pattern": "^[0-9]{4}\\.[0-9]{2}\\.[0-9]{2}\\.[1-9][0-9]*$"}, "artifacts": {"type": "array", "minItems": 5, "maxItems": 5}, }, "$defs": { "artifact": {"type": "object", "additionalProperties": False}, "taxonomy": { "type": "object", "additionalProperties": False, "properties": {"full_pass": {"properties": {"implementation_status": {}, "execution_verdict": {}, "evidence_mode": {}}}}, }, "run": { "type": "object", "additionalProperties": False, "required": ["id", "command_id", "flow_ids", "execution_verdict", "evidence_mode", "accepted", "source_report_sha256", "source_revision", "environment_class", "collector"], }, "defect": { "type": "object", "additionalProperties": False, "required": ["id", "severity", "steps", "contract", "owner", "flow_ids", "next_action"], }, "manual_result": {"type": "object", "additionalProperties": False}, "results": {"type": "object", "additionalProperties": False}, }, } dump(root / "docs/schemas/capability-baseline.schema.json", schema) manifest = { "schema_version": 1, "baseline_version": VERSION, "artifacts": [ {"kind": kind, "path": str(path.relative_to(root)), "sha256": sha256(path)} for kind, path in paths.items() ], } dump(root / "docs/capability-baseline/manifest.json", manifest) subprocess.run(["git", "init", "-q", str(root)], check=True) subprocess.run( ["git", "-C", str(root), "add", *[str(path.relative_to(root)) for path in paths.values()]], check=True, ) return temporary, root def run_validator(self, root: Path) -> subprocess.CompletedProcess[str]: return subprocess.run( ["python3", str(VALIDATOR), "--root", str(root), "--manifest", "docs/capability-baseline/manifest.json", "--schema", "docs/schemas/capability-baseline.schema.json"], text=True, capture_output=True, check=False, ) def mutate_results(self, root: Path, mutate) -> None: path = root / "docs/capability-baseline/results.json" value = json.loads(path.read_text(encoding="utf-8")) mutate(value) dump(path, value) manifest_path = root / "docs/capability-baseline/manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) next(item for item in manifest["artifacts"] if item["kind"] == "results")["sha256"] = sha256(path) dump(manifest_path, manifest) def mutate_artifact(self, root: Path, kind: str, mutate) -> None: manifest_path = root / "docs/capability-baseline/manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) item = next(entry for entry in manifest["artifacts"] if entry["kind"] == kind) path = root / item["path"] value = json.loads(path.read_text(encoding="utf-8")) mutate(value) dump(path, value) item["sha256"] = sha256(path) dump(manifest_path, manifest) def test_valid_snapshot_passes(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) result = self.run_validator(root) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("baseline_version=2026.08.08.1", result.stdout) def test_checksum_drift_fails_closed(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) with (root / "docs/capability-baseline/manual-checklist.md").open("a", encoding="utf-8") as file: file.write("\nchanged\n") result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("CHECKSUM_MISMATCH", result.stderr) def test_unknown_flow_and_non_pass_cannot_be_accepted(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) self.mutate_results(root, lambda value: value["runs"][0].update(flow_ids=["missing-flow"], execution_verdict="flaky")) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("UNKNOWN_FLOW_ID", result.stderr) self.assertIn("NON_PASS_RECORDED_AS_PASS", result.stderr) def test_severe_defect_requires_blocked_inventory_flow(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) self.mutate_results(root, lambda value: value["defects"].append({ "id": "DEF-001", "severity": "High", "flow_ids": ["api-operation-test-run"], "contract": "Admin test run", "owner": "runtime-community", "steps": ["Run bounded fixture"], "next_action": "Fix downstream", })) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("SEVERE_DEFECT_FLOW_NOT_BLOCKED", result.stderr) def test_version_mismatch_and_broken_artifact_path_are_safe(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) manifest_path = root / "docs/capability-baseline/manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["baseline_version"] = "2026.08.08.2" manifest["artifacts"][0]["path"] = "../outside-secret.txt" dump(manifest_path, manifest) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("BROKEN_ARTIFACT_LINK", result.stderr) self.assertNotIn(str(root), result.stderr) def test_unsupported_schema_applicators_fail_closed_at_any_depth(self) -> None: for mutation in ( lambda schema: schema.update({"not": {}}), lambda schema: schema["properties"]["artifacts"].update({"allOf": []}), ): with self.subTest(mutation=mutation): temporary, root = self.make_root() self.addCleanup(temporary.cleanup) schema_path = root / "docs/schemas/capability-baseline.schema.json" schema = json.loads(schema_path.read_text(encoding="utf-8")) mutation(schema) dump(schema_path, schema) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("INVALID_SCHEMA_CONTRACT", result.stderr) def test_required_surfaces_cannot_omit_or_duplicate_implemented_flow(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) self.mutate_artifact(root, "required_surfaces", lambda value: value.update(required_flow_ids=[], surface_groups=[])) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("INVALID_REQUIRED_SURFACES", result.stderr) def test_manual_results_and_taxonomy_truth_table_are_enforced(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) self.mutate_results(root, lambda value: value["manual_results"][0].pop("next_evidence")) self.mutate_artifact(root, "taxonomy", lambda value: value["full_pass"].update(implementation_status="planned")) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("INVALID_MANUAL_EVIDENCE", result.stderr) self.assertIn("/taxonomy/full_pass", result.stderr) def test_accepted_run_requires_known_command_matching_provenance_and_implemented_flow(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) self.mutate_results(root, lambda value: value["runs"][0].update(command_id="fabricated", source_revision="f" * 40, flow_ids=[{}])) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("UNKNOWN_COMMAND", result.stderr) self.assertIn("PROVENANCE_MISMATCH", result.stderr) self.assertNotIn("Traceback", result.stderr) def test_planned_flow_cannot_receive_accepted_evidence(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) self.mutate_artifact(root, "inventory", lambda value: value["flows"].append({ **value["flows"][0], "id": "planned-resource-read", "status": "planned", })) self.mutate_results(root, lambda value: value["runs"][0].update(flow_ids=["planned-resource-read"])) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("NON_PASS_RECORDED_AS_PASS", result.stderr) def test_untracked_canonical_artifact_fails_closed(self) -> None: temporary, root = self.make_root() self.addCleanup(temporary.cleanup) subprocess.run(["git", "-C", str(root), "rm", "--cached", "docs/capability-baseline/results.json"], check=True, capture_output=True) result = self.run_validator(root) self.assertNotEqual(result.returncode, 0) self.assertIn("UNTRACKED_ARTIFACT", result.stderr) if __name__ == "__main__": unittest.main()