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
61 changes: 53 additions & 8 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,23 @@ def _rev_resolves(rev: str) -> bool:
return result.returncode == 0


def _local_author(sha: str) -> tuple[str, str] | None:
"""Author (name, email) of a commit the clone holds, or ``None``."""
try:
result = subprocess.run(
["git", "show", "-s", "--format=%an%x1f%ae", sha],
capture_output=True,
encoding="utf-8",
check=False,
)
except OSError:
return None
name, sep, email = result.stdout.strip("\n").partition("\x1f")
if result.returncode != 0 or not sep:
return None
return name, email


def _fetch_commit(sha: str) -> bool:
"""Fetch one commit, without its history, into a clone that lacks it."""
try:
Expand Down Expand Up @@ -489,6 +506,10 @@ def get_messages_from_head_ref(base_ref: str) -> list[Commit]:
#: The most commits GitHub's pull request commits endpoint returns.
API_COMMIT_LIMIT = 250

#: Author (name, email) of each commit listed by the API, by hash. Those
#: commits are not in the clone, so this is the only record of who wrote them.
API_AUTHORS: dict[str, tuple[str, str]] = {}


def get_messages_from_api() -> list[Commit]:
"""The pull request's commits from the REST API, for a clone without them.
Expand All @@ -509,7 +530,12 @@ def get_messages_from_api() -> list[Commit]:
pull = (
Github(auth=Auth.Token(token), base_url=api_url).get_repo(repo).get_pull(number)
)
commits = [(c.sha, c.commit.message) for c in pull.get_commits()]
listed = list(pull.get_commits())
commits = [(c.sha, c.commit.message) for c in listed]
for c in listed:
author = c.commit.author
if author is not None:
API_AUTHORS[c.sha] = (author.name or "", author.email or "")
# A force push after the event can leave the same count but other commits.
head = get_pr_head_sha()
if len(commits) != total or (head and commits[-1][0] != head):
Expand Down Expand Up @@ -578,7 +604,9 @@ def _relay_cli_notices(text: str) -> None:


def run_check_json(
args: list[str], input_text: str | None = None
args: list[str],
input_text: str | None = None,
env: dict[str, str] | None = None,
) -> tuple[int, dict[str, Any] | None, str]:
"""Run ``commit-check --format json`` and return (exit code, parsed JSON, raw output).

