fix(ci-visibility): fix ITR test skipping [backport 1.17] (#6499)

Backport
[ddf8d41](ddf8d41)
from #6306 to 1.17.

CI Visibility:

- Fix for test or suite skipping (controlled by env var to use the
latter)
- Create spans for ITR-skipped items (as agreed at ITR skipped
tests/suites RFC)

## Checklist
- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Romain Komorn <[email protected]>

Author: gnufede

ddtrace/contrib/pytest/plugin.py

@@ -29,22 +29,26 @@
 from ddtrace.ext import SpanTypes
 from ddtrace.ext import test
 from ddtrace.internal.ci_visibility import CIVisibility as _CIVisibility
+from ddtrace.internal.ci_visibility.constants import COVERAGE_TAG_NAME
 from ddtrace.internal.ci_visibility.constants import EVENT_TYPE as _EVENT_TYPE
 from ddtrace.internal.ci_visibility.constants import MODULE_ID as _MODULE_ID
 from ddtrace.internal.ci_visibility.constants import MODULE_TYPE as _MODULE_TYPE
 from ddtrace.internal.ci_visibility.constants import SESSION_ID as _SESSION_ID
 from ddtrace.internal.ci_visibility.constants import SESSION_TYPE as _SESSION_TYPE
+from ddtrace.internal.ci_visibility.constants import SUITE
 from ddtrace.internal.ci_visibility.constants import SUITE_ID as _SUITE_ID
 from ddtrace.internal.ci_visibility.constants import SUITE_TYPE as _SUITE_TYPE
-from ddtrace.internal.ci_visibility.coverage import _coverage_end
-from ddtrace.internal.ci_visibility.coverage import _coverage_start
-from ddtrace.internal.ci_visibility.coverage import _initialize
+from ddtrace.internal.ci_visibility.constants import TEST
+from ddtrace.internal.ci_visibility.coverage import _initialize_coverage
+from ddtrace.internal.ci_visibility.coverage import build_payload as build_coverage_payload
+from ddtrace.internal.ci_visibility.recorder import _get_test_skipping_level
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.logger import get_logger
 
 
 SKIPPED_BY_ITR = "Skipped by Datadog Intelligent Test Runner"
 PATCH_ALL_HELP_MSG = "Call ddtrace.patch_all before running tests."
+
 log = get_logger(__name__)
 
 
@@ -70,6 +74,22 @@ def _store_span(item, span):
     setattr(item, "_datadog_span", span)
 
 
+def _attach_coverage(item):
+    coverage = _initialize_coverage(str(item.config.rootdir))
+    setattr(item, "_coverage", coverage)
+    coverage.start()
+
+
+def _detach_coverage(item, span):
+    if not hasattr(item, "_coverage"):
+        return
+    span_id = str(span.trace_id)
+    item._coverage.stop()
+    span.set_tag(COVERAGE_TAG_NAME, build_coverage_payload(item._coverage, test_id=span_id))
+    item._coverage.erase()
+    del item._coverage
+
+
 def _extract_module_span(item):
     """Extract span from `pytest.Item` instance."""
     return getattr(item, "_datadog_span_module", None)
@@ -201,15 +221,14 @@ def _start_test_module_span(pytest_package_item=None, pytest_module_item=None):
     return test_module_span, is_package
 
 
-def _start_test_suite_span(item, test_module_span):
+def _start_test_suite_span(item, test_module_span, should_enable_coverage=False):
     """
     Starts a test suite span at the start of a new pytest test module.
     Note that ``item`` is a ``pytest.Module`` object referencing the test file being run.
     """
     test_session_span = _extract_span(item.session)
     if test_module_span is None and isinstance(item.parent, pytest.Package):
         test_module_span = _extract_span(item.parent)
-
     parent_span = test_module_span
     if parent_span is None:
         parent_span = test_session_span
@@ -237,6 +256,9 @@ def _start_test_suite_span(item, test_module_span):
         test_suite_span.set_tag_str(test.MODULE_PATH, test_module_path)
     test_suite_span.set_tag_str(test.SUITE, _get_suite_name(item, test_module_path))
     _store_span(item, test_suite_span)
+
+    if should_enable_coverage:
+        _attach_coverage(item)
     return test_suite_span
 
 
@@ -350,7 +372,7 @@ def pytest_collection_modifyitems(session, config, items):
     if _CIVisibility.test_skipping_enabled():
         skip = pytest.mark.skip(reason=SKIPPED_BY_ITR)
         for item in items:
-            if _CIVisibility._instance._should_skip_path(str(get_fslocation_from_item(item)[0])):
+            if _CIVisibility._instance._should_skip_path(str(get_fslocation_from_item(item)[0]), item.name):
                 item.add_marker(skip)
 
 
@@ -366,97 +388,113 @@ def pytest_runtest_protocol(item, nextitem):
         if "reason" in marker.kwargs and marker.kwargs["reason"] == SKIPPED_BY_ITR
     ]
 
