feat(ci-insights): Detect newly added tests (#178)

Fixes: MRGFY-5824

Author: remyduthu

pytest_mergify/__init__.py

@@ -102,6 +102,9 @@ def _report_flaky_detection(
         self,
         terminalreporter: _pytest.terminal.TerminalReporter,
     ) -> None:
+        if not self.mergify_ci.is_flaky_detection_enabled():
+            return
+
         if self.mergify_ci.flaky_detection_error_message:
             terminalreporter.write_line(
                 f"Unable to perform flaky detection. Error: {self.mergify_ci.flaky_detection_error_message}",
@@ -110,11 +113,20 @@ def _report_flaky_detection(
 
             return
 
-        # NOTE(remyduthu): This is a temporary log. These names form the
-        # baseline to later detect newly added tests.
+        # NOTE(remyduthu):
+        # The following logs are temporary.
+        # We'll automatically retry newly added tests to detect flakiness.
         terminalreporter.write_line(
-            f"Found {len(self.mergify_ci.existing_test_names)} existing tests",
+            f"Fetched {len(self.mergify_ci.existing_test_names)} existing tests",
         )
+        terminalreporter.write_line(
+            f"Detected {len(self.mergify_ci.new_test_durations_by_name)} new tests"
+        )
+        for (
+            test_name,
+            test_duration_ms,
+        ) in self.mergify_ci.new_test_durations_by_name.items():
+            terminalreporter.write_line(f"  - {test_name} ({test_duration_ms}ms)")
 
     @property
     def tracer(self) -> typing.Optional[opentelemetry.trace.Tracer]:
@@ -252,6 +264,27 @@ def pytest_exception_interact(
                 )
             )
 
+    def _handle_flaky_detection_for_report(
+        self,
+        report: _pytest.reports.TestReport,
+    ) -> None:
+        if not self.mergify_ci.is_flaky_detection_active():
+            return
+
+        test_duration_ms = int(report.duration * 1000)
+        self.mergify_ci.total_test_durations_ms += test_duration_ms
+
+        test_name = report.nodeid
+        if test_name in self.mergify_ci.existing_test_names:
+            return
+
+        self.mergify_ci.add_new_test_duration(test_name, test_duration_ms)
+
+        if self.tracer:
+            opentelemetry.trace.get_current_span().set_attributes(
+                {"cicd.test.new": True}
+            )
+
     def pytest_runtest_logreport(self, report: _pytest.reports.TestReport) -> None:
         if self.tracer is None:
             return
@@ -278,6 +311,8 @@ def pytest_runtest_logreport(self, report: _pytest.reports.TestReport) -> None:
             }
         )
 
+        self._handle_flaky_detection_for_report(report)
+
 
 def pytest_addoption(parser: _pytest.config.argparsing.Parser) -> None:
     group = parser.getgroup("pytest-mergify", "Mergify support for pytest")

pytest_mergify/ci_insights.py

@@ -87,6 +87,10 @@ class MergifyCIInsights:
         init=False,
         default=None,
     )
