Files
carbon-lang/github_tools/check_dependent_pr_test.py
T
Chandler Carruth 824d7e3c83 Create a GitHub action to automatically handle dependent PRs (#7101)
This should detect when a PR has a dependency of another open PR and add
a comment and label describing it. The comment will even do a
best-effort to compute the best starting commit for review.

Whenever PRs are closed, it will also scan the open depnedent PRs and
try to either remove the PRs in the comment or if it reaches zero the
label.

It works to update a single comment on a PR rather than adding more
comments.

Assisted-by: Antigravity with Gemini
2026-04-24 19:52:07 +00:00

406 lines
14 KiB
Python

"""Tests for check_dependent_pr.py."""
__copyright__ = """
Part of the Carbon Language project, under the Apache License v2.0 with LLVM
Exceptions. See /LICENSE for license information.
SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
"""
import json
import unittest
from unittest import mock
from typing import Any
import check_dependent_pr
import github_helpers
_OID1 = "1" * 40
_OID2 = "2" * 40
_OID3 = "3" * 40
_OID4 = "4" * 40
_OID9 = "9" * 40
class TestCheckDependentPR(unittest.TestCase):
def setUp(self) -> None:
self.mock_client = mock.MagicMock(spec=github_helpers.Client)
def _make_comment(
self,
open_deps: list[int],
merged_deps: list[int] = None,
first_commit: str = None,
comment_id: str = "comment_id",
) -> dict[str, str]:
"""Builds a boilerplate PR comment."""
state: dict[str, Any] = {
"open": open_deps,
"merged": merged_deps if merged_deps else [],
}
if first_commit:
state["first_commit"] = first_commit
return {
"id": comment_id,
"body": f"<!-- check_dependent_pr {json.dumps(state)} -->",
}
def _make_pr_response(
self,
pr_id: str,
head_ref_oid: str,
commits: list[str],
comments: list[dict[str, str]] = None,
has_dependent_label: bool = False,
) -> dict[str, Any]:
"""Builds a boilerplate GitHub response for a PR."""
labels = (
[{"name": "dependent", "id": "label_dependent"}]
if has_dependent_label
else []
)
return {
"repository": {
"pullRequest": {
"id": pr_id,
"headRefOid": head_ref_oid,
"labels": {"nodes": labels},
"commits": {
"nodes": [{"commit": {"oid": oid}} for oid in commits]
},
"comments": {"nodes": comments if comments else []},
}
}
}
def test_process_pr_no_overlap(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_1",
head_ref_oid=_OID1,
commits=[_OID1],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=1,
pr_to_commits={1: [_OID1]},
open_pr_numbers={1},
label_id="label_id",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_with_overlap(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_2",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=2,
pr_to_commits={1: [_OID1], 2: [_OID1, _OID2]},
open_pr_numbers={1, 2},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 3)
calls = self.mock_client.execute.call_args_list
self.assertIn("addLabelsToLabelable", calls[1][0][0])
self.assertIn("addComment", calls[2][0][0])
def test_process_pr_dependencies_merged(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_3",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
comments=[self._make_comment(open_deps=[1])],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=3,
pr_to_commits={3: [_OID1, _OID2]},
open_pr_numbers={3},
label_id="label_dependent",
dry_run=False,
)
calls = self.mock_client.execute.call_args_list
self.assertIn("removeLabelsFromLabelable", calls[1][0][0])
self.assertIn("updateIssueComment", calls[2][0][0])
def test_process_pr_dependency_got_new_commits(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_3",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
comments=[self._make_comment(open_deps=[1, 2])],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=3,
pr_to_commits={1: [_OID1, _OID4], 3: [_OID1, _OID2]},
open_pr_numbers={1, 3},
label_id="label_dependent",
dry_run=False,
)
calls = self.mock_client.execute.call_args_list
update_mutation = calls[1][0][0]
self.assertIn("updateIssueComment", update_mutation)
variable_values = calls[1][1]["variable_values"]
self.assertIn('"open": [1]', variable_values["body"])
self.assertIn('"merged": [2]', variable_values["body"])
def test_process_pr_non_coherent_prefix(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_10",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=10,
pr_to_commits={10: [_OID1, _OID2], 11: [_OID1, _OID3]},
open_pr_numbers={10, 11},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_overlap_only_on_head_ref(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_9",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=9,
pr_to_commits={1: [_OID2], 9: [_OID1, _OID2]},
open_pr_numbers={1, 9},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 3)
calls = self.mock_client.execute.call_args_list
self.assertIn("addLabelsToLabelable", calls[1][0][0])
self.assertIn("addComment", calls[2][0][0])
def test_process_pr_scanning_no_add(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_7",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=7,
pr_to_commits={1: [_OID1], 7: [_OID1, _OID2]},
open_pr_numbers={1, 7},
label_id="label_dependent",
dry_run=False,
scanning=True,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_no_changes_needed(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_6",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
comments=[self._make_comment(open_deps=[1], first_commit=_OID2)],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=6,
pr_to_commits={1: [_OID1], 6: [_OID1, _OID2]},
open_pr_numbers={1, 6},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_invalid_marker(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_5",
head_ref_oid=_OID1,
commits=[_OID1],
comments=[
{
"id": "comment_id",
"body": "<!-- check_dependent_pr {invalid_json} -->",
}
],
)
import json
self.assertRaises(
json.decoder.JSONDecodeError,
check_dependent_pr._process_pr,
self.mock_client,
pr_number=5,
pr_to_commits={5: [_OID1]},
open_pr_numbers={5},
label_id="label_dependent",
dry_run=False,
)
def test_process_pr_hidden_comment(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_14",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
comments=[
{
"id": "hidden_comment_id",
"body": '<!-- check_dependent_pr {"open": [1]} -->',
"isMinimized": True,
}
],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=14,
pr_to_commits={1: [_OID1], 14: [_OID1, _OID2]},
open_pr_numbers={1, 14},
label_id="label_dependent",
dry_run=False,
)
calls = self.mock_client.execute.call_args_list
self.assertEqual(self.mock_client.execute.call_count, 2)
self.assertIn("addComment", calls[1][0][0])
def test_process_pr_sticky_first_commit(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_11",
head_ref_oid=_OID3,
commits=[_OID1, _OID2, _OID3],
comments=[self._make_comment(open_deps=[1, 2], first_commit=_OID2)],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=11,
pr_to_commits={1: [_OID1], 11: [_OID1, _OID2, _OID3]},
open_pr_numbers={1, 11},
label_id="label_dependent",
dry_run=False,
)
calls = self.mock_client.execute.call_args_list
variable_values = calls[1][1]["variable_values"]
self.assertIn(_OID2[:8], variable_values["body"])
self.assertNotIn(_OID1[:8], variable_values["body"])
def test_process_pr_rebase_first_commit(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_12",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
comments=[self._make_comment(open_deps=[1, 2])],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=12,
pr_to_commits={1: [_OID9], 12: [_OID1, _OID2]},
open_pr_numbers={1, 12},
label_id="label_dependent",
dry_run=False,
)
calls = self.mock_client.execute.call_args_list
variable_values = calls[1][1]["variable_values"]
self.assertIn(_OID1[:8], variable_values["body"])
def test_process_pr_fallback_no_independent_commit(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_13",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
comments=[self._make_comment(open_deps=[1, 2])],
has_dependent_label=True,
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=13,
pr_to_commits={1: [_OID1, _OID2], 13: [_OID1, _OID2]},
open_pr_numbers={1, 13},
label_id="label_dependent",
dry_run=False,
)
calls = self.mock_client.execute.call_args_list
variable_values = calls[1][1]["variable_values"]
self.assertIn(
"unable to identify starting review commit", variable_values["body"]
)
def test_process_pr_sequence_failure(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_1",
head_ref_oid=_OID1,
commits=[_OID1],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=1,
pr_to_commits={1: [_OID1], 2: [_OID1, _OID2]},
open_pr_numbers={1, 2},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_no_overlap_different_commits(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_2",
head_ref_oid=_OID2,
commits=[_OID2],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=2,
pr_to_commits={1: [_OID1], 2: [_OID2]},
open_pr_numbers={1, 2},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_no_unique_commit(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_2",
head_ref_oid=_OID2,
commits=[_OID1, _OID2],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=2,
pr_to_commits={1: [_OID1, _OID2, _OID3], 2: [_OID1, _OID2]},
open_pr_numbers={1, 2},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 1)
def test_process_pr_multiple_non_overlapping_commits(self) -> None:
self.mock_client.execute.return_value = self._make_pr_response(
pr_id="pr_2",
head_ref_oid=_OID4,
commits=[_OID1, _OID3, _OID4],
)
check_dependent_pr._process_pr(
self.mock_client,
pr_number=2,
pr_to_commits={1: [_OID1, _OID2], 2: [_OID1, _OID3, _OID4]},
open_pr_numbers={1, 2},
label_id="label_dependent",
dry_run=False,
)
self.assertEqual(self.mock_client.execute.call_count, 3)
calls = self.mock_client.execute.call_args_list
self.assertIn("addLabelsToLabelable", calls[1][0][0])
if __name__ == "__main__":
unittest.main()