-    if is_skipped_by_itr:
-        yield
-    else:
-        test_session_span = _extract_span(item.session)
+    test_session_span = _extract_span(item.session)
 
-        pytest_module_item = _find_pytest_item(item, pytest.Module)
-        pytest_package_item = _find_pytest_item(pytest_module_item, pytest.Package)
+    pytest_module_item = _find_pytest_item(item, pytest.Module)
+    pytest_package_item = _find_pytest_item(pytest_module_item, pytest.Package)
+
+    module_is_package = True
+
+    test_module_span = _extract_span(pytest_package_item)
+    if not test_module_span:
+        test_module_span = _extract_module_span(pytest_module_item)
+        if test_module_span:
+            module_is_package = False
+
+    if test_module_span is None:
+        test_module_span, module_is_package = _start_test_module_span(pytest_package_item, pytest_module_item)
+
+    test_suite_span = _extract_span(pytest_module_item)
+    if pytest_module_item is not None and test_suite_span is None:
+        # Start coverage for the test suite if coverage is enabled
+        test_suite_span = _start_test_suite_span(
+            pytest_module_item,
+            test_module_span,
+            should_enable_coverage=(
+                _get_test_skipping_level() == SUITE
+                and _CIVisibility._instance._collect_coverage_enabled
+                and not is_skipped_by_itr
+            ),
+        )
 
-        module_is_package = True
+    with _CIVisibility._instance.tracer._start_span(
+        ddtrace.config.pytest.operation_name,
+        service=_CIVisibility._instance._service,
+        resource=item.nodeid,
+        span_type=SpanTypes.TEST,
+        activate=True,
+    ) as span:
+        span.set_tag_str(COMPONENT, "pytest")
+        span.set_tag_str(SPAN_KIND, KIND)
+        span.set_tag_str(test.FRAMEWORK, FRAMEWORK)
+        span.set_tag_str(_EVENT_TYPE, SpanTypes.TEST)
+        span.set_tag_str(test.NAME, item.name)
+        span.set_tag_str(test.COMMAND, _get_pytest_command(item.config))
+        span.set_tag_str(_SESSION_ID, str(test_session_span.span_id))
+
+        span.set_tag_str(_MODULE_ID, str(test_module_span.span_id))
+        span.set_tag_str(test.MODULE, test_module_span.get_tag(test.MODULE))
+        span.set_tag_str(test.MODULE_PATH, test_module_span.get_tag(test.MODULE_PATH))
+
+        span.set_tag_str(_SUITE_ID, str(test_suite_span.span_id))
+        test_class_hierarchy = _get_test_class_hierarchy(item)
+        if test_class_hierarchy:
+            span.set_tag_str(test.CLASS_HIERARCHY, test_class_hierarchy)
+        if hasattr(item, "dtest") and isinstance(item.dtest, DocTest):
+            span.set_tag_str(test.SUITE, "{}.py".format(item.dtest.globs["__name__"]))
+        else:
+            span.set_tag_str(test.SUITE, test_suite_span.get_tag(test.SUITE))
 
-        test_module_span = _extract_span(pytest_package_item)
-        if not test_module_span:
-            test_module_span = _extract_module_span(pytest_module_item)
-            if test_module_span:
-                module_is_package = False
+        span.set_tag_str(test.TYPE, SpanTypes.TEST)
+        span.set_tag_str(test.FRAMEWORK_VERSION, pytest.__version__)
 
