chore(testing): add test discovery mode for pytest

ddtrace/internal/settings/_supported_configurations.py

@@ -459,6 +459,8 @@
         "DD_TEST_DEBUG",
         "DD_TEST_MANAGEMENT_ATTEMPT_TO_FIX_RETRIES",
         "DD_TEST_MANAGEMENT_ENABLED",
+        "DD_TEST_OPTIMIZATION_DISCOVERY_ENABLED",
+        "DD_TEST_OPTIMIZATION_DISCOVERY_FILE",
         "DD_TEST_OPTIMIZATION_ENV_DATA_FILE",
         "DD_TEST_OPTIMIZATION_MANIFEST_FILE",
         "DD_TEST_OPTIMIZATION_PAYLOADS_IN_FILES",

ddtrace/testing/internal/constants.py

@@ -25,5 +25,9 @@ class ITRSkippingLevel(Enum):
 DD_TEST_OPTIMIZATION_ENV_DATA_FILE = "DD_TEST_OPTIMIZATION_ENV_DATA_FILE"
 TEST_UNDECLARED_OUTPUTS_DIR = "TEST_UNDECLARED_OUTPUTS_DIR"
 
+# Test discovery environment variables
+DD_TEST_OPTIMIZATION_DISCOVERY_ENABLED = "DD_TEST_OPTIMIZATION_DISCOVERY_ENABLED"
+DD_TEST_OPTIMIZATION_DISCOVERY_FILE = "DD_TEST_OPTIMIZATION_DISCOVERY_FILE"
+
 # The only supported .testoptimization manifest version
 SUPPORTED_MANIFEST_VERSION = 1

ddtrace/testing/internal/pytest/_discovery.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+import typing as t
+
+import pytest
+
+from ddtrace.internal.settings import env
+from ddtrace.testing.internal.constants import DD_TEST_OPTIMIZATION_DISCOVERY_ENABLED
+from ddtrace.testing.internal.constants import DD_TEST_OPTIMIZATION_DISCOVERY_FILE
+from ddtrace.testing.internal.git import get_workspace_path
+from ddtrace.testing.internal.pytest.utils import _get_test_parameters_json
+from ddtrace.testing.internal.pytest.utils import item_to_test_ref
+from ddtrace.testing.internal.utils import asbool
+
+
+log = logging.getLogger(__name__)
+
+_DEFAULT_OUTPUT_PATH = ".testoptimization/tests-discovery/tests.json"
+
+
+def is_discovery_mode_enabled() -> bool:
+    return asbool(env.get(DD_TEST_OPTIMIZATION_DISCOVERY_ENABLED))
+
+
+def _get_output_path() -> Path:
+    return Path(env.get(DD_TEST_OPTIMIZATION_DISCOVERY_FILE, _DEFAULT_OUTPUT_PATH))
+
+
+def _is_item_skipped(item: pytest.Item) -> bool:
+    """Return True if the item will definitely be skipped at execution time.
+
+    Handles pytest.mark.skip (unconditional) and pytest.mark.skipif with
+    non-string conditions. String conditions are not evaluated (would require
+    exec in the test module's namespace) so those tests are conservatively
+    included.
+    """
+    if item.get_closest_marker("skip") is not None:
+        return True
+    for marker in item.iter_markers("skipif"):
+        condition = marker.args[0] if marker.args else marker.kwargs.get("condition")
+        if condition is None or isinstance(condition, str):
+            # String conditions require eval in the test module's namespace, which
+            # we can't do safely at collection time. None conditions arise when
+            # @pytest.mark.skipif is used with only a reason= kwarg and no positional
+            # condition — pytest 9 accepts this and skips the test at runtime, but
+            # does not signal it during collection.
+            # Conservatively include in both cases rather than risk hiding a test.
+            continue
+        if condition:
+            return True
+    return False
+
+
+def _get_suite_source_file(item: pytest.Item, workspace_path: t.Optional[Path]) -> str:
+    item_path = Path(item.path if hasattr(item, "path") else getattr(item, "fspath", "")).absolute()
+    if workspace_path is not None:
+        try:
+            return item_path.relative_to(workspace_path).as_posix()
+        except ValueError:
+            pass
+    return str(item_path)
+
+
+@pytest.hookimpl(tryfirst=True)  # type: ignore[misc]
+def pytest_collection_finish(session: pytest.Session) -> None:
+    if not is_discovery_mode_enabled():
+        return
+
+    workspace_path: t.Optional[Path] = None
+    try:
+        workspace_path = get_workspace_path()
+    except Exception:
+        log.debug("Could not determine workspace path for test discovery", exc_info=True)
+
+    output_path = _get_output_path()
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with open(output_path, "w", encoding="utf-8") as f:
+        for item in session.items:
+            if _is_item_skipped(item):
+                continue
+
+            test_ref = item_to_test_ref(item)
+            module = test_ref.suite.module.name
+            suite = test_ref.suite.name
+            name = test_ref.name
+            parameters = _get_test_parameters_json(item)
+            suite_source_file = _get_suite_source_file(item, workspace_path)
+
+            test_info: dict[str, t.Any] = {
+                "name": name,
+                "suite": suite,
+                "module": module,
+                "parameters": parameters,
+                "suiteSourceFile": suite_source_file,
+            }
+            f.write(json.dumps(test_info) + "\n")
+
+    log.info("Test discovery complete: wrote tests to %s", output_path)
+    pytest.exit("Test discovery complete", returncode=0)

