feat: query Mergify's Quarantine API to check for ignorable failed tests (#142)

Fixes MRGFY-5519

Author: Greesb

poetry.lock

@@ -22,6 +22,7 @@ poethepoet = ">=0.30"
 codespell = "^2.3.0"
 yamllint = "^1.35.1"
 anys = "^0.3.1"
+responses = "^0.25.7"
 
 [build-system]
 requires = ["poetry-core", "poetry-dynamic-versioning"]
@@ -48,6 +49,10 @@ module = [
 ]
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "requests"
+ignore_missing_imports = true
+
 [tool.poetry-dynamic-versioning]
 enable = true
 vcs = "git"

pyproject.toml

@@ -54,6 +54,22 @@ def pytest_terminal_summary(
             )
             return
 
+        # CI Insights Quarantine warning logs
+        if not self.mergify_ci.branch_name:
+            terminalreporter.write_line(
+                "No valid branch name found, unable to setup CI Insights Quarantine",
+                yellow=True,
+            )
+
+        if (
+            self.mergify_ci.quarantined_tests is not None
+            and self.mergify_ci.quarantined_tests.init_error_msg
+        ):
+            terminalreporter.write_line(
+                self.mergify_ci.quarantined_tests.init_error_msg, yellow=True
+            )
+
+        # CI Insights Traces upload logs
         if self.mergify_ci.tracer_provider is None:
             terminalreporter.write_line(
                 "Mergify Tracer didn't start for unexpected reason (please contact Mergify support); test results will not be uploaded",
@@ -171,6 +187,8 @@ def pytest_runtest_protocol(
             else:
                 skip_attributes = {}
 
+                self.mergify_ci.mark_test_as_quarantined_if_needed(item)
+
             context = opentelemetry.trace.set_span_in_context(self.session_span)
             with self.tracer.start_as_current_span(
                 item.nodeid,

pytest_mergify/__init__.py

@@ -1,9 +1,10 @@
 import dataclasses
-import os
+import _pytest.nodes
 import random
+import os
 import typing
 
-import requests  # type: ignore[import-untyped]
+import requests
 import opentelemetry.sdk.resources
 from opentelemetry.sdk.trace import export
 from opentelemetry.sdk.trace import TracerProvider, SpanProcessor, ReadableSpan
@@ -13,8 +14,10 @@
 )
 
 from pytest_mergify import utils
+import pytest_mergify.quarantine
 
 import pytest_mergify.resources.ci as resources_ci
+from opentelemetry.semconv._incubating.attributes import vcs_attributes
 import pytest_mergify.resources.github_actions as resources_gha
 import pytest_mergify.resources.pytest as resources_pytest
 import pytest_mergify.resources.mergify as resources_mergify
@@ -60,6 +63,10 @@ class MergifyCIInsights:
             "MERGIFY_API_URL", "https://api.mergify.com"
         )
     )
+    branch_name: typing.Optional[str] = dataclasses.field(
+        init=False,
+        default=None,
+    )
     exporter: typing.Optional[export.SpanExporter] = dataclasses.field(
         init=False, default=None
     )
@@ -70,7 +77,14 @@ class MergifyCIInsights:
         dataclasses.field(init=False, default=None)
     )
     test_run_id: str = dataclasses.field(
-        default_factory=lambda: random.getrandbits(64).to_bytes(8, "big").hex()
+        init=False,
+        default_factory=lambda: random.getrandbits(64).to_bytes(8, "big").hex(),
+    )
+    quarantined_tests: typing.Optional[pytest_mergify.quarantine.Quarantine] = (
+        dataclasses.field(
+            init=False,
+            default=None,
+        )
     )
 
     def __post_init__(self) -> None:
@@ -122,3 +136,24 @@ def __post_init__(self) -> None:
 
         self.tracer_provider.add_span_processor(span_processor)
         self.tracer = self.tracer_provider.get_tracer("pytest-mergify")
+
+        # Retrieve the branch name based on the detected resources's attributes
+        branch_name = resource.attributes.get(
+            vcs_attributes.VCS_REF_BASE_NAME,
+            resource.attributes.get(vcs_attributes.VCS_REF_HEAD_NAME),
+        )
+        if branch_name is not None:
+            # `str` cast just for `mypy`
+            self.branch_name = str(branch_name)
+
+        if self.token and self.repo_name and self.branch_name:
+            self.quarantined_tests = pytest_mergify.quarantine.Quarantine(
+                self.api_url,
+                self.token,
+                self.repo_name,
+                self.branch_name,
+            )
+
+    def mark_test_as_quarantined_if_needed(self, item: _pytest.nodes.Item) -> None:
+        if self.quarantined_tests is not None and item in self.quarantined_tests:
+            self.quarantined_tests.mark_test_as_quarantined(item)

pytest_mergify/ci_insights.py

