feat(ci_visibility): coverage itr backfill

ddtrace/testing/internal/api_client.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 import gzip
 import json
 import logging
@@ -22,6 +23,7 @@
 from ddtrace.testing.internal.test_data import SuiteRef
 from ddtrace.testing.internal.test_data import TestRef
 from ddtrace.testing.internal.test_data import TestTag
+from ddtrace.testing.internal.tracer_api.coverage import CoverageLines
 
 
 log = logging.getLogger(__name__)
@@ -387,7 +389,9 @@ def send_git_pack_file(self, packfile: Path) -> t.Optional[int]:
 
         return len(content)
 
-    def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Optional[str]]:
+    def get_skippable_tests(
+        self,
+    ) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Optional[str], dict[str, CoverageLines]]:
         telemetry = self.telemetry_api.with_request_metric_names(
             count="itr_skippable_tests.request",
             duration="itr_skippable_tests.request_ms",
@@ -415,7 +419,7 @@ def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Option
             log.warning("Git info not available, cannot get skippable items (missing key: %s)", e)
             telemetry.record_error(ErrorType.UNKNOWN)
             self.configuration_errors[TestTag.LIBRARY_CONFIGURATION_ERROR_SKIPPABLE_TESTS] = "true"
-            return set(), None
+            return set(), None, {}
 
         try:
             result = self.connector.post_json("/api/v2/ci/tests/skippable", request_data, telemetry=telemetry)
@@ -424,13 +428,17 @@ def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Option
         except Exception as e:
             log.warning("Error getting skippable tests from API: %s", e)
             self.configuration_errors[TestTag.LIBRARY_CONFIGURATION_ERROR_SKIPPABLE_TESTS] = "true"
-            return set(), None
+            return set(), None, {}
 
         try:
             skippable_items: set[t.Union[SuiteRef, TestRef]] = set()
 
             for item in result.parsed_response["data"]:
-                if item["type"] in ("test", "suite"):
+                # Tests/suites with _missing_line_code_coverage lack stored line coverage in the
+                # backend, so the meta.coverage bitmap is incomplete for them.  Don't skip these
+                # — let them run so their coverage is collected fresh (matches Java behaviour).
+                has_missing = item["attributes"].get("_missing_line_code_coverage", False)
+                if item["type"] in ("test", "suite") and not has_missing:
                     module_ref = ModuleRef(item["attributes"].get("configurations", {}).get("test.bundle", EMPTY_NAME))
                     suite_ref = SuiteRef(module_ref, item["attributes"].get("suite", EMPTY_NAME))
                     if item["type"] == "suite" and self.itr_skipping_level == ITRSkippingLevel.SUITE:
@@ -441,15 +449,26 @@ def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Option
 
             correlation_id = result.parsed_response["meta"]["correlation_id"]
 
+            covered_files: dict[str, CoverageLines] = {}
+            raw_coverage = result.parsed_response.get("meta", {}).get("coverage", {})
+            for file_path, b64_bitmap in raw_coverage.items():
+                try:
+                    raw_bytes = base64.b64decode(b64_bitmap)
+                    # Normalize path: API omits leading "/", but session coverage keys include it
+                    normalized_path = f"/{file_path}" if not file_path.startswith("/") else file_path
+                    covered_files[normalized_path] = CoverageLines.from_bytearray(bytearray(raw_bytes))
+                except Exception:
+                    log.warning("Failed to decode coverage bitmap for file: %s", file_path)
+
         except Exception as e:
             log.warning("Failed to parse skippable tests data from API: %s", e)
             telemetry.record_error(ErrorType.BAD_JSON)
             self.configuration_errors[TestTag.LIBRARY_CONFIGURATION_ERROR_SKIPPABLE_TESTS] = "true"
-            return set(), None
+            return set(), None, {}
 
         self.telemetry_api.record_skippable_count(count=len(skippable_items), level=self.itr_skipping_level)
 
-        return skippable_items, correlation_id
+        return skippable_items, correlation_id, covered_files
 
     def upload_coverage_report(
         self, coverage_report_bytes: bytes, coverage_format: str, tags: t.Optional[dict[str, str]] = None

ddtrace/testing/internal/cached_file_provider.py

@@ -24,6 +24,7 @@
 from ddtrace.testing.internal.test_data import ModuleRef
 from ddtrace.testing.internal.test_data import SuiteRef
 from ddtrace.testing.internal.test_data import TestRef
+from ddtrace.testing.internal.tracer_api.coverage import CoverageLines
 from ddtrace.testing.internal.utils import asbool
 
 
@@ -48,7 +49,9 @@ def get_test_management_properties(
         self, statuses: t.Optional[tuple[str, ...]] = None
     ) -> dict[TestRef, TestProperties]: ...
 
-    def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Optional[str]]: ...
+    def get_skippable_tests(
+        self,
+    ) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Optional[str], dict[str, CoverageLines]]: ...
 
     def get_known_commits(self, latest_commits: list[str]) -> t.Optional[list[str]]: ...
 
