Fixes for tests.

juliomenendez · juliomenendez · commit 5f8e109b4793 · 2025-12-10T12:05:56.000-07:00
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/execute_tool_scope.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/execute_tool_scope.py
@@ -42,7 +42,7 @@ def start(
         Returns:
             A new ExecuteToolScope instance
         """
-        return ExecuteToolScope(details, agent_details, tenant_details)
+        return ExecuteToolScope(details, agent_details, tenant_details, request)
 
     def __init__(
         self,
diff --git a/tests/observability/core/test_execute_tool_scope.py b/tests/observability/core/test_execute_tool_scope.py
@@ -1,12 +1,17 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import os
+from pathlib import Path
+import sys
 import unittest
+import pytest
 
 from microsoft_agents_a365.observability.core import (
     AgentDetails,
     ExecutionType,
     ExecuteToolScope,
+    OpenTelemetryScope,
     Request,
     SourceMetadata,
     TenantDetails,
@@ -30,6 +35,14 @@ class TestExecuteToolScope(unittest.TestCase):
     def setUpClass(cls):
         """Set up test environment once for all tests."""
         # Configure Microsoft Agent 365 for testing
+        os.environ["ENABLE_A365_OBSERVABILITY"] = "true"
+
+        # Set up tracer to capture spans
+        cls.span_exporter = InMemorySpanExporter()
+        tracer_provider = TracerProvider()
+        tracer_provider.add_span_processor(SimpleSpanProcessor(cls.span_exporter))
+        trace.set_tracer_provider(tracer_provider)
+
         configure(
             service_name="test-execute-tool-service",
             service_namespace="test-namespace",
@@ -60,11 +73,6 @@ def test_record_response_method_exists(self):
 
     def test_request_metadata_set_on_span(self):
         """Test that request source metadata is set on span attributes."""
-        span_exporter = InMemorySpanExporter()
-        tracer_provider = TracerProvider()
-        tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
-        trace.set_tracer_provider(tracer_provider)
-
         request = Request(
             content="Execute tool with request metadata",
             execution_type=ExecutionType.AGENT_TO_AGENT,
@@ -79,24 +87,33 @@ def test_request_metadata_set_on_span(self):
         if scope is not None:
             scope.dispose()
 
-        finished_spans = span_exporter.get_finished_spans()
+        finished_spans = self.span_exporter.get_finished_spans()
+        self.assertTrue(finished_spans, "Expected at least one span to be created")
 
-        if finished_spans:
-            span = finished_spans[-1]
-            span_attributes = getattr(span, "attributes", {}) or {}
+        span = finished_spans[-1]
+        span_attributes = getattr(span, "attributes", {}) or {}
 
-            if GEN_AI_EXECUTION_SOURCE_NAME_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_SOURCE_NAME_KEY],
-                    request.source_metadata.name,
-                )
+        self.assertIn(
+            GEN_AI_EXECUTION_SOURCE_NAME_KEY,
+            span_attributes,
+            "Expected source name to be set on span",
+        )
+        self.assertEqual(
+            span_attributes[GEN_AI_EXECUTION_SOURCE_NAME_KEY],
+            request.source_metadata.name,
+        )
 
-            if GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY],
-                    request.source_metadata.description,
-                )
+        self.assertIn(
+            GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY,
+            span_attributes,
+            "Expected source description to be set on span",
+        )
+        self.assertEqual(
+            span_attributes[GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY],
+            request.source_metadata.description,
+        )
 
 
 if __name__ == "__main__":
-    unittest.main(verbosity=2)
+    # Run pytest only on the current file
+    sys.exit(pytest.main([str(Path(__file__))] + sys.argv[1:]))
diff --git a/tests/observability/core/test_inference_scope.py b/tests/observability/core/test_inference_scope.py
@@ -1,13 +1,18 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import os
+from pathlib import Path
+import sys
 import unittest
+import pytest
 
 from microsoft_agents_a365.observability.core import (
     ExecutionType,
     InferenceCallDetails,
     InferenceOperationType,
     InferenceScope,
+    OpenTelemetryScope,
     Request,
     SourceMetadata,
     TenantDetails,
@@ -31,6 +36,14 @@ class TestInferenceScope(unittest.TestCase):
     def setUpClass(cls):
         """Set up test environment once for all tests."""
         # Configure Microsoft Agent 365 for testing
+        os.environ["ENABLE_A365_OBSERVABILITY"] = "true"
+
+        # Set up tracer to capture spans
+        cls.span_exporter = InMemorySpanExporter()
+        tracer_provider = TracerProvider()
+        tracer_provider.add_span_processor(SimpleSpanProcessor(cls.span_exporter))
+        trace.set_tracer_provider(tracer_provider)
+
         configure(
             service_name="test-inference-service",
             service_namespace="test-namespace",
@@ -135,33 +148,36 @@ def test_request_metadata_set_on_span(self):
             source_metadata=SourceMetadata(name="Channel 1", description="Link to channel"),
         )
 
-        span_exporter = InMemorySpanExporter()
-        tracer_provider = TracerProvider()
-        tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
-        trace.set_tracer_provider(tracer_provider)
-
         scope = InferenceScope.start(details, self.agent_details, self.tenant_details, request)
 
         if scope is not None:
             scope.dispose()
 
-        finished_spans = span_exporter.get_finished_spans()
+        finished_spans = self.span_exporter.get_finished_spans()
+        self.assertTrue(finished_spans, "Expected at least one span to be created")
 
-        if finished_spans:
-            span = finished_spans[-1]
-            span_attributes = getattr(span, "attributes", {}) or {}
+        span = finished_spans[-1]
+        span_attributes = getattr(span, "attributes", {}) or {}
 
-            if GEN_AI_EXECUTION_SOURCE_NAME_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_SOURCE_NAME_KEY],
-                    request.source_metadata.name,
-                )
+        self.assertIn(
+            GEN_AI_EXECUTION_SOURCE_NAME_KEY,
+            span_attributes,
+            "Expected source name to be set on span",
+        )
+        self.assertEqual(
+            span_attributes[GEN_AI_EXECUTION_SOURCE_NAME_KEY],
+            request.source_metadata.name,
+        )
 
-            if GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY],
-                    request.source_metadata.description,
-                )
+        self.assertIn(
+            GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY,
+            span_attributes,
+            "Expected source description to be set on span",
+        )
+        self.assertEqual(
+            span_attributes[GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY],
+            request.source_metadata.description,
+        )
 
     def test_inference_scope_context_manager(self):
         """Test InferenceScope as context manager."""
@@ -308,5 +324,5 @@ def test_record_thought_process(self):
 
 
 if __name__ == "__main__":
-    # Run the tests
-    unittest.main(verbosity=2)
+    # Run pytest only on the current file
+    sys.exit(pytest.main([str(Path(__file__))] + sys.argv[1:]))
diff --git a/tests/observability/core/test_invoke_agent_scope.py b/tests/observability/core/test_invoke_agent_scope.py
@@ -1,7 +1,11 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import os
+from pathlib import Path
+import sys
 import unittest
+import pytest
 from urllib.parse import urlparse
 
 from microsoft_agents_a365.observability.core import (
@@ -35,6 +39,14 @@ class TestInvokeAgentScope(unittest.TestCase):
     def setUpClass(cls):
         """Set up test environment once for all tests."""
         # Configure Microsoft Agent 365 for testing
+        os.environ["ENABLE_A365_OBSERVABILITY"] = "true"
+
+        # Set up tracer to capture spans
+        cls.span_exporter = InMemorySpanExporter()
+        tracer_provider = TracerProvider()
+        tracer_provider.add_span_processor(SimpleSpanProcessor(cls.span_exporter))
+        trace.set_tracer_provider(tracer_provider)
+
         configure(
             service_name="test-invoke-agent-service",
             service_namespace="test-namespace",
@@ -121,12 +133,6 @@ def test_record_output_messages_method_exists(self):
 
     def test_request_attributes_set_on_span(self):
         """Test that request parameters from mock data are available on span attributes."""
-        # Set up tracer to capture spans
-        span_exporter = InMemorySpanExporter()
-        tracer_provider = TracerProvider()
-        tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
-        trace.set_tracer_provider(tracer_provider)
-
         # Create scope with request
         scope = InvokeAgentScope.start(
             invoke_agent_details=self.invoke_details,
@@ -138,42 +144,42 @@ def test_request_attributes_set_on_span(self):
             scope.dispose()
 
         # Check if mock data parameters are available in span attributes
-        finished_spans = span_exporter.get_finished_spans()
-
-        if finished_spans:
-            # Get attributes from the span
-            span = finished_spans[-1]
-            span_attributes = getattr(span, "attributes", {}) or {}
-
-            # Verify mock data request parameters are in span attributes
-            # Check source channel name from mock data
-            if GEN_AI_EXECUTION_SOURCE_NAME_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_SOURCE_NAME_KEY],
-                    self.source_metadata.name,  # From cls.source_metadata.name
-                )
-
-            # Check source channel description from mock data
-            if GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY],
-                    self.source_metadata.description,  # From cls.source_metadata.description
-                )
-
-            # Check execution type from mock data
-            if GEN_AI_EXECUTION_TYPE_KEY in span_attributes:
-                self.assertEqual(
-                    span_attributes[GEN_AI_EXECUTION_TYPE_KEY],
-                    self.test_request.execution_type.value,  # From cls.test_request.execution_type
-                )
-
-            # Check input messages contain request content from mock data
-            if GEN_AI_INPUT_MESSAGES_KEY in span_attributes:
-                input_messages = span_attributes[GEN_AI_INPUT_MESSAGES_KEY]
-                self.assertIn(
-                    self.test_request.content,  # From cls.test_request.content
-                    input_messages,
-                )
+        finished_spans = self.span_exporter.get_finished_spans()
+        self.assertTrue(finished_spans, "Expected at least one span to be created")
+
+        # Get attributes from the span
+        span = finished_spans[-1]
+        span_attributes = getattr(span, "attributes", {}) or {}
+
+        # Verify mock data request parameters are in span attributes
+        # Check source channel name from mock data
+        if GEN_AI_EXECUTION_SOURCE_NAME_KEY in span_attributes:
+            self.assertEqual(
+                span_attributes[GEN_AI_EXECUTION_SOURCE_NAME_KEY],
+                self.source_metadata.name,  # From cls.source_metadata.name
+            )
+
+        # Check source channel description from mock data
+        if GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY in span_attributes:
+            self.assertEqual(
+                span_attributes[GEN_AI_EXECUTION_SOURCE_DESCRIPTION_KEY],
+                self.source_metadata.description,  # From cls.source_metadata.description
+            )
+
+        # Check execution type from mock data
+        if GEN_AI_EXECUTION_TYPE_KEY in span_attributes:
+            self.assertEqual(
+                span_attributes[GEN_AI_EXECUTION_TYPE_KEY],
+                self.test_request.execution_type.value,  # From cls.test_request.execution_type
+            )
+
+        # Check input messages contain request content from mock data
+        if GEN_AI_INPUT_MESSAGES_KEY in span_attributes:
+            input_messages = span_attributes[GEN_AI_INPUT_MESSAGES_KEY]
+            self.assertIn(
+                self.test_request.content,  # From cls.test_request.content
+                input_messages,
+            )
 
     def test_caller_agent_client_ip_in_scope(self):
         """Test that caller agent client IP is properly handled when creating InvokeAgentScope."""
@@ -209,4 +215,5 @@ def test_caller_agent_client_ip_in_scope(self):
 
 
 if __name__ == "__main__":
-    unittest.main(verbosity=2)
+    # Run pytest only on the current file
+    sys.exit(pytest.main([str(Path(__file__))] + sys.argv[1:]))