@@ -0,0 +1,64 @@
+import dataclasses
+import pytest
+import _pytest.nodes
+import requests
+import typing
+
+
+@dataclasses.dataclass
+class Quarantine:
+    api_url: str
+    token: str
+    repo_name: str
+    branch_name: str
+    quarantined_tests: typing.List[str] = dataclasses.field(
+        init=False, default_factory=list
+    )
+    init_error_msg: typing.Optional[str] = dataclasses.field(init=False, default=None)
+
+    def __post_init__(self) -> None:
+        try:
+            owner, repository = self.repo_name.split("/")
+        except ValueError:
+            self.init_error_msg = f"Repository name '{self.repo_name}' has an unexpected format (expected 'owner/repository'), skipping CI Insights Quarantine setup"
+            return
+
+        url = f"{self.api_url}/v1/ci/{owner}/repositories/{repository}/quarantines"
+
+        try:
+            quarantine_resp: requests.Response = requests.get(
+                url,
+                headers={"Authorization": f"Bearer {self.token}"},
+                params={"branch": self.branch_name},
+                timeout=10,
+            )
+        except requests.ConnectionError as exc:
+            self.init_error_msg = f"Failed to connect to Mergify's API, tests won't be quarantined. Error: {str(exc)}"
+            return
+
+        if quarantine_resp.status_code == 402:
+            # No CI Insights Quarantine subscription, skip it.
+            return
+
+        try:
+            quarantine_resp.raise_for_status()
+        except requests.HTTPError as exc:
+            self.init_error_msg = f"Error when querying Mergify's API, tests won't be quarantined. Error: {str(exc)}"
+            return
+
+        self.quarantined_tests = quarantine_resp.json()["quarantined_tests"]
+
+    def __contains__(self, item: _pytest.nodes.Item) -> bool:
+        return item.nodeid in self.quarantined_tests
+
+    @staticmethod
+    def mark_test_as_quarantined(test_item: _pytest.nodes.Item) -> None:
+        test_item.add_marker(
+            pytest.mark.xfail(
+                reason="Test is quarantined from Mergify CI Insights",
+                raises=None,
+                run=True,
+                strict=False,
+            ),
+            append=True,
+        )

pytest_mergify/quarantine.py

@@ -1,13 +1,18 @@
 import typing
+import re
+import os
+import responses
 import http.server
 import socketserver
 import threading
 
 import pytest
+from pytest_mergify import utils
 import _pytest.pytester
 from opentelemetry.sdk import trace
 
 import pytest_mergify
+import pytest_mergify.quarantine
 
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
     InMemorySpanExporter,
@@ -35,6 +40,7 @@ def __call__(
         self,
         code: str = ...,
         setenv: typing.Optional[typing.Dict[str, typing.Optional[str]]] = ...,
+        quarantined_tests: typing.Optional[typing.List[str]] = None,
     ) -> PytesterWithSpanReturnT: ...
 
 
@@ -43,20 +49,41 @@ def __call__(
 
 @pytest.fixture
 def pytester_with_spans(
-    pytester: _pytest.pytester.Pytester, monkeypatch: pytest.MonkeyPatch
+    pytester: _pytest.pytester.Pytester,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> PytesterWithSpanT:
+    @responses.activate
     def _run(
         code: str = _DEFAULT_PYTESTER_CODE,
         setenv: typing.Optional[typing.Dict[str, typing.Optional[str]]] = None,
+        quarantined_tests: typing.Optional[typing.List[str]] = None,
     ) -> PytesterWithSpanReturnT:
         monkeypatch.delenv("PYTEST_MERGIFY_DEBUG", raising=False)
         monkeypatch.setenv("CI", "true")
         monkeypatch.setenv("_PYTEST_MERGIFY_TEST", "true")
+
         for k, v in (setenv or {}).items():
             if v is None:
                 monkeypatch.delenv(k, raising=False)
             else:
                 monkeypatch.setenv(k, v)
+
+        api_url = os.getenv("MERGIFY_API_URL")
+        responses.add(
+            responses.GET,
+            re.compile(rf"{api_url}/v1/ci/.*/repositories/.*/quarantines\?branch=.*"),
+            status=200,
+            json={"quarantined_tests": quarantined_tests or []},
+        )
+
+        full_repository = utils.get_repository_name()
+        passthrough = responses.Response(
+            responses.POST,
+            f"{api_url}/v1/repos/{full_repository}/ci/traces",
+            passthrough=True,
+        )
+        responses.add(passthrough)
+
         plugin = pytest_mergify.PytestMergify()
         pytester.makepyfile(code)
         result = pytester.runpytest_inprocess(plugins=[plugin])

tests/conftest.py

@@ -147,6 +147,7 @@ def test_errors_logs(
     monkeypatch.setenv("CI", "1")
     monkeypatch.setenv("GITHUB_ACTIONS", "true")
     monkeypatch.setenv("GITHUB_REPOSITORY", "foo/bar")
+    monkeypatch.setenv("GITHUB_REF", "main")
     pytester.makepyfile(
         """
         def test_pass():
@@ -180,12 +181,14 @@ def test_errors_logs_403(
     monkeypatch.setenv("GITHUB_ACTIONS", "true")
     monkeypatch.setenv("GITHUB_REPOSITORY", "foo/bar")
     monkeypatch.setenv("MERGIFY_API_URL", http_server)
+    monkeypatch.setenv("GITHUB_BASE_REF", "main")
     pytester.makepyfile(
         """
         def test_pass():
             pass
         """
     )
+
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(passed=1)
     assert any(