playbook/test/test_commit_message_policy.py

"""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 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"
CI_ENTRY = ROOT / ".gitea" / "ci" / "commit_message_lint.py"
EVENT_ENV_KEYS = (
    "GITHUB_EVENT_NAME",
    "GITHUB_EVENT_PATH",
    "GITEA_EVENT_NAME",
    "GITEA_EVENT_PATH",
)


def run(script, *args, env=None):
    merged = os.environ.copy()
    for key in EVENT_ENV_KEYS:
        merged.pop(key, None)
    merged.update(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,
    )


class CommitPolicySingleOwnerTests(unittest.TestCase):
    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()