Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions commit_check/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ def _load_from_url(url: str) -> dict[str, Any]:
import io

return toml_load(io.BytesIO(data))
except (urllib.error.URLError, urllib.error.HTTPError, Exception):
except urllib.error.URLError:
return {}


Expand Down Expand Up @@ -137,7 +137,7 @@ def load_config(path_hint: str = "") -> dict[str, Any]:
URL before applying local overrides.
"""
if path_hint:
p = Path(path_hint)
p = Path(path_hint).resolve()
if not p.exists():
raise FileNotFoundError(f"Specified config file not found: {path_hint}")
with open(p, "rb") as f:
Expand Down
78 changes: 44 additions & 34 deletions commit_check/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,42 +98,52 @@ def _should_skip_validation(self, context: ValidationContext) -> bool:
and not has_commits()
)

def _should_skip_commit_validation(self, context: ValidationContext) -> bool:
"""
Determine if commit validation should be skipped.

Skip if the current author or any co-author is in the ignore_authors list
for commits, or if no stdin_text, no commit_file, and no commits exist.
"""
def _author_in_ignore_list(self, context: ValidationContext) -> bool:
"""Check if the current author or any co-author is in the ignore list."""
import re

ignore_authors = context.config.get("commit", {}).get("ignore_authors", [])
if not ignore_authors:
return False

current_author = get_commit_info("an")
if current_author and current_author in ignore_authors:
return True

# Check co-authors from the commit message body
if ignore_authors:
message = ""
if context.stdin_text:
message = context.stdin_text
elif context.commit_file:
try:
with open(context.commit_file, "r") as f:
message = f.read()
except (OSError, IOError):
pass
else:
message = get_commit_info("b")
if message:
co_authors = re.findall(
r"^Co-authored-by:\s*([^<\n]+?)\s*(?:<|$)",
message,
re.MULTILINE,
)
for co_author in co_authors:
if co_author.strip() in ignore_authors:
return True
message = self._get_commit_body(context)
if not message:
return False

co_authors = re.findall(
r"^Co-authored-by:\s*([^<\n]+)\s*(?:<|$)",
message,
re.MULTILINE,
)
return any(co_author.strip() in ignore_authors for co_author in co_authors)

@staticmethod
def _get_commit_body(context: ValidationContext) -> str:
"""Retrieve the commit message body from context or git."""
if context.stdin_text:
return context.stdin_text
if context.commit_file:
try:
with open(context.commit_file, "r") as f:
return f.read()
except (OSError, IOError):
pass
return get_commit_info("b")

def _should_skip_commit_validation(self, context: ValidationContext) -> bool:
"""
Determine if commit validation should be skipped.

Skip if the current author or any co-author is in the ignore_authors list
for commits, or if no stdin_text, no commit_file, and no commits exist.
"""
if self._author_in_ignore_list(context):
return True

return (
context.stdin_text is None
Expand Down Expand Up @@ -242,7 +252,7 @@ def _get_subject(self, context: ValidationContext) -> str:

return get_commit_info("s")

def _validate_subject(self, subject: str) -> ValidationResult:
def _validate_subject(self, _subject: str) -> ValidationResult:
"""Override in subclasses for specific validation logic."""
return ValidationResult.PASS

Expand Down Expand Up @@ -307,11 +317,11 @@ def _validate_subject(self, subject: str) -> ValidationResult:
length = len(subject)
constraint_value = self.rule.value

if self.rule.check == "subject_max_length" and length <= constraint_value:
return ValidationResult.PASS
elif self.rule.check == "subject_min_length" and length >= constraint_value:
return ValidationResult.PASS
elif self.rule.check not in ["subject_max_length", "subject_min_length"]:
if (
(self.rule.check == "subject_max_length" and length <= constraint_value)
or (self.rule.check == "subject_min_length" and length >= constraint_value)
or self.rule.check not in ["subject_max_length", "subject_min_length"]
):
return ValidationResult.PASS

self._print_failure(subject, f"length={length}, constraint={constraint_value}")
Expand Down
179 changes: 98 additions & 81 deletions commit_check/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -376,6 +376,89 @@ def _get_message_content(
return None


def _resolve_commit_message_source(
args: argparse.Namespace,
stdin_reader: StdinReader,
) -> tuple[str | None, str | None]:
"""Determine commit message source: file path or stdin content.

