✨ feat(commit-message): bundle policy and validator
This commit is contained in:
@@ -0,0 +1,540 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Pattern
|
||||
|
||||
TYPE_RE = re.compile(r"^[a-z][a-z0-9-]*$")
|
||||
EMOJI_RE = re.compile(r"^:[a-z0-9_+-]+:$")
|
||||
GIT_OBJECT_ID_RE = re.compile(r"^[0-9a-fA-F]{40}(?:[0-9a-fA-F]{24})?$")
|
||||
HEADER_RE = re.compile(
|
||||
r"^(?:(?P<emoji>:[a-z0-9_+-]+:)\s+)?"
|
||||
r"(?P<type>[a-z][a-z0-9-]*)"
|
||||
r"(?:\((?P<scope>[^()]*)\))?"
|
||||
r":\s+(?P<text>.+)$"
|
||||
)
|
||||
|
||||
|
||||
class PolicyError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
class InputError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CommitPolicy:
|
||||
source: Path
|
||||
types: dict[str, str]
|
||||
scope_optional: bool
|
||||
scope_pattern: Pattern[str]
|
||||
subject_max_length: int
|
||||
subject_forbidden_suffixes: tuple[str, ...]
|
||||
emoji_required_by_default: bool
|
||||
emoji_requirement_env: str
|
||||
emoji_false_values: frozenset[str]
|
||||
|
||||
|
||||
def _eprint(*args: object) -> None:
|
||||
print(*args, file=sys.stderr)
|
||||
|
||||
|
||||
def resolve_policy_path(cli_path: str | None) -> Path:
|
||||
if cli_path is not None:
|
||||
return Path(cli_path).expanduser()
|
||||
environment_path = os.getenv("COMMIT_POLICY_PATH")
|
||||
if environment_path:
|
||||
return Path(environment_path).expanduser()
|
||||
return Path(__file__).resolve().parent.parent / "references" / "commit_policy.json"
|
||||
|
||||
|
||||
def _load_json(path: Path) -> object:
|
||||
try:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
||||
raise PolicyError(f"cannot read policy {path}: {exc}") from exc
|
||||
|
||||
|
||||
def load_policy(path: Path) -> CommitPolicy:
|
||||
source = path.resolve()
|
||||
raw = _load_json(source)
|
||||
if not isinstance(raw, dict):
|
||||
raise PolicyError(f"policy root must be an object: {source}")
|
||||
|
||||
schema_version = raw.get("schema_version")
|
||||
if type(schema_version) is not int:
|
||||
raise PolicyError(f"policy schema_version must be the integer 1: {source}")
|
||||
if schema_version != 1:
|
||||
raise PolicyError(
|
||||
f"unsupported policy schema_version {schema_version}: {source}"
|
||||
)
|
||||
|
||||
types = raw.get("types")
|
||||
if not isinstance(types, dict) or not types:
|
||||
raise PolicyError("policy.types must be a non-empty object")
|
||||
if not all(
|
||||
isinstance(name, str)
|
||||
and TYPE_RE.fullmatch(name)
|
||||
and isinstance(emoji, str)
|
||||
and EMOJI_RE.fullmatch(emoji)
|
||||
for name, emoji in types.items()
|
||||
):
|
||||
raise PolicyError(
|
||||
"policy.types must map lowercase type names to colon-delimited emoji codes"
|
||||
)
|
||||
|
||||
scope = raw.get("scope")
|
||||
if not isinstance(scope, dict):
|
||||
raise PolicyError("policy.scope must be an object")
|
||||
scope_optional = scope.get("optional")
|
||||
scope_expression = scope.get("pattern")
|
||||
if not isinstance(scope_optional, bool):
|
||||
raise PolicyError("policy.scope.optional must be boolean")
|
||||
if not isinstance(scope_expression, str) or not scope_expression:
|
||||
raise PolicyError("policy.scope.pattern must be a non-empty string")
|
||||
try:
|
||||
scope_pattern = re.compile(scope_expression)
|
||||
except re.error as exc:
|
||||
raise PolicyError(f"invalid policy.scope.pattern: {exc}") from exc
|
||||
|
||||
subject = raw.get("subject")
|
||||
if not isinstance(subject, dict):
|
||||
raise PolicyError("policy.subject must be an object")
|
||||
max_length = subject.get("max_length")
|
||||
forbidden_suffixes = subject.get("forbidden_suffixes")
|
||||
if type(max_length) is not int or max_length <= 0:
|
||||
raise PolicyError("policy.subject.max_length must be a positive integer")
|
||||
if not isinstance(forbidden_suffixes, list) or not all(
|
||||
isinstance(suffix, str) and suffix for suffix in forbidden_suffixes
|
||||
):
|
||||
raise PolicyError(
|
||||
"policy.subject.forbidden_suffixes must contain non-empty strings"
|
||||
)
|
||||
|
||||
emoji = raw.get("emoji")
|
||||
if not isinstance(emoji, dict):
|
||||
raise PolicyError("policy.emoji must be an object")
|
||||
required_by_default = emoji.get("required_by_default")
|
||||
requirement_env = emoji.get("requirement_env")
|
||||
false_values = emoji.get("false_values")
|
||||
if not isinstance(required_by_default, bool):
|
||||
raise PolicyError("policy.emoji.required_by_default must be boolean")
|
||||
if not isinstance(requirement_env, str) or not requirement_env.strip():
|
||||
raise PolicyError("policy.emoji.requirement_env must be a non-empty string")
|
||||
if not isinstance(false_values, list) or not false_values or not all(
|
||||
isinstance(value, str) and value.strip() for value in false_values
|
||||
):
|
||||
raise PolicyError("policy.emoji.false_values must contain non-empty strings")
|
||||
|
||||
return CommitPolicy(
|
||||
source=source,
|
||||
types=dict(types),
|
||||
scope_optional=scope_optional,
|
||||
scope_pattern=scope_pattern,
|
||||
subject_max_length=max_length,
|
||||
subject_forbidden_suffixes=tuple(forbidden_suffixes),
|
||||
emoji_required_by_default=required_by_default,
|
||||
emoji_requirement_env=requirement_env.strip(),
|
||||
emoji_false_values=frozenset(
|
||||
value.strip().casefold() for value in false_values
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def emoji_is_required(policy: CommitPolicy) -> bool:
|
||||
override = os.getenv(policy.emoji_requirement_env)
|
||||
if override is None:
|
||||
return policy.emoji_required_by_default
|
||||
return override.strip().casefold() not in policy.emoji_false_values
|
||||
|
||||
|
||||
def validate_subject(
|
||||
line: str,
|
||||
policy: CommitPolicy,
|
||||
*,
|
||||
require_emoji: bool,
|
||||
) -> str | None:
|
||||
if "\n" in line or "\r" in line:
|
||||
return "subject must be a single line"
|
||||
|
||||
if not line.strip():
|
||||
return "empty subject"
|
||||
if line != line.strip():
|
||||
return "subject must not have leading or trailing whitespace"
|
||||
subject = line
|
||||
|
||||
match = HEADER_RE.fullmatch(subject)
|
||||
if not match:
|
||||
return "does not match ':emoji: type(scope): subject' or 'type(scope): subject'"
|
||||
|
||||
type_name = match.group("type")
|
||||
scope = match.group("scope")
|
||||
text = match.group("text").rstrip()
|
||||
if type_name not in policy.types:
|
||||
return f"unknown type: {type_name}"
|
||||
if scope is None and not policy.scope_optional:
|
||||
return "missing required scope"
|
||||
if scope is not None and policy.scope_pattern.fullmatch(scope) is None:
|
||||
return f"invalid scope: {scope}"
|
||||
|
||||
supplied_emoji = match.group("emoji")
|
||||
if supplied_emoji is None and require_emoji:
|
||||
false_values = ", ".join(
|
||||
repr(value) for value in sorted(policy.emoji_false_values)
|
||||
)
|
||||
return (
|
||||
"missing emoji "
|
||||
f"(set {policy.emoji_requirement_env} to one of {false_values} to allow)"
|
||||
)
|
||||
expected_emoji = policy.types[type_name]
|
||||
if supplied_emoji is not None and supplied_emoji != expected_emoji:
|
||||
return (
|
||||
"emoji/type mismatch: "
|
||||
f"got {supplied_emoji} {type_name}, expected {expected_emoji}"
|
||||
)
|
||||
|
||||
if not text:
|
||||
return "empty subject"
|
||||
if len(text) > policy.subject_max_length:
|
||||
return f"subject exceeds {policy.subject_max_length} characters"
|
||||
forbidden_suffix = next(
|
||||
(
|
||||
suffix
|
||||
for suffix in policy.subject_forbidden_suffixes
|
||||
if text.endswith(suffix)
|
||||
),
|
||||
None,
|
||||
)
|
||||
if forbidden_suffix is not None:
|
||||
if forbidden_suffix in {".", "。"}:
|
||||
return (
|
||||
"subject should not end with a period "
|
||||
f"(forbidden suffix {forbidden_suffix!r})"
|
||||
)
|
||||
return f"subject must not end with forbidden suffix {forbidden_suffix!r}"
|
||||
return None
|
||||
|
||||
|
||||
def _read_message_file(path: Path) -> str:
|
||||
try:
|
||||
content = path.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeError) as exc:
|
||||
raise InputError(f"cannot read message file {path}: {exc}") from exc
|
||||
lines = content.splitlines()
|
||||
return lines[0] if lines else ""
|
||||
|
||||
|
||||
def _parse_args(argv: list[str] | None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="Validate commit-message subjects")
|
||||
parser.add_argument("--policy", help="path to a versioned commit policy JSON")
|
||||
candidates = parser.add_mutually_exclusive_group()
|
||||
candidates.add_argument("--subject", help="single commit subject to validate")
|
||||
candidates.add_argument(
|
||||
"--message-file",
|
||||
type=Path,
|
||||
help="UTF-8 commit message file whose first line is validated",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def _requires_event_payload(event_name: str) -> bool:
|
||||
return event_name == "push" or event_name.startswith("pull_request")
|
||||
|
||||
|
||||
def _load_event_payload() -> tuple[str, dict[str, object] | None]:
|
||||
event_name = os.getenv("GITHUB_EVENT_NAME") or os.getenv("GITEA_EVENT_NAME") or ""
|
||||
event_path = os.getenv("GITHUB_EVENT_PATH") or os.getenv("GITEA_EVENT_PATH") or ""
|
||||
if not event_path:
|
||||
if _requires_event_payload(event_name):
|
||||
raise InputError(f"{event_name} event requires an event payload path")
|
||||
return event_name, None
|
||||
|
||||
path = Path(event_path)
|
||||
try:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
||||
if _requires_event_payload(event_name):
|
||||
raise InputError(f"cannot parse {event_name} event payload {path}: {exc}") from exc
|
||||
_eprint(f"WARN: failed to parse event payload: {path} ({exc})")
|
||||
return event_name, None
|
||||
if not isinstance(payload, dict):
|
||||
if _requires_event_payload(event_name):
|
||||
raise InputError(f"{event_name} event payload must be an object: {path}")
|
||||
_eprint(f"WARN: event payload is not an object: {path}")
|
||||
return event_name, None
|
||||
return event_name, payload
|
||||
|
||||
|
||||
def _message_subject(message: str, label: str) -> str:
|
||||
if not message:
|
||||
raise InputError(f"{label}.message must be a non-empty string")
|
||||
lines = message.splitlines()
|
||||
if not lines:
|
||||
raise InputError(f"{label}.message must contain a subject line")
|
||||
return lines[0]
|
||||
|
||||
|
||||
def _git_log_subjects(
|
||||
revisions: list[str],
|
||||
description: str,
|
||||
*,
|
||||
allow_empty: bool = False,
|
||||
label_prefix: str = "push.commit",
|
||||
) -> list[tuple[str, str]]:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "log", "--reverse", "--format=%H%x00%s", *revisions],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise InputError(f"cannot read {description}: {exc}") from exc
|
||||
if result.returncode != 0:
|
||||
detail = result.stderr.strip()
|
||||
suffix = f": {detail}" if detail else ""
|
||||
raise InputError(f"cannot read {description}{suffix}")
|
||||
|
||||
subjects: list[tuple[str, str]] = []
|
||||
for line in result.stdout.splitlines():
|
||||
sha, separator, subject = line.partition("\0")
|
||||
if not separator or not sha or not subject:
|
||||
raise InputError(f"invalid git log output for {description}")
|
||||
subjects.append((f"{label_prefix} {sha[:7]}", subject))
|
||||
if not subjects and not allow_empty:
|
||||
raise InputError(f"{description} is empty")
|
||||
return subjects
|
||||
|
||||
|
||||
def _require_complete_git_history(context: str) -> None:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "--is-shallow-repository"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise InputError(f"cannot inspect repository depth for {context}: {exc}") from exc
|
||||
if result.returncode != 0:
|
||||
detail = result.stderr.strip()
|
||||
suffix = f": {detail}" if detail else ""
|
||||
raise InputError(f"cannot inspect repository depth for {context}{suffix}")
|
||||
shallow_state = result.stdout.strip()
|
||||
if shallow_state == "true":
|
||||
raise InputError(f"{context} requires complete, non-shallow Git history")
|
||||
if shallow_state != "false":
|
||||
raise InputError(f"cannot determine repository depth for {context}")
|
||||
|
||||
|
||||
def _zero_before_subjects(
|
||||
payload: dict[str, object], after: str
|
||||
) -> list[tuple[str, str]]:
|
||||
target_ref = payload.get("ref")
|
||||
if not isinstance(target_ref, str) or not target_ref.startswith("refs/heads/"):
|
||||
raise InputError("zero-before push event requires a refs/heads/* ref")
|
||||
try:
|
||||
check_ref = subprocess.run(
|
||||
["git", "check-ref-format", target_ref],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
refs_result = subprocess.run(
|
||||
[
|
||||
"git",
|
||||
"for-each-ref",
|
||||
"--format=%(refname)%00%(symref)",
|
||||
"refs/heads",
|
||||
"refs/remotes",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise InputError(f"cannot inspect refs for zero-before push: {exc}") from exc
|
||||
if check_ref.returncode != 0:
|
||||
raise InputError(f"invalid zero-before push ref: {target_ref}")
|
||||
if refs_result.returncode != 0:
|
||||
detail = refs_result.stderr.strip()
|
||||
suffix = f": {detail}" if detail else ""
|
||||
raise InputError(f"cannot inspect refs for zero-before push{suffix}")
|
||||
|
||||
branch_name = target_ref.removeprefix("refs/heads/")
|
||||
exclusions: list[str] = []
|
||||
for line in refs_result.stdout.splitlines():
|
||||
ref_name, separator, symbolic_target = line.partition("\0")
|
||||
if not separator or not ref_name:
|
||||
raise InputError("invalid git ref output for zero-before push")
|
||||
if symbolic_target:
|
||||
continue
|
||||
if ref_name == target_ref:
|
||||
continue
|
||||
if ref_name.startswith("refs/remotes/"):
|
||||
remote_parts = ref_name.split("/", 3)
|
||||
if len(remote_parts) == 4 and remote_parts[3] == branch_name:
|
||||
continue
|
||||
exclusions.append(ref_name)
|
||||
|
||||
revisions = [after]
|
||||
if exclusions:
|
||||
revisions.extend(["--not", *exclusions])
|
||||
return _git_log_subjects(
|
||||
revisions,
|
||||
f"zero-before push history for {target_ref}",
|
||||
allow_empty=True,
|
||||
)
|
||||
|
||||
|
||||
def _push_range_subjects(payload: dict[str, object]) -> list[tuple[str, str]]:
|
||||
before = payload.get("before")
|
||||
after = payload.get("after")
|
||||
if (
|
||||
not isinstance(before, str)
|
||||
or not isinstance(after, str)
|
||||
or GIT_OBJECT_ID_RE.fullmatch(before) is None
|
||||
or GIT_OBJECT_ID_RE.fullmatch(after) is None
|
||||
):
|
||||
raise InputError("push event before/after must be full Git object IDs")
|
||||
if set(after) == {"0"}:
|
||||
raise InputError("push event after cannot be the zero object ID")
|
||||
_require_complete_git_history("push event")
|
||||
if set(before) == {"0"}:
|
||||
return _zero_before_subjects(payload, after)
|
||||
|
||||
revision_range = f"{before}..{after}"
|
||||
return _git_log_subjects(
|
||||
[revision_range], f"push commit range {revision_range}"
|
||||
)
|
||||
|
||||
|
||||
def _gather_ci_subjects() -> tuple[str, list[tuple[str, str]]]:
|
||||
event_name, payload = _load_event_payload()
|
||||
subjects: list[tuple[str, str]] = []
|
||||
|
||||
if event_name.startswith("pull_request"):
|
||||
if not isinstance(payload, dict):
|
||||
raise InputError(f"{event_name} event payload is required")
|
||||
pull_request = payload.get("pull_request")
|
||||
if not isinstance(pull_request, dict):
|
||||
raise InputError("pull_request event payload.pull_request must be an object")
|
||||
title = pull_request.get("title")
|
||||
if not isinstance(title, str) or not title:
|
||||
raise InputError("pull_request.title must be a non-empty string")
|
||||
subjects.append(("pull_request.title", _message_subject(title, "pull_request.title")))
|
||||
base = pull_request.get("base")
|
||||
head = pull_request.get("head")
|
||||
if not isinstance(base, dict) or not isinstance(head, dict):
|
||||
raise InputError("pull_request.base/head must be objects")
|
||||
base_sha = base.get("sha")
|
||||
head_sha = head.get("sha")
|
||||
if (
|
||||
not isinstance(base_sha, str)
|
||||
or not isinstance(head_sha, str)
|
||||
or GIT_OBJECT_ID_RE.fullmatch(base_sha) is None
|
||||
or GIT_OBJECT_ID_RE.fullmatch(head_sha) is None
|
||||
):
|
||||
raise InputError("pull_request.base/head.sha must be full Git object IDs")
|
||||
_require_complete_git_history("pull request event")
|
||||
revision_range = f"{base_sha}..{head_sha}"
|
||||
subjects.extend(
|
||||
_git_log_subjects(
|
||||
[revision_range],
|
||||
f"pull request commit range {revision_range}",
|
||||
allow_empty=True,
|
||||
label_prefix="pull_request.commit",
|
||||
)
|
||||
)
|
||||
return event_name, subjects
|
||||
|
||||
if event_name == "push":
|
||||
if not isinstance(payload, dict):
|
||||
raise InputError("push event payload is required")
|
||||
commits = payload.get("commits")
|
||||
if not isinstance(commits, list):
|
||||
raise InputError("push event payload.commits must be a list")
|
||||
for index, commit in enumerate(commits):
|
||||
item_label = f"push.commits[{index}]"
|
||||
if not isinstance(commit, dict):
|
||||
raise InputError(f"{item_label} must be an object")
|
||||
message = commit.get("message")
|
||||
if not isinstance(message, str):
|
||||
raise InputError(f"{item_label}.message must be a non-empty string")
|
||||
subject = _message_subject(message, item_label)
|
||||
sha = commit.get("id") or commit.get("sha") or ""
|
||||
sha_text = str(sha)
|
||||
label = f"push.commit {sha_text[:7]}" if sha_text else "push.commit"
|
||||
subjects.append((label, subject))
|
||||
range_subjects = _push_range_subjects(payload)
|
||||
return event_name, range_subjects
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "log", "-1", "--format=%s", "HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise InputError(f"cannot read HEAD subject: {exc}") from exc
|
||||
lines = result.stdout.splitlines()
|
||||
if result.returncode == 0 and lines:
|
||||
return event_name, [("HEAD", lines[0])]
|
||||
|
||||
detail = result.stderr.strip()
|
||||
suffix = f": {detail}" if detail else ""
|
||||
raise InputError(f"no commit subject found in event payload or HEAD{suffix}")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = _parse_args(argv)
|
||||
try:
|
||||
policy = load_policy(resolve_policy_path(args.policy))
|
||||
if args.subject is not None:
|
||||
event_name = ""
|
||||
subjects = [("--subject", args.subject)]
|
||||
elif args.message_file is not None:
|
||||
event_name = ""
|
||||
subjects = [(str(args.message_file), _read_message_file(args.message_file))]
|
||||
else:
|
||||
event_name, subjects = _gather_ci_subjects()
|
||||
except (InputError, PolicyError) as exc:
|
||||
_eprint(f"ERROR: {exc}")
|
||||
return 2
|
||||
|
||||
require_emoji = emoji_is_required(policy)
|
||||
print(f"commit policy: {policy.source}")
|
||||
if event_name:
|
||||
print(f"event: {event_name}")
|
||||
print(f"require emoji: {require_emoji}")
|
||||
print(f"checks: {len(subjects)} subject(s)")
|
||||
for label, _subject in subjects:
|
||||
print(f"input: {label}")
|
||||
|
||||
errors: list[str] = []
|
||||
for label, subject in subjects:
|
||||
error = validate_subject(subject, policy, require_emoji=require_emoji)
|
||||
if error:
|
||||
errors.append(f"- {label}: {error}\n subject: {subject}")
|
||||
|
||||
if errors:
|
||||
_eprint("ERROR: commit message lint failed:")
|
||||
for error in errors:
|
||||
_eprint(error)
|
||||
return 1
|
||||
|
||||
print("OK")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user