fix(ci_visibility): honor attempt-to-fix failures (#17784)

## Description

This fixes a regression in the new `ddtrace.testing` pytest plugin where tests marked as Attempt to Fix could still have failures hidden when combined with test management states such as quarantine or disable.

Attempt to Fix now takes precedence over quarantine/disable handling in the new plugin, so failing attempts are reported as pytest failures instead of being masked. The final Attempt to Fix status is also computed as passing only when all attempts pass.

## Testing

Unit tests

## Risks

Low. The change is scoped to the new `ddtrace.testing` pytest plugin path and updates the behavior to expose failures that were previously hidden.

## Additional Notes

This is the first PR in a two-part stack. A follow-up PR will refactor and clean up, but doesn't need to be backported.

Co-authored-by: federico.mon <federico.mon@datadoghq.com>

Author: gnufede

ddtrace/testing/internal/pytest/plugin.py

@@ -459,17 +459,18 @@ def _apply_test_management_markers(self, item: pytest.Item, test: "Test") -> Non
         """Apply test management markers for the base plugin (used when an external rerun plugin drives execution).
 
         ATF retries are not supported in this mode — the external plugin controls the protocol and we cannot intercept
-        individual test runs to capture real FAIL statuses. Disabled+ATF tests therefore get the same xfail treatment
-        as quarantined tests: failures won't break the pipeline, but ATF retry semantics are silently unavailable.
-        If ATF support is needed, disable the external rerun plugin (see the warning emitted at configure time).
+        individual test runs to capture real FAIL statuses. ATF therefore takes precedence over quarantine/disable
+        markers so failures are still reported by pytest.
 
         Overridden in TestOptPluginWithProtocol, which drives retries itself and needs real FAIL outcomes for ATF.
         """
         if not self.manager.settings.test_management.enabled:
             return
         if test.is_disabled() and not test.is_attempt_to_fix():
             item.add_marker(pytest.mark.skip(reason=DISABLED_BY_TEST_MANAGEMENT_REASON))
-        elif test.is_quarantined() or (test.is_disabled() and test.is_attempt_to_fix()):
+        elif test.is_attempt_to_fix():
+            return
+        elif test.is_quarantined():
             # Use xfail so failures don't break the pipeline. Works regardless of who drives test execution.
             item.add_marker(pytest.mark.xfail(strict=False, reason="dd_quarantined", run=True))
 
@@ -882,27 +883,17 @@ class TestOptPluginWithProtocol(TestOptPlugin):
     def _apply_test_management_markers(self, item: pytest.Item, test: "Test") -> None:
         """Apply test management markers for the plugin that drives retries itself.
 
-        ATF tests must NOT use xfail here: xfail converts failures into SKIP outcomes, which prevents
-        AttemptToFixHandler.get_final_status from seeing real FAIL counts and computing correct tags
-        (HAS_FAILED_ALL_RETRIES, ATTEMPT_TO_FIX_PASSED). Instead, ATF tests set a user property so
-        _get_test_outcome captures the true status, and reports are mangled to "skipped" afterwards so
-        failures still don't break the pipeline.
-
-        The outer condition mirrors the original logic: only quarantined tests and disabled+ATF tests
-        are treated as "run but don't fail the pipeline" — plain ATF (neither disabled nor quarantined)
-        runs normally without any report suppression.
+        ATF tests must NOT use skip or xfail here: ATF takes precedence over quarantine/disable markers, and any
+        failed attempt should fail the test from pytest's point of view.
         """
         if not self.manager.settings.test_management.enabled:
             return
         if test.is_disabled() and not test.is_attempt_to_fix():
             item.add_marker(pytest.mark.skip(reason=DISABLED_BY_TEST_MANAGEMENT_REASON))
-        elif test.is_quarantined() or (test.is_disabled() and test.is_attempt_to_fix()):
-            # NOTE: keep is_attempt_to_fix() check inside this branch, not outside — plain ATF tests that are
-            # neither disabled nor quarantined run normally and must not get report mangling or a user property.
-            if test.is_attempt_to_fix():
-                item.user_properties += [("dd_disabled_attempt_to_fix", True)]
-            else:
-                item.add_marker(pytest.mark.xfail(strict=False, reason="dd_quarantined", run=True))
+        elif test.is_attempt_to_fix():
+            return
+        elif test.is_quarantined():
+            item.add_marker(pytest.mark.xfail(strict=False, reason="dd_quarantined", run=True))
 
     @catch_and_log_exceptions()
     def pytest_runtest_protocol(self, item: pytest.Item, nextitem: t.Optional[pytest.Item]) -> bool:

ddtrace/testing/internal/retry_handlers.py

@@ -198,12 +198,12 @@ def get_final_status(self, test: Test) -> TestStatus:
             status_counts[test_run.get_status()] += 1
             total_count += 1
 
-        if status_counts[TestStatus.PASS] > 0:
+        if status_counts[TestStatus.PASS] == total_count:
             final_status = TestStatus.PASS
-        elif status_counts[TestStatus.FAIL] > 0:
-            final_status = TestStatus.FAIL
-        else:
+        elif status_counts[TestStatus.SKIP] == total_count:
             final_status = TestStatus.SKIP
+        else:
+            final_status = TestStatus.FAIL
 
         if status_counts[TestStatus.PASS] == total_count:
             final_tags[TestTag.ATTEMPT_TO_FIX_PASSED] = TAG_TRUE

releasenotes/notes/fix-ci-visibility-atf-failures-ed2860ed6d3b9d83.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    CI Visibility: Fixes an issue where tests marked as attempt-to-fix could have failures hidden when they were also quarantined or disabled.

tests/contrib/pytest/test_pytest_configure_plugin_selection.py

@@ -173,12 +173,10 @@ def test_quarantine_and_disabled_applied_when_test_management_enabled(self, is_q
         "is_quarantined,is_disabled,is_attempt_to_fix",
         [
             (True, False, False),  # quarantined
-            (False, True, True),  # disabled + ATF (ATF retries unavailable, treated same as quarantine)
-            (True, False, True),  # quarantined + ATF (same: ATF retries unavailable in base class)
         ],
     )
-    def test_base_plugin_uses_xfail_for_quarantine_and_atf(self, is_quarantined, is_disabled, is_attempt_to_fix):
-        """Base plugin (no retry control) always uses xfail for quarantine/ATF — ATF retries can't fire anyway."""
+    def test_base_plugin_uses_xfail_for_quarantine(self, is_quarantined, is_disabled, is_attempt_to_fix):
+        """Base plugin (no retry control) uses xfail for quarantined non-ATF tests."""
         plugin = mock.MagicMock()
         plugin.manager.settings.test_management.enabled = True
 
@@ -208,8 +206,10 @@ def test_base_plugin_uses_xfail_for_quarantine_and_atf(self, is_quarantined, is_
             (True, False, True),  # quarantined + ATF
         ],
     )
