refactor(ci_visibility): remove core usage

ddtrace/contrib/internal/pytest/_plugin_v1.py

@@ -472,7 +472,7 @@ def pytest_sessionstart(session):
         global _global_skipped_elements
         _global_skipped_elements = 0
 
-        workspace_path = _CIVisibility.get_workspace_path()
+        workspace_path = _CIVisibility._instance.get_workspace_path()
         if workspace_path is None:
             workspace_path = session.config.rootdir
 
@@ -492,7 +492,7 @@ def pytest_sessionstart(session):
         test_session_span.set_tag_str(test.COMMAND, test_command)
         test_session_span.set_tag_str(_SESSION_ID, str(test_session_span.span_id))
 
-        _CIVisibility.set_test_session_name(test_command=test_command)
+        _CIVisibility._instance.set_test_session_name(test_command=test_command)
 
         if _CIVisibility.test_skipping_enabled():
             test_session_span.set_tag_str(test.ITR_TEST_SKIPPING_ENABLED, "true")

ddtrace/contrib/internal/pytest/_plugin_v2.py

@@ -441,7 +441,7 @@ def _pytest_runtest_protocol_pre_yield(item) -> t.Optional[ModuleCodeCollector.C
     _handle_test_management(item, test_id)
     _handle_itr_should_skip(item, test_id)
 
-    item_will_skip = _pytest_marked_to_skip(item) or InternalTest.was_skipped_by_itr(test_id)
+    item_will_skip = _pytest_marked_to_skip(item) or InternalTest.was_itr_skipped(test_id)
 
     collect_test_coverage = InternalTestSession.should_collect_coverage() and not item_will_skip
 
@@ -470,7 +470,7 @@ def _pytest_runtest_protocol_post_yield(item, nextitem, coverage_collector):
     # - we trust that the next item is in the same module if it is in the same suite
     next_test_id = _get_test_id_from_item(nextitem) if nextitem else None
     if next_test_id is None or next_test_id.parent_id != suite_id:
-        if InternalTestSuite.is_itr_skippable(suite_id) and not InternalTestSuite.was_forced_run(suite_id):
+        if InternalTestSuite.is_itr_skippable(suite_id) and not InternalTestSuite.was_itr_forced_run(suite_id):
             InternalTestSuite.mark_itr_skipped(suite_id)
         else:
             _handle_coverage_dependencies(suite_id)
@@ -611,7 +611,7 @@ def _process_result(item, result) -> _TestOutcome:
     # If run with --runxfail flag, tests behave as if they were not marked with xfail,
     # that's why no XFAIL_REASON or test.RESULT tags will be added.
     if result.skipped:
-        if InternalTest.was_skipped_by_itr(test_id):
+        if InternalTest.was_itr_skipped(test_id):
             # Items that were skipped by ITR already have their status and reason set
             return _TestOutcome()
 
@@ -794,7 +794,7 @@ def _pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
         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.set_itr_skipped_count(skipped_count)
 
     InternalTestSession.finish(
         force_finish_children=True,

ddtrace/contrib/internal/pytest/_report_links.py

@@ -3,7 +3,7 @@
 from urllib.parse import quote
 
 from ddtrace.ext import ci
-from ddtrace.internal.ci_visibility import CIVisibility
+from ddtrace.internal.ci_visibility.service_registry import require_ci_visibility_service
 
 
 DEFAULT_DATADOG_SITE = "datadoghq.com"
@@ -16,10 +16,11 @@ def print_test_report_links(terminalreporter):
     base_url = _get_base_url(
         dd_site=os.getenv("DD_SITE", DEFAULT_DATADOG_SITE), dd_subdomain=os.getenv("DD_SUBDOMAIN", "")
     )
-    ci_tags = CIVisibility.get_ci_tags()
-    settings = CIVisibility.get_session_settings()
+    ci_visibility_instance = require_ci_visibility_service()
+    ci_tags = ci_visibility_instance.get_ci_tags()
+    settings = ci_visibility_instance.get_session_settings()
     service = settings.test_service
-    env = CIVisibility.get_dd_env()
+    env = ci_visibility_instance.get_dd_env()
 
     redirect_test_commit_url = _build_test_commit_redirect_url(base_url, ci_tags, service, env)
     test_runs_url = _build_test_runs_url(base_url, ci_tags)

ddtrace/contrib/internal/unittest/patch.py

@@ -659,7 +659,7 @@ def _start_test_session_span(instance) -> ddtrace.trace.Span:
         "true" if _CIVisibility._instance._collect_coverage_enabled else "false",
     )
 
-    _CIVisibility.set_test_session_name(test_command=test_command)
+    _CIVisibility._instance.set_test_session_name(test_command=test_command)
 
     if _CIVisibility.test_skipping_enabled():
         _set_test_skipping_tags_to_span(test_session_span)

ddtrace/ext/test_visibility/_test_visibility_base.py

@@ -2,11 +2,7 @@
 import dataclasses
 from enum import Enum
 from pathlib import Path
-from typing import Any
-from typing import Dict
 from typing import Generic
-from typing import List
-from typing import NamedTuple
 from typing import Optional
 from typing import TypeVar
 from typing import Union
@@ -61,46 +57,57 @@ def get_parent_id(self) -> PT:
         return self.parent_id
 
 
-TestVisibilityItemId = TypeVar(
-    "TestVisibilityItemId", bound=Union[_TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId]
-)
+@dataclasses.dataclass(frozen=True)
+class TestModuleId(_TestVisibilityRootItemIdBase):
+    name: str
 
+    def __repr__(self):
+        return "TestModuleId(module={})".format(
+            self.name,
+        )
 
-class _TestVisibilityAPIBase(abc.ABC):
-    __test__ = False
 
-    class GetTagArgs(NamedTuple):
-        item_id: Union[_TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId]
-        name: str
+@dataclasses.dataclass(frozen=True)
+class TestSuiteId(_TestVisibilityChildItemIdBase[TestModuleId]):
+    def __repr__(self):
+        return "TestSuiteId(module={}, suite={})".format(self.parent_id.name, self.name)
 
-    class SetTagArgs(NamedTuple):
-        item_id: Union[_TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId]
-        name: str
-        value: Any
 
-    class DeleteTagArgs(NamedTuple):
-        item_id: Union[_TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId]
-        name: str
+@dataclasses.dataclass(frozen=True)
+class TestId(_TestVisibilityChildItemIdBase[TestSuiteId]):
+    parameters: Optional[str] = None  # For hashability, a JSON string of a dictionary of parameters
 
-    class SetTagsArgs(NamedTuple):
-        item_id: Union[_TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId]
-        tags: Dict[str, Any]
+    def __repr__(self):
+        return "TestId(module={}, suite={}, test={}, parameters={})".format(
+            self.parent_id.parent_id.name,
+            self.parent_id.name,
+            self.name,
+            self.parameters,
+        )
 
-    class DeleteTagsArgs(NamedTuple):
-        item_id: Union[_TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId]
-        names: List[str]
+
+TestVisibilityItemId = TypeVar(
+    "TestVisibilityItemId",
+    bound=Union[
+        _TestVisibilityChildItemIdBase, _TestVisibilityRootItemIdBase, TestSessionId, TestModuleId, TestSuiteId, TestId
+    ],
+)
+
+
+class _TestVisibilityAPIBase(abc.ABC):
+    __test__ = False
 
     def __init__(self):
         raise NotImplementedError("This class is not meant to be instantiated")
 
     @staticmethod
     @abc.abstractmethod
-    def discover(item_id: TestVisibilityItemId, *args, **kwargs):
+    def discover(*args, **kwargs):
         pass
 
     @staticmethod
     @abc.abstractmethod
-    def start(item_id: TestVisibilityItemId, *args, **kwargs):
+    def start(*args, **kwargs):
         pass
 
     @staticmethod

ddtrace/ext/test_visibility/_utils.py

@@ -1,22 +1,19 @@
 from functools import wraps
-from typing import Any
-from typing import Dict
-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 import config as ddconfig
 from ddtrace.internal.logger import get_logger
 
 
 log = get_logger(__name__)
 
 
+def _noop_decorator(func):
+    return func
+
+
 def _catch_and_log_exceptions(func):
     """This decorator is meant to be used around all methods of the Test Visibility classes.
-
     It accepts an optional parameter to allow it to be used on functions and methods.
-
     No uncaught errors should ever reach the integration-side, and potentially cause crashes.
     """
 
@@ -30,36 +27,5 @@ 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 _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_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),))
-
-
-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 _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 _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
+if ddconfig._raise:
+    _catch_and_log_decorator = _noop_decorator