wip

Author: gnufede

ddtrace/contrib/internal/coverage/patch.py

@@ -52,6 +52,10 @@ def unpatch():
     coverage._datadog_patch = False
 
 
+def _is_coverage_available():
+    return coverage is not None
+
+
 def coverage_report_wrapper(func: Any, instance: Any, args: tuple, kwargs: dict) -> Any:
     """Wrapper to cache percentage when report() is called."""
     global _cached_coverage_percentage

ddtrace/contrib/internal/coverage/utils.py

@@ -1,11 +1,19 @@
+from pathlib import Path
 import sys
+import tempfile
+from typing import Callable
 from typing import List
+from typing import Optional
 
 from ddtrace.contrib.internal.coverage.data import _original_sys_argv_command
+from ddtrace.internal.logger import get_logger
 from ddtrace.internal.settings._config import _get_config
 from ddtrace.internal.utils.formats import asbool
 
 
+log = get_logger(__name__)
+
+
 def is_coverage_loaded() -> bool:
     return "coverage" in sys.modules
 
@@ -25,3 +33,86 @@ def _is_coverage_invoked_by_coverage_run() -> bool:
     if _get_config("COVERAGE_RUN", False, asbool):
         return True
     return _command_invokes_coverage_run(_original_sys_argv_command)
+
+
+def handle_coverage_report(
+    session,
+    upload_func: Callable[[bytes, str], bool],
+    is_pytest_cov_enabled_func: Callable,
+    stop_coverage_func: Optional[Callable] = None,
+) -> None:
+    """
+    Shared coverage report upload handling for pytest plugins.
+
+    Args:
+        session: pytest session object
+        upload_func: Function to call for uploading (signature: upload_func(bytes, format) -> bool)
+        is_pytest_cov_enabled_func: Function to check if pytest-cov is enabled
+        stop_coverage_func: Optional function to stop coverage collection
+    """
+    try:
+        from ddtrace.contrib.internal.coverage.patch import generate_lcov_report
+        from ddtrace.contrib.internal.coverage.patch import is_coverage_running
+        from ddtrace.contrib.internal.coverage.patch import set_coverage_instance
+
+        log.debug("Coverage report upload is enabled, checking for coverage data")
+
+        # If pytest-cov is enabled but coverage detection fails, register the pytest-cov instance
+        if is_pytest_cov_enabled_func(session.config) and not is_coverage_running():
+            # Try to get coverage from pytest-cov plugin and register it
+            for plugin in session.config.pluginmanager.list_name_plugin():
+                _, plugin_instance = plugin
+                if hasattr(plugin_instance, "cov_controller") and plugin_instance.cov_controller:
+                    set_coverage_instance(plugin_instance.cov_controller.cov)
+                    log.debug("Registered pytest-cov coverage instance with ddtrace")
+                    break
+
+        # Now check if coverage is available (either ddtrace started or pytest-cov registered)
+        if is_coverage_running():
+            try:
+                coverage_format = "lcov"  # Default to LCOV
+                # Generate the report in a temporary file
+                with tempfile.NamedTemporaryFile(mode="wb", suffix=".lcov", delete=False) as tmp_file:
+                    tmp_path = Path(tmp_file.name)
+
+                # Generate LCOV report. This returns the percentage and also stores it
+                # in _coverage_data, so we don't need to generate a second report just for the percentage.
+                pct_covered = generate_lcov_report(outfile=str(tmp_path))
+                log.debug("Generated LCOV coverage report: %s (%.1f%% coverage)", tmp_path, pct_covered)
+
+                # Read the report file
+                coverage_report_bytes = tmp_path.read_bytes()
+                log.debug("Read coverage report: %d bytes", len(coverage_report_bytes))
+
+                # Upload the report using provided upload function
+                upload_success = upload_func(coverage_report_bytes, coverage_format)
+
+                if upload_success:
+                    log.info("Successfully uploaded coverage report")
+                else:
+                    log.warning("Failed to upload coverage report")
+
+                # Clean up temporary file
+                try:
+                    tmp_path.unlink()
+                except Exception as e:
+                    log.debug("Failed to clean up temporary coverage report file: %s", e)
+
+                # Stop coverage AFTER generating and uploading the report (if we started it ourselves)
+                if stop_coverage_func and not is_pytest_cov_enabled_func(session.config):
+                    log.debug("Stopping coverage.py collection")
+                    stop_coverage_func(save=True)
+
+            except Exception as e:
+                log.exception("Error generating or uploading coverage report: %s", e)
+                # Still try to stop coverage even if report generation failed
+                if stop_coverage_func and not is_pytest_cov_enabled_func(session.config):
+                    try:
+                        stop_coverage_func(save=True)
+                    except Exception:
+                        log.debug("Could not stop coverage after error", exc_info=True)
+        else:
+            log.debug("Coverage is not running, skipping coverage report upload")
+
+    except Exception as e:
+        log.exception("Error in coverage report upload handling: %s", e)

