diff --git a/main.py b/main.py index 9130586..975a298 100755 --- a/main.py +++ b/main.py @@ -839,6 +839,36 @@ def run_pr_message_checks(pr_commits: list[Commit]) -> list[ScopeResult]: return results +def run_pr_author_checks( + flags: list[str], pr_commits: list[Commit] +) -> list[ScopeResult]: + """Run the author checks on every author of the pull request's commits. + + Checking only the branch tip let an earlier commit by another author, + say a GitHub noreply address that ``author_email_pattern`` forbids, go + through. Each distinct author is checked once, on the newest commit + that carries it, so a one-person pull request still reports one scope + per check. A commit the clone lacks is fetched without its history. + + Returns ``[]`` when no author can be read, or a commit cannot be had + even by fetching it, so the caller falls back to the branch tip as before. + """ + newest: dict[tuple[str, str], str] = {} + for sha, _msg in pr_commits: + author = API_AUTHORS.get(sha) or (_local_author(sha) if sha else None) + if author: + newest.pop(author, None) + newest[author] = sha + revs = [sha for sha in newest.values() if _rev_resolves(sha) or _fetch_commit(sha)] + if len(revs) != len(newest): + return [] + return [ + check_scope(CHECK_LABELS[flag], [flag, "--rev", sha], sha=sha) + for flag in flags + for sha in revs + ] + + def check_empty_message( label: str, sha: str, env: dict[str, str] | None ) -> ScopeResult: @@ -895,7 +925,8 @@ def run_commit_check() -> tuple[int, list[ScopeResult]]: event; with only ``ai-label`` set it is read for the label, and no scope is reported) 3. Individual PR commit messages (when ``message: true`` and in a PR event) - 4. All remaining checks (branch, author name/email, etc.) + 4. Author checks on every PR commit's author, then the remaining checks + (branch; outside a PR, the HEAD commit and its author) Outside of a PR event all enabled checks are handed to the CLI at once. """ @@ -922,6 +953,7 @@ def run_commit_check() -> tuple[int, list[ScopeResult]]: PR_BODY_AI_TOOLS[:] = scope.ai_tools # ---- 3. Commit message checks ----------------------------------------- + pr_commits: list[Commit] | None = None if MESSAGE_ENABLED: pr_commits = get_pr_commit_messages() if pr_commits: @@ -943,6 +975,14 @@ def run_commit_check() -> tuple[int, list[ScopeResult]]: results.append(check_scope("Commit message", ["--message"], sha=head_sha())) args = [a for a in args if a != "--message"] rev = None + author_flags = [flag for flag in args if flag in AUTHOR_FLAGS] + if is_pr_event() and author_flags: + if pr_commits is None: + pr_commits = get_pr_commit_messages() + author_scopes = run_pr_author_checks(author_flags, pr_commits) + if author_scopes: + results.extend(author_scopes) + args = [a for a in args if a not in AUTHOR_FLAGS] if is_pr_event() and any(flag in AUTHOR_FLAGS for flag in args): rev = pr_head_rev() if rev is None: diff --git a/main_test.py b/main_test.py index 850a200..542de69 100644 --- a/main_test.py +++ b/main_test.py @@ -4110,6 +4110,109 @@ def test_a_missing_pygithub_warns(self): unittest.main() +class TestPrAuthorChecksEveryCommit(unittest.TestCase): + """The author checks read every PR commit's author, not only the tip.""" + + def setUp(self): + main.API_AUTHORS.clear() + self.addCleanup(main.API_AUTHORS.clear) + + def _run(self, commits, authors, resolves=True, message=False): + commands: list[list[str]] = [] + + def run(command, **_kwargs): + commands.append(command) + if command[:2] == ["git", "rev-parse"]: + return MagicMock(returncode=0 if resolves else 1, stdout="") + if command[:2] == ["git", "fetch"]: + return MagicMock(returncode=1, stdout="") + if "--message" in command: + return MagicMock( + returncode=0, stdout=json_output(make_check("message")) + ) + check = command[3].lstrip("-").replace("-", "_") + sha = command[-1] + status = "fail" if "noreply" in authors[sha][1] else "pass" + return MagicMock( + returncode=0, + stdout=json_output(make_check(check, status=status, value=sha)), + ) + + with ( + patch("main.MESSAGE_ENABLED", message), + patch("main.BRANCH_ENABLED", False), + patch("main.AUTHOR_NAME_ENABLED", False), + patch("main.AUTHOR_EMAIL_ENABLED", True), + patch.dict(os.environ, {"GITHUB_EVENT_NAME": "pull_request"}), + patch("main.get_pr_commit_messages", return_value=commits) as listed, + patch("main._local_author", side_effect=lambda sha: authors.get(sha)), + patch("main.get_pr_head_sha", return_value=commits[-1][0]), + patch("main.subprocess.run", side_effect=run), + patch("sys.stdout", io.StringIO()), + ): + rc, results = main.run_commit_check() + self.listed_calls = listed.call_count + return rc, results, commands + + def test_the_commits_are_listed_once_for_messages_and_authors(self): + commits = [("a1", "feat: one")] + _rc, results, _commands = self._run( + commits, {"a1": ("Dev", "dev@example.org")}, message=True + ) + self.assertEqual(self.listed_calls, 1) + self.assertEqual([s.label for s in results], ["Commit 1/1", "Author email"]) + + def test_a_commit_whose_author_cannot_be_read_is_passed_over(self): + commits = [("a1", "feat: one"), ("b2", "feat: two")] + _rc, results, _commands = self._run(commits, {"b2": ("Dev", "dev@example.org")}) + self.assertEqual([(s.label, s.sha) for s in results], [("Author email", "b2")]) + + def test_an_earlier_commit_by_another_author_is_checked(self): + """FreeBSD atf forbids noreply emails; the tip alone hid this one.""" + commits = [("a1", "feat: one"), ("b2", "feat: two")] + authors = { + "a1": ("Dev", "dev@users.noreply.github.com"), + "b2": ("Dev", "dev@example.org"), + } + rc, results, commands = self._run(commits, authors) + self.assertEqual(rc, 1) + self.assertEqual( + [(s.label, s.sha, s.status) for s in results], + [("Author email", "a1", "fail"), ("Author email", "b2", "pass")], + ) + self.assertIn( + ["commit-check", "--format", "json", "--author-email", "--rev", "a1"], + commands, + ) + + def test_one_author_is_checked_once_on_the_newest_commit(self): + commits = [("a1", "feat: one"), ("b2", "feat: two"), ("c3", "feat: three")] + same = ("Dev", "dev@example.org") + rc, results, _commands = self._run( + commits, dict.fromkeys(("a1", "b2", "c3"), same) + ) + self.assertEqual(rc, 0) + self.assertEqual([(s.label, s.sha) for s in results], [("Author email", "c3")]) + + def test_an_api_listed_author_is_used(self): + main.API_AUTHORS["a1"] = ("Dev", "dev@users.noreply.github.com") + _rc, results, _ = self._run( + [("a1", "feat: one")], {"a1": ("Dev", "dev@users.noreply.github.com")} + ) + self.assertEqual([(s.sha, s.status) for s in results], [("a1", "fail")]) + + def test_a_commit_that_cannot_be_had_falls_back_to_the_tip(self): + """Neither in the clone nor fetchable: the tip path decides (here: skip).""" + commits = [("a1", "feat: one")] + _rc, results, commands = self._run( + commits, {"a1": ("Dev", "dev@example.org")}, resolves=False + ) + self.assertEqual( + [(s.label, s.status) for s in results], [("Author email", "skip")] + ) + self.assertFalse([c for c in commands if c[0] == "commit-check"], commands) + + class TestBooleanInputs(unittest.TestCase): """A boolean input is true or false; anything else fails the step."""