Expand All @@ -602,6 +630,7 @@ def run_check_json(
text=True,
encoding="utf-8",
check=False,
env={**os.environ, **env} if env else None,
)
out = result.stdout or ""
err = result.stderr or ""
Expand All @@ -621,28 +650,44 @@ def run_check_json(


def check_scope(
label: str, args: list[str], input_text: str | None = None, sha: str = ""
label: str,
args: list[str],
input_text: str | None = None,
sha: str = "",
env: dict[str, str] | None = None,
) -> ScopeResult:
"""Run commit-check for one scope and wrap the outcome in a ScopeResult.

``sha`` names the commit a message scope checked; it rides along on both
outcomes so an unparsable CLI response still says which commit it was.
"""
_rc, data, raw = run_check_json(args, input_text=input_text)
_rc, data, raw = run_check_json(args, input_text=input_text, env=env)
if isinstance(data, dict):
return ScopeResult(label=label, checks=data.get("checks", []), sha=sha)
return ScopeResult(label=label, raw_text=raw, sha=sha)


def run_pr_message_checks(pr_commits: list[Commit]) -> list[ScopeResult]:
"""Check each PR commit message individually via commit-check --message."""
"""Check each PR commit individually via commit-check --message.

The raw message is piped in, so every commit is read the same way
whether the clone holds it or only the API listed it. Its own author
goes with it as git's author variables, which is where the CLI reads the
author of a piped message: the AI author rule (CC017) and
``ignore_authors`` then judge that commit's author, not whoever HEAD or
the runner's git config names.
"""
results: list[ScopeResult] = []
total = len(pr_commits)
for index, (sha, msg) in enumerate(pr_commits, start=1):
label = f"Commit {index}/{total}"
author = API_AUTHORS.get(sha) or (_local_author(sha) if sha else None)
env = None
if author:
names = ("GIT_AUTHOR_NAME", "GIT_AUTHOR_EMAIL")
env = {key: value for key, value in zip(names, author) if value} or None
results.append(
check_scope(
f"Commit {index}/{total}", ["--message"], input_text=msg, sha=sha
)
check_scope(label, ["--message"], input_text=msg, sha=sha, env=env)
Comment thread
shenxianpeng marked this conversation as resolved.
Resolved
)
return results

Expand Down
168 changes: 167 additions & 1 deletion main_test.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
"""Unit tests for main.py."""

import functools
import importlib.metadata
import io
import json
import os
import re
import runpy
import shutil
import subprocess
import sys
import tempfile
import unittest
Expand Down Expand Up @@ -390,6 +392,68 @@ def test_invalid_json_falls_back_to_raw_text(self):


class TestRunPrMessageChecks(unittest.TestCase):
"""Each commit's raw message is piped in, with that commit's own author."""

def setUp(self):
local = patch("main._local_author", return_value=None)
self.local_author = local.start()
self.addCleanup(local.stop)
main.API_AUTHORS.clear()
self.addCleanup(main.API_AUTHORS.clear)

def test_a_commit_in_the_clone_carries_its_own_author(self):
"""The raw message is still piped, so it reads as before (no --rev)."""
self.local_author.return_value = ("Renovate Bot", "bot@renovateapp.com")
mock_result = MagicMock(returncode=0, stdout=json_output(make_check("message")))
with patch("main.subprocess.run", return_value=mock_result) as mock_run:
scopes = main.run_pr_message_checks([(SHA_A, "fix: something")])
self.assertEqual(scopes[0].status, "pass")
self.assertEqual(scopes[0].sha, SHA_A)
self.local_author.assert_called_once_with(SHA_A)
self.assertEqual(
mock_run.call_args[0][0],
["commit-check", "--format", "json", "--message"],
)
self.assertEqual(mock_run.call_args[1]["input"], "fix: something")
env = mock_run.call_args[1]["env"]
self.assertEqual(env["GIT_AUTHOR_NAME"], "Renovate Bot")
self.assertEqual(env["GIT_AUTHOR_EMAIL"], "bot@renovateapp.com")

def test_a_blank_author_field_is_left_to_git(self):
self.local_author.return_value = ("Renovate Bot", "")
mock_result = MagicMock(returncode=0, stdout=json_output(make_check("message")))
with patch("main.subprocess.run", return_value=mock_result) as mock_run:
main.run_pr_message_checks([(SHA_A, "fix: something")])
env = mock_run.call_args[1]["env"]
self.assertEqual(env["GIT_AUTHOR_NAME"], "Renovate Bot")
self.assertNotEqual(env.get("GIT_AUTHOR_EMAIL"), "")

def test_no_author_at_all_passes_no_environment(self):
self.local_author.return_value = ("", "")
mock_result = MagicMock(returncode=0, stdout=json_output(make_check("message")))
with patch("main.subprocess.run", return_value=mock_result) as mock_run:
main.run_pr_message_checks([(SHA_A, "fix: something")])
self.assertIsNone(mock_run.call_args[1]["env"])

def test_a_commit_without_a_hash_is_not_looked_up(self):
mock_result = MagicMock(returncode=0, stdout=json_output(make_check("message")))
with patch("main.subprocess.run", return_value=mock_result):
main.run_pr_message_checks([("", "fix: something")])
self.local_author.assert_not_called()

def test_an_api_listed_commit_carries_its_author(self):
"""No commit to read, so the API's author goes in git's variables."""
main.API_AUTHORS[SHA_A] = ("google-labs-jules[bot]", "jules@example.com")
mock_result = MagicMock(returncode=0, stdout=json_output(make_check("message")))
with patch("main.subprocess.run", return_value=mock_result) as mock_run:
main.run_pr_message_checks([(SHA_A, "fix: something")])
self.assertEqual(mock_run.call_args[1]["input"], "fix: something")
env = mock_run.call_args[1]["env"]
self.assertEqual(env["GIT_AUTHOR_NAME"], "google-labs-jules[bot]")
self.assertEqual(env["GIT_AUTHOR_EMAIL"], "jules@example.com")
# The rest of the environment is kept: PATH finds the CLI.
self.assertEqual(env.get("PATH"), os.environ.get("PATH"))

def test_single_message_pass(self):
mock_result = MagicMock(returncode=0, stdout=json_output(make_check("message")))
with patch("main.subprocess.run", return_value=mock_result) as mock_run:
Expand Down Expand Up @@ -451,6 +515,69 @@ def test_empty_list(self):
mock_run.assert_not_called()


class TestLocalAuthor(unittest.TestCase):
"""The author of a commit the clone holds, read from git."""

def test_reads_name_and_email(self):
result = MagicMock(returncode=0, stdout="Jane Dev\x1fjane@example.com\n")
with patch("main.subprocess.run", return_value=result) as mock_run:
self.assertEqual(
main._local_author(SHA_A), ("Jane Dev", "jane@example.com")
)
self.assertEqual(
mock_run.call_args[0][0],
["git", "show", "-s", "--format=%an%x1f%ae", SHA_A],
)

def test_a_commit_git_cannot_show_has_no_author(self):
result = MagicMock(returncode=128, stdout="")
with patch("main.subprocess.run", return_value=result):
self.assertIsNone(main._local_author(SHA_A))

def test_output_without_the_separator_has_no_author(self):
result = MagicMock(returncode=0, stdout="unexpected\n")
with patch("main.subprocess.run", return_value=result):
self.assertIsNone(main._local_author(SHA_A))

def test_missing_git_has_no_author(self):
with patch("main.subprocess.run", side_effect=OSError("no git")):
self.assertIsNone(main._local_author(SHA_A))

def test_reads_a_real_commit(self):
with tempfile.TemporaryDirectory() as repo:
env = {
**os.environ,
"GIT_AUTHOR_NAME": "Renovate Bot",
"GIT_AUTHOR_EMAIL": "bot@renovateapp.com",
"GIT_COMMITTER_NAME": "Renovate Bot",
"GIT_COMMITTER_EMAIL": "bot@renovateapp.com",
}
run = functools.partial(
subprocess.run, cwd=repo, env=env, check=True, capture_output=True
)
run(["git", "init", "-q"])
run(
[
"git",
"-c",
"commit.gpgsign=false",
"commit",
"-q",
"--allow-empty",
"-m",
"chore: bump",
]
)
sha = run(["git", "rev-parse", "HEAD"], text=True).stdout.strip()
cwd = os.getcwd()
os.chdir(repo)
try:
author = main._local_author(sha)
finally:
os.chdir(cwd)
self.assertEqual(author, ("Renovate Bot", "bot@renovateapp.com"))


class TestRunOtherChecks(unittest.TestCase):
def test_empty_args_returns_no_scopes(self):
with patch("main.subprocess.run") as mock_run:
Expand Down Expand Up @@ -1158,9 +1285,17 @@ class TestGetMessagesFromApi(unittest.TestCase):
def _run(self, total, api_commits, head="b2"):
pull = MagicMock()
pull.get_commits.return_value = [
MagicMock(sha=sha, commit=MagicMock(message=msg))
MagicMock(
sha=sha,
commit=MagicMock(
message=msg,
author=MagicMock(email=f"{sha}@example.com"),
),
)
for sha, msg in api_commits
]
for listed in pull.get_commits.return_value:
listed.commit.author.name = f"author {listed.sha}"
github_module = MagicMock()
github_module.Github.return_value.get_repo.return_value.get_pull.return_value = pull
env = {
Expand All @@ -1185,6 +1320,37 @@ def test_lists_the_commits_oldest_first(self):
"https://ghe.example.com/api/v3",
)

def test_records_each_commit_author(self):
main.API_AUTHORS.clear()
self.addCleanup(main.API_AUTHORS.clear)
self._run(2, [("a1", "feat: one"), ("b2", "fix: two")])
self.assertEqual(
main.API_AUTHORS,
{
"a1": ("author a1", "a1@example.com"),
"b2": ("author b2", "b2@example.com"),
},
)

def test_a_commit_without_an_author_records_nothing(self):
main.API_AUTHORS.clear()
self.addCleanup(main.API_AUTHORS.clear)
pull = MagicMock()
pull.get_commits.return_value = [
MagicMock(sha="b2", commit=MagicMock(message="fix: two", author=None))
]
github_module = MagicMock()
github_module.Github.return_value.get_repo.return_value.get_pull.return_value = pull
env = {"GITHUB_TOKEN": "token", "GITHUB_REPOSITORY": "owner/repo"}
with (
patch.dict(os.environ, env),
patch("main.get_pr_event", return_value={"number": 7, "commits": 1}),
patch("main.get_pr_head_sha", return_value="b2"),
patch.dict(sys.modules, {"github": github_module}),
):
self.assertEqual(main.get_messages_from_api(), [("b2", "fix: two")])
self.assertEqual(main.API_AUTHORS, {})

def test_a_list_that_ends_elsewhere_is_not_used(self):
# A force push after the event: same count, other commits.
result, _ = self._run(2, [("a1", "feat: one"), ("c3", "fix: other")])
Expand Down
Loading