chore(ci_visibility): add missing config env vars to new plugin (#15615)

## Description

Add support for missing Test Optimization environment variables.

## Testing

Unit tests.

## Risks

None.

## Additional Notes

None.

Author: vitor-de-araujo

ddtrace/testing/internal/env_tags.py

@@ -33,6 +33,7 @@ def get_env_tags() -> t.Dict[str, str]:
         git.get_git_tags_from_git_command(),
         ci.get_ci_tags(os.environ),
         git.get_git_tags_from_dd_variables(os.environ),
+        get_custom_dd_tags(os.environ),
     )
 
     if head_sha := tags.get(GitTag.COMMIT_HEAD_SHA):
@@ -66,3 +67,45 @@ def normalize_git_tags(tags: _TagDict) -> None:
         tags[GitTag.TAG] = git.normalize_ref(tag)
 
     tags[GitTag.REPOSITORY_URL] = _filter_sensitive_info(tags.get(GitTag.REPOSITORY_URL))
+
+
+def parse_tags_str(tags_str: t.Optional[str]) -> t.Dict[str, str]:
+    """
+    Parses a string containing key-value pairs and returns a dictionary.
+    Key-value pairs are delimited by ':', and pairs are separated by whitespace, comma, OR BOTH.
+
+    This implementation aligns with the way tags are parsed by the Agent and other Datadog SDKs
+
+    :param tags_str: A string of the above form to parse tags from.
+    :return: A dict containing the tags that were parsed.
+    """
+    tags: t.Dict[str, str] = {}
+    if not tags_str:
+        return tags
+
+    # falling back to comma as separator
+    separator = "," if "," in tags_str else " "
+
+    for tag in tags_str.split(separator):
+        tag = tag.strip()
+        if not tag:
+            # skip empty tags
+            continue
+        elif ":" in tag:
+            # if tag contains a colon, split on the first colon
+            key, val = tag.split(":", 1)
+        else:
+            # if tag does not contain a colon, use the whole string as the key
+            key, val = tag, ""
+        key, val = key.strip(), val.strip()
+        if key:
+            # only add the tag if the key is not empty
+            tags[key] = val
+    return tags
+
+
+def get_custom_dd_tags(env: t.MutableMapping[str, str]) -> _TagDict:
+    tags: _TagDict = {}
+    tags.update(parse_tags_str(env.get("DD_TAGS")))
+    tags.update(parse_tags_str(env.get("_CI_DD_TAGS")))
+    return tags

ddtrace/testing/internal/http.py