-        if test_module_span is None:
-            test_module_span, module_is_package = _start_test_module_span(pytest_package_item, pytest_module_item)
+        if item.location and item.location[0]:
+            _CIVisibility.set_codeowners_of(item.location[0], span=span)
 
-        test_suite_span = _extract_span(pytest_module_item)
-        if pytest_module_item is not None and test_suite_span is None:
-            test_suite_span = _start_test_suite_span(pytest_module_item, test_module_span)
-            # Start coverage for the test suite if coverage is enabled
-            if _CIVisibility._instance._collect_coverage_enabled:
-                _initialize(str(item.config.rootdir))
-                _coverage_start()
+        # We preemptively set FAIL as a status, because if pytest_runtest_makereport is not called
+        # (where the actual test status is set), it means there was a pytest error
+        span.set_tag_str(test.STATUS, test.Status.FAIL.value)
 
-        with _CIVisibility._instance.tracer._start_span(
-            ddtrace.config.pytest.operation_name,
-            service=_CIVisibility._instance._service,
-            resource=item.nodeid,
-            span_type=SpanTypes.TEST,
-            activate=True,
-        ) as span:
-            span.set_tag_str(COMPONENT, "pytest")
-            span.set_tag_str(SPAN_KIND, KIND)
-            span.set_tag_str(test.FRAMEWORK, FRAMEWORK)
-            span.set_tag_str(_EVENT_TYPE, SpanTypes.TEST)
-            span.set_tag_str(test.NAME, item.name)
-            span.set_tag_str(test.COMMAND, _get_pytest_command(item.config))
-            span.set_tag_str(_SESSION_ID, str(test_session_span.span_id))
-
-            span.set_tag_str(_MODULE_ID, str(test_module_span.span_id))
-            span.set_tag_str(test.MODULE, test_module_span.get_tag(test.MODULE))
-            span.set_tag_str(test.MODULE_PATH, test_module_span.get_tag(test.MODULE_PATH))
-
-            span.set_tag_str(_SUITE_ID, str(test_suite_span.span_id))
-            test_class_hierarchy = _get_test_class_hierarchy(item)
-            if test_class_hierarchy:
-                span.set_tag_str(test.CLASS_HIERARCHY, test_class_hierarchy)
-            if hasattr(item, "dtest") and isinstance(item.dtest, DocTest):
-                span.set_tag_str(test.SUITE, "{}.py".format(item.dtest.globs["__name__"]))
-            else:
-                span.set_tag_str(test.SUITE, test_suite_span.get_tag(test.SUITE))
-
-            span.set_tag_str(test.TYPE, SpanTypes.TEST)
-            span.set_tag_str(test.FRAMEWORK_VERSION, pytest.__version__)
-
-            if item.location and item.location[0]:
-                _CIVisibility.set_codeowners_of(item.location[0], span=span)
-
-            # We preemptively set FAIL as a status, because if pytest_runtest_makereport is not called
-            # (where the actual test status is set), it means there was a pytest error
-            span.set_tag_str(test.STATUS, test.Status.FAIL.value)
-
-            # Parameterized test cases will have a `callspec` attribute attached to the pytest Item object.
-            # Pytest docs: https://docs.pytest.org/en/6.2.x/reference.html#pytest.Function
-            if getattr(item, "callspec", None):
-                parameters = {"arguments": {}, "metadata": {}}  # type: Dict[str, Dict[str, str]]
-                for param_name, param_val in item.callspec.params.items():
-                    try:
-                        parameters["arguments"][param_name] = encode_test_parameter(param_val)
-                    except Exception:
-                        parameters["arguments"][param_name] = "Could not encode"
-                        log.warning("Failed to encode %r", param_name, exc_info=True)
-                span.set_tag_str(test.PARAMETERS, json.dumps(parameters))
-
-            markers = [marker.kwargs for marker in item.iter_markers(name="dd_tags")]
-            for tags in markers:
-                span.set_tags(tags)
-            _store_span(item, span)
-
-            # Run the actual test
-            yield
+        # Parameterized test cases will have a `callspec` attribute attached to the pytest Item object.
+        # Pytest docs: https://docs.pytest.org/en/6.2.x/reference.html#pytest.Function
+        if getattr(item, "callspec", None):
+            parameters = {"arguments": {}, "metadata": {}}  # type: Dict[str, Dict[str, str]]
+            for param_name, param_val in item.callspec.params.items():
+                try:
+                    parameters["arguments"][param_name] = encode_test_parameter(param_val)
+                except Exception:
+                    parameters["arguments"][param_name] = "Could not encode"
+                    log.warning("Failed to encode %r", param_name, exc_info=True)
+            span.set_tag_str(test.PARAMETERS, json.dumps(parameters))
+
+        markers = [marker.kwargs for marker in item.iter_markers(name="dd_tags")]
+        for tags in markers:
+            span.set_tags(tags)
+        _store_span(item, span)
+
+        coverage_per_test = (
+            _get_test_skipping_level() == TEST
+            and _CIVisibility._instance._collect_coverage_enabled
+            and not is_skipped_by_itr
+        )
+        if coverage_per_test:
+            _attach_coverage(item)
+        # Run the actual test
+        yield
+        # Finish coverage for the test suite if coverage is enabled
+        if coverage_per_test:
+            _detach_coverage(item, span)
 
         nextitem_pytest_module_item = _find_pytest_item(nextitem, pytest.Module)
         if nextitem is None or nextitem_pytest_module_item != pytest_module_item and not test_suite_span.finished:
             _mark_test_status(pytest_module_item, test_suite_span)
             # Finish coverage for the test suite if coverage is enabled
