fix(static-quality-gates): really skip per-PR size thresholds on main

tasks/git.py

@@ -56,13 +56,8 @@ def push_signed_commits(ctx: Context, branch: str, commit_message: str, source_b
 def get_ancestor(ctx, branch: str) -> str:
     base_branch = get_ancestor_base_branch(branch)
     ancestor = get_common_ancestor(ctx, "HEAD", base_branch)
-    current_commit = get_commit_sha(ctx)
-    # Detect if we're on main branch (ancestor == current commit means merge-base is HEAD itself)
-    # This is used to determine if bypass tolerance should apply (only for PRs, not main)
-    is_on_main_branch = ancestor == current_commit
-    # When on main/release branch, get_common_ancestor returns HEAD itself since merge-base of HEAD and origin/<branch>
-    # is the current commit. In this case, use the parent commit as the ancestor instead.
-    if is_on_main_branch:
+    # When HEAD is the tip of its own base branch, the merge-base is HEAD itself: use the parent commit instead
+    if ancestor == get_commit_sha(ctx):
         ancestor = get_commit_sha(ctx, commit="HEAD~1")
-        print(color_message(f"On main branch, using parent commit {ancestor} as ancestor", "cyan"))
+        print(color_message(f"On {base_branch}, using parent commit {ancestor} as ancestor", "cyan"))
     return ancestor

tasks/quality_gates.py

@@ -11,7 +11,7 @@
 from tasks.libs.ciproviders.github_api import GithubAPI
 from tasks.libs.common.color import color_message
 from tasks.libs.common.git import (
-    get_commit_sha,
+    get_default_branch,
     is_a_release_branch,
 )
 from tasks.libs.package.size import InfraError
@@ -99,14 +99,15 @@ def parse_and_trigger_gates(ctx, config_path: str = GATE_CONFIG_PATH) -> list[St
 
     nightly_run = os.environ.get("BUCKET_BRANCH") == "nightly"
     branch = os.environ["CI_COMMIT_BRANCH"]
+    is_on_main_branch = branch == get_default_branch()
 
     # Early PR lookup - cache for later use in metrics and PR comment
-    # Skip for release branches since they don't have associated PRs
+    # Skip for release branches since they don't have associated PRs, and for main whose open PRs target other branches
     pr = None
     pr_number = None
     pr_author = None
     if not is_a_release_branch(ctx, branch):
-        pr = get_pr_for_branch(branch)
+        pr = None if is_on_main_branch else get_pr_for_branch(branch)
         if pr:
             print(color_message(f"Found PR #{pr.number}: {pr.title}", "cyan"))
             pr_number = str(pr.number)
@@ -174,8 +175,6 @@ def parse_and_trigger_gates(ctx, config_path: str = GATE_CONFIG_PATH) -> list[St
     ancestor = get_ancestor(ctx, branch)
     metric_handler.generate_relative_size(ancestor=ancestor)
     metric_handler.send_metrics_to_datadog()
-    current_commit = get_commit_sha(ctx)
-    is_on_main_branch = ancestor == current_commit
     is_merge_queue = branch.startswith("mq-working-branch-")
 
     # Take a decision on gate results based on measurements

tasks/unit_tests/static_quality_gates/decisions_tests.py

@@ -94,37 +94,6 @@ class TestBypassOnlyAppliesToPRs(unittest.TestCase):
     These tests document the expected behavior at the integration level.
     """
 
-    def test_main_branch_detection_logic(self):
-        """
-        Document: On main branch, ancestor == current_commit, so is_on_main_branch = True.
-
-        When is_on_main_branch is True, the bypass loop in parse_and_trigger_gates
-        is skipped entirely, meaning all failures remain blocking regardless of delta.
-        """
-        # This test documents the detection logic:
-        # ancestor = get_common_ancestor(ctx, "HEAD", base_branch)
-        # is_on_main_branch = ancestor == current_commit
-        # On main, merge-base of HEAD and origin/main is HEAD itself
-
-        # Simulate: on main branch, ancestor equals current commit
-        ancestor = "abc123"
-        current_commit = "abc123"
-        is_on_main_branch = ancestor == current_commit
-        self.assertTrue(is_on_main_branch)
-
-    def test_pr_branch_detection_logic(self):
-        """
-        Document: On PR branches, ancestor != current_commit, so is_on_main_branch = False.
-
-        When is_on_main_branch is False, the bypass loop runs and failures with
-        delta <= 2KiB threshold can be marked non-blocking.
-        """
-        # Simulate: on PR branch, ancestor is different from current commit
-        ancestor = "abc123"  # Common ancestor with main
-        current_commit = "def456"  # PR's HEAD
-        is_on_main_branch = ancestor == current_commit
-        self.assertFalse(is_on_main_branch)
-
     def test_bypass_logic_skipped_on_main_conceptually(self):
         """
         Document: The bypass logic should NOT run on main branch.

tasks/unit_tests/static_quality_gates/tasks_tests.py

@@ -33,6 +33,7 @@ def gitlab_ref_slug(git_ref):
 
 _PR_BRANCH_NAME = "feature/branch"
 _PR_BUCKET_BRANCH = 'dev'
+_MAIN_BRANCH = 'main'
 _CI_PIPELINE_ID = '999999999'
 _CI_COMMIT_SHA = '1234567890abcdef'
 _ANCESTOR_SHA = 'ancestor-sha'
@@ -47,6 +48,11 @@ def gitlab_ref_slug(git_ref):
     'CI_COMMIT_SHA': _CI_COMMIT_SHA,
 }
 
+_MAIN_ENV_VARS = _PR_ENV_VARS | {
+    'CI_COMMIT_BRANCH': _MAIN_BRANCH,
+    'CI_COMMIT_REF_SLUG': _MAIN_BRANCH,
+}
+
 
 class FakeGithubAPI:
     PR = SimpleNamespace(
@@ -137,6 +143,7 @@ def _docker_measure(ctx, config):
             yield config_path
 
 
+@patch("tasks.quality_gates.get_default_branch", new=MagicMock(return_value=_MAIN_BRANCH))
 class TestQualityGatesIntegration(unittest.TestCase):
     def _assert_gate_metrics(self, sent, gate_name, *, on_disk, on_wire, max_disk, max_wire, rel_disk, rel_wire):
         metric_prefix = "datadog.agent.static_quality_gate."
@@ -172,7 +179,6 @@ def test_pr_all_gates_pass_and_send_metrics(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.static_quality_gates.github.GithubAPI", new=FakeGithubAPI),
             patch("tasks.static_quality_gates.gates.send_metrics") as mock_send_metrics,
             patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
@@ -251,7 +257,6 @@ def test_gate_fails_absolute_disk_limit(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.static_quality_gates.github.GithubAPI", new=FakeGithubAPI),
             patch("tasks.static_quality_gates.gates.GateMetricHandler.generate_relative_size"),
             patch(
@@ -294,7 +299,6 @@ def test_gate_absolute_wire_limit_does_not_block(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.static_quality_gates.github.GithubAPI", new=FakeGithubAPI),
             patch("tasks.static_quality_gates.gates.GateMetricHandler.generate_relative_size"),
             patch(
@@ -337,7 +341,6 @@ def test_gate_fails_per_pr_disk_threshold(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.static_quality_gates.github.GithubAPI", new=FakeGithubAPI),
             patch("tasks.static_quality_gates.gates.send_metrics"),
             patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
@@ -386,7 +389,6 @@ def test_gate_fails_per_pr_disk_threshold_exception_approval(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.quality_gates.get_pr_for_branch", return_value=approved_pr),
             patch("tasks.static_quality_gates.gates.send_metrics"),
             patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
@@ -435,7 +437,6 @@ def test_gate_fails_per_pr_wire_threshold(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.static_quality_gates.github.GithubAPI", new=FakeGithubAPI),
             patch("tasks.static_quality_gates.gates.send_metrics"),
             patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
@@ -484,7 +485,6 @@ def test_gate_fails_per_pr_wire_threshold_exception_approval(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.quality_gates.get_pr_for_branch", return_value=approved_pr),
             patch("tasks.static_quality_gates.gates.send_metrics"),
             patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
@@ -533,7 +533,6 @@ def test_gate_fails_absolute_limit_non_blocking_unchanged_from_ancestor(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.static_quality_gates.github.GithubAPI", new=FakeGithubAPI),
             patch("tasks.static_quality_gates.gates.send_metrics"),
             patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
@@ -624,7 +623,6 @@ def test_per_pr_threshold_skipped_on_merge_queue(self):
         with (
             _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
             patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
-            patch("tasks.quality_gates.get_commit_sha", return_value=_CI_COMMIT_SHA),
             patch("tasks.quality_gates.get_pr_for_branch", return_value=None),
             patch("tasks.quality_gates.get_pr_number_from_commit", return_value=None),
             patch("tasks.static_quality_gates.gates.send_metrics"),
@@ -636,6 +634,47 @@ def test_per_pr_threshold_skipped_on_merge_queue(self):
             # Should NOT raise Exit — per-PR threshold is skipped on merge queue branches
             parse_and_trigger_gates(ctx, config_path)
 
+    @patch.dict('os.environ', _MAIN_ENV_VARS, clear=True)
+    def test_per_pr_threshold_skipped_on_main(self):
+        """Per-PR threshold check is not applied on main, where deltas belong to the merged PR."""
+        gate_scenarios = [
+            GateScenario(
+                name=_DEB_GATE,
+                ancestor_disk=100 * _MiB,
+                ancestor_wire=50 * _MiB,
+                current_disk=100 * _MiB,
+                current_wire=50 * _MiB + 5.1 * _MiB,  # > 5 MiB wire threshold
+                max_disk=105 * _MiB,
+                max_wire=60 * _MiB,
+            ),
+            GateScenario(
+                name=_DOCKER_GATE,
+                ancestor_disk=600 * _MiB,
+                ancestor_wire=240 * _MiB,
+                current_disk=600 * _MiB + 700 * _KiB,  # > 600 KiB disk threshold
+                current_wire=240 * _MiB,
+                max_disk=700 * _MiB,
+                max_wire=250 * _MiB,
+            ),
+        ]
+        with (
+            _gate_scenarios_fixture(*gate_scenarios, ancestor_sha=_ANCESTOR_SHA) as config_path,
+            patch("tasks.quality_gates.get_ancestor", return_value=_ANCESTOR_SHA),
+            patch("tasks.quality_gates.get_pr_for_branch") as mock_get_pr_for_branch,
+            patch("tasks.quality_gates.get_pr_number_from_commit", return_value=None),
+            patch("tasks.static_quality_gates.gates.send_metrics"),
+            patch("tasks.static_quality_gates.pr_comment.pr_commenter") as mock_pr_commenter,
+            patch("tasks.static_quality_gates.gates.GateMetricHandler.generate_metric_reports"),
+        ):
+            ctx = MockContext(
+                run={"datadog-ci tag --level job --tags static_quality_gates:\"success\"": Result("Done")}
+            )
+            # Should NOT raise Exit — per-PR thresholds are skipped on main
+            parse_and_trigger_gates(ctx, config_path)
+            # Open PRs whose head branch is main are none of main's business
+            mock_get_pr_for_branch.assert_not_called()
+            mock_pr_commenter.assert_not_called()
+
 
 if __name__ == '__main__':
     unittest.main()