Testcases

Author: UtkarshMishra-Microsoft

src/backend/agents/tech_support.py

@@ -157,19 +157,23 @@ async def input_task_endpoint(input_task: InputTask, request: Request):
     track_event(
         "InputTaskProcessed",
         {
-            "status": f"Plan created:\n {plan.summary}"
-            if plan.id
-            else "Error occurred: Plan ID is empty",
+            "status": (
+                f"Plan created:\n {plan.summary}"
+                if plan.id
+                else "Error occurred: Plan ID is empty"
+            ),
             "session_id": input_task.session_id,
             "plan_id": plan.id,
             "description": input_task.description,
         },
     )
 
     return {
-        "status": f"Plan created:\n {plan.summary}"
-        if plan.id
-        else "Error occurred: Plan ID is empty",
+        "status": (
+            f"Plan created:\n {plan.summary}"
+            if plan.id
+            else "Error occurred: Plan ID is empty"
+        ),
         "session_id": input_task.session_id,
         "plan_id": plan.id,
         "description": input_task.description,

src/backend/app.py

@@ -1,6 +1,5 @@
 from opentelemetry import trace
-from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import \
-    OTLPSpanExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
@@ -13,4 +12,4 @@ def configure_oltp_tracing(endpoint: str = None) -> trace.TracerProvider:
     tracer_provider.add_span_processor(processor)
     trace.set_tracer_provider(tracer_provider)
 
-    return tracer_provider
+    return tracer_provider

src/backend/otlp_tracing.py

@@ -4,7 +4,8 @@
 import json  # Fix for missing import
 from unittest.mock import AsyncMock, MagicMock, patch
 from pydantic import ValidationError
-sys.modules['azure.monitor.events.extension'] = MagicMock()
+
+sys.modules["azure.monitor.events.extension"] = MagicMock()
 
 # Set environment variables to mock Config dependencies before any import
 os.environ["COSMOSDB_ENDPOINT"] = "https://mock-endpoint"
@@ -15,14 +16,15 @@
 os.environ["AZURE_OPENAI_API_VERSION"] = "2023-01-01"
 os.environ["AZURE_OPENAI_ENDPOINT"] = "https://mock-openai-endpoint"
 
-from autogen_core.components.models import AssistantMessage, AzureOpenAIChatCompletionClient
+from autogen_core.components.models import (
+    AssistantMessage,
+    AzureOpenAIChatCompletionClient,
+)
 from src.backend.models.messages import Step
 from src.backend.context.cosmos_memory import CosmosBufferedChatCompletionContext
 from src.backend.agents.agentutils import extract_and_update_transition_states
 
 
-
-
 @pytest.mark.asyncio
 async def test_extract_and_update_transition_states_invalid_response():
     """Test handling of invalid JSON response from model client."""
@@ -41,8 +43,11 @@ async def test_extract_and_update_transition_states_invalid_response():
     cosmos_mock = MagicMock()
 
     model_client.create.return_value = MagicMock(content="invalid_json")
-    
-    with patch("src.backend.context.cosmos_memory.CosmosBufferedChatCompletionContext", cosmos_mock):
+
+    with patch(
+        "src.backend.context.cosmos_memory.CosmosBufferedChatCompletionContext",
+        cosmos_mock,
+    ):
         with pytest.raises(json.JSONDecodeError):
             await extract_and_update_transition_states(
                 step=step,
@@ -54,6 +59,7 @@ async def test_extract_and_update_transition_states_invalid_response():
 
     cosmos_mock.update_step.assert_not_called()
 
+
 @pytest.mark.asyncio
 async def test_extract_and_update_transition_states_validation_error():
     """Test handling of a response missing required fields."""
@@ -71,12 +77,15 @@ async def test_extract_and_update_transition_states_validation_error():
     model_client = AsyncMock()
     cosmos_mock = MagicMock()
 
-    invalid_response = {"identifiedTargetState": "state1"}  # Missing 'identifiedTargetTransition'
-    model_client.create.return_value = MagicMock(
-        content=json.dumps(invalid_response)
-    )
-    
-    with patch("src.backend.context.cosmos_memory.CosmosBufferedChatCompletionContext", cosmos_mock):
+    invalid_response = {
+        "identifiedTargetState": "state1"
+    }  # Missing 'identifiedTargetTransition'
+    model_client.create.return_value = MagicMock(content=json.dumps(invalid_response))
+
+    with patch(
+        "src.backend.context.cosmos_memory.CosmosBufferedChatCompletionContext",
+        cosmos_mock,
+    ):
         with pytest.raises(ValidationError):
             await extract_and_update_transition_states(
                 step=step,
@@ -88,6 +97,7 @@ async def test_extract_and_update_transition_states_validation_error():
 
     cosmos_mock.update_step.assert_not_called()
 
+
 def test_step_initialization():
     """Test Step initialization with valid data."""
     step = Step(
@@ -110,6 +120,7 @@ def test_step_initialization():
     assert step.status == "planned"
     assert step.human_approval_status == "requested"
 
+
 def test_step_missing_required_fields():
     """Test Step initialization with missing required fields."""
     with pytest.raises(ValidationError):

src/backend/tests/agents/test_agentutils.py

@@ -4,7 +4,8 @@
 import json  # Fix for missing import
 from unittest.mock import AsyncMock, MagicMock, patch
 from pydantic import ValidationError
-sys.modules['azure.monitor.events.extension'] = MagicMock()
+
+sys.modules["azure.monitor.events.extension"] = MagicMock()
 
 # Set environment variables to mock Config dependencies before any import
 os.environ["COSMOSDB_ENDPOINT"] = "https://mock-endpoint"
@@ -26,7 +27,6 @@
 from src.backend.agents.generic import get_generic_tools, GenericAgent, dummy_function
 
 
-
 class TestGenericAgent(unittest.TestCase):
     def setUp(self):
         self.mock_model_client = MagicMock(spec=AzureOpenAIChatCompletionClient)
@@ -36,7 +36,7 @@ def setUp(self):
         self.mock_tools = get_generic_tools()
         self.mock_agent_id = MagicMock(spec=AgentId)
 
- 
+
 class TestDummyFunction(unittest.IsolatedAsyncioTestCase):
     async def test_dummy_function(self):
         result = await dummy_function()