-            if _CIVisibility._instance._collect_coverage_enabled:
-                _coverage_end(test_suite_span)
+            if (
+                _get_test_skipping_level() == SUITE
+                and _CIVisibility._instance._collect_coverage_enabled
+                and not is_skipped_by_itr
+            ):
+                _detach_coverage(pytest_module_item, test_suite_span)
             test_suite_span.finish()
 
             if not module_is_package:

ddtrace/internal/ci_visibility/constants.py

@@ -1,6 +1,9 @@
 from enum import IntEnum
 
 
+SUITE = "suite"
+TEST = "test"
+
 EVENT_TYPE = "type"
 
 

ddtrace/internal/ci_visibility/coverage.py

@@ -1,16 +1,17 @@
 from itertools import groupby
 import json
 import os
-from typing import Dict
-from typing import Iterable
-from typing import List
-from typing import Optional
-from typing import Tuple
+from typing import TYPE_CHECKING
 
 from ddtrace.internal.logger import get_logger
 
-from .constants import COVERAGE_TAG_NAME
 
+if TYPE_CHECKING:  # pragma: no cover
+    from typing import Dict
+    from typing import Iterable
+    from typing import List
+    from typing import Optional
+    from typing import Tuple
 
 log = get_logger(__name__)
 
@@ -24,42 +25,27 @@
     Coverage = None  # type: ignore[misc,assignment]
     EXECUTE_ATTR = ""
 
-COVERAGE_SINGLETON = None
 ROOT_DIR = None
 
 
 def is_coverage_available():
     return Coverage is not None
 
 
-def _initialize(root_dir):
+def _initialize_coverage(root_dir):
     global ROOT_DIR
     if ROOT_DIR is None:
         ROOT_DIR = root_dir
 
-    global COVERAGE_SINGLETON
-    if COVERAGE_SINGLETON is None:
-        coverage_kwargs = {
-            "data_file": None,
-            "source": [root_dir],
-            "config_file": False,
-            "omit": [
-                "*/site-packages/*",
-            ],
-        }
-        COVERAGE_SINGLETON = Coverage(**coverage_kwargs)
-
-
-def _coverage_start():
-    COVERAGE_SINGLETON.start()
-
-
-def _coverage_end(span):
-    span_id = str(span.trace_id)
-    COVERAGE_SINGLETON.stop()
-    span.set_tag(COVERAGE_TAG_NAME, build_payload(COVERAGE_SINGLETON, test_id=span_id))
-    COVERAGE_SINGLETON._collector._clear_data()
-    COVERAGE_SINGLETON._collector.data.clear()
+    coverage_kwargs = {
+        "data_file": None,
+        "source": [ROOT_DIR],
+        "config_file": False,
+        "omit": [
+            "*/site-packages/*",
+        ],
+    }
+    return Coverage(**coverage_kwargs)
 
 
 def segments(lines):