Fix hosting user agent missing

python/packages/core/tests/core/test_telemetry.py

@@ -1,14 +1,21 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from unittest.mock import patch
+import os
+from unittest.mock import MagicMock, patch
 
+import agent_framework._telemetry as _telemetry_mod
 from agent_framework import (
     AGENT_FRAMEWORK_USER_AGENT,
     USER_AGENT_KEY,
     USER_AGENT_TELEMETRY_DISABLED_ENV_VAR,
     prepend_agent_framework_to_user_agent,
 )
-from agent_framework._telemetry import user_agent_prefix
+from agent_framework._telemetry import (
+    _FOUNDRY_HOSTING_ENV_VAR,
+    _HOSTED_USER_AGENT_PREFIX,
+    _add_user_agent_prefix,
+    _detect_hosted_environment,
+)
 
 # region Test constants
 
@@ -83,7 +90,7 @@ def test_prepend_to_empty_headers():
 
 def test_prepend_to_empty_dict():
     """Test prepending to empty headers dict."""
-    headers = {}
+    headers: dict[str, str] = {}
     result = prepend_agent_framework_to_user_agent(headers)
 
     assert "User-Agent" in result
@@ -99,54 +106,145 @@ def test_modifies_original_dict():
     assert "User-Agent" in headers
 
 
-# region Test user_agent_prefix context manager
+# region Test _add_user_agent_prefix
 
 
-def test_user_agent_prefix_adds_prefix():
-    """Test that the context manager adds a prefix within its scope."""
-    with user_agent_prefix("test-host"):
-        result = prepend_agent_framework_to_user_agent()
-        assert result["User-Agent"].startswith("test-host/")
-        assert AGENT_FRAMEWORK_USER_AGENT in result["User-Agent"]
-
-    # Prefix is removed after exiting the context
+def test_add_user_agent_prefix_adds_prefix():
+    """Test that _add_user_agent_prefix permanently adds a prefix."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    _add_user_agent_prefix("test-host")
     result = prepend_agent_framework_to_user_agent()
-    assert result["User-Agent"] == AGENT_FRAMEWORK_USER_AGENT
+    assert result["User-Agent"].startswith("test-host/")
+    assert AGENT_FRAMEWORK_USER_AGENT in result["User-Agent"]
+    _telemetry_mod._user_agent_prefixes.clear()
 
 
-def test_user_agent_prefix_ignores_duplicates():
-    """Test that duplicate prefixes are not added within nested scopes."""
-    with user_agent_prefix("test-host"), user_agent_prefix("test-host"):
-        result = prepend_agent_framework_to_user_agent()
-        assert result["User-Agent"].count("test-host") == 1
+def test_add_user_agent_prefix_ignores_duplicates():
+    """Test that duplicate prefixes are not added."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    _add_user_agent_prefix("test-host")
+    _add_user_agent_prefix("test-host")
+    result = prepend_agent_framework_to_user_agent()
+    assert result["User-Agent"].count("test-host") == 1
+    _telemetry_mod._user_agent_prefixes.clear()
 
 
-def test_user_agent_prefix_ignores_empty():
+def test_add_user_agent_prefix_ignores_empty():
     """Test that empty strings are not added as prefixes."""
-    with user_agent_prefix(""):
-        result = prepend_agent_framework_to_user_agent()
-        assert result["User-Agent"] == AGENT_FRAMEWORK_USER_AGENT
-
-
-def test_user_agent_prefix_restores_on_exit():
-    """Test that prefixes are fully restored after the context manager exits."""
-    with user_agent_prefix("test-host"):
-        pass
+    _telemetry_mod._user_agent_prefixes.clear()
+    _add_user_agent_prefix("")
     result = prepend_agent_framework_to_user_agent()
     assert result["User-Agent"] == AGENT_FRAMEWORK_USER_AGENT
