[SDK Parity] Add scope and domain overrides for observability pre-prod testing

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/agent365_exporter.py

@@ -18,6 +18,7 @@
 from opentelemetry.trace import StatusCode
 
 from .utils import (
+    get_validated_domain_override,
     hex_span_id,
     hex_trace_id,
     kind_name,
@@ -60,6 +61,8 @@ def __init__(
         self._token_resolver = token_resolver
         self._cluster_category = cluster_category
         self._use_s2s_endpoint = use_s2s_endpoint
+        # Read domain override once at initialization
+        self._domain_override = get_validated_domain_override()
 
     # ------------- SpanExporter API -----------------
 
@@ -86,8 +89,11 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                 body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
 
                 # Resolve endpoint + token
-                discovery = PowerPlatformApiDiscovery(self._cluster_category)
-                endpoint = discovery.get_tenant_island_cluster_endpoint(tenant_id)
+                if self._domain_override:
+                    endpoint = self._domain_override
+                else:
+                    discovery = PowerPlatformApiDiscovery(self._cluster_category)
+                    endpoint = discovery.get_tenant_island_cluster_endpoint(tenant_id)
                 endpoint_path = (
                     f"/maven/agent365/service/agents/{agent_id}/traces"
                     if self._use_s2s_endpoint
@@ -142,6 +148,8 @@ def shutdown(self) -> None:
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         return True
 
+    # ------------- Helper methods -------------------
+
     # ------------- HTTP helper ----------------------
 
     @staticmethod

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/utils.py

@@ -142,6 +142,28 @@ def partition_by_identity(
     return groups
 
 
+def get_validated_domain_override() -> str | None:
+    """
+    Get and validate the domain override from environment variable.
+
+    Returns:
+        The validated domain override, or None if not set or invalid.
+    """
+    domain_override = os.getenv("A365_OBSERVABILITY_DOMAIN_OVERRIDE", "").strip()
+    if not domain_override:
+        return None
+
+    # Basic validation: ensure domain doesn't contain protocol or path separators
+    if "://" in domain_override or "/" in domain_override:
+        logger.warning(
+            f"Invalid domain override '{domain_override}': "
+            "domain should not contain protocol (://) or path separators (/)"
+        )
+        return None
+
+    return domain_override
+
+
 def is_agent365_exporter_enabled() -> bool:
     """Check if Agent 365 exporter is enabled."""
     # Check environment variable

libraries/microsoft-agents-a365-runtime/microsoft_agents_a365/runtime/environment_utils.py

@@ -21,10 +21,14 @@ def get_observability_authentication_scope() -> list[str]:
     """
     Returns the scope for authenticating to the observability service based on the current environment.
 
+    The scope can be overridden via the A365_OBSERVABILITY_SCOPE_OVERRIDE environment variable
+    to enable testing against pre-production environments.
+
     Returns:
         list[str]: The authentication scope for the current environment.
     """
-    return [PROD_OBSERVABILITY_SCOPE]
+    override_scope = os.getenv("A365_OBSERVABILITY_SCOPE_OVERRIDE", "").strip()
+    return [override_scope] if override_scope else [PROD_OBSERVABILITY_SCOPE]
 
 
 def is_development_environment() -> bool:

tests/observability/core/test_agent365_exporter.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import json
+import os
 import unittest
 from unittest.mock import Mock, patch
 
@@ -20,11 +21,25 @@ def setUp(self):
         self.mock_token_resolver = Mock()
         self.mock_token_resolver.return_value = "test_token_123"
 
-        # Don't patch the class in setUp, do it per test
+        # Store original environment variable values for cleanup
+        self._original_domain_override = os.environ.get("A365_OBSERVABILITY_DOMAIN_OVERRIDE")
+
+        # Ensure no override is set by default for most tests
+        os.environ.pop("A365_OBSERVABILITY_DOMAIN_OVERRIDE", None)
+
+        # Create default exporter for tests that don't need special setup
         self.exporter = _Agent365Exporter(
             token_resolver=self.mock_token_resolver, cluster_category="test"
         )
 
+    def tearDown(self):
+        """Clean up test environment."""
+        # Restore original environment variable value
+        if self._original_domain_override is None:
+            os.environ.pop("A365_OBSERVABILITY_DOMAIN_OVERRIDE", None)
+        else:
+            os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = self._original_domain_override
+
     def _create_mock_span(
         self,
         name: str = "test_span",
@@ -384,6 +399,172 @@ def test_exporter_is_internal(self):
             "Exporter class should be prefixed with underscore to indicate it's private/internal",
         )
 
+    def test_export_uses_domain_override_when_env_var_set(self):
+        """Test that domain override is used when A365_OBSERVABILITY_DOMAIN_OVERRIDE is set."""
+        # Arrange
+        override_domain = "override.example.com"
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = override_domain
+
+        # Create exporter after setting environment variable so it reads the override
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("override_test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should not be called when override is set)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            # Mock the _post_with_retries method
+            with patch.object(exporter, "_post_with_retries", return_value=True) as mock_post:
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                mock_post.assert_called_once()
+
+                # Verify the call arguments - should use override domain with complete URL
+                args, kwargs = mock_post.call_args
+                url, body, headers = args
+
+                expected_url = f"https://{override_domain}/maven/agent365/agents/test-agent-456/traces?api-version=1"
+                self.assertEqual(url, expected_url)
+
+                # Verify PowerPlatformApiDiscovery was not instantiated
+                mock_discovery_class.assert_not_called()
+
+    def test_export_uses_default_domain_when_no_override(self):
+        """Test that default domain resolution is used when no override is set."""
+        # Arrange
+        # Ensure override is not set
+        os.environ.pop("A365_OBSERVABILITY_DOMAIN_OVERRIDE", None)
+
+        # Create exporter after clearing environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("default_domain_span")]
+
+        # Mock the PowerPlatformApiDiscovery class
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            mock_discovery = Mock()
+            mock_discovery.get_tenant_island_cluster_endpoint.return_value = "default-endpoint.com"
+            mock_discovery_class.return_value = mock_discovery
+
+            # Mock the _post_with_retries method
+            with patch.object(exporter, "_post_with_retries", return_value=True) as mock_post:
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                mock_post.assert_called_once()
+
+                # Verify the call arguments - should use default domain
+                args, kwargs = mock_post.call_args
+                url, body, headers = args
+
+                self.assertIn("default-endpoint.com", url)
+                self.assertIn("/maven/agent365/agents/test-agent-456/traces", url)
+
+                # Verify PowerPlatformApiDiscovery was called
+                mock_discovery_class.assert_called_once_with("test")
+                mock_discovery.get_tenant_island_cluster_endpoint.assert_called_once_with(
+                    "test-tenant-123"
+                )
+
+    def test_export_ignores_empty_domain_override(self):
+        """Test that empty or whitespace-only domain override is ignored."""
+        # Arrange
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "   "  # whitespace only
+
+        # Create exporter after setting environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should be called since override is invalid)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            mock_discovery = Mock()
+            mock_discovery.get_tenant_island_cluster_endpoint.return_value = "default-endpoint.com"
+            mock_discovery_class.return_value = mock_discovery
+
+            with patch.object(exporter, "_post_with_retries", return_value=True):
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                # Verify PowerPlatformApiDiscovery was called (override was ignored)
+                mock_discovery_class.assert_called_once_with("test")
+
+    def test_export_ignores_invalid_domain_with_protocol(self):
+        """Test that domain override containing protocol is ignored."""
+        # Arrange
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "https://invalid.example.com"
+
+        # Create exporter after setting environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should be called since override is invalid)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            mock_discovery = Mock()
+            mock_discovery.get_tenant_island_cluster_endpoint.return_value = "default-endpoint.com"
+            mock_discovery_class.return_value = mock_discovery
+
+            with patch.object(exporter, "_post_with_retries", return_value=True):
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                # Verify PowerPlatformApiDiscovery was called (override was ignored)
+                mock_discovery_class.assert_called_once_with("test")
+
+    def test_export_ignores_invalid_domain_with_path(self):
+        """Test that domain override containing path separator is ignored."""
+        # Arrange
+        os.environ["A365_OBSERVABILITY_DOMAIN_OVERRIDE"] = "invalid.example.com/path"
+
+        # Create exporter after setting environment variable
+        exporter = _Agent365Exporter(
+            token_resolver=self.mock_token_resolver, cluster_category="test"
+        )
+
+        spans = [self._create_mock_span("test_span")]
+
+        # Mock the PowerPlatformApiDiscovery class (should be called since override is invalid)
+        with patch(
+            "microsoft_agents_a365.observability.core.exporters.agent365_exporter.PowerPlatformApiDiscovery"
+        ) as mock_discovery_class:
+            mock_discovery = Mock()
+            mock_discovery.get_tenant_island_cluster_endpoint.return_value = "default-endpoint.com"
+            mock_discovery_class.return_value = mock_discovery
+
+            with patch.object(exporter, "_post_with_retries", return_value=True):
+                # Act
+                result = exporter.export(spans)
+
+                # Assert
+                self.assertEqual(result, SpanExportResult.SUCCESS)
+                # Verify PowerPlatformApiDiscovery was called (override was ignored)
+                mock_discovery_class.assert_called_once_with("test")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/runtime/test_environment_utils.py

@@ -18,6 +18,40 @@ def test_get_observability_authentication_scope():
     assert result == [PROD_OBSERVABILITY_SCOPE]
 
 
+def test_get_observability_authentication_scope_with_override(monkeypatch):
+    """Test get_observability_authentication_scope returns override when env var is set."""
+    override_scope = "https://override.example.com/.default"
+    monkeypatch.setenv("A365_OBSERVABILITY_SCOPE_OVERRIDE", override_scope)
+
+    result = get_observability_authentication_scope()
+    assert result == [override_scope]
+
+
+def test_get_observability_authentication_scope_ignores_empty_override(monkeypatch):
+    """Test get_observability_authentication_scope ignores empty string override."""
+    monkeypatch.setenv("A365_OBSERVABILITY_SCOPE_OVERRIDE", "")
+
+    result = get_observability_authentication_scope()
+    assert result == [PROD_OBSERVABILITY_SCOPE]
+
+
+def test_get_observability_authentication_scope_ignores_whitespace_override(monkeypatch):
+    """Test get_observability_authentication_scope ignores whitespace-only override."""
+    monkeypatch.setenv("A365_OBSERVABILITY_SCOPE_OVERRIDE", "   ")
+
+    result = get_observability_authentication_scope()
+    assert result == [PROD_OBSERVABILITY_SCOPE]
+
+
+def test_get_observability_authentication_scope_trims_whitespace(monkeypatch):
+    """Test get_observability_authentication_scope trims whitespace from override."""
+    override_scope = "  https://override.example.com/.default  "
+    monkeypatch.setenv("A365_OBSERVABILITY_SCOPE_OVERRIDE", override_scope)
+
+    result = get_observability_authentication_scope()
+    assert result == [override_scope.strip()]
+
+
 @pytest.mark.parametrize(
     "env_value,expected",
     [