bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat: harden community production foundation through story 1.5

Browse Source
This commit is contained in:
bsodfather
2026-08-14 00:21:59 +03:00
parent c30461cc92
commit f6fc2e5c9b
161 changed files with 16758 additions and 2515 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_validate_capability_baseline.py
+283
View File
@@ -0,0 +1,283 @@
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()