ddtrace/contrib/internal/pytest/_plugin_v2.py

@@ -7,9 +7,11 @@
 
 from ddtrace import DDTraceDeprecationWarning
 from ddtrace import config as dd_config
+from ddtrace.contrib.internal.coverage.patch import _is_coverage_available
 from ddtrace.contrib.internal.coverage.patch import get_coverage_percentage
 from ddtrace.contrib.internal.coverage.patch import patch as patch_coverage
 from ddtrace.contrib.internal.coverage.patch import run_coverage_report
+from ddtrace.contrib.internal.coverage.patch import start_coverage
 from ddtrace.contrib.internal.coverage.utils import _is_coverage_invoked_by_coverage_run
 from ddtrace.contrib.internal.coverage.utils import _is_coverage_patched
 from ddtrace.contrib.internal.pytest._benchmark_utils import _set_benchmark_data_from_item
@@ -171,6 +173,30 @@ def _handle_itr_xdist_skipped_suite(item, suite_id) -> bool:
     return True
 
 
+def _is_coverage_report_upload_enabled() -> bool:
+    """
+    Check if coverage report upload is enabled, following V3 pattern:
+    1. First, get setting from API (via CI visibility service)
+    2. Then, allow environment variable to override
+
+    This should only be called after pytest_sessionstart when settings are available.
+    """
+    # Get API setting (should be available after InternalTestSession.discover())
+    coverage_report_upload_enabled = False
+    try:
+        ci_visibility_service = require_ci_visibility_service()
+        if hasattr(ci_visibility_service, "_api_settings") and ci_visibility_service._api_settings:
+            coverage_report_upload_enabled = ci_visibility_service._api_settings.coverage_report_upload_enabled
+    except Exception:
+        log.debug("Unable to check if coverage report upload is enabled from settings", exc_info=True)
+
+    # Allow environment variable to override (same pattern as V3)
+    if asbool(os.getenv("DD_CIVISIBILITY_CODE_COVERAGE_REPORT_UPLOAD_ENABLED", "false")):
+        coverage_report_upload_enabled = True
+
+    return coverage_report_upload_enabled
+
+
 def _handle_test_management(item, test_id):
     """Add a user property to identify quarantined tests, and mark them for skipping if quarantine is enabled in
     skipping mode.
@@ -381,6 +407,32 @@ def _is_pytest_cov_enabled(config) -> bool:
     return cov_option
 
 
+def _handle_coverage_patch_early(config):
+    """
+    Handle coverage patching in pytest_configure (must be early for pytest-cov).
+    Coverage start decision is deferred to pytest_sessionstart when API settings are available.
+    """
+    pytest_cov_enabled = _is_pytest_cov_enabled(config)
+
+    # Check environment variable (API settings not available yet)
+    env_coverage_upload = asbool(os.getenv("DD_CIVISIBILITY_CODE_COVERAGE_REPORT_UPLOAD_ENABLED", "false"))
+
+    # Patch if pytest-cov is enabled OR env var suggests we might need coverage
+    if not (pytest_cov_enabled or env_coverage_upload):
+        return
+
+    # Patch coverage.py early (needed for pytest-cov to work)
+    patch_coverage()
+
+    # Verify patching worked
+    if not _is_coverage_available():
+        log.warning("Coverage requested but coverage.py not available - install with 'pip install coverage'")
+        return
+
+    if pytest_cov_enabled:
+        log.debug("pytest-cov detected, coverage.py patched successfully")
+
+
 def pytest_configure(config: pytest_Config) -> None:
     global skip_pytest_runtest_protocol
 
@@ -399,8 +451,8 @@ def pytest_configure(config: pytest_Config) -> None:
         if is_enabled(config):
             unpatch_unittest()
             enable_test_visibility(config=dd_config.pytest)
-            if _is_pytest_cov_enabled(config):
-                patch_coverage()
+
+            _handle_coverage_patch_early(config)
 
             skip_pytest_runtest_protocol = False
 
@@ -475,6 +527,19 @@ def pytest_sessionstart(session: pytest.Session) -> None:
 
         InternalTestSession.set_library_capabilities(library_capabilities)
 
+        # Start coverage if needed (API settings are now available after discover())
+        if (
+            not _is_pytest_cov_enabled(session.config)
+            and _is_coverage_available()
+            and _is_coverage_report_upload_enabled()
+        ):
+            workspace_path = InternalTestSession.get_workspace_path()
+            if workspace_path:
+                start_coverage(source=[str(workspace_path)])
+                log.debug("Started coverage.py for report upload")
+            else:
+                log.warning("Coverage report upload enabled but workspace path not available")
+
         extracted_context = None
         distributed_children = False
         if hasattr(session.config, "workerinput"):
@@ -755,6 +820,33 @@ def _pytest_run_one_test(item, nextitem):
     item.ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)
 
 
+def _handle_coverage_report_upload(session: pytest.Session) -> None:
+    """
+    Handle coverage report upload if enabled using shared implementation.
+    """
+    from ddtrace.contrib.internal.coverage.utils import handle_coverage_report
+
+    def upload_func(coverage_report_bytes: bytes, coverage_format: str) -> bool:
+        """Upload coverage report using native V2 writer/recorder infrastructure."""
+        try:
+            # Get the CI visibility service (recorder)
+            ci_visibility_service = require_ci_visibility_service()
+
+            # Use the recorder's native upload method which integrates with writer infrastructure
+            return ci_visibility_service.upload_coverage_report(coverage_report_bytes, coverage_format)
+
+        except Exception as e:
+            log.exception("Error in native coverage upload: %s", e)
+            return False
+
+    handle_coverage_report(
+        session=session,
+        upload_func=upload_func,
+        is_pytest_cov_enabled_func=_is_pytest_cov_enabled,
+        stop_coverage_func=None,  # V2 plugin doesn't need to stop coverage manually
+    )
+
+
 def _process_reports_dict(item, reports) -> _TestOutcome:
     final_outcome = None
 
@@ -963,6 +1055,8 @@ def _pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
 
     invoked_by_coverage_run_status = _is_coverage_invoked_by_coverage_run()
     pytest_cov_status = _is_pytest_cov_enabled(session.config)
+    coverage_report_upload_enabled = _is_coverage_report_upload_enabled()
+
     if _is_coverage_patched() and (pytest_cov_status or invoked_by_coverage_run_status):
         if invoked_by_coverage_run_status and not pytest_cov_status:
             run_coverage_report()
@@ -981,6 +1075,10 @@ def _pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
         else:
             InternalTestSession.set_covered_lines_pct(lines_pct_value)
 
+    # Handle coverage report upload if enabled
+    if coverage_report_upload_enabled:
+        _handle_coverage_report_upload(session)
+
     if ModuleCodeCollector.is_installed():
         ModuleCodeCollector.uninstall()