Returns a tuple of (stdin_content, commit_file_path).
"""
if not args.message:
return None, None

if args.commit_msg_file:
return None, args.commit_msg_file

stdin_content = stdin_reader.read_piped_input()
return stdin_content or None, None


def _resolve_stdin_for_non_message(
args: argparse.Namespace, stdin_reader: StdinReader
) -> str | None:
"""Resolve stdin content for non-message validation types."""
has_non_message_check = any(
[args.branch, args.author_name, args.author_email, args.no_force_push]
)
if not has_non_message_check:
return None

stdin_content = stdin_reader.read_piped_input()
if args.no_force_push and stdin_content is None:
return _build_pre_commit_push_input()
return stdin_content


def _get_requested_checks(args: argparse.Namespace) -> list[str]:
"""Build the list of requested validation checks based on CLI args."""
requested_checks: list[str] = []

if args.message:
requested_checks.extend(
[
"message",
"subject_imperative",
"subject_max_length",
"subject_min_length",
"require_signed_off_by",
"subject_capitalized",
"require_body",
"allow_merge_commits",
"allow_revert_commits",
"allow_empty_commits",
"allow_fixup_commits",
"allow_wip_commits",
]
)
if args.branch:
requested_checks.extend(["branch", "merge_base"])
if args.author_name:
requested_checks.append("author_name")
if args.author_email:
requested_checks.append("author_email")
if args.no_force_push:
requested_checks.append("no_force_push")

return requested_checks


def _run_json_output(engine: ValidationEngine, context: ValidationContext) -> int:
"""Run validation and print JSON output."""
outcomes: list[CheckOutcome] = engine.validate_all_detailed(context)
overall = "fail" if any(o.status == "fail" for o in outcomes) else "pass"
print(
json.dumps(
{
"status": overall,
"checks": [o.to_dict() for o in outcomes],
},
indent=2,
)
)
return 0 if overall == "pass" else 1


def main() -> int:
"""The main entrypoint of commit-check program."""
parser = _get_parser()
Expand All @@ -387,6 +470,10 @@ def main() -> int:
stdin_reader = StdinReader()

try:
# Handle positional commit_msg_file argument for pre-commit compatibility
if args.commit_msg_file:
args.message = True

# Load and merge configuration from all sources: CLI > Env > TOML > Defaults
config_data = ConfigMerger.from_all_sources(args, args.config)

Expand All @@ -399,81 +486,24 @@ def main() -> int:
rule_builder = RuleBuilder(config_data)
all_rules = rule_builder.build_all_rules()

# Handle positional commit_msg_file argument for pre-commit compatibility
# Store the file path separately from the boolean flag
commit_msg_file_path = None
if args.commit_msg_file:
commit_msg_file_path = args.commit_msg_file
# If a file was provided positionally, always enable message checking
args.message = True

# Filter rules based on CLI arguments
requested_checks = []
if args.message: # args.message is now a boolean flag
# Add commit message related checks
requested_checks.extend(
[
"message",
"subject_imperative",
"subject_max_length",
"subject_min_length",
"require_signed_off_by",
"subject_capitalized",
"require_body",
"allow_merge_commits",
"allow_revert_commits",
"allow_empty_commits",
"allow_fixup_commits",
"allow_wip_commits",
]
)
if args.branch:
requested_checks.extend(["branch", "merge_base"])
if args.author_name:
requested_checks.append("author_name")
if args.author_email:
requested_checks.append("author_email")
if args.no_force_push:
requested_checks.append("no_force_push")

# If no specific checks requested, show help
# Determine which checks to run
requested_checks = _get_requested_checks(args)
if not requested_checks:
parser.print_help()
return 0

# Filter rules to only include requested checks
filtered_rules = [rule for rule in all_rules if rule.check in requested_checks]

# Create validation engine with filtered rules
engine = ValidationEngine(filtered_rules)

# Create validation context
stdin_content = None
commit_file_path = None

if args.message: # args.message is a boolean flag
# Check if we have a file path from positional argument
if commit_msg_file_path:
commit_file_path = commit_msg_file_path
else:
# No file path provided, try reading from stdin
stdin_content = stdin_reader.read_piped_input()
if not stdin_content:
# No stdin and no file - let validators get data from git themselves
stdin_content = None
elif not any(
[args.branch, args.author_name, args.author_email, args.no_force_push]
):
# If no specific validation type is requested, don't read stdin
pass
else:
# For non-message validations (branch, author, push), check for stdin input
stdin_content = stdin_reader.read_piped_input()
if args.no_force_push and stdin_content is None:
stdin_content = _build_pre_commit_push_input()

# Reset banner state for this run so that multiple main() calls
# in the same process (e.g. tests) don't share banner state.
# Resolve validation context inputs
stdin_content, commit_file_path = _resolve_commit_message_source(
args, stdin_reader
)
if not args.message:
stdin_content = _resolve_stdin_for_non_message(args, stdin_reader)

# Reset banner state for this run
from commit_check.util import print_error_header as _peh

_peh.has_been_called = False
Expand All @@ -490,22 +520,9 @@ def main() -> int:
# Run validation – choose output mode based on --format
output_format: str = getattr(args, "output_format", "text")
if output_format == "json":
outcomes: list[CheckOutcome] = engine.validate_all_detailed(context)
overall = "fail" if any(o.status == "fail" for o in outcomes) else "pass"
print(
json.dumps(
{
"status": overall,
"checks": [o.to_dict() for o in outcomes],
},
indent=2,
)
)
return 0 if overall == "pass" else 1
return _run_json_output(engine, context)

result = engine.validate_all(context)

# Return appropriate exit code
return 0 if result == ValidationResult.PASS else 1

except FileNotFoundError as e:
Expand Down
16 changes: 8 additions & 8 deletions commit_check/rule_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,13 +259,13 @@ def _build_boolean_rule(

# For "allow_*" rules, only create rule if they're disabled (False)
# For "require_*" rules, only create rule if they're enabled (True)
if check.startswith("allow_") and config_value is True:
return None
elif check.startswith("require_") and config_value is False:
return None
elif (
check in ["subject_capitalized", "subject_imperative"]
and config_value is False
if (
(check.startswith("allow_") and config_value is True)
or (check.startswith("require_") and config_value is False)
or (
check in ["subject_capitalized", "subject_imperative"]
and config_value is False
)
):
return None

Expand Down Expand Up @@ -295,7 +295,7 @@ def _get_allowed_branch_names(self) -> list[str]:
def _build_conventional_commit_regex(self, allowed_types: list[str]) -> str:
"""Build regex for conventional commit messages."""
types_pattern = "|".join(sorted(set(allowed_types)))
return rf"^({types_pattern}){{1}}(\([\w\-\.]+\))?(!)?: ([\w ])+([\s\S]*)|(Merge).*|(fixup!.*)"
return rf"^({types_pattern})(\([\w\-\.]+\))?(!)?: ([\w ])+([\s\S]*)|(Merge).*|(fixup!.*)"

def _build_conventional_branch_regex(
self, allowed_types: list[str], allowed_names: list[str]
Expand Down
4 changes: 2 additions & 2 deletions docs/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@
}

object_description_options = [
("py:parameter", dict(include_in_toc=False)),
("py:parameter", {"include_in_toc": False}),
]

sphinx_immaterial_custom_admonitions = [
Expand All @@ -102,7 +102,7 @@
]
for name in ("hint", "tip", "important"):
sphinx_immaterial_custom_admonitions.append(
dict(name=name, icon="material/school", override=True)
{"name": name, "icon": "material/school", "override": True}
)


Expand Down
Loading
Loading