ddtrace/testing/internal/pytest/plugin.py

@@ -2,10 +2,8 @@
 
 from collections import defaultdict
 from io import StringIO
-import json
 import logging
 from pathlib import Path
-import re
 import traceback
 import typing as t
 
@@ -27,11 +25,13 @@
 from ddtrace.testing.internal.logging import catch_and_log_exceptions
 from ddtrace.testing.internal.logging import setup_logging
 from ddtrace.testing.internal.offline_mode import get_offline_mode
+from ddtrace.testing.internal.pytest._discovery import is_discovery_mode_enabled
 from ddtrace.testing.internal.pytest.bdd import BddTestOptPlugin
 from ddtrace.testing.internal.pytest.benchmark import BenchmarkData
 from ddtrace.testing.internal.pytest.benchmark import get_benchmark_tags_and_metrics
 from ddtrace.testing.internal.pytest.hookspecs import TestOptHooks
 from ddtrace.testing.internal.pytest.report_links import print_test_report_links
+from ddtrace.testing.internal.pytest.utils import _get_test_parameters_json
 from ddtrace.testing.internal.pytest.utils import item_to_test_ref
 from ddtrace.testing.internal.retry_handlers import RetryHandler
 from ddtrace.testing.internal.session_manager import SessionManager
@@ -1074,6 +1074,9 @@ def _is_enabled_early(early_config: pytest.Config, args: list[str]) -> bool:
     if _is_test_optimization_disabled_by_kill_switch():
         return False
 
+    if is_discovery_mode_enabled():
+        return False
+
     if _is_option_true("no-ddtrace", early_config, args):
         return False
 
@@ -1138,6 +1141,18 @@ def pytest_configure(config: pytest.Config) -> None:
     if _is_test_optimization_disabled_by_kill_switch():
         return
 
+    if is_discovery_mode_enabled():
+        # Register hook specs so item_to_test_ref can call the custom name hooks during
+        # discovery, giving the same module/suite/name resolution as a real test run.
+        # AIDEV-NOTE: BddTestOptPlugin is not registered here, so pytest-bdd tests will
+        # fall back to nodeid-based names rather than feature-file names during discovery.
+        # TODO: register BddTestOptPlugin in discovery mode to support pytest-bdd.
+        import ddtrace.testing.internal.pytest._discovery as _ddtrace_discovery
+
+        config.pluginmanager.add_hookspecs(TestOptHooks)
+        config.pluginmanager.register(_ddtrace_discovery, "_ddtrace_discovery")
+        return
+
     session_manager = config.stash.get(SESSION_MANAGER_STASH_KEY, None)
     if not session_manager:
         log.debug("Session manager not initialized (plugin was not enabled)")
@@ -1243,38 +1258,10 @@ def _get_user_property(report: pytest.TestReport, user_property: str) -> t.Optio
     return None
 
 
