refactor(ci_visibility): remove core usage

ddtrace/contrib/internal/pytest/_plugin_v2.py

@@ -113,6 +113,13 @@ def pytest_configure_node(self, node):
 
         node.workerinput["root_span"] = root_span
 
+    @pytest.hookimpl
+    def pytest_testnodedown(self, node, error):
+        if hasattr(node, "workeroutput") and "itr_skipped_count" in node.workeroutput:
+            if not hasattr(pytest, "global_worker_itr_results"):
+                pytest.global_worker_itr_results = 0
+            pytest.global_worker_itr_results += node.workeroutput["itr_skipped_count"]
+
 
 def _handle_itr_should_skip(item, test_id) -> bool:
     """Checks whether a test should be skipped
@@ -135,6 +142,13 @@ def _handle_itr_should_skip(item, test_id) -> bool:
         InternalTest.mark_itr_skipped(test_id)
         # Marking the test as skipped by ITR so that it appears in pytest's output
         item.add_marker(pytest.mark.skip(reason=SKIPPED_BY_ITR_REASON))  # TODO don't rely on internal for reason
+
+        # If we're in a worker process, count the skipped test
+        if hasattr(item.config, "workeroutput"):
+            if "itr_skipped_count" not in item.config.workeroutput:
+                item.config.workeroutput["itr_skipped_count"] = 0
+            item.config.workeroutput["itr_skipped_count"] += 1
+
         return True
 
     return False
@@ -267,6 +281,9 @@ def pytest_configure(config: pytest_Config) -> None:
 
             if config.pluginmanager.hasplugin("xdist"):
                 config.pluginmanager.register(XdistHooks())
+
+                if not hasattr(config, "workerinput"):  # Main process
+                    pytest.global_worker_itr_results = 0
         else:
             # If the pytest ddtrace plugin is not enabled, we should disable CI Visibility, as it was enabled during
             # pytest_load_initial_conftests
@@ -315,6 +332,7 @@ def pytest_sessionstart(session: pytest.Session) -> None:
         InternalTestSession.set_library_capabilities(library_capabilities)
 
         extracted_context = None
+        distributed_children = False
         if hasattr(session.config, "workerinput"):
             from ddtrace._trace.context import Context
             from ddtrace.constants import USER_KEEP
@@ -332,8 +350,10 @@ def pytest_sessionstart(session: pytest.Session) -> None:
                     "pytest_sessionstart: Could not convert root_span %s to int",
                     received_root_span,
                 )
+        elif hasattr(pytest, "global_worker_itr_results"):
+            distributed_children = True
 
-        InternalTestSession.start(extracted_context)
+        InternalTestSession.start(distributed_children, extracted_context)
 
         if InternalTestSession.efd_enabled() and not _pytest_version_supports_efd():
             log.warning("Early Flake Detection disabled: pytest version is not supported")
@@ -453,7 +473,7 @@ def _pytest_runtest_protocol_post_yield(item, nextitem, coverage_collector):
             InternalTestSuite.mark_itr_skipped(suite_id)
         else:
             _handle_coverage_dependencies(suite_id)
-            InternalTestSuite.finish(suite_id)
+        InternalTestSuite.finish(suite_id)
         if nextitem is None or (next_test_id is not None and next_test_id.parent_id.parent_id != module_id):
             InternalTestModule.finish(module_id)
 
@@ -767,6 +787,14 @@ def _pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
     if ModuleCodeCollector.is_installed():
         ModuleCodeCollector.uninstall()
 
+    # Count ITR skipped tests from workers if we're in the main process
+    if hasattr(pytest, "global_worker_itr_results"):
+        skipped_count = pytest.global_worker_itr_results
+        if skipped_count > 0:
+            # Update the session's internal _itr_skipped_count so that when _set_itr_tags() is called
+            # during session finishing, it will use the correct worker-aggregated count
+            InternalTestSession.set_itr_tags(skipped_count)
+
     InternalTestSession.finish(
         force_finish_children=True,
         override_status=TestStatus.FAIL if session.exitstatus == pytest.ExitCode.TESTS_FAILED else None,

ddtrace/ext/test_visibility/_utils.py

@@ -4,8 +4,6 @@
 from typing import List
 
 from ddtrace.ext.test_visibility._test_visibility_base import TestVisibilityItemId
-from ddtrace.ext.test_visibility._test_visibility_base import _TestVisibilityAPIBase
-from ddtrace.internal import core
 from ddtrace.internal.logger import get_logger
 
 
@@ -30,36 +28,43 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
-def _get_item_tag(item_id: TestVisibilityItemId, tag_name: str) -> Any:
-    log.debug("Getting tag for item %s: %s", item_id, tag_name)
-    tag_value = core.dispatch_with_results(
-        "test_visibility.item.get_tag", (_TestVisibilityAPIBase.GetTagArgs(item_id, tag_name),)
-    ).tag_value.value
-    return tag_value
+def _get_item_tag(item_id: TestVisibilityItemId, name: str) -> Any:
+    # Lazy import to avoid circular dependency
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
 
+    return CIVisibility.get_item_by_id(item_id).get_tag(name)
 
-def _set_item_tag(item_id: TestVisibilityItemId, tag_name: str, tag_value: Any, recurse: bool = False):
-    log.debug("Setting tag for item %s: %s=%s", item_id, tag_name, tag_value)
-    core.dispatch("test_visibility.item.set_tag", (_TestVisibilityAPIBase.SetTagArgs(item_id, tag_name, tag_value),))
 
+def _set_item_tag(item_id: TestVisibilityItemId, name: str, value: Any, recurse: bool = False) -> None:
+    # Lazy import to avoid circular dependency
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
 
-def _set_item_tags(item_id: TestVisibilityItemId, tags: Dict[str, Any], recurse: bool = False):
-    log.debug("Setting tags for item %s: %s", item_id, tags)
-    core.dispatch("test_visibility.item.set_tags", (_TestVisibilityAPIBase.SetTagsArgs(item_id, tags),))
+    CIVisibility.get_item_by_id(item_id).set_tag(name, value)
 
 
-def _delete_item_tag(item_id: TestVisibilityItemId, tag_name: str, recurse: bool = False):
-    log.debug("Deleting tag for item %s: %s", item_id, tag_name)
-    core.dispatch("test_visibility.item.delete_tag", (_TestVisibilityAPIBase.DeleteTagArgs(item_id, tag_name),))
+def _set_item_tags(item_id: TestVisibilityItemId, tags: Dict[str, Any], recurse: bool = False) -> None:
+    # Lazy import to avoid circular dependency
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
 
+    CIVisibility.get_item_by_id(item_id).set_tags(tags)
 
-def _delete_item_tags(item_id: TestVisibilityItemId, tag_names: List[str], recurse: bool = False):
-    log.debug("Deleting tags for item %s: %s", item_id, tag_names)
-    core.dispatch("test_visibility.item.delete_tags", (_TestVisibilityAPIBase.DeleteTagsArgs(item_id, tag_names),))
+
+def _delete_item_tag(item_id: TestVisibilityItemId, name: str, recurse: bool = False) -> None:
+    # Lazy import to avoid circular dependency
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
+
+    CIVisibility.get_item_by_id(item_id).delete_tag(name)
+
+
+def _delete_item_tags(item_id: TestVisibilityItemId, names: List[str], recurse: bool = False) -> None:
+    # Lazy import to avoid circular dependency
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
+
+    CIVisibility.get_item_by_id(item_id).delete_tags(names)
 
 
 def _is_item_finished(item_id: TestVisibilityItemId) -> bool:
-    log.debug("Checking if item %s is finished", item_id)
-    _is_finished = bool(core.dispatch_with_results("test_visibility.item.is_finished", (item_id,)).is_finished.value)
-    log.debug("Item %s is finished: %s", item_id, _is_finished)
-    return _is_finished
+    # Lazy import to avoid circular dependency
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
+
+    return CIVisibility.get_item_by_id(item_id).is_finished()

ddtrace/ext/test_visibility/api.py

@@ -39,7 +39,6 @@
 from ddtrace.ext.test_visibility._utils import _is_item_finished
 from ddtrace.ext.test_visibility._utils import _set_item_tag
 from ddtrace.ext.test_visibility._utils import _set_item_tags
-from ddtrace.internal import core
 from ddtrace.internal.logger import get_logger as _get_logger
 
 
@@ -86,21 +85,27 @@ class TestExcInfo:
 @_catch_and_log_exceptions
 def enable_test_visibility(config: Optional[Any] = None):
     log.debug("Enabling Test Visibility with config: %s", config)
-    core.dispatch("test_visibility.enable", (config,))
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
+
+    CIVisibility.enable(config=config)
 
     if not is_test_visibility_enabled():
         log.warning("Failed to enable Test Visibility")
 
 
 @_catch_and_log_exceptions
 def is_test_visibility_enabled():
-    return core.dispatch_with_results("test_visibility.is_enabled").is_enabled.value
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
+
+    return CIVisibility.enabled
 
 
 @_catch_and_log_exceptions
 def disable_test_visibility():
     log.debug("Disabling Test Visibility")
-    core.dispatch("test_visibility.disable")
+    from ddtrace.internal.ci_visibility.recorder import CIVisibility
+
+    CIVisibility.disable()
     if is_test_visibility_enabled():
         log.warning("Failed to disable Test Visibility")
 
@@ -161,28 +166,29 @@ def discover(
             log.debug("Test Visibility is not enabled, session not registered.")
             return
 
-        core.dispatch(
-            "test_visibility.session.discover",
-            (
-                TestSession.DiscoverArgs(
-                    test_command,
-                    reject_duplicates,
-                    test_framework,
-                    test_framework_version,
-                    session_operation_name,
-                    module_operation_name,
-                    suite_operation_name,
-                    test_operation_name,
-                    root_dir,
-                ),
-            ),
+        from ddtrace.internal.ci_visibility.recorder import on_discover_session
+
+        on_discover_session(
+            TestSession.DiscoverArgs(
+                test_command,
+                reject_duplicates,
+                test_framework,
+                test_framework_version,
+                session_operation_name,
+                module_operation_name,
+                suite_operation_name,
+                test_operation_name,
+                root_dir,
+            )
         )
 
     @staticmethod
     @_catch_and_log_exceptions
-    def start(context: Optional[Context] = None):
+    def start(distributed_children: bool = False, context: Optional[Context] = None):
         log.debug("Starting session")
-        core.dispatch("test_visibility.session.start", (context,))
+        from ddtrace.internal.ci_visibility.recorder import on_start_session
+
+        on_start_session(distributed_children, context)
 
     class FinishArgs(NamedTuple):
         force_finish_children: bool
@@ -196,9 +202,9 @@ def finish(
     ):
         log.debug("Finishing session, force_finish_session_modules: %s", force_finish_children)
 
-        core.dispatch(
-            "test_visibility.session.finish", (TestSession.FinishArgs(force_finish_children, override_status),)
-        )
+        from ddtrace.internal.ci_visibility.recorder import on_finish_session
+
+        on_finish_session(TestSession.FinishArgs(force_finish_children, override_status))
 
     @staticmethod
     def get_tag(tag_name: str) -> Any:
@@ -235,13 +241,17 @@ class FinishArgs(NamedTuple):
     @_catch_and_log_exceptions
     def discover(item_id: TestModuleId, module_path: Optional[Path] = None):
         log.debug("Registered module %s", item_id)
-        core.dispatch("test_visibility.module.discover", (TestModule.DiscoverArgs(item_id, module_path),))
+        from ddtrace.internal.ci_visibility.recorder import on_discover_module
+
+        on_discover_module(TestModule.DiscoverArgs(item_id, module_path))
 
     @staticmethod
     @_catch_and_log_exceptions
     def start(item_id: TestModuleId):
         log.debug("Starting module %s", item_id)
-        core.dispatch("test_visibility.module.start", (item_id,))
+        from ddtrace.internal.ci_visibility.recorder import on_start_module
+
+        on_start_module(item_id)
 
     @staticmethod
     @_catch_and_log_exceptions
@@ -256,9 +266,9 @@ def finish(
             override_status,
             force_finish_children,
         )
-        core.dispatch(
-            "test_visibility.module.finish", (TestModule.FinishArgs(item_id, override_status, force_finish_children),)
-        )
+        from ddtrace.internal.ci_visibility.recorder import on_finish_module
+
+        on_finish_module(TestModule.FinishArgs(item_id, override_status, force_finish_children))
 
 
 class TestSuite(TestBase):
@@ -276,15 +286,17 @@ def discover(
     ):
         """Registers a test suite with the Test Visibility service."""
         log.debug("Registering suite %s, source: %s", item_id, source_file_info)
-        core.dispatch(
-            "test_visibility.suite.discover", (TestSuite.DiscoverArgs(item_id, codeowners, source_file_info),)
-        )
+        from ddtrace.internal.ci_visibility.recorder import on_discover_suite
+
+        on_discover_suite(TestSuite.DiscoverArgs(item_id, codeowners, source_file_info))
 
     @staticmethod
     @_catch_and_log_exceptions
     def start(item_id: TestSuiteId):
         log.debug("Starting suite %s", item_id)
-        core.dispatch("test_visibility.suite.start", (item_id,))
+        from ddtrace.internal.ci_visibility.recorder import on_start_suite
+
+        on_start_suite(item_id)
 
     class FinishArgs(NamedTuple):
         suite_id: TestSuiteId
@@ -304,10 +316,9 @@ def finish(
             force_finish_children,
             override_status,
         )
-        core.dispatch(
-            "test_visibility.suite.finish",
-            (TestSuite.FinishArgs(item_id, force_finish_children, override_status),),
-        )
+        from ddtrace.internal.ci_visibility.recorder import on_finish_suite
+
+        on_finish_suite(TestSuite.FinishArgs(item_id, force_finish_children, override_status))
 
 
 class Test(TestBase):
@@ -333,15 +344,17 @@ def discover(
             source_file_info,
             resource,
         )
-        core.dispatch(
-            "test_visibility.test.discover", (Test.DiscoverArgs(item_id, codeowners, source_file_info, resource),)
-        )
+        from ddtrace.internal.ci_visibility.recorder import on_discover_test
+
+        on_discover_test(Test.DiscoverArgs(item_id, codeowners, source_file_info, resource))
 
     @staticmethod
     @_catch_and_log_exceptions
     def start(item_id: TestId):
         log.debug("Starting test %s", item_id)
-        core.dispatch("test_visibility.test.start", (item_id,))
+        from ddtrace.internal.ci_visibility.recorder import on_start_test
+
+        on_start_test(item_id)
 
     class FinishArgs(NamedTuple):
         test_id: TestId
@@ -364,16 +377,17 @@ def finish(
             skip_reason,
             exc_info,
         )
-        core.dispatch(
-            "test_visibility.test.finish",
-            (Test.FinishArgs(item_id, status, skip_reason=skip_reason, exc_info=exc_info),),
-        )
+        from ddtrace.internal.ci_visibility.recorder import on_finish_test
+
+        on_finish_test(Test.FinishArgs(item_id, status, skip_reason=skip_reason, exc_info=exc_info))
 
     @staticmethod
     @_catch_and_log_exceptions
     def set_parameters(item_id: TestId, params: str):
         log.debug("Setting test %s parameters to %s", item_id, params)
-        core.dispatch("test_visibility.test.set_parameters", (item_id, params))
+        from ddtrace.internal.ci_visibility.recorder import on_set_test_parameters
+
+        on_set_test_parameters(item_id, params)
 
     @staticmethod
     @_catch_and_log_exceptions

ddtrace/internal/ci_visibility/api/_base.py

@@ -549,6 +549,7 @@ def __init__(
     ) -> None:
         super().__init__(name, session_settings, operation_name, initial_tags)
         self._children: Dict[CIDT, CITEMT] = {}
+        self._distributed_children = False
 
     def get_status(self) -> Union[TestStatus, SPECIAL_STATUS]:
         """Recursively computes status based on all children's status
@@ -659,5 +660,8 @@ def _set_itr_tags(self, itr_enabled: bool) -> None:
         self.set_tag(test.ITR_TEST_SKIPPING_TESTS_SKIPPED, self._itr_skipped_count > 0)
 
         # Only parent items set skipped counts because tests would always be 1 or 0
-        if self._children:
+        if self._children or self._distributed_children:
             self.set_tag(test.ITR_TEST_SKIPPING_COUNT, self._itr_skipped_count)
+
+    def set_distributed_children(self) -> None:
+        self._distributed_children = True

ddtrace/internal/ci_visibility/api/_session.py

@@ -107,6 +107,10 @@ def _telemetry_record_event_finished(self):
     def add_coverage_data(self, *args, **kwargs):
         raise NotImplementedError("Coverage data cannot be added to session.")
 
+    def set_skipped_count(self, skipped_count: int):
+        self._itr_skipped_count = skipped_count
+        self._set_itr_tags(self._session_settings.itr_test_skipping_enabled)
+
     def set_covered_lines_pct(self, coverage_pct: float):
         self.set_tag(test.TEST_LINES_PCT, coverage_pct)