ci: add commit message linting to ci workflow

Author: ayushiahjolia

.github/workflows/ci.yml

@@ -12,6 +12,20 @@ on:
     branches: [ main ]
 
 jobs:
+  lint-commits:
+    if: github.event_name == 'pull_request'
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+      with:
+        fetch-depth: 0
+    - name: Set up Python
+      uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+      with:
+        python-version: "3.12"
+    - name: Lint commit messages
+      run: python ops/lintcommit.py --range "origin/${{ github.event.pull_request.base.ref }}..${{ github.event.pull_request.head.sha }}"
+
   build:
     runs-on: ubuntu-latest
     strategy:

ops/lintcommit.py

@@ -8,8 +8,11 @@
 
 from __future__ import annotations
 
+import argparse
 import re
+import subprocess
 import sys
+from dataclasses import dataclass, field
 
 TYPES: set[str] = {
     "build",
@@ -124,75 +127,141 @@ def validate_message(message: str) -> tuple[str | None, list[str]]:
     return (error, warnings)
 
 
-def run_local() -> None:
-    """Validate local commit messages ahead of origin/main.
+@dataclass
+class CommitResult:
+    """Result of validating a single commit."""
 
-    If there are uncommitted changes, prints a warning and skips validation.
-    """
-    import subprocess
+    sha: str
+    subject: str
+    error: str | None = None
+    warnings: list[str] = field(default_factory=list)
 
-    # Check for uncommitted changes
-    status: subprocess.CompletedProcess[str] = subprocess.run(
-        ["git", "status", "--porcelain"],
-        capture_output=True,
-        text=True,
-    )
-    if status.stdout.strip():
-        print(
-            "WARNING: uncommitted changes detected, skipping commit message validation.\n"
-            "Commit your changes and re-run to validate."
+
+@dataclass
+class LintResult:
+    """Result of linting a range of commits."""
+
+    commits: list[CommitResult] = field(default_factory=list)
+    skipped: bool = False
+    skip_reason: str = ""
+    empty: bool = False
+    git_error: str = ""
+
+    @property
+    def has_errors(self) -> bool:
+        return bool(self.git_error) or any(c.error for c in self.commits)
+
+
+def lint_range(git_range: str, *, skip_dirty_check: bool = False) -> LintResult:
+    """Validate commit messages in a git range (e.g. 'origin/main..HEAD').
+
+    Args:
+        git_range: A git revision range like 'origin/main..HEAD'.
+        skip_dirty_check: When True, skip the uncommitted changes check
+            (useful in CI where the worktree may be clean by definition).
+
+    Returns:
+        A LintResult with per-commit validation results.
+    """
+    if not skip_dirty_check:
+        status = subprocess.run(
+            ["git", "status", "--porcelain"],
+            capture_output=True,
+            text=True,
         )
-        return
+        if status.stdout.strip():
+            return LintResult(
+                skipped=True,
+                skip_reason=(
+                    "uncommitted changes detected, skipping commit message validation.\n"
+                    "Commit your changes and re-run to validate."
+                ),
+            )
 
-    # Get all commit messages ahead of origin/main
-    result: subprocess.CompletedProcess[str] = subprocess.run(
-        ["git", "log", "origin/main..HEAD", "--format=%H%n%B%n---END---"],
+    result = subprocess.run(
+        ["git", "log", "--no-merges", git_range, "-z", "--format=%H%n%B"],
         capture_output=True,
         text=True,
     )
     if result.returncode != 0:
-        print(f"git log failed: {result.stderr}", file=sys.stderr)
-        sys.exit(1)
-
-    raw: str = result.stdout.strip()
-    if not raw:
-        print("No local commits ahead of origin/main")
-        return
+        return LintResult(git_error=result.stderr.strip())
 
-    blocks: list[str] = raw.split("---END---")
-    has_errors: bool = False
+    if not result.stdout.strip():
+        return LintResult(empty=True)
 
-    for block in blocks:
-        block = block.strip()
-        if not block:
+    commits: list[CommitResult] = []
+    for record in result.stdout.split("\0"):
+        if not record.strip():
             continue
-
-        lines: list[str] = block.splitlines()
-        sha: str = lines[0][:7]
-        message: str = "\n".join(lines[1:]).strip()
-
+        sha, _, message = record.partition("\n")
+        message = message.strip()
         if not message:
             continue
 
         error, warnings = validate_message(message)
-        subject: str = message.splitlines()[0]
+        subject = message.splitlines()[0]
+        commits.append(
+            CommitResult(
+                sha=sha[:7],
+                subject=subject,
+                error=error,
+                warnings=warnings,
+            )
+        )
+
+    return LintResult(commits=commits)
+
+
+def write_output(lint_result: LintResult, git_range: str) -> None:
+    """Write lint results to stdout/stderr."""
+    if lint_result.skipped:
+        print(f"WARNING: {lint_result.skip_reason}")
+        return
 
-        if error:
-            print(f"FAIL {sha}: {subject}", file=sys.stderr)
-            print(f"  Error: {error}", file=sys.stderr)
-            has_errors = True
+    if lint_result.git_error:
+        print(f"git log failed: {lint_result.git_error}", file=sys.stderr)
+        return
+
+    if lint_result.empty:
+        print(f"No commits in range {git_range}")
+        return
+
+    for commit in lint_result.commits:
+        if commit.error:
+            print(f"FAIL {commit.sha}: {commit.subject}", file=sys.stderr)
+            print(f"  Error: {commit.error}", file=sys.stderr)
         else:
-            print(f"PASS {sha}: {subject}")
+            print(f"PASS {commit.sha}: {commit.subject}")
 
-        for warning in warnings:
+        for warning in commit.warnings:
             print(f"  Warning: {warning}")
 
-    if has_errors:
+
+def run_range(git_range: str, *, skip_dirty_check: bool = False) -> None:
+    """Validate commit messages in a git range and exit on errors."""
+    lint_result = lint_range(git_range, skip_dirty_check=skip_dirty_check)
+    write_output(lint_result, git_range)
+    if lint_result.has_errors:
         sys.exit(1)
 
 
 def main() -> None:
-    run_local()
+    parser = argparse.ArgumentParser(
+        description="Lint commit messages for conventional commits compliance."
+    )
+    parser.add_argument(
+        "--range",
+        default=None,
+        dest="git_range",
+        help="Validate all commits in a git revision range (e.g. 'origin/main..HEAD'). "
+        "Skips the uncommitted-changes check (useful in CI).",
+    )
+    args = parser.parse_args()
+
+    if args.git_range is not None:
+        run_range(args.git_range, skip_dirty_check=True)
+    else:
+        run_range("origin/main..HEAD")
 
 
 if __name__ == "__main__":

ops/tests/test_lintcommit.py

@@ -1,6 +1,13 @@
 #!/usr/bin/env python3
 
-from ops.lintcommit import validate_message, validate_subject
+from __future__ import annotations
+
+from subprocess import CompletedProcess
+from unittest.mock import patch
+
+import pytest
+
+from ops.lintcommit import lint_range, validate_message, validate_subject
 
 
 # region validate_subject: valid subjects
@@ -151,3 +158,104 @@ def test_empty_message() -> None:
 def test_invalid_subject_in_message() -> None:
     error, _ = validate_message("invalid title")
     assert error == "missing colon (:) char"
+
+
+# region lint_range
+
+
+def _make_git_log_output(*messages: str) -> str:
+    """Build fake ``git log --no-merges -z --format=%H%n%B`` output.
+
+    Records are separated by null characters.
+    """
+    records: list[str] = []
+    for i, msg in enumerate(messages):
+        sha = f"abc{i:04d}" + "0" * 33  # 40-char fake SHA
+        records.append(f"{sha}\n{msg}\n")
+    return "\0".join(records)
+
+
+def _completed(
+    stdout: str = "", stderr: str = "", returncode: int = 0
+) -> CompletedProcess[str]:
+    """Shorthand for a ``subprocess.CompletedProcess``."""
+    return CompletedProcess(
+        args=[], returncode=returncode, stdout=stdout, stderr=stderr
+    )
+
+
+@patch("subprocess.run")
+def test_lint_range_all_valid(mock_run) -> None:
+    log_output = _make_git_log_output(
+        "feat: add new feature",
+        "fix(sdk): resolve issue",
+    )
+    mock_run.return_value = _completed(stdout=log_output)
+
+    result = lint_range("origin/main..HEAD", skip_dirty_check=True)
+
+    assert not result.has_errors
+    assert len(result.commits) == 2
+    assert all(c.error is None for c in result.commits)
+
+
+@patch("subprocess.run")
+def test_lint_range_with_invalid_commit(mock_run) -> None:
+    log_output = _make_git_log_output(
+        "feat: add new feature",
+        "bad commit no colon",
+    )
+    mock_run.return_value = _completed(stdout=log_output)
+
+    result = lint_range("origin/main..HEAD", skip_dirty_check=True)
+
+    assert result.has_errors
+    assert result.commits[0].error is None
+    assert result.commits[1].error == "missing colon (:) char"
+
+
+@patch("subprocess.run")
+def test_lint_range_empty(mock_run) -> None:
+    mock_run.return_value = _completed(stdout="")
+
+    result = lint_range("origin/main..HEAD", skip_dirty_check=True)
+
+    assert result.empty
+    assert not result.has_errors
+
+
+@patch("subprocess.run")
+def test_lint_range_git_failure(mock_run) -> None:
+    mock_run.return_value = _completed(returncode=1, stderr="fatal: bad range")
+
+    result = lint_range("bad..range", skip_dirty_check=True)
+
+    assert result.has_errors
+    assert result.git_error == "fatal: bad range"
+
+
+@patch("subprocess.run")
+def test_lint_range_dirty_worktree_skips(mock_run) -> None:
+    """When skip_dirty_check=False and worktree is dirty, validation is skipped."""
+    mock_run.return_value = _completed(stdout=" M ops/lintcommit.py\n")
+
+    result = lint_range("origin/main..HEAD", skip_dirty_check=False)
+
+    assert result.skipped
+    assert "uncommitted changes" in result.skip_reason
+    # git log should never have been called (only git status)
+    mock_run.assert_called_once()
+
+
+@patch("subprocess.run")
+def test_lint_range_warnings_collected(mock_run) -> None:
+    log_output = _make_git_log_output(
+        "feat: add thing\n\n" + "x" * 80,
+    )
+    mock_run.return_value = _completed(stdout=log_output)
+
+    result = lint_range("origin/main..HEAD", skip_dirty_check=True)
+
+    assert not result.has_errors
+    assert len(result.commits) == 1
+    assert any("exceeds 72 chars" in w for w in result.commits[0].warnings)