Consolidating StreamingResponse definitions and slight fixes

dev/microsoft-agents-testing/microsoft_agents/testing/core/utils.py

@@ -90,4 +90,4 @@ def generate_token_from_config(sdk_config: dict, connection_name: str = "SERVICE
 
     if not client_id or not client_secret or not tenant_id:
         raise ValueError("Incorrect configuration provided for token generation.")
-    return generate_token(client_id, client_secret, tenant_id)
+    return generate_token(client_id, client_secret, tenant_id)

dev/microsoft-agents-testing/microsoft_agents/testing/utils.py

@@ -102,4 +102,4 @@ def resolve_scenario(scenario_or_str: Scenario | str ) -> Scenario:
         else:
             return scenario_registry.get(scenario_or_str)
     else:
-        raise TypeError("Input must be a Scenario instance or a string key.")
+        raise TypeError("Input must be a Scenario instance or a string key.")

dev/tests/sdk/test_streaming_response.py

@@ -0,0 +1,122 @@
+import pytest
+import asyncio
+
+from microsoft_agents.activity import (
+    Activity,
+    ActivityTypes,
+    Channels,
+    Entity
+)
+
+from microsoft_agents.hosting.core import (
+    TurnContext,
+    TurnState,
+)
+
+from microsoft_agents.testing import (
+    AgentClient,
+    AgentEnvironment,
+    AiohttpScenario,
+)
+
+FULL_TEXT = "This is a streaming response."
+CHUNKS = FULL_TEXT.split()
+
+def get_streaminfo(activity: Activity) -> Entity:
+    for entity in activity.entities:
+        if isinstance(entity, dict) and entity.get("type") == "streaminfo":
+            return Entity.model_validate(entity)
+        elif isinstance(entity, Entity) and entity.type == "streaminfo":
+            return entity
+    raise ValueError("No streaminfo entity found")
+
+async def init_agent(env: AgentEnvironment):
+
+    app = env.agent_application
+
+    @app.message("/stream")
+    async def stream_handler(context: TurnContext, state: TurnState):
+
+        assert context.streaming_response is not None
+
+        context.streaming_response.queue_informative_update("Starting stream...")
+
+        for chunk in CHUNKS:
+            await asyncio.sleep(1.0)  # Simulate delay between chunks
+            context.streaming_response.queue_text_chunk(chunk)
+
+        await asyncio.sleep(1.0)
+
+        await context.streaming_response.end_stream()
+
+_SCENARIO = AiohttpScenario(init_agent=init_agent, use_jwt_middleware=False)
+
+@pytest.mark.asyncio
+@pytest.mark.agent_test(_SCENARIO)
+async def test_basic_streaming_response_non_streaming_channel(agent_client: AgentClient):
+
+    expected_len = len(FULL_TEXT.split())
+
+    agent_client.template = agent_client.template.with_updates(channel_id=Channels.emulator)
+
+    # give enough time for all the activities to send
+    await agent_client.send("/stream", wait=expected_len * 2.0)
+
+    stream_activities = agent_client.select().where(
+        entities=lambda x: any(e["type"] == "streaminfo" for e in x)
+    ).get()
+
+    assert len(stream_activities) == 1
+
+    final_streaminfo = get_streaminfo(stream_activities[0])
+
+    assert final_streaminfo.stream_sequence == 1
+    assert final_streaminfo.stream_type == "final"
+    assert stream_activities[0].text == FULL_TEXT.replace(" ", "")
+
+
+
+@pytest.mark.asyncio
+@pytest.mark.agent_test(_SCENARIO)
+async def test_basic_streaming_response_streaming_channel(agent_client: AgentClient):
+
+    expected_len = len(FULL_TEXT.split())
+
+    agent_client.template = agent_client.template.with_updates(channel_id=Channels.webchat)
+
+    # give enough time for all the activities to send
+    await agent_client.send("/stream", wait=expected_len * 2.0)
+
+    stream_activities = agent_client.select().where(
+        entities=lambda x: any(e["type"] == "streaminfo" for e in x)
+    ).get()
+
+    assert len(stream_activities) == len(CHUNKS) + 2
+
+    informative = stream_activities[0]
+    informative_streaminfo = get_streaminfo(informative)
+
+    assert informative_streaminfo.stream_type == "informative"
+    assert informative_streaminfo.stream_sequence == 1
+    assert informative.text == "Starting stream..."
+    assert informative.type == ActivityTypes.typing
+
+    t = ""
+    for i, chunk in enumerate(CHUNKS):
+        t += chunk
+
+        j = i + 1
+
+        streaminfo = get_streaminfo(stream_activities[j])
+
+        assert stream_activities[j].text == t
+        assert stream_activities[j].type == ActivityTypes.typing
+        assert streaminfo.stream_type == "streaming"
+        assert streaminfo.stream_sequence == j + 1
+
+    final_streaminfo = get_streaminfo(stream_activities[-1])
+
+    assert final_streaminfo.stream_sequence == len(stream_activities)
+    assert final_streaminfo.stream_type == "final"
+    assert stream_activities[-1].text == FULL_TEXT.replace(" ", "")
+