-def _get_test_parameters_json(item: pytest.Item) -> t.Optional[str]:
-    callspec: t.Optional[pytest.python.CallSpec2] = getattr(item, "callspec", None)
-
-    if callspec is None:
-        return None
-
-    parameters: dict[str, dict[str, str]] = {"arguments": {}, "metadata": {}}
-    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)
-
-    try:
-        return json.dumps(parameters, sort_keys=True)
-    except TypeError:
-        log.warning("Failed to serialize parameters for test %s", item, exc_info=True)
-        return None
-
-
 def _get_test_original_name(item: pytest.Item) -> t.Optional[str]:
     return getattr(item, "originalname", None)
 
 
-def _encode_test_parameter(parameter: t.Any) -> str:
-    param_repr = repr(parameter)
-    # if the representation includes an id() we'll remove it
-    # because it isn't constant across executions
-    return re.sub(r" at 0[xX][0-9a-fA-F]+", "", param_repr)
-
-
 def _get_skipif_condition(marker: pytest.Mark) -> t.Any:
     # DEV: pytest allows the condition to be a string to be evaluated. We currently don't support this.
     if marker.args:

ddtrace/testing/internal/pytest/utils.py

@@ -1,4 +1,9 @@
+from __future__ import annotations
+
+import json
+import logging
 import re
+import typing as t
 
 import pytest
 
@@ -7,6 +12,8 @@
 from ddtrace.testing.internal.test_data import TestRef
 
 
+log = logging.getLogger(__name__)
+
 _NODEID_REGEX = re.compile("^(((?P<module>.*)/)?(?P<suite>[^/]*?))::(?P<name>.*?)$")
 
 
@@ -40,3 +47,31 @@ def item_to_test_ref(item: pytest.Item) -> TestRef:
     test_ref = TestRef(suite_ref, custom_test or default_test)
 
     return test_ref
+
+
+def _encode_test_parameter(parameter: t.Any) -> str:
+    param_repr = repr(parameter)
+    # if the representation includes an id() we'll remove it
+    # because it isn't constant across executions
+    return re.sub(r" at 0[xX][0-9a-fA-F]+", "", param_repr)
+
+
+def _get_test_parameters_json(item: pytest.Item) -> t.Optional[str]:
+    callspec: t.Optional[pytest.python.CallSpec2] = getattr(item, "callspec", None)
+
+    if callspec is None:
+        return None
+
+    parameters: dict[str, dict[str, str]] = {"arguments": {}, "metadata": {}}
+    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)
+
+    try:
+        return json.dumps(parameters, sort_keys=True)
+    except TypeError:
+        log.warning("Failed to serialize parameters for test %s", item, exc_info=True)
+        return None

supported-configurations.json

@@ -3516,6 +3516,20 @@
         "default": null
       }
     ],
+    "DD_TEST_OPTIMIZATION_DISCOVERY_ENABLED": [
+      {
+        "implementation": "A",
+        "type": "boolean",
+        "default": "false"
+      }
+    ],
+    "DD_TEST_OPTIMIZATION_DISCOVERY_FILE": [
+      {
+        "implementation": "A",
+        "type": "string",
+        "default": null
+      }
+    ],
     "DD_TEST_OPTIMIZATION_ENV_DATA_FILE": [
       {
         "implementation": "A",

tests/testing/internal/pytest/test_plugin.py

@@ -16,7 +16,6 @@
 from ddtrace.testing.internal.pytest.plugin import SKIPPED_BY_ITR_REASON
 from ddtrace.testing.internal.pytest.plugin import TestOptPlugin
 from ddtrace.testing.internal.pytest.plugin import XdistTestOptPlugin
-from ddtrace.testing.internal.pytest.plugin import _encode_test_parameter
 from ddtrace.testing.internal.pytest.plugin import _get_exception_tags
 from ddtrace.testing.internal.pytest.plugin import _get_module_path_from_item
 from ddtrace.testing.internal.pytest.plugin import _get_source_lines
@@ -25,6 +24,7 @@
 from ddtrace.testing.internal.pytest.plugin import _get_test_original_name
 from ddtrace.testing.internal.pytest.plugin import _get_test_parameters_json
 from ddtrace.testing.internal.pytest.plugin import _get_user_property
+from ddtrace.testing.internal.pytest.utils import _encode_test_parameter
 from ddtrace.testing.internal.pytest.utils import nodeid_to_names
 from ddtrace.testing.internal.test_data import TestStatus
 from ddtrace.testing.internal.test_data import TestTag