@@ -165,7 +168,9 @@ def get_test_management_properties(
             log.warning("Error parsing cached test management file: %s", e)
             return {}
 
-    def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Optional[str]]:
+    def get_skippable_tests(
+        self,
+    ) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Optional[str], dict[str, CoverageLines]]:
         # WARNING: this guard assumes Bazel ALWAYS sets DD_TEST_OPTIMIZATION_PAYLOADS_IN_FILES
         # together with DD_TEST_OPTIMIZATION_MANIFEST_FILE. If a Bazel workflow ever sets only
         # the manifest file without the payload-files flag, cached skippable decisions would be
@@ -175,10 +180,10 @@ def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Option
         # In Bazel payload-files mode the build system handles test selection;
         # applying cached skippable decisions here would skip tests Bazel expects to run.
         if asbool(env.get(DD_TEST_OPTIMIZATION_PAYLOADS_IN_FILES)):
-            return set(), None
+            return set(), None, {}
         cached = _read_cache_json(self._cache_path("cache/http/skippable_tests.json"))
         if cached is None:
-            return set(), None
+            return set(), None, {}
         try:
             skippable_items: set[t.Union[SuiteRef, TestRef]] = set()
             for item in cached["data"]:
@@ -193,8 +198,8 @@ def get_skippable_tests(self) -> tuple[set[t.Union[SuiteRef, TestRef]], t.Option
             correlation_id = cached.get("meta", {}).get("correlation_id")
         except Exception as e:
             log.warning("Error parsing cached skippable tests file: %s", e)
-            return set(), None
-        return skippable_items, correlation_id
+            return set(), None, {}
+        return skippable_items, correlation_id, {}
 
     # --- no-ops for methods unreachable in manifest mode ---
 

ddtrace/testing/internal/pytest/plugin.py

@@ -51,6 +51,7 @@
 from ddtrace.testing.internal.tracer_api.context import enable_all_ddtrace_integrations
 from ddtrace.testing.internal.tracer_api.context import install_global_trace_filter
 from ddtrace.testing.internal.tracer_api.context import trace_context
+from ddtrace.testing.internal.tracer_api.coverage import CoverageLines
 from ddtrace.testing.internal.tracer_api.coverage import coverage_collection
 from ddtrace.testing.internal.tracer_api.coverage import get_coverage_percentage
 from ddtrace.testing.internal.tracer_api.coverage import install_coverage
@@ -298,6 +299,14 @@ def __init__(self, session_manager: SessionManager) -> None:
         # pytest items whose test exhausted Auto Test Retries and are eligible to be retried out of session.
         self._osr_candidates: list[pytest.Item] = []
 
+        # ITR coverage backfilling: accumulate per-test coverage bitmaps so we can OR-merge
+        # them with meta.coverage from the skippable API at session end.
+        self._session_coverage: dict[str, CoverageLines] = {}
+        # Whether to backfill: set whenever ITR skipping is enabled.  Tests with
+        # _missing_line_code_coverage are excluded from skippable_items (they run fresh),
+        # so their coverage is always collected — no need for a separate flag.
+        self._should_backfill_coverage: bool = self.session.itr_skipping_enabled
+
     def pytest_sessionstart(self, session: pytest.Session) -> None:
         if xdist_worker_input := getattr(session.config, "workerinput", None):
             if session_id := xdist_worker_input.get("dd_session_id"):
@@ -368,6 +377,12 @@ def pytest_sessionfinish(self, session: pytest.Session) -> None:
         if self.is_xdist_worker and hasattr(session.config, "workeroutput"):
             # Propagate number of skipped tests to the main process.
             session.config.workeroutput["tests_skipped_by_itr"] = self.session.tests_skipped_by_itr
+            # Propagate session coverage bitmaps for backfill OR-merge on the main process.
+            # execnet serializes bytes but not bytearray; to_bytes() returns bytes.
+            if self._should_backfill_coverage and self._session_coverage:
+                session.config.workeroutput["dd_session_coverage"] = {
+                    path: cl.to_bytes() for path, cl in self._session_coverage.items()
+                }
 
         # If coverage report upload is enabled, generate and upload the report.
         # NOTE: Skip in payload-files mode (Bazel): coverage data is already
@@ -394,6 +409,24 @@ def upload_func(coverage_report_bytes: bytes, coverage_format: str) -> bool:
             # Clean up external coverage instance registration
             clear_coverage_instance()
 
+        if self._should_backfill_coverage:
+            # OR-merge the session's live coverage with meta.coverage from the skippable API
+            # (the union of all skipped tests' line coverage) and recompute the percentage.
+            merged = dict(self._session_coverage)
+            for path, skipped_cl in self.manager.itr_covered_files.items():
+                if path in merged:
+                    merged[path].update(skipped_cl)
+                else:
+                    merged[path] = CoverageLines.from_bytearray(bytearray(skipped_cl.to_bytes()))
+
+            if merged:
+                total_bits = sum(len(cl.to_bytes()) * 8 for cl in merged.values())
+                set_bits = sum(bin(b).count("1") for cl in merged.values() for b in cl.to_bytes())
+                if total_bits > 0:
+                    self.session.metrics[TestTag.CODE_COVERAGE_LINES_PCT] = set_bits / total_bits * 100.0
+
+            self.session.tags[TestTag.CODE_COVERAGE_BACKFILLED] = "true"
+
         self.session.finish()
 
         TelemetryAPI.get().record_session_finished(
@@ -571,19 +604,31 @@ def pytest_runtest_protocol_wrapper(
 
         test.finish()
 
-        self.manager.coverage_writer.put_coverage(
-            test.last_test_run, coverage_data.get_coverage_bitmaps(relative_to=self.manager.workspace_path)
-        )
+        # Materialize the generator so we can both accumulate session coverage and write per-test bitmaps.
+        bitmaps = list(coverage_data.get_coverage_bitmaps(relative_to=self.manager.workspace_path))
+
+        for path, bitmap_bytes in bitmaps:
+            cl = CoverageLines.from_bytearray(bytearray(bitmap_bytes))
+            if path in self._session_coverage:
+                self._session_coverage[path].update(cl)
+            else:
+                self._session_coverage[path] = cl
+
+        self.manager.coverage_writer.put_coverage(test.last_test_run, bitmaps)
 
         if not next_test_ref or test_ref.suite != next_test_ref.suite:
             self.manager._set_suite_source_location(test_suite)
             if codeowners := test.tags.get(TestTag.CODEOWNERS):
                 test_suite.tags[TestTag.CODEOWNERS] = codeowners
+            if self._should_backfill_coverage:
+                test_suite.tags[TestTag.CODE_COVERAGE_BACKFILLED] = "true"
             test_suite.finish()
             self.manager.writer.put_item(test_suite)
             TelemetryAPI.get().record_suite_finished(test_framework=TEST_FRAMEWORK)
 
         if not next_test_ref or test_ref.suite.module != next_test_ref.suite.module:
+            if self._should_backfill_coverage:
+                test_module.tags[TestTag.CODE_COVERAGE_BACKFILLED] = "true"
             test_module.finish()
             self.manager.writer.put_item(test_module)
             TelemetryAPI.get().record_module_finished(test_framework=TEST_FRAMEWORK)
@@ -1155,6 +1200,7 @@ def get_extra_failed_report(self, test: Test, final_status: TestStatus) -> t.Opt
 class XdistTestOptPlugin:
     def __init__(self, main_plugin: TestOptPlugin) -> None:
         self.main_plugin = main_plugin
+        self._workers_session_coverage: dict[str, CoverageLines] = {}
 
     @pytest.hookimpl
     def pytest_configure_node(self, node: t.Any) -> None:
@@ -1166,14 +1212,34 @@ def pytest_configure_node(self, node: t.Any) -> None:
     @pytest.hookimpl
     def pytest_testnodedown(self, node: t.Any, error: t.Any) -> None:
         """
-        Collect count of tests skipped by ITR from a worker node and add it to the main process' session.
+        Collect per-worker ITR skip counts and session coverage bitmaps.
         """
         if not hasattr(node, "workeroutput"):
             return
 
         if tests_skipped_by_itr := node.workeroutput.get("tests_skipped_by_itr"):
             self.main_plugin.session.tests_skipped_by_itr += tests_skipped_by_itr
 
+        # Merge this worker's session coverage for backfill OR-merge.
+        worker_coverage = node.workeroutput.get("dd_session_coverage")
+        if worker_coverage:
+            for path, bitmap_bytes in worker_coverage.items():
+                cl = CoverageLines.from_bytearray(bytearray(bitmap_bytes))
+                if path in self._workers_session_coverage:
+                    self._workers_session_coverage[path].update(cl)
+                else:
+                    self._workers_session_coverage[path] = cl
+
+    @pytest.hookimpl(tryfirst=True)
+    def pytest_sessionfinish(self, session: pytest.Session) -> None:
+        """Transfer accumulated worker coverage to the main plugin before it processes session finish."""
+        if self._workers_session_coverage:
+            for path, cl in self._workers_session_coverage.items():
+                if path in self.main_plugin._session_coverage:
+                    self.main_plugin._session_coverage[path].update(cl)
+                else:
+                    self.main_plugin._session_coverage[path] = cl
+
 
 def _make_reports_dict(reports: list[pytest.TestReport]) -> _ReportGroup:
     return {report.when: report for report in reports}

ddtrace/testing/internal/session_manager.py

@@ -44,6 +44,7 @@
 from ddtrace.testing.internal.test_data import TestSuite
 from ddtrace.testing.internal.test_data import TestTag
 from ddtrace.testing.internal.tracer_api import Codeowners
+from ddtrace.testing.internal.tracer_api.coverage import CoverageLines
 from ddtrace.testing.internal.utils import asbool
 from ddtrace.testing.internal.writer import PayloadFileCoverageWriter
 from ddtrace.testing.internal.writer import PayloadFileTestOptWriter
@@ -104,6 +105,7 @@ def __init__(self, session: TestSession) -> None:
         self.collected_tests: set[TestRef] = set()
         self.skippable_items: set[t.Union[SuiteRef, TestRef]] = set()
         self.itr_correlation_id: t.Optional[str] = None
+        self.itr_covered_files: dict[str, CoverageLines] = {}
         self.itr_skipping_level = ITRSkippingLevel.TEST  # TODO: SUITE level not supported at the moment.
 
         self.is_user_provided_service: bool
@@ -166,7 +168,11 @@ def __init__(self, session: TestSession) -> None:
 
         self.upload_git_data()
         if self.settings.itr_enabled:
-            self.skippable_items, self.itr_correlation_id = self.api_client.get_skippable_tests()
+            (
+                self.skippable_items,
+                self.itr_correlation_id,
+                self.itr_covered_files,
+            ) = self.api_client.get_skippable_tests()
         else:
             self.skippable_items = set()
             self.itr_correlation_id = None

ddtrace/testing/internal/test_data.py

@@ -479,6 +479,7 @@ class TestTag:
     BROWSER_DRIVER = "test.browser.driver"
 
     CODE_COVERAGE_LINES_PCT = "test.code_coverage.lines_pct"
+    CODE_COVERAGE_BACKFILLED = "test.code_coverage.backfilled"
 
     # Library configuration error tags — set when backend requests fail.
     LIBRARY_CONFIGURATION_ERROR_SETTINGS = "_dd.ci.library_configuration_error.settings"

ddtrace/testing/internal/tracer_api/coverage.py

@@ -1,8 +1,12 @@
 """
 API for code coverage collection for use by ddtrace.testing.
 
-The rest of ddtrace.testing should only use the interface exposed in this file to set up code coverage and get
-coverage data.
+The rest of ddtrace.testing should only use the interface exposed in this file to set up code
+coverage and get coverage data.
+
+Exports: CoverageData, CoverageLines, coverage_collection, get_coverage_percentage,
+         install_coverage, install_coverage_percentage, uninstall_coverage_percentage,
+         clear_coverage_instance, stop_coverage.
 """
 
 import contextlib
@@ -18,7 +22,7 @@
 from ddtrace.contrib.internal.coverage.utils import _is_coverage_patched
 from ddtrace.internal.coverage.code import ModuleCodeCollector
 import ddtrace.internal.coverage.installer
-from ddtrace.internal.test_visibility.coverage_lines import CoverageLines
+from ddtrace.internal.test_visibility.coverage_lines import CoverageLines as CoverageLines
 from ddtrace.testing.internal.logging import catch_and_log_exceptions
 
 

pyproject.toml

@@ -9,7 +9,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ddtrace"
-version = "4.12.0rc1"
+version = "4.12.0rc2"
 description = "Datadog APM client library"
 readme = "README.md"
 license = { text = "LICENSE.BSD3" }