+    total_test_durations_ms: int = dataclasses.field(init=False, default=0)
+    new_test_durations_by_name: typing.Dict[str, int] = dataclasses.field(
+        init=False, default_factory=dict
+    )
     quarantined_tests: typing.Optional[pytest_mergify.quarantine.Quarantine] = (
         dataclasses.field(
             init=False,
@@ -168,9 +172,24 @@ def __post_init__(self) -> None:
                 self.branch_name,
             )
 
-    def _load_flaky_detection(self) -> None:
+    def add_new_test_duration(self, test_name: str, test_duration_ms: int) -> None:
+        if test_name in self.new_test_durations_by_name:
+            return
+
+        self.new_test_durations_by_name[test_name] = test_duration_ms
+
+    def is_flaky_detection_enabled(self) -> bool:
         # NOTE(remyduthu): Hide behind a feature flag for now.
-        if not utils.is_env_truthy("_MERGIFY_TEST_NEW_FLAKY_DETECTION"):
+        return utils.is_env_truthy("_MERGIFY_TEST_NEW_FLAKY_DETECTION")
+
+    def is_flaky_detection_active(self) -> bool:
+        return (
+            self.is_flaky_detection_enabled()
+            and self.flaky_detection_error_message is None
+        )
+
+    def _load_flaky_detection(self) -> None:
+        if not self.is_flaky_detection_enabled():
             return
 
         try:

tests/test_ci_insights.py

@@ -5,6 +5,8 @@
 
 from pytest_mergify import ci_insights
 
+from . import conftest
+
 
 def _set_test_environment(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("_MERGIFY_TEST_NEW_FLAKY_DETECTION", "true")
@@ -66,3 +68,50 @@ def test_load_flaky_detection_error(monkeypatch: pytest.MonkeyPatch) -> None:
     assert not client.existing_test_names
     assert client.flaky_detection_error_message is not None
     assert "500 Server Error" in client.flaky_detection_error_message
+
+
+@responses.activate
+def test_flaky_detection_detects_new_tests(
+    monkeypatch: pytest.MonkeyPatch,
+    pytester_with_spans: conftest.PytesterWithSpanT,
+) -> None:
+    _set_test_environment(monkeypatch)
+    _make_quarantine_mock()
+    _make_test_names_mock(
+        [
+            "test_flaky_detection_detects_new_tests.py::test_foo",
+            "test_flaky_detection_detects_new_tests.py::test_unknown",
+        ]
+    )
+
+    result, spans = pytester_with_spans(
+        code="""
+        def test_foo():
+            assert True
+
+        def test_bar():
+            assert True
+        
+        def test_baz():
+            assert True
+        """
+    )
+    result.assert_outcomes(passed=3)
+
+    assert """Fetched 2 existing tests
+Detected 2 new tests
+  - test_flaky_detection_detects_new_tests.py::test_bar (0ms)
+  - test_flaky_detection_detects_new_tests.py::test_baz (0ms)""" in result.stdout.str()
+
+    assert spans is not None
+    for test_name, expected in {
+        "test_flaky_detection_detects_new_tests.py::test_foo": False,
+        "test_flaky_detection_detects_new_tests.py::test_bar": True,
+        "test_flaky_detection_detects_new_tests.py::test_baz": True,
+    }.items():
+        span = spans.get(test_name)
+
+        assert span is not None
+        assert span.attributes is not None
+        if expected:
+            assert span.attributes.get("cicd.test.new")

tests/test_plugin.py

@@ -1,6 +1,5 @@
 import _pytest.config
 import pytest
-import responses
 from _pytest.pytester import Pytester
 
 import pytest_mergify
@@ -202,39 +201,3 @@ def test_pass():
         line.startswith("::notice title=Mergify CI::MERGIFY_TEST_RUN_ID=")
         for line in result.stdout.lines
     )
-
-
-@responses.activate
-def test_summary_shows_flaky_test_detection(
-    monkeypatch: pytest.MonkeyPatch,
-    pytester: Pytester,
-) -> None:
-    monkeypatch.setenv("_MERGIFY_TEST_NEW_FLAKY_DETECTION", "true")
-    monkeypatch.setenv("_PYTEST_MERGIFY_TEST", "true")
-    monkeypatch.setenv("CI", "true")
-    monkeypatch.setenv("GITHUB_ACTIONS", "true")
-    monkeypatch.setenv("GITHUB_BASE_REF", "main")
-    monkeypatch.setenv("GITHUB_REPOSITORY", "Mergifyio/pytest-mergify")
-    monkeypatch.setenv("MERGIFY_API_URL", "https://example.com")
-    monkeypatch.setenv("MERGIFY_TOKEN", "my_token")
-
-    responses.add(
-        method=responses.GET,
-        url="https://example.com/v1/ci/Mergifyio/tests/names",
-        match=[
-            responses.matchers.query_param_matcher(
-                {"repository": "pytest-mergify", "branch": "main"}
-            )
-        ],
-        json={"test_names": ["x::test_a", "x::test_b"]},
-        status=200,
-    )
-
-    pytester.makepyfile(
-        """
-        def test_foo():
-            assert True
-        """
-    )
-    result = pytester.runpytest_inprocess()
-    assert "Found 2 existing tests" in result.stdout.str()