"""Guards that commit-message rules have exactly one implementation. The skill validator owns every rule and reads `references/commit_policy.json`. `docs/common/commit_message.md` is human guidance; the CI entry point must delegate rather than re-derive rules from that document. """ import ast import json import os import re import subprocess import sys import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SKILL_ROOT = ROOT / "skills" / "commit-message" VALIDATOR = SKILL_ROOT / "scripts" / "validate_commit_message.py" POLICY = SKILL_ROOT / "references" / "commit_policy.json" CI_ENTRY = ROOT / ".gitea" / "ci" / "commit_message_lint.py" SPEC = ROOT / "docs" / "common" / "commit_message.md" TABLE_ROW_RE = re.compile(r"^\|(?P[^|]*)\|(?P[^|]*)\|") def run(script, *args, env=None): merged = {**os.environ, **(env or {})} merged.pop("COMMIT_POLICY_PATH", None) return subprocess.run( [sys.executable, str(script), *args], cwd=ROOT, capture_output=True, text=True, env=merged, ) def spec_type_emoji_mapping(): mapping = {} for line in SPEC.read_text(encoding="utf-8").splitlines(): match = TABLE_ROW_RE.match(line.strip()) if not match: continue type_cell = re.search(r"`([a-z][a-z0-9-]*)`", match.group("type")) emoji_cell = re.search(r"`(:[a-z0-9_+-]+:)`", match.group("emoji")) if type_cell and emoji_cell: mapping[type_cell.group(1)] = emoji_cell.group(1) return mapping class CommitPolicySingleOwnerTests(unittest.TestCase): def test_policy_and_spec_table_agree(self): policy_types = json.loads(POLICY.read_text(encoding="utf-8"))["types"] spec_types = spec_type_emoji_mapping() self.assertTrue(spec_types, f"no type/emoji table parsed from {SPEC}") self.assertEqual( policy_types, spec_types, "commit_policy.json and docs/common/commit_message.md disagree; " "update both when adding or renaming a type", ) def test_ci_entry_point_does_not_reimplement_rules(self): source = CI_ENTRY.read_text(encoding="utf-8") self.assertIn("validate_commit_message.py", source) tree = ast.parse(source) imported = { alias.name.split(".")[0] for node in ast.walk(tree) if isinstance(node, ast.Import) for alias in node.names } | { node.module.split(".")[0] for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module } self.assertNotIn( "re", imported, f"{CI_ENTRY.name} must delegate; a regex means it parses rules itself", ) called = set() for node in ast.walk(tree): if not isinstance(node, ast.Call): continue function = node.func if isinstance(function, ast.Attribute): called.add(function.attr) elif isinstance(function, ast.Name): called.add(function.id) for reader in ("read_text", "open", "read"): self.assertNotIn( reader, called, f"{CI_ENTRY.name} must not read a rule source; it only delegates", ) def test_ci_entry_point_reports_the_validator_it_delegates_to(self): result = run(CI_ENTRY) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("validator:", result.stdout) self.assertIn("validate_commit_message.py", result.stdout) self.assertIn("commit policy:", result.stdout) class SubjectRuleTests(unittest.TestCase): def assert_subject(self, subject, expected_rc): result = run(VALIDATOR, "--subject", subject) self.assertEqual( result.returncode, expected_rc, f"{subject!r}\nstdout={result.stdout}\nstderr={result.stderr}", ) def test_valid_subjects_pass(self): for subject in ( ":bug: fix(core): repair the thing", ":sparkles: feat(tsl-api-reference): add lookup", ":wrench: chore(ci_gitea): retune runner", ":memo: docs: describe the flow", ): with self.subTest(subject=subject): self.assert_subject(subject, 0) def test_malformed_scopes_are_rejected(self): for scope in ("-a", "a-", "a--b", "_", "A", "a b"): with self.subTest(scope=scope): self.assert_subject(f":bug: fix({scope}): repair the thing", 1) def test_overlong_subject_is_rejected(self): self.assert_subject(":bug: fix(core): " + "a" * 73, 1) self.assert_subject(":bug: fix(core): " + "a" * 72, 0) def test_period_suffixes_are_rejected(self): self.assert_subject(":bug: fix(core): repair the thing.", 1) self.assert_subject(":bug: fix(core): 修好了这个东西。", 1) def test_unknown_type_and_emoji_mismatch_are_rejected(self): self.assert_subject(":bug: nope(core): repair the thing", 1) self.assert_subject(":memo: fix(core): repair the thing", 1) def test_missing_policy_reports_deployment_error(self): self.assert_no_such_policy() def assert_no_such_policy(self): result = run(VALIDATOR, "--policy", "no/such/policy.json", "--subject", "x") self.assertEqual(result.returncode, 2, result.stdout) self.assertIn("ERROR", result.stderr) class WorkflowRunEventTests(unittest.TestCase): def head_sha(self): return subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True ).strip() def run_with_payload(self, payload): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "event.json" path.write_text(json.dumps(payload), encoding="utf-8") return run( VALIDATOR, env={ "GITHUB_EVENT_NAME": "workflow_run", "GITHUB_EVENT_PATH": str(path), }, ) def test_upstream_push_validates_the_upstream_head_commit(self): result = self.run_with_payload( { "workflow_run": { "head_sha": self.head_sha(), "head_branch": "main", "event": "push", } } ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("event: workflow_run", result.stdout) self.assertIn("workflow_run(push).commit", result.stdout) self.assertNotIn("input: HEAD", result.stdout) def test_malformed_payload_is_an_input_error(self): for run_payload in ({}, {"head_sha": "nope", "event": "push"}, {"head_sha": "0" * 40}): with self.subTest(payload=run_payload): result = self.run_with_payload({"workflow_run": run_payload}) self.assertEqual(result.returncode, 2, result.stdout) def test_missing_payload_path_is_an_input_error(self): result = run(VALIDATOR, env={"GITHUB_EVENT_NAME": "workflow_run"}) self.assertEqual(result.returncode, 2, result.stdout) if __name__ == "__main__": unittest.main()