import copy import importlib.util import json import re import shutil import subprocess import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] VALIDATOR = ROOT / "scripts" / "validate-capability-inventory.py" CANONICAL_INVENTORY = ROOT / "docs" / "capability-inventory.json" CANONICAL_SCHEMA = ROOT / "docs" / "schemas" / "capability-inventory.schema.json" COMMUNITY_CHECKER = ROOT / "scripts" / "check-community-scope.py" COMMUNITY_README = ROOT / "README.md" DOCS_INDEX = ROOT / "docs" / "README.md" COMMUNITY_INTRO = ROOT / "docs" / "intro.md" COMMUNITY_README_EN = ROOT / "docs" / "en" / "README.md" JUSTFILE = ROOT / "justfile" CI_WORKFLOW = ROOT / ".gitea" / "workflows" / "ci.yml" RELEASE_WORKFLOW = ROOT / ".gitea" / "workflows" / "release.yml" REQUIRED_SURFACES = ROOT / "docs" / "capability-baseline" / "required-surfaces.json" def load_validator_module(): spec = importlib.util.spec_from_file_location("capability_inventory_validator", VALIDATOR) if spec is None or spec.loader is None: raise RuntimeError("cannot load capability inventory validator") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module class CapabilityInventoryValidatorTests(unittest.TestCase): def setUp(self) -> None: self._temporary_directory = tempfile.TemporaryDirectory() self.root = Path(self._temporary_directory.name) self.inventory_path = self.root / "docs" / "capability-inventory.json" self.schema_path = ( self.root / "docs" / "schemas" / "capability-inventory.schema.json" ) self.inventory_path.parent.mkdir(parents=True) self.schema_path.parent.mkdir(parents=True) def tearDown(self) -> None: self._temporary_directory.cleanup() def write_evidence(self, relative_path: str, content: str = "verified\n") -> str: path = self.root / relative_path path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") return relative_path def valid_flow( self, flow_id: str, flow_type: str, requirement: str, status: str = "implemented", capabilities: list[str] | None = None, ) -> dict[str, object]: automated = self.write_evidence( f"evidence/automated/{flow_id}.txt", f"{flow_id}: automated acceptance evidence\n", ) manual = self.write_evidence( f"evidence/manual/{flow_id}.md", f"# {flow_id}\n\nManual verification complete.\n", ) return { "id": flow_id, "type": flow_type, "requirements": [requirement], "user_outcome": f"User completes {flow_id.replace('-', ' ')}.", "owner": "community-release", "status": status, "capabilities": capabilities or ["tools"], "evidence": { "automated": [automated], "manual": [manual], }, } def write_inventory(self, flows: list[dict[str, object]]) -> None: payload = { "schema_version": 1, "product": "crank-community", "flows": flows, } self.inventory_path.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) def install_canonical_schema(self) -> None: self.assertTrue( CANONICAL_SCHEMA.is_file(), f"Story 1.1 must create {CANONICAL_SCHEMA.relative_to(ROOT)}", ) shutil.copyfile(CANONICAL_SCHEMA, self.schema_path) def run_validator( self, required_frs: list[str], install_schema: bool = True, ) -> subprocess.CompletedProcess[str]: if install_schema: self.install_canonical_schema() command = [ sys.executable, str(VALIDATOR), "--root", str(self.root), "--inventory", str(self.inventory_path.relative_to(self.root)), "--schema", str(self.schema_path.relative_to(self.root)), ] for requirement in required_frs: command.extend(["--required-fr", requirement]) return subprocess.run( command, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) def test_accepts_complete_inventory_with_all_flow_types_and_statuses(self) -> None: flows = [ self.valid_flow("ui-create-method", "ui", "FR-1", "implemented"), self.valid_flow("api-create-method", "api", "FR-2", "planned"), self.valid_flow("mcp-tool-call", "mcp", "FR-3", "gap"), self.valid_flow("mcp-resource-read", "mcp", "FR-4", "blocked"), ] self.write_inventory(flows) result = self.run_validator(["FR-1", "FR-2", "FR-3", "FR-4"]) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("Capability inventory validation passed", result.stdout) self.assertRegex(result.stdout, r"\btotal=4\b") self.assertRegex(result.stdout, r"\bimplemented=1\b") self.assertRegex(result.stdout, r"\bplanned=1\b") self.assertRegex(result.stdout, r"\bgap=1\b") self.assertRegex(result.stdout, r"\bblocked=1\b") def test_counts_only_implemented_flows_as_pass(self) -> None: flows = [ self.valid_flow("implemented-flow", "api", "FR-1", "implemented"), self.valid_flow("planned-flow", "ui", "FR-2", "planned"), self.valid_flow("gap-flow", "mcp", "FR-3", "gap"), self.valid_flow("blocked-flow", "mcp", "FR-4", "blocked"), ] self.write_inventory(flows) result = self.run_validator(["FR-1", "FR-2", "FR-3", "FR-4"]) self.assertEqual(result.returncode, 0, result.stderr) self.assertRegex(result.stdout, r"\bpass=1\b") self.assertNotRegex(result.stdout, r"\bpass=4\b") def test_rejects_duplicate_id_unknown_enum_and_missing_required_fields(self) -> None: base = self.valid_flow("flow-a", "api", "FR-1") cases: list[tuple[str, list[dict[str, object]], str]] = [] duplicate = copy.deepcopy(base) duplicate["user_outcome"] = "A distinct outcome with a duplicate identifier." cases.append(("duplicate ID", [base, duplicate], "DUPLICATE_FLOW_ID")) unknown_status = copy.deepcopy(base) unknown_status["status"] = "done" cases.append(("unknown status", [unknown_status], "UNKNOWN_STATUS")) unknown_type = copy.deepcopy(base) unknown_type["type"] = "worker" cases.append(("unknown type", [unknown_type], "UNKNOWN_FLOW_TYPE")) no_owner = copy.deepcopy(base) no_owner.pop("owner") cases.append(("missing owner", [no_owner], "MISSING_OWNER")) no_outcome = copy.deepcopy(base) no_outcome.pop("user_outcome") cases.append(("missing required field", [no_outcome], "MISSING_REQUIRED_FIELD")) for label, flows, expected_code in cases: with self.subTest(label=label): self.write_inventory(flows) result = self.run_validator(["FR-1"]) self.assertNotEqual(result.returncode, 0) self.assertIn(f"error[{expected_code}]", result.stderr) def test_rejects_missing_mandatory_requirement(self) -> None: self.write_inventory([self.valid_flow("only-fr-one", "api", "FR-1")]) result = self.run_validator(["FR-1", "FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn("error[MISSING_REQUIRED_FR]", result.stderr) self.assertIn("FR-46", result.stderr) def test_rejects_missing_non_regular_and_escaping_evidence_links(self) -> None: base = self.valid_flow("evidence-flow", "mcp", "FR-46") outside = self.root.parent / f"{self.root.name}-outside.txt" outside.write_text("outside\n", encoding="utf-8") self.addCleanup(outside.unlink, missing_ok=True) internal_target = self.write_evidence("evidence/automated/internal-target.txt") internal_link = self.root / "evidence" / "automated" / "internal-link.txt" internal_link.symlink_to(self.root / internal_target) outside_link = self.root / "evidence" / "automated" / "outside-link.txt" outside_link.symlink_to(outside) broken_cases = [ "evidence/automated/does-not-exist.txt", "evidence/automated", f"../{outside.name}", str(outside), "evidence/automated/internal-link.txt", "evidence/automated/outside-link.txt", ] for evidence_path in broken_cases: with self.subTest(evidence_path=evidence_path): flow = copy.deepcopy(base) flow["evidence"] = { "automated": [evidence_path], "manual": ["evidence/manual/evidence-flow.md"], } self.write_inventory([flow]) result = self.run_validator(["FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn("error[BROKEN_EVIDENCE_LINK]", result.stderr) def test_report_is_deterministic_bounded_sorted_and_redacted(self) -> None: secret_canary = "SECRET_CANARY_sk-live-4hQ9Vw8x" flows = [] for index in range(1_200, 0, -1): flow = self.valid_flow( f"invalid-flow-{index:04d}", "api", f"FR-{index}", ) flow["status"] = "unknown" flow["user_outcome"] = f"{secret_canary}:{index}" flows.append(flow) self.write_inventory(flows) raw_inventory = self.inventory_path.read_text(encoding="utf-8") first = self.run_validator(["FR-46"]) second = self.run_validator(["FR-46"]) self.assertNotEqual(first.returncode, 0) self.assertEqual(first.stderr, second.stderr) self.assertLessEqual(len(first.stderr.encode("utf-8")), 64 * 1024) self.assertNotIn(secret_canary, first.stderr) self.assertNotIn(raw_inventory, first.stderr) diagnostics = re.findall(r"error\[([^]]+)\] ([^:]+):", first.stderr) self.assertTrue(diagnostics) self.assertEqual(diagnostics[:-1], sorted(diagnostics[:-1])) self.assertEqual(diagnostics[-1][0], "TRUNCATED") self.assertNotRegex(first.stderr, r"invalid-flow-\d{4}") def test_rejects_malformed_documents_and_schema_contract_drift(self) -> None: self.write_inventory([self.valid_flow("valid-flow", "api", "FR-46")]) self.inventory_path.write_text("{not-json", encoding="utf-8") malformed_inventory = self.run_validator(["FR-46"]) self.assertNotEqual(malformed_inventory.returncode, 0) self.assertIn("error[INVALID_JSON]", malformed_inventory.stderr) self.write_inventory([self.valid_flow("valid-flow", "api", "FR-46")]) self.schema_path.write_text("{not-json", encoding="utf-8") malformed_schema = self.run_validator(["FR-46"], install_schema=False) self.assertNotEqual(malformed_schema.returncode, 0) self.assertIn("error[INVALID_SCHEMA]", malformed_schema.stderr) self.install_canonical_schema() schema = json.loads(self.schema_path.read_text(encoding="utf-8")) schema["$schema"] = "https://json-schema.org/draft/2019-09/schema" self.schema_path.write_text(json.dumps(schema), encoding="utf-8") drifted_schema = self.run_validator(["FR-46"], install_schema=False) self.assertNotEqual(drifted_schema.returncode, 0) self.assertIn("error[INVALID_SCHEMA_CONTRACT]", drifted_schema.stderr) def test_rejects_duplicate_nonstandard_recursive_and_large_integer_json(self) -> None: malformed_documents = { "duplicate member": ( '{"schema_version":1,"schema_version":1,' '"product":"crank-community","flows":[]}', "INVALID_JSON", ), "non-standard number": ( '{"schema_version":1,"product":NaN,"flows":[]}', "INVALID_JSON", ), "recursive": ("[" * 100_000 + "0" + "]" * 100_000, "INVALID_JSON"), "large integer": ("9" * 5_000, "INVALID_JSON"), } for label, (content, expected_code) in malformed_documents.items(): with self.subTest(label=label): self.inventory_path.write_text(content, encoding="utf-8") result = self.run_validator(["FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn(f"error[{expected_code}]", result.stderr) self.assertNotIn("Traceback", result.stderr) self.assertLessEqual(len(result.stderr.encode("utf-8")), 64 * 1024) def test_schema_parity_rejects_required_type_ref_and_pattern_drift(self) -> None: self.write_inventory([self.valid_flow("schema-parity", "api", "FR-46")]) mutations = { "root required missing": lambda schema: schema.pop("required"), "root contract inversion": lambda schema: schema.update({"not": {}}), "malformed evidence required": lambda schema: schema["$defs"]["evidence"].update( {"required": [{}]} ), "flow ref drift": lambda schema: schema["properties"]["flows"]["items"].update( {"$ref": "#/$defs/evidence"} ), "owner type drift": lambda schema: schema["$defs"]["flow"]["properties"][ "owner" ].update({"type": "integer"}), "owner const drift": lambda schema: schema["$defs"]["flow"]["properties"][ "owner" ].update({"const": "impossible"}), "nested applicator drift": lambda schema: schema["$defs"]["evidencePaths"][ "items" ].update({"anyOf": [{"type": "integer"}]}), "evidence pattern drift": lambda schema: schema["$defs"]["evidencePaths"][ "items" ].update({"pattern": ".+"}), } for label, mutate in mutations.items(): with self.subTest(label=label): self.install_canonical_schema() schema = json.loads(self.schema_path.read_text(encoding="utf-8")) mutate(schema) self.schema_path.write_text(json.dumps(schema), encoding="utf-8") result = self.run_validator(["FR-46"], install_schema=False) self.assertNotEqual(result.returncode, 0) self.assertIn("error[INVALID_SCHEMA_CONTRACT]", result.stderr) self.assertNotIn("Traceback", result.stderr) self.install_canonical_schema() schema = json.loads(self.schema_path.read_text(encoding="utf-8")) schema["$defs"]["flow"]["properties"]["status"]["enum"].reverse() self.schema_path.write_text(json.dumps(schema), encoding="utf-8") reordered_enum = self.run_validator(["FR-46"], install_schema=False) self.assertEqual(reordered_enum.returncode, 0, reordered_enum.stderr) def test_rejects_boolean_schema_version_and_whitespace_required_text(self) -> None: base = self.valid_flow("strict-scalars", "api", "FR-46") payload_cases: list[tuple[str, dict[str, object], str]] = [] boolean_version = { "schema_version": True, "product": "crank-community", "flows": [copy.deepcopy(base)], } payload_cases.append(("boolean inventory version", boolean_version, "MISSING_REQUIRED_FIELD")) whitespace_owner = copy.deepcopy(base) whitespace_owner["owner"] = " \t\n " payload_cases.append( ( "whitespace owner", {"schema_version": 1, "product": "crank-community", "flows": [whitespace_owner]}, "MISSING_OWNER", ) ) whitespace_outcome = copy.deepcopy(base) whitespace_outcome["user_outcome"] = " \t\n " payload_cases.append( ( "whitespace outcome", {"schema_version": 1, "product": "crank-community", "flows": [whitespace_outcome]}, "LIMIT_EXCEEDED", ) ) for label, payload, expected_code in payload_cases: with self.subTest(label=label): self.inventory_path.write_text(json.dumps(payload), encoding="utf-8") result = self.run_validator(["FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn(f"error[{expected_code}]", result.stderr) self.write_inventory([base]) self.install_canonical_schema() schema = json.loads(self.schema_path.read_text(encoding="utf-8")) schema["properties"]["schema_version"]["const"] = True self.schema_path.write_text(json.dumps(schema), encoding="utf-8") schema_result = self.run_validator(["FR-46"], install_schema=False) self.assertNotEqual(schema_result.returncode, 0) self.assertIn("error[INVALID_SCHEMA_CONTRACT]", schema_result.stderr) def test_diagnostic_accumulation_is_capped_counted_and_redacted(self) -> None: secret_canary = "SECRET_CANARY_unknown-field" flow = self.valid_flow("bounded-diagnostics", "api", "FR-46") for index in range(2_000): flow[f"{secret_canary}-{index:04d}"] = True self.write_inventory([flow]) result = self.run_validator(["FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn("error[TRUNCATED] /: diagnostics omitted=1000", result.stderr) self.assertEqual(result.stderr.count("error[UNKNOWN_FIELD]"), 1_000) self.assertNotIn(secret_canary, result.stderr) self.assertLessEqual(len(result.stderr.encode("utf-8")), 64 * 1024) def test_rejects_oversized_document_and_field_limits(self) -> None: self.inventory_path.write_bytes(b" " * (4_194_304 + 1)) oversized = self.run_validator(["FR-46"]) self.assertNotEqual(oversized.returncode, 0) self.assertIn("error[INPUT_TOO_LARGE]", oversized.stderr) flow = self.valid_flow("bounded-flow", "api", "FR-46") flow["user_outcome"] = "x" * 4_097 self.write_inventory([flow]) overlong = self.run_validator(["FR-46"]) self.assertNotEqual(overlong.returncode, 0) self.assertIn("error[LIMIT_EXCEEDED]", overlong.stderr) def test_named_string_limits_cover_below_at_and_above_boundaries(self) -> None: string_cases = [ ("id", 128, lambda length: "a" * length), ("owner", 256, lambda length: "o" * length), ("user_outcome", 4_096, lambda length: "u" * length), ("notes", 4_096, lambda length: "n" * length), ("requirements", 128, lambda length: "FR-" + "1" * (length - 3)), ] for field, limit, make_value in string_cases: for delta in (-1, 0, 1): with self.subTest(field=field, length=limit + delta): safe_field = field.replace("_", "-") flow = self.valid_flow(f"bounded-{safe_field}-{delta + 1}", "api", "FR-46") value = make_value(limit + delta) if field == "requirements": flow[field] = [value] required_frs = [value] else: flow[field] = value required_frs = ["FR-46"] self.write_inventory([flow]) result = self.run_validator(required_frs) if delta <= 0: self.assertEqual(result.returncode, 0, result.stderr) else: self.assertNotEqual(result.returncode, 0) self.assertIn("error[LIMIT_EXCEEDED]", result.stderr) def test_collection_limits_cover_below_at_and_above_boundaries(self) -> None: module = load_validator_module() automated = self.write_evidence("evidence/shared/automated.txt") manual = self.write_evidence("evidence/shared/manual.md") def direct_flow(index: int) -> dict[str, object]: return { "id": f"flow-{index:05d}", "type": "api", "requirements": ["FR-46"], "user_outcome": "A bounded flow succeeds.", "owner": "community-release", "status": "implemented", "capabilities": ["tools"], "evidence": {"automated": [automated], "manual": [manual]}, } all_flows = [direct_flow(index) for index in range(10_001)] for count in (9_999, 10_000, 10_001): with self.subTest(collection="flows", count=count): inventory = { "schema_version": 1, "product": "crank-community", "flows": all_flows[:count], } diagnostics, _ = module.validate_inventory(self.root, inventory, ["FR-46"]) has_limit_error = any( item.code == "LIMIT_EXCEEDED" and item.pointer == "/flows" for item in diagnostics.ordered() ) self.assertEqual(has_limit_error, count > 10_000) for count in (63, 64, 65): with self.subTest(collection="requirements", count=count): flow = self.valid_flow(f"requirements-{count}", "api", "FR-1") flow["requirements"] = [f"FR-{index}" for index in range(1, count + 1)] self.write_inventory([flow]) result = self.run_validator(["FR-1"]) self.assertEqual(result.returncode == 0, count <= 64, result.stderr) evidence_paths = [ self.write_evidence(f"evidence/bounds/path-{index:02d}.txt") for index in range(65) ] for count in (63, 64, 65): with self.subTest(collection="evidence", count=count): flow = self.valid_flow(f"evidence-{count}", "api", "FR-46") flow["evidence"] = { "automated": evidence_paths[:count], "manual": [manual], } self.write_inventory([flow]) result = self.run_validator(["FR-46"]) self.assertEqual(result.returncode == 0, count <= 64, result.stderr) for count in (15, 16, 17): with self.subTest(collection="capability-bound", count=count): diagnostics = module.validate_string_list( [f"capability-{index}" for index in range(count)], "/capabilities", 16, 128, ) has_limit_error = any(item.code == "LIMIT_EXCEEDED" for item in diagnostics) self.assertEqual(has_limit_error, count > 16) def test_evidence_path_limit_covers_below_at_and_above_boundary(self) -> None: def path_of_length(length: int) -> str: segment_count = (length + 199) // 201 character_count = length - (segment_count - 1) lengths: list[int] = [] remaining = character_count for index in range(segment_count): slots_left = segment_count - index - 1 current = min(200, remaining - slots_left) lengths.append(current) remaining -= current path = "/".join("p" * size for size in lengths) self.assertEqual(len(path), length) return path manual = self.write_evidence("evidence/manual/path-limit.md") for length in (1_023, 1_024, 1_025): with self.subTest(length=length): evidence_path = path_of_length(length) if length <= 1_024: self.write_evidence(evidence_path) flow = self.valid_flow(f"path-limit-{length}", "api", "FR-46") flow["evidence"] = { "automated": [evidence_path], "manual": [manual], } self.write_inventory([flow]) result = self.run_validator(["FR-46"]) self.assertEqual(result.returncode == 0, length <= 1_024, result.stderr) def test_semantic_validator_enforces_schema_shape_uniqueness_and_fr_format(self) -> None: cases: list[tuple[str, object, str]] = [] unknown_top = { "schema_version": 1, "product": "crank-community", "flows": [self.valid_flow("unknown-top", "api", "FR-46")], "unexpected": True, } cases.append(("unknown top-level field", unknown_top, "UNKNOWN_FIELD")) unknown_flow = self.valid_flow("unknown-flow", "api", "FR-46") unknown_flow["unexpected"] = True cases.append( ( "unknown flow field", {"schema_version": 1, "product": "crank-community", "flows": [unknown_flow]}, "UNKNOWN_FIELD", ) ) duplicate_requirement = self.valid_flow("duplicate-fr", "api", "FR-46") duplicate_requirement["requirements"] = ["FR-46", "FR-46"] cases.append( ( "duplicate requirement", { "schema_version": 1, "product": "crank-community", "flows": [duplicate_requirement], }, "DUPLICATE_LIST_ITEM", ) ) malformed_requirement = self.valid_flow("malformed-fr", "api", "FR-46") malformed_requirement["requirements"] = ["FR-0"] cases.append( ( "malformed requirement", { "schema_version": 1, "product": "crank-community", "flows": [malformed_requirement], }, "INVALID_FORMAT", ) ) unknown_evidence = self.valid_flow("unknown-evidence", "api", "FR-46") evidence = unknown_evidence["evidence"] self.assertIsInstance(evidence, dict) evidence["unexpected"] = [] cases.append( ( "unknown evidence field", { "schema_version": 1, "product": "crank-community", "flows": [unknown_evidence], }, "UNKNOWN_FIELD", ) ) for label, payload, expected_code in cases: with self.subTest(label=label): self.inventory_path.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) result = self.run_validator(["FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn(f"error[{expected_code}]", result.stderr) def test_accepts_planned_modern_mcp_and_load_run_capabilities(self) -> None: capabilities = ["resources", "prompts", "tasks", "load_runs"] flows = [ self.valid_flow( f"planned-{capability.replace('_', '-')}", "mcp" if capability != "load_runs" else "ui", f"FR-{index}", "planned", [capability], ) for index, capability in enumerate(capabilities, start=47) ] self.write_inventory(flows) result = self.run_validator([f"FR-{index}" for index in range(47, 51)]) self.assertEqual(result.returncode, 0, result.stderr) for capability in capabilities: self.assertNotIn(f"error[FORBIDDEN_COMMUNITY_CAPABILITY]: {capability}", result.stderr) def test_rejects_forbidden_community_capabilities_despite_allow_text(self) -> None: forbidden = [ "multi_workspace", # community-scope: allow=multi-workspace "enterprise_rbac", # community-scope: allow=enterprise "sso", # community-scope: allow=sso "non_rest_upstream", # community-scope: allow=non-rest-upstream "arbitrary_distributed_load_targets", # community-scope: allow=distributed-load-targets ] for index, capability in enumerate(forbidden, start=1): with self.subTest(capability=capability): flow = self.valid_flow( f"forbidden-{index}", "api", "FR-46", "planned", [capability], ) flow["notes"] = f"community-scope: allow={capability}" self.write_inventory([flow]) result = self.run_validator(["FR-46"]) self.assertNotEqual(result.returncode, 0) self.assertIn("error[FORBIDDEN_COMMUNITY_CAPABILITY]", result.stderr) self.assertIn(capability, result.stderr) def test_scope_scanner_receives_inventory_files_explicitly_and_cannot_be_bypassed(self) -> None: clean_inventory = self.valid_flow( "planned-resources", "mcp", "FR-47", "planned", ["resources"], ) self.write_inventory([clean_inventory]) explicit_files = [ str(self.inventory_path.relative_to(self.root)), "docs/new-untracked-scope-note.md", ] note = self.root / explicit_files[1] note.write_text("Community Resources remain planned.\n", encoding="utf-8") clean = subprocess.run( [ sys.executable, str(COMMUNITY_CHECKER), "--root", str(self.root), "--files", *explicit_files, ], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) self.assertEqual(clean.returncode, 0, clean.stderr) self.assertIn("2 files scanned", clean.stdout) note.write_text( "community-scope: allow=enterprise\n" "enterprise_rbac remains outside Community scope.\n", # community-scope: allow=enterprise encoding="utf-8", ) rejected = subprocess.run( [ sys.executable, str(COMMUNITY_CHECKER), "--root", str(self.root), "--files", *explicit_files, ], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) self.assertNotEqual(rejected.returncode, 0) self.assertIn("docs/new-untracked-scope-note.md:2", rejected.stderr) self.assertIn("forbidden marker", rejected.stderr) self.assertIn("enterprise", rejected.stderr) # community-scope: allow=enterprise def test_canonical_inventory_seed_does_not_promote_planned_work(self) -> None: for path in [ CANONICAL_INVENTORY, CANONICAL_SCHEMA, ]: self.assertTrue(path.is_file(), f"Missing canonical artifact: {path}") inventory = json.loads(CANONICAL_INVENTORY.read_text(encoding="utf-8")) schema = json.loads(CANONICAL_SCHEMA.read_text(encoding="utf-8")) flow_schema = schema["$defs"]["flow"] evidence_paths = schema["$defs"]["evidencePaths"] self.assertEqual( schema["$schema"], "https://json-schema.org/draft/2020-12/schema", ) self.assertEqual(schema["properties"]["flows"]["maxItems"], 10_000) self.assertEqual( set(flow_schema["properties"]["type"]["enum"]), {"ui", "api", "mcp"}, ) self.assertEqual( set(flow_schema["properties"]["status"]["enum"]), {"implemented", "planned", "gap", "blocked"}, ) self.assertEqual(evidence_paths["minItems"], 1) self.assertEqual(evidence_paths["maxItems"], 64) target_capabilities = {"resources", "prompts", "tasks", "load_runs"} matching_flows = [ flow for flow in inventory["flows"] if target_capabilities.intersection(flow.get("capabilities", [])) ] observed_capabilities = { capability for flow in matching_flows for capability in flow.get("capabilities", []) if capability in target_capabilities } self.assertEqual(observed_capabilities, target_capabilities) self.assertTrue(matching_flows) self.assertTrue( all(flow["status"] != "implemented" for flow in matching_flows), "Story 1.1 must not claim target capabilities are implemented.", ) def test_canonical_inventory_covers_all_requirements_and_required_surfaces(self) -> None: inventory = json.loads(CANONICAL_INVENTORY.read_text(encoding="utf-8")) required_surfaces = json.loads(REQUIRED_SURFACES.read_text(encoding="utf-8")) flows = {flow["id"]: flow for flow in inventory["flows"]} observed_requirements = { requirement for flow in flows.values() for requirement in flow["requirements"] } self.assertEqual( {f"FR-{number}" for number in range(1, 55)} - observed_requirements, set(), ) self.assertEqual( set(required_surfaces["required_flow_ids"]) - set(flows), set(), ) for flow_id in required_surfaces["required_flow_ids"]: flow = flows[flow_id] self.assertIn(flow["status"], {"implemented", "blocked"}) for evidence_kind in ("automated", "manual"): for evidence_path in flow["evidence"][evidence_kind]: resolved = ROOT / evidence_path self.assertTrue(resolved.is_file(), evidence_path) self.assertFalse(resolved.is_symlink(), evidence_path) def test_tracked_documentation_describes_planned_scope(self) -> None: for documentation in [ COMMUNITY_README, DOCS_INDEX, COMMUNITY_INTRO, COMMUNITY_README_EN, ]: text = documentation.read_text(encoding="utf-8") self.assertIn("docs/capability-inventory.json", text) self.assertRegex(text.lower(), r"\b(planned|gap|blocked)\b") def test_no_story_acceptance_scenario_remains_skipped(self) -> None: for name, method in vars(type(self)).items(): if not name.startswith("test_") or name == self._testMethodName: continue self.assertFalse( getattr(method, "__unittest_skip__", False), f"Story acceptance scenario remains skipped: {name}", ) def test_canonical_validator_is_a_ci_and_release_prerequisite(self) -> None: justfile = JUSTFILE.read_text(encoding="utf-8") ci = CI_WORKFLOW.read_text(encoding="utf-8") release = RELEASE_WORKFLOW.read_text(encoding="utf-8") command = "scripts/validate-capability-inventory.py" self.assertIn("capability-inventory-check:", justfile) self.assertIn("just capability-inventory-check", justfile) self.assertLess(ci.index("Run tooling unit tests"), ci.index(command)) self.assertLess(ci.index(command), ci.index("Check Community scope")) self.assertIn("python3 --version", release) self.assertLess(release.index("Run tooling unit tests"), release.index(command)) self.assertLess(release.index(command), release.index("Check Community scope")) self.assertLess(release.index("Check Community scope"), release.index("Build release binaries")) if __name__ == "__main__": unittest.main()