-    def test_child_plugin_uses_user_property_for_atf(self, is_quarantined, is_disabled, is_attempt_to_fix):
-        """Child plugin (drives retries) uses user property for ATF so real FAIL status is captured."""
+    def test_attempt_to_fix_takes_precedence_over_test_management_markers(
+        self, is_quarantined, is_disabled, is_attempt_to_fix
+    ):
+        """ATF tests are not skipped or xfailed, even when quarantined or disabled."""
         plugin = mock.MagicMock()
         plugin.manager.settings.test_management.enabled = True
 
@@ -221,10 +221,14 @@ def test_child_plugin_uses_user_property_for_atf(self, is_quarantined, is_disabl
         item = mock.MagicMock()
         item.user_properties = []
 
-        TestOptPluginWithProtocol._apply_test_management_markers(plugin, item=item, test=test)
+        for plugin_class in (TestOptPlugin, TestOptPluginWithProtocol):
+            item.add_marker.reset_mock()
+            item.user_properties = []
 
-        item.add_marker.assert_not_called()
-        assert ("dd_disabled_attempt_to_fix", True) in item.user_properties
+            plugin_class._apply_test_management_markers(plugin, item=item, test=test)
+
+            item.add_marker.assert_not_called()
+            assert item.user_properties == []
 
     def test_child_plugin_uses_xfail_for_quarantine_without_atf(self):
         """Child plugin uses xfail for quarantined non-ATF tests (same as base)."""

tests/testing/internal/pytest/test_pytest_atf.py

@@ -3,6 +3,7 @@
 from unittest.mock import patch
 
 from _pytest.pytester import Pytester
+import pytest
 
 from ddtrace.testing.internal.settings_data import TestProperties
 from ddtrace.testing.internal.test_data import ModuleRef
@@ -94,7 +95,9 @@ def test_broken(self):
             setup_standard_mocks(),
         ):
             with EventCapture.capture() as event_capture:
-                pytester.inline_run("--ddtrace", "-v", "-s")
+                result = pytester.inline_run("--ddtrace", "-v", "-s")
+
+        assert result.ret == 1
 
         test_events = list(event_capture.events_by_test_name("TestNotFixed::test_broken"))
         # 1 initial (pass) + 1 retry (fail) = 2 events, NOT 1 + 20 retries
@@ -106,6 +109,7 @@ def test_broken(self):
         assert test_events[1]["content"]["meta"].get("test.status") == "fail"
         assert test_events[1]["content"]["meta"].get("test.is_retry") == "true"
         assert test_events[1]["content"]["meta"].get("test.retry_reason") == "attempt_to_fix"
+        assert test_events[1]["content"]["meta"].get("test.test_management.attempt_to_fix_passed") == "false"
 
     def test_atf_always_failing_exits_after_first(self, pytester: Pytester) -> None:
         """Test that ATF stops retrying an always-failing test after the first run."""
@@ -144,3 +148,46 @@ def test_always_fails():
         # Final ATF tags must still be set even without retries
         assert test_events[0]["content"]["meta"].get("test.test_management.attempt_to_fix_passed") == "false"
         assert test_events[0]["content"]["meta"].get("test.has_failed_all_retries") == "true"
+
+    @pytest.mark.parametrize(
+        "test_properties",
+        [
+            TestProperties(quarantined=True, attempt_to_fix=True),
+            TestProperties(disabled=True, attempt_to_fix=True),
+        ],
+    )
+    def test_atf_failures_are_not_masked_by_quarantine_or_disable(
+        self, pytester: Pytester, test_properties: TestProperties
+    ) -> None:
+        """Test that ATF takes precedence over quarantine/disable and failed attempts fail pytest."""
+        pytester.makepyfile(
+            test_foo="""
+            def test_not_fixed():
+                assert False
+        """
+        )
+
+        test_ref = TestRef(SuiteRef(ModuleRef(""), "test_foo.py"), "test_not_fixed")
+        known_tests: set[TestRef] = {test_ref}
+
+        with (
+            patch(
+                "ddtrace.testing.internal.session_manager.APIClient",
+                return_value=mock_api_client_settings(
+                    test_management_enabled=True,
+                    known_tests_enabled=True,
+                    known_tests=known_tests,
+                    test_management_properties={test_ref: test_properties},
+                ),
+            ),
+            setup_standard_mocks(),
+        ):
+            with EventCapture.capture() as event_capture:
+                result = pytester.inline_run("--ddtrace", "-v", "-s")
+
+        assert result.ret == 1
+
+        test_events = list(event_capture.events_by_test_name("test_not_fixed"))
+        assert len(test_events) == 1
+        assert test_events[0]["content"]["meta"].get("test.status") == "fail"
+        assert test_events[0]["content"]["meta"].get("test.test_management.attempt_to_fix_passed") == "false"