import importlib.util import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[2] CHECKER = ROOT / "scripts" / "check-community-scope.py" def load_checker_module(): spec = importlib.util.spec_from_file_location("community_scope_checker", CHECKER) if spec is None or spec.loader is None: raise RuntimeError("cannot load Community scope checker") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module class CommunityScopeCheckTests(unittest.TestCase): def run_checker(self, root: Path, files: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, str(CHECKER), "--root", str(root), "--files", *files], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) def run_default_checker(self, root: Path) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, str(CHECKER), "--root", str(root)], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) def test_passes_for_clean_community_content(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "README.md").write_text( "Crank publishes REST API endpoints as MCP tools.\n", encoding="utf-8", ) result = self.run_checker(root, ["README.md"]) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("Community scope check passed", result.stdout) def test_fails_for_forbidden_product_marker(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "apps").mkdir() (root / "apps" / "ui.md").write_text( "Hidden GraphQL protocol card must not be here.\n", encoding="utf-8", ) result = self.run_checker(root, ["apps/ui.md"]) self.assertNotEqual(result.returncode, 0) self.assertIn("apps/ui.md:1", result.stderr) self.assertIn("graphql", result.stderr.lower()) def test_inline_directive_allows_only_declared_marker(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "protocol.rs").write_text( 'assert_rejected_protocol("grpc"); ' "// community-scope: allow=grpc\n" 'assert_rejected_protocol("grpc");\n' 'assert_rejected_protocol("graphql"); ' "// community-scope: allow=grpc\n", encoding="utf-8", ) result = self.run_checker(root, ["protocol.rs"]) self.assertNotEqual(result.returncode, 0) self.assertIn("protocol.rs:2: forbidden marker `grpc`", result.stderr) self.assertIn("protocol.rs:3: forbidden marker `graphql`", result.stderr) self.assertNotIn("protocol.rs:1:", result.stderr) def test_rejects_marker_without_inline_directive(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "README.md").write_text( "Enable grpc transport.\n", encoding="utf-8", ) result = self.run_checker(root, ["README.md"]) self.assertNotEqual(result.returncode, 0) self.assertIn("forbidden marker `grpc`", result.stderr) def test_ignores_default_excluded_paths(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "LICENSE").write_text( "The license text may mention commercial distribution terms.\n", encoding="utf-8", ) result = self.run_checker(root, ["LICENSE"]) self.assertEqual(result.returncode, 0, result.stderr) def test_excluded_explicit_paths_are_validated_before_skip(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "target").mkdir() (root / "target" / "generated.txt").write_text( "generated output\n", encoding="utf-8" ) valid_excluded = self.run_checker(root, ["target/generated.txt"]) self.assertEqual(valid_excluded.returncode, 0, valid_excluded.stderr) for raw_path in ["LICENSE", "target/missing.txt", "target/../missing.txt"]: with self.subTest(raw_path=raw_path): rejected = self.run_checker(root, [raw_path]) self.assertNotEqual(rejected.returncode, 0) self.assertIn("invalid explicit file path", rejected.stderr) def test_default_discovery_ignores_tracked_deletions(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) subprocess.run(["git", "init", "-q"], cwd=root, check=True) deleted = root / "deleted.txt" deleted.write_text("clean\n", encoding="utf-8") subprocess.run(["git", "add", "deleted.txt"], cwd=root, check=True) deleted.unlink() result = self.run_default_checker(root) self.assertEqual(result.returncode, 0, result.stderr) def test_skips_binary_files(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "logo.png").write_bytes(b"\x89PNG\x00enterprise\x00") result = self.run_checker(root, ["logo.png"]) self.assertEqual(result.returncode, 0, result.stderr) def test_allows_planned_community_capability_identifiers(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "inventory.txt").write_text( "resources prompts tasks load_runs\n", encoding="utf-8", ) result = self.run_checker(root, ["inventory.txt"]) self.assertEqual(result.returncode, 0, result.stderr) def test_rejects_forbidden_identifier_after_narrow_inline_allow(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "scope.txt").write_text( "enterprise_rbac enforcement fixture " "# community-scope: allow=enterprise\n" "enterprise_rbac product capability\n", encoding="utf-8", ) result = self.run_checker(root, ["scope.txt"]) self.assertNotEqual(result.returncode, 0) self.assertNotIn("scope.txt:1:", result.stderr) self.assertIn("scope.txt:2: forbidden marker `enterprise`", result.stderr) def test_explicit_paths_fail_closed_without_disclosing_host_paths(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) outside = root.parent / "SECRET_CANARY_scope-outside.txt" outside.write_text("clean\n", encoding="utf-8") self.addCleanup(outside.unlink, missing_ok=True) internal_target = root / "target.txt" internal_target.write_text("clean\n", encoding="utf-8") internal_link = root / "internal-link.txt" internal_link.symlink_to(internal_target) outside_link = root / "outside-link.txt" outside_link.symlink_to(outside) cases = [ "missing.txt", f"../{outside.name}", str(outside), "internal-link.txt", "outside-link.txt", ] for raw_path in cases: with self.subTest(raw_path=raw_path): result = self.run_checker(root, [raw_path]) self.assertNotEqual(result.returncode, 0) self.assertIn("invalid explicit file path", result.stderr) self.assertNotIn("SECRET_CANARY", result.stderr) def test_invalid_path_report_is_bounded_and_deterministically_truncated(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) missing = [f"missing-{index:04d}.txt" for index in range(1_200, 0, -1)] first = self.run_checker(root, missing) second = self.run_checker(root, missing) self.assertNotEqual(first.returncode, 0) self.assertEqual(first.stderr, second.stderr) self.assertLessEqual(len(first.stderr.encode("utf-8")), 64 * 1024) self.assertIn("error: findings truncated", first.stderr) self.assertNotIn("missing-", first.stderr) def test_single_file_finding_limit_emits_truncation_marker(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "scope.txt").write_text( "grpc\n" * 1_001, encoding="utf-8", ) result = self.run_checker(root, ["scope.txt"]) self.assertNotEqual(result.returncode, 0) self.assertIn("error: findings truncated", result.stderr) self.assertLessEqual(len(result.stderr.encode("utf-8")), 64 * 1024) def test_exact_finding_limit_is_not_reported_as_truncated(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "scope.txt").write_text("grpc\n" * 1_000, encoding="utf-8") result = self.run_checker(root, ["scope.txt"]) self.assertNotEqual(result.returncode, 0) self.assertNotIn("error: findings truncated", result.stderr) def test_rejects_camel_and_pascal_case_forbidden_identifiers(self) -> None: cases = { "EnterpriseRbac": "enterprise", "MultiWorkspace": "multi-workspace", "NonRestUpstream": "non-rest-upstream", "ArbitraryDistributedLoadTargets": "distributed-load-targets", } with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) for index, (identifier, marker) in enumerate(cases.items()): path = root / f"scope-{index}.txt" path.write_text(identifier + "\n", encoding="utf-8") with self.subTest(identifier=identifier): result = self.run_checker(root, [path.name]) self.assertNotEqual(result.returncode, 0) self.assertIn(f"forbidden marker `{marker}`", result.stderr) def test_invalid_utf8_text_fails_closed(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "scope.txt").write_bytes(b"enterpr\xffise capability\n") result = self.run_checker(root, ["scope.txt"]) self.assertNotEqual(result.returncode, 0) self.assertIn("invalid explicit file path", result.stderr) self.assertNotIn("Traceback", result.stderr) def test_control_characters_in_explicit_paths_cannot_forge_report_lines(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) forged_path = "forged\nerror: injected.txt" (root / forged_path).write_text("grpc\n", encoding="utf-8") result = self.run_checker(root, [forged_path]) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stderr.count("\n"), 2) self.assertNotIn("forged", result.stderr) self.assertNotIn("injected", result.stderr) def test_oversized_single_line_fails_closed_without_materializing_it(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "scope.txt").write_text("x" * 65_537, encoding="utf-8") result = self.run_checker(root, ["scope.txt"]) self.assertNotEqual(result.returncode, 0) self.assertIn("invalid explicit file path", result.stderr) self.assertNotIn("Traceback", result.stderr) def test_open_failure_after_validation_returns_a_safe_failure(self) -> None: checker = load_checker_module() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "scope.txt").write_text("clean\n", encoding="utf-8") with mock.patch.object(Path, "open", side_effect=OSError("host path canary")): findings, valid, truncated, scanned_bytes = checker.scan_file( root, "scope.txt" ) self.assertEqual(findings, []) self.assertFalse(valid) self.assertFalse(truncated) self.assertEqual(scanned_bytes, 0) if __name__ == "__main__": unittest.main()