cshandClaude Fable 5
7408c532f0
CI ran .gitea/ci/commit_message_lint.py, an independent reimplementation that parsed the type/emoji mapping out of docs/common/commit_message.md. Its scope pattern accepted fix(-a), fix(a-), fix(a--b) and fix(_), and it had no subject length check, so the gate that actually blocks merges enforced weaker rules than the Validation this skill reports. The CI entry is now a wrapper that locates the skill validator and delegates to it with no arguments; commit_policy.json becomes the only rule source and explanatory docs stop being a machine policy input. The validator also learns the workflow_run event, whose payload carries neither a PR title nor a before/after range. An upstream pull_request now validates <integration-branch>..head_sha instead of HEAD alone, taking the branch name from COMMIT_LINT_MAIN_BRANCH, and degrades to the upstream head commit with a WARN rather than guessing a base. Alongside: --help now documents the no-argument CI mode it had always supported silently, CI wiring detail moves to references/ci-wiring.md, and the description gains negative boundaries. test/test_commit_message_policy.py asserts policy/spec-table equality and uses ast to assert the CI entry imports no regex and reads no file, so a second implementation cannot reappear unnoticed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
665 lines
25 KiB
Python
665 lines
25 KiB
Python
#!/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>.+)$"
|
|
)
|
|
|
|
HELP_EPILOG = """\
|
|
三种模式（互斥；未给出 --subject / --message-file 时进入 CI 输入校验）:
|
|
|
|
--subject "<候选>" 校验单个候选主题；交互式起草提交信息时用这一种
|
|
--message-file <path> 校验提交信息文件的首行；body/footer 不做机械校验
|
|
(无参数) CI 输入校验：从事件 payload 取回全部待校验主题
|
|
|
|
策略来源固定优先级：--policy > $COMMIT_POLICY_PATH > 本脚本旁的
|
|
references/commit_policy.json。策略缺失、损坏或 schema_version 不支持时报部署
|
|
错误，不退回任何内置约定。本脚本是提交信息规则的唯一实现；说明性文档不是机器
|
|
策略来源。
|
|
|
|
CI 输入校验按事件展开：
|
|
|
|
pull_request* pull_request.title 加 base.sha..head.sha 的完整提交范围
|
|
push before..after 完整范围；before 为零对象时按分支图计算新增提交
|
|
workflow_run 上游为 pull_request* 时，取 <集成分支>..workflow_run.head_sha
|
|
的完整范围；其它上游事件只校验 workflow_run.head_sha 一条
|
|
其它 / 本地 回退 HEAD 一条
|
|
|
|
集成分支名取 $COMMIT_LINT_MAIN_BRANCH，默认 main；依次尝试
|
|
refs/remotes/origin/<名字> 与 refs/heads/<名字>。无法解析、或上游 head_branch
|
|
就是集成分支时，退化为只校验上游 head 一条并打印 WARN，不猜测 base。
|
|
|
|
workflow_run payload 不含 PR 标题，该架构下标题不参与校验；完整提交范围校验取代
|
|
它，且强度更高。
|
|
|
|
已识别事件缺失或损坏 payload 时失败；shallow repository 一律失败。
|
|
|
|
退出码：0 全部通过；1 存在不合规主题；2 策略、参数或输入错误。
|
|
"""
|
|
|
|
|
|
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",
|
|
epilog=HELP_EPILOG,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
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 == "workflow_run"
|
|
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 _other_branch_refs(branch_name: str, context: str) -> list[str]:
|
|
"""Non-symbolic branch refs that are not `branch_name` on any remote."""
|
|
try:
|
|
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 {context}: {exc}") from exc
|
|
if refs_result.returncode != 0:
|
|
detail = refs_result.stderr.strip()
|
|
suffix = f": {detail}" if detail else ""
|
|
raise InputError(f"cannot inspect refs for {context}{suffix}")
|
|
|
|
own_ref = f"refs/heads/{branch_name}"
|
|
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(f"invalid git ref output for {context}")
|
|
if symbolic_target:
|
|
continue
|
|
if ref_name == own_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)
|
|
return exclusions
|
|
|
|
|
|
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,
|
|
)
|
|
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}")
|
|
|
|
branch_name = target_ref.removeprefix("refs/heads/")
|
|
exclusions = _other_branch_refs(branch_name, "zero-before push")
|
|
|
|
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 _integration_branch() -> tuple[str, str | None]:
|
|
"""The configured integration branch name and its first resolvable ref."""
|
|
name = (os.getenv("COMMIT_LINT_MAIN_BRANCH") or "").strip() or "main"
|
|
for ref in (f"refs/remotes/origin/{name}", f"refs/heads/{name}"):
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--verify", "--quiet", ref],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except OSError as exc:
|
|
raise InputError(f"cannot resolve {ref}: {exc}") from exc
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
return name, ref
|
|
return name, None
|
|
|
|
|
|
def _workflow_run_subjects(payload: dict[str, object]) -> list[tuple[str, str]]:
|
|
"""Subjects for a run triggered by another run's completion.
|
|
|
|
A workflow_run payload carries no PR title and no before/after range, so the
|
|
upstream head SHA is the only anchor. For an upstream pull_request the range
|
|
is taken against the integration branch; every other case validates the
|
|
upstream head commit alone rather than guessing a base.
|
|
"""
|
|
run = payload.get("workflow_run")
|
|
if not isinstance(run, dict):
|
|
raise InputError("workflow_run event payload.workflow_run must be an object")
|
|
head_sha = run.get("head_sha")
|
|
if not isinstance(head_sha, str) or GIT_OBJECT_ID_RE.fullmatch(head_sha) is None:
|
|
raise InputError("workflow_run.head_sha must be a full Git object ID")
|
|
upstream_event = run.get("event")
|
|
if not isinstance(upstream_event, str) or not upstream_event:
|
|
raise InputError("workflow_run.event must be a non-empty string")
|
|
_require_complete_git_history("workflow_run event")
|
|
|
|
label_prefix = f"workflow_run({upstream_event}).commit"
|
|
head_only = f"validating only upstream head commit {head_sha[:7]}"
|
|
if upstream_event.startswith("pull_request"):
|
|
head_branch = run.get("head_branch")
|
|
if not isinstance(head_branch, str) or not head_branch:
|
|
raise InputError("workflow_run.head_branch must be a non-empty string")
|
|
main_name, main_ref = _integration_branch()
|
|
if main_ref is None:
|
|
_eprint(f"WARN: no {main_name} ref to compare against; {head_only}")
|
|
elif head_branch == main_name:
|
|
_eprint(f"WARN: upstream head_branch is {main_name}; {head_only}")
|
|
else:
|
|
revision_range = f"{main_ref}..{head_sha}"
|
|
subjects = _git_log_subjects(
|
|
[head_sha, "--not", main_ref],
|
|
f"workflow_run pull-request range {revision_range}",
|
|
allow_empty=True,
|
|
label_prefix=label_prefix,
|
|
)
|
|
if subjects:
|
|
return subjects
|
|
_eprint(f"WARN: {revision_range} is empty; {head_only}")
|
|
|
|
return _git_log_subjects(
|
|
["-1", head_sha],
|
|
f"workflow_run head commit {head_sha[:7]}",
|
|
label_prefix=label_prefix,
|
|
)
|
|
|
|
|
|
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)
|
|
# The loop above only asserts payload shape; the Git range is authority.
|
|
return event_name, range_subjects
|
|
|
|
if event_name == "workflow_run":
|
|
if not isinstance(payload, dict):
|
|
raise InputError("workflow_run event payload is required")
|
|
return event_name, _workflow_run_subjects(payload)
|
|
|
|
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())
|