+    _telemetry_mod._user_agent_prefixes.clear()
 
 
-def test_user_agent_prefix_nesting():
-    """Test that nested context managers compose prefixes correctly."""
-    with user_agent_prefix("outer"):
-        with user_agent_prefix("inner"):
-            result = prepend_agent_framework_to_user_agent()
-            assert "outer" in result["User-Agent"]
-            assert "inner" in result["User-Agent"]
-        # Inner prefix removed
-        result = prepend_agent_framework_to_user_agent()
-        assert "outer" in result["User-Agent"]
-        assert "inner" not in result["User-Agent"]
-    # Both removed
+def test_add_user_agent_prefix_multiple():
+    """Test that multiple prefixes compose correctly."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    _add_user_agent_prefix("outer")
+    _add_user_agent_prefix("inner")
     result = prepend_agent_framework_to_user_agent()
-    assert result["User-Agent"] == AGENT_FRAMEWORK_USER_AGENT
+    assert "outer" in result["User-Agent"]
+    assert "inner" in result["User-Agent"]
+    _telemetry_mod._user_agent_prefixes.clear()
+
+
+# region Test _detect_hosted_environment
+
+
+def test_detect_hosted_env_var_truthy_adds_prefix():
+    """Test that a truthy FOUNDRY_HOSTING_ENVIRONMENT env var adds the prefix."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    with patch.dict("os.environ", {_FOUNDRY_HOSTING_ENV_VAR: "production"}):
+        _detect_hosted_environment()
+    assert _HOSTED_USER_AGENT_PREFIX in _telemetry_mod._user_agent_prefixes
+    _telemetry_mod._user_agent_prefixes.clear()
+
+
+def test_detect_hosted_env_var_empty_skips_prefix():
+    """Test that an empty FOUNDRY_HOSTING_ENVIRONMENT env var does NOT add the prefix."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    with patch.dict("os.environ", {_FOUNDRY_HOSTING_ENV_VAR: ""}):
+        _detect_hosted_environment()
+    assert _HOSTED_USER_AGENT_PREFIX not in _telemetry_mod._user_agent_prefixes
+    _telemetry_mod._user_agent_prefixes.clear()
+
+
+def test_detect_hosted_env_var_set_skips_agent_config_fallback():
+    """Test that when the env var is set, AgentConfig is never consulted even if import would fail."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    import builtins
+
+    real_import = builtins.__import__
+
+    def _block_agentconfig(name: str, *args, **kwargs):  # type: ignore[no-untyped-def]
+        if "agentserver" in name:
+            raise AssertionError("AgentConfig should not be imported when env var is set")
+        return real_import(name, *args, **kwargs)
+
+    with (
+        patch.dict("os.environ", {_FOUNDRY_HOSTING_ENV_VAR: "prod"}),
+        patch("builtins.__import__", side_effect=_block_agentconfig),
+    ):
+        _detect_hosted_environment()
+    assert _HOSTED_USER_AGENT_PREFIX in _telemetry_mod._user_agent_prefixes
+    _telemetry_mod._user_agent_prefixes.clear()
+
+
+def _mock_agent_config(*, is_hosted: bool) -> MagicMock:
+    """Create a mock azure.ai.agentserver.core module with AgentConfig."""
+    mock_config = MagicMock()
+    mock_config.is_hosted = is_hosted
+    mock_module = MagicMock()
+    mock_module.AgentConfig.from_env.return_value = mock_config
+    return mock_module
+
+
+def test_detect_hosted_fallback_agent_config_is_hosted():
+    """Test that AgentConfig fallback adds the prefix when is_hosted is True."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    env = {k: v for k, v in os.environ.items() if k != _FOUNDRY_HOSTING_ENV_VAR}
+    mock_module = _mock_agent_config(is_hosted=True)
+    with (
+        patch.dict("os.environ", env, clear=True),
+        patch.dict("sys.modules", {"azure.ai.agentserver.core": mock_module}),
+    ):
+        _detect_hosted_environment()
+    assert _HOSTED_USER_AGENT_PREFIX in _telemetry_mod._user_agent_prefixes
+    _telemetry_mod._user_agent_prefixes.clear()
+
+
+def test_detect_hosted_fallback_agent_config_not_hosted():
+    """Test that AgentConfig fallback does NOT add the prefix when is_hosted is False."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    mock_module = _mock_agent_config(is_hosted=False)
+    env = {k: v for k, v in os.environ.items() if k != _FOUNDRY_HOSTING_ENV_VAR}
+    with (
+        patch.dict("os.environ", env, clear=True),
+        patch.dict("sys.modules", {"azure.ai.agentserver.core": mock_module}),
+    ):
+        _detect_hosted_environment()
+    assert _HOSTED_USER_AGENT_PREFIX not in _telemetry_mod._user_agent_prefixes
+    _telemetry_mod._user_agent_prefixes.clear()
+
+
+def test_detect_hosted_fallback_import_error():
+    """Test that ImportError from AgentConfig is silently handled."""
+    _telemetry_mod._user_agent_prefixes.clear()
+    env = {k: v for k, v in os.environ.items() if k != _FOUNDRY_HOSTING_ENV_VAR}
+    with patch.dict("os.environ", env, clear=True):
+        # The real import may succeed or fail depending on the environment;
+        # force the ImportError path by making the import raise.
+        import builtins
+
+        real_import = builtins.__import__
+
+        def _block_agentconfig(name: str, *args, **kwargs):  # type: ignore[no-untyped-def]
+            if "agentserver" in name:
+                raise ImportError("mocked")
+            return real_import(name, *args, **kwargs)
+
+        with patch("builtins.__import__", side_effect=_block_agentconfig):
+            _detect_hosted_environment()
+    assert _HOSTED_USER_AGENT_PREFIX not in _telemetry_mod._user_agent_prefixes
+    _telemetry_mod._user_agent_prefixes.clear()