@@ -98,7 +98,7 @@ def _detect_agentless_setup(cls) -> BackendConnectorSetup:
         Detect settings for agentless backend connection mode.
         """
         site = os.environ.get("DD_SITE") or DEFAULT_SITE
-        api_key = os.environ.get("DD_API_KEY")
+        api_key = os.environ.get("_CI_DD_API_KEY") or os.environ.get("DD_API_KEY")
 
         if not api_key:
             raise SetupError("DD_API_KEY environment variable is not set")
@@ -110,7 +110,7 @@ def _detect_evp_proxy_setup(cls) -> BackendConnectorSetup:
         """
         Detect settings for EVP proxy mode backend connection mode.
         """
-        agent_url = os.environ.get("DD_TRACE_AGENT_URL")
+        agent_url = os.environ.get("_CI_DD_AGENT_URL") or os.environ.get("DD_TRACE_AGENT_URL")
         if not agent_url:
             user_provided_host = os.environ.get("DD_TRACE_AGENT_HOSTNAME") or os.environ.get("DD_AGENT_HOST")
             user_provided_port = os.environ.get("DD_TRACE_AGENT_PORT") or os.environ.get("DD_AGENT_PORT")

ddtrace/testing/internal/logging.py

@@ -14,7 +14,9 @@
 def setup_logging() -> None:
     testing_logger.propagate = False
 
-    log_level = logging.DEBUG if asbool(os.getenv("DD_TEST_DEBUG")) else logging.INFO
+    debug_enabled = asbool(os.getenv("DD_TEST_DEBUG")) or asbool(os.getenv("DD_TRACE_DEBUG"))
+
+    log_level = logging.DEBUG if debug_enabled else logging.INFO
     testing_logger.setLevel(log_level)
 
     for handler in list(testing_logger.handlers):

ddtrace/testing/internal/pytest/plugin.py

@@ -37,6 +37,7 @@
 from ddtrace.testing.internal.tracer_api.coverage import coverage_collection
 from ddtrace.testing.internal.tracer_api.coverage import install_coverage
 from ddtrace.testing.internal.utils import TestContext
+from ddtrace.testing.internal.utils import asbool
 
 
 DISABLED_BY_TEST_MANAGEMENT_REASON = "Flaky test is disabled by Datadog"
@@ -659,7 +660,14 @@ def pytest_addoption(parser: pytest.Parser) -> None:
     parser.addini("ddtrace-patch-all", "Enable all integrations with ddtrace", type="bool")
 
 
+def _is_test_optimization_disabled_by_kill_switch() -> bool:
+    return not asbool(os.environ.get("DD_CIVISIBILITY_ENABLED", "true"))
+
+
 def _is_enabled_early(early_config: pytest.Config, args: t.List[str]) -> bool:
+    if _is_test_optimization_disabled_by_kill_switch():
+        return False
+
     if _is_option_true("no-ddtrace", early_config, args):
         return False
 
@@ -708,6 +716,9 @@ def setup_coverage_collection() -> None:
 
 
 def pytest_configure(config: pytest.Config) -> None:
+    if _is_test_optimization_disabled_by_kill_switch():
+        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)")

ddtrace/testing/internal/session_manager.py

@@ -66,7 +66,7 @@ def __init__(self, session: TestSession) -> None:
 
         self.is_auto_injected = bool(os.getenv("DD_CIVISIBILITY_AUTO_INSTRUMENTATION_PROVIDER", ""))
 
-        self.env = os.environ.get("DD_ENV") or DEFAULT_ENV_NAME
+        self.env = os.environ.get("_CI_DD_ENV") or os.environ.get("DD_ENV") or DEFAULT_ENV_NAME
 
         self.api_client = APIClient(
             service=self.service,
@@ -78,6 +78,8 @@ def __init__(self, session: TestSession) -> None:
             telemetry_api=self.telemetry_api,
         )
         self.settings = self.api_client.get_settings()
+        self.override_settings_with_env_vars()
+
         self.known_tests = self.api_client.get_known_tests() if self.settings.known_tests_enabled else set()
         self.test_properties = (
             self.api_client.get_test_management_properties() if self.settings.test_management.enabled else {}
@@ -151,7 +153,7 @@ def setup_retry_handlers(self) -> None:
             else:
                 log.info("Not enabling Early Flake Detection: no known tests")
 
-        if self.settings.auto_test_retries.enabled and asbool(os.getenv("DD_CIVISIBILITY_FLAKY_RETRY_ENABLED", "true")):
+        if self.settings.auto_test_retries.enabled:
             self.retry_handlers.append(AutoTestRetriesHandler(self))
 
     def start(self) -> None:
@@ -310,6 +312,33 @@ def is_skippable_test(self, test_ref: TestRef) -> bool:
     def has_codeowners(self) -> bool:
         return self.codeowners is not None
 
+    def override_settings_with_env_vars(self) -> None:
+        # Kill switches.
+        # These variables default to true, and if explicitly given a false value, disable a feature.
+        if not asbool(os.environ.get("DD_CIVISIBILITY_ITR_ENABLED", "true")):
+            log.debug("Test Impact Analysis is disabled by environment variable")
+            self.settings.itr_enabled = False
+
+        if not asbool(os.environ.get("DD_CIVISIBILITY_EARLY_FLAKE_DETECTION_ENABLED", "true")):
+            log.debug("Early Flake Detection is disabled by environment variable")
+            self.settings.early_flake_detection.enabled = False
+
+        if not asbool(os.environ.get("DD_CIVISIBILITY_FLAKY_RETRY_ENABLED", "true")):
+            log.debug("Auto Test Retries is disabled by environment variable")
+            self.settings.auto_test_retries.enabled = False
+
+        # "Reverse" kill switches.
+        # These variables default to false, and if explicitly given a true value, disable a feature.
+        if asbool(os.environ.get("_DD_CIVISIBILITY_ITR_PREVENT_TEST_SKIPPING", "false")):
+            log.debug("TIA test skipping is disabled by environment variable")
+            self.settings.skipping_enabled = False
+
+        # Other overrides.
+        # These variables default to false, and if explicitly given a true value, enable a feature.
+        if asbool(os.environ.get("_DD_CIVISIBILITY_ITR_FORCE_ENABLE_COVERAGE", "false")):
+            log.debug("TIA code coverage collection is enabled by environment variable")
+            self.settings.coverage_enabled = True
+
 
 def _get_service_name_from_git_repo(env_tags: t.Dict[str, str]) -> t.Optional[str]:
     repo_name = env_tags.get(GitTag.REPOSITORY_URL)

tests/testing/internal/test_env_tags.py

@@ -319,3 +319,17 @@ def test_falsey_ci_provider_values_overwritten_by_git_executable(
     assert tags["git.commit.author.name"] == "John Doe"
     assert tags["git.commit.author.email"] == "[email protected]"
     assert tags["ci.workspace_path"] == git_repo
+
+
+def test_custom_dd_tags(monkeypatch: pytest.MonkeyPatch, git_repo: str) -> None:
+    env = {
+        "DD_TAGS": "foo:1 bar.baz:2",
+    }
+
+    monkeypatch.setattr(os, "environ", env)
+    monkeypatch.chdir(git_repo)
+
+    tags = get_env_tags()
+
+    assert tags["foo"] == "1"
+    assert tags["bar.baz"] == "2"

tests/testing/internal/test_session_manager.py

@@ -1,13 +1,20 @@
+from contextlib import contextmanager
 import os
+import typing as t
+from unittest.mock import patch
 
 import pytest
 
 from ddtrace.testing.internal.ci import CITag
+from ddtrace.testing.internal.session_manager import SessionManager
 from ddtrace.testing.internal.test_data import ModuleRef
 from ddtrace.testing.internal.test_data import SuiteRef
 from ddtrace.testing.internal.test_data import TestRef
+from ddtrace.testing.internal.test_data import TestSession
 from tests.testing.mocks import MockDefaults
+from tests.testing.mocks import mock_api_client_settings
 from tests.testing.mocks import session_manager_mock
+from tests.testing.mocks import setup_standard_mocks
 
 
 class TestSessionManagerIsSkippableTest:
@@ -200,3 +207,62 @@ def test_session_name_from_test_command(self, monkeypatch: pytest.MonkeyPatch) -
         expected_name = "pytest"
         assert session_manager._get_test_session_name() == expected_name
         assert session_manager.writer.metadata["*"]["test_session.name"] == expected_name
+
+
+class TestSessionManagerEnvVarOverrides:
+    def setup_method(self) -> None:
+        self.session = TestSession("pytest")
+        self.session.set_attributes(
+            test_command="pytest --ddtrace", test_framework="pytest", test_framework_version="9.0.0"
+        )
+
+    @contextmanager
+    def mock_settings(self, **kwargs) -> t.Generator[None, None, None]:
+        with (
+            patch(
+                "ddtrace.testing.internal.session_manager.APIClient",
+                return_value=mock_api_client_settings(**kwargs),
+            ),
+            setup_standard_mocks(),
+        ):
+            yield
+
+    @pytest.mark.parametrize("env_var_value, expected_setting", [(None, True), ("true", True), ("false", False)])
+    def test_session_manager_efd_kill_switch(self, monkeypatch, env_var_value, expected_setting):
+        with self.mock_settings(efd_enabled=True):
+            if env_var_value is not None:
+                monkeypatch.setenv("DD_CIVISIBILITY_EARLY_FLAKE_DETECTION_ENABLED", env_var_value)
+            session_manager = SessionManager(self.session)
+            assert session_manager.settings.early_flake_detection.enabled is expected_setting
+
+    @pytest.mark.parametrize("env_var_value, expected_setting", [(None, True), ("true", True), ("false", False)])
+    def test_session_manager_atr_kill_switch(self, monkeypatch, env_var_value, expected_setting):
+        with self.mock_settings(auto_retries_enabled=True):
+            if env_var_value is not None:
+                monkeypatch.setenv("DD_CIVISIBILITY_FLAKY_RETRY_ENABLED", env_var_value)
+            session_manager = SessionManager(self.session)
+            assert session_manager.settings.auto_test_retries.enabled is expected_setting
+
+    @pytest.mark.parametrize("env_var_value, expected_setting", [(None, True), ("true", True), ("false", False)])
+    def test_session_manager_itr_kill_switch(self, monkeypatch, env_var_value, expected_setting):
+        with self.mock_settings(skipping_enabled=True):
+            if env_var_value is not None:
+                monkeypatch.setenv("DD_CIVISIBILITY_ITR_ENABLED", env_var_value)
+            session_manager = SessionManager(self.session)
+            assert session_manager.settings.itr_enabled is expected_setting
+
+    @pytest.mark.parametrize("env_var_value, expected_setting", [(None, True), ("true", False), ("false", True)])
+    def test_session_manager_skipping_kill_switch(self, monkeypatch, env_var_value, expected_setting):
+        with self.mock_settings(skipping_enabled=True):
+            if env_var_value is not None:
+                monkeypatch.setenv("_DD_CIVISIBILITY_ITR_PREVENT_TEST_SKIPPING", env_var_value)
+            session_manager = SessionManager(self.session)
+            assert session_manager.settings.skipping_enabled is expected_setting
+
+    @pytest.mark.parametrize("env_var_value, expected_setting", [(None, False), ("true", True), ("false", False)])
+    def test_session_manager_force_coverage(self, monkeypatch, env_var_value, expected_setting):
+        with self.mock_settings():
+            if env_var_value is not None:
+                monkeypatch.setenv("_DD_CIVISIBILITY_ITR_FORCE_ENABLE_COVERAGE", env_var_value)
+            session_manager = SessionManager(self.session)
+            assert session_manager.settings.coverage_enabled is expected_setting

tests/testing/test_integration.py

@@ -115,6 +115,31 @@ def test_simple():
         assert result.ret == 0
         result.assert_outcomes(passed=1)
 
+    def test_simple_plugin_disabled_by_kill_switch(self, pytester: Pytester, monkeypatch: MonkeyPatch) -> None:
+        """Test that plugin does not run when DD_CIVISIBILITY_ENABLED is false."""
+        # Create a simple test file
+        pytester.makepyfile(
+            """
+            def test_simple():
+                '''A simple test.'''
+                assert True
+        """
+        )
+
+        monkeypatch.setenv("DD_CIVISIBILITY_ENABLED", "false")
+
+        # Use network mocks to prevent all real HTTP calls
+        with network_mocks(), patch("ddtrace.testing.internal.session_manager.APIClient") as mock_api_client:
+            mock_api_client.return_value = mock_api_client_settings()
+
+            result = pytester.runpytest("--ddtrace", "-v")
+
+        assert mock_api_client.call_count == 0
+
+        # Test should pass
+        assert result.ret == 0
+        result.assert_outcomes(passed=1)
+
     def test_retry_functionality_with_pytester(self, pytester: Pytester, monkeypatch: MonkeyPatch) -> None:
         """Test that failing tests are retried when auto retry is enabled."""
         # Create a test file with a failing test