mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Streaming sample for azurefunctions (#3057)
* Streaming sample for azurefunctions * Fixed links and sample name * Addressed feedback * Addressed feedback * Fixed integration tests * Updated test
This commit is contained in:
Gavin Aguiar
committed by
GitHub
Unverified
GitHub
parent
a118fd5c07
commit
f4ab586f11
@@ -1,102 +0,0 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""
|
||||
Integration Tests for Callbacks Sample
|
||||
|
||||
Tests the callbacks sample for event tracking and management.
|
||||
|
||||
The function app is automatically started by the test fixture.
|
||||
|
||||
Prerequisites:
|
||||
- Azure OpenAI credentials configured (see packages/azurefunctions/tests/integration_tests/.env.example)
|
||||
- Azurite or Azure Storage account configured
|
||||
|
||||
Usage:
|
||||
uv run pytest packages/azurefunctions/tests/integration_tests/test_03_callbacks.py -v
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from .testutils import (
|
||||
TIMEOUT,
|
||||
SampleTestHelper,
|
||||
skip_if_azure_functions_integration_tests_disabled,
|
||||
)
|
||||
|
||||
# Module-level markers - applied to all tests in this file
|
||||
pytestmark = [
|
||||
pytest.mark.sample("03_callbacks"),
|
||||
pytest.mark.usefixtures("function_app_for_test"),
|
||||
skip_if_azure_functions_integration_tests_disabled,
|
||||
]
|
||||
|
||||
|
||||
class TestSampleCallbacks:
|
||||
"""Tests for 03_callbacks sample."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _set_base_url(self, base_url: str) -> None:
|
||||
"""Provide the callback agent base URL for each test."""
|
||||
self.base_url = f"{base_url}/api/agents/CallbackAgent"
|
||||
|
||||
@staticmethod
|
||||
def _wait_for_callback_events(base_url: str, thread_id: str) -> list[dict[str, Any]]:
|
||||
events: list[dict[str, Any]] = []
|
||||
response = SampleTestHelper.get(f"{base_url}/callbacks/{thread_id}")
|
||||
if response.status_code == 200:
|
||||
events = response.json()
|
||||
return events
|
||||
|
||||
def test_agent_with_callbacks(self) -> None:
|
||||
"""Test agent execution with callback tracking."""
|
||||
thread_id = "test-callback"
|
||||
|
||||
response = SampleTestHelper.post_json(
|
||||
f"{self.base_url}/run",
|
||||
{"message": "Tell me about Python", "thread_id": thread_id},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "success"
|
||||
|
||||
events = self._wait_for_callback_events(self.base_url, thread_id)
|
||||
|
||||
assert events
|
||||
assert any(event.get("event_type") == "final" for event in events)
|
||||
|
||||
def test_get_callbacks(self) -> None:
|
||||
"""Test retrieving callback events."""
|
||||
thread_id = "test-callback-retrieve"
|
||||
|
||||
# Send a message first
|
||||
SampleTestHelper.post_json(
|
||||
f"{self.base_url}/run",
|
||||
{"message": "Hello", "thread_id": thread_id, "wait_for_response": False},
|
||||
)
|
||||
|
||||
# Get callbacks
|
||||
response = SampleTestHelper.get(f"{self.base_url}/callbacks/{thread_id}")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_delete_callbacks(self) -> None:
|
||||
"""Test clearing callback events."""
|
||||
thread_id = "test-callback-delete"
|
||||
|
||||
# Send a message first
|
||||
SampleTestHelper.post_json(
|
||||
f"{self.base_url}/run",
|
||||
{"message": "Test", "thread_id": thread_id, "wait_for_response": False},
|
||||
)
|
||||
|
||||
# Delete callbacks
|
||||
response = requests.delete(f"{self.base_url}/callbacks/{thread_id}", timeout=TIMEOUT)
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -0,0 +1,125 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
"""
|
||||
Integration Tests for Reliable Streaming Sample
|
||||
|
||||
Tests the reliable streaming sample using Redis Streams for persistent message delivery.
|
||||
|
||||
The function app is automatically started by the test fixture.
|
||||
|
||||
Prerequisites:
|
||||
- Azure OpenAI credentials configured (see packages/azurefunctions/tests/integration_tests/.env.example)
|
||||
- Azurite or Azure Storage account configured
|
||||
- Redis running (docker run -d --name redis -p 6379:6379 redis:latest)
|
||||
|
||||
Usage:
|
||||
uv run pytest packages/azurefunctions/tests/integration_tests/test_03_reliable_streaming.py -v
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from .testutils import (
|
||||
SampleTestHelper,
|
||||
skip_if_azure_functions_integration_tests_disabled,
|
||||
)
|
||||
|
||||
# Module-level markers - applied to all tests in this file
|
||||
pytestmark = [
|
||||
pytest.mark.sample("03_reliable_streaming"),
|
||||
pytest.mark.usefixtures("function_app_for_test"),
|
||||
skip_if_azure_functions_integration_tests_disabled,
|
||||
]
|
||||
|
||||
|
||||
class TestSampleReliableStreaming:
|
||||
"""Tests for 03_reliable_streaming sample."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _set_base_url(self, base_url: str) -> None:
|
||||
"""Provide the base URL for each test."""
|
||||
self.base_url = base_url
|
||||
self.agent_url = f"{base_url}/api/agents/TravelPlanner"
|
||||
self.stream_url = f"{base_url}/api/agent/stream"
|
||||
|
||||
def test_agent_run_and_stream(self) -> None:
|
||||
"""Test agent execution with Redis streaming."""
|
||||
# Start agent run
|
||||
response = SampleTestHelper.post_json(
|
||||
f"{self.agent_url}/run",
|
||||
{"message": "Plan a 1-day trip to Seattle in 1 sentence", "wait_for_response": False},
|
||||
)
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
|
||||
thread_id = data.get("thread_id")
|
||||
|
||||
# Wait a moment for the agent to start writing to Redis
|
||||
time.sleep(2)
|
||||
|
||||
# Stream response from Redis with shorter timeout
|
||||
# Note: We use text/plain to avoid SSE parsing complexity
|
||||
stream_response = requests.get(
|
||||
f"{self.stream_url}/{thread_id}",
|
||||
headers={"Accept": "text/plain"},
|
||||
timeout=30, # Shorter timeout for test
|
||||
)
|
||||
assert stream_response.status_code == 200
|
||||
|
||||
def test_stream_with_sse_format(self) -> None:
|
||||
"""Test streaming with Server-Sent Events format."""
|
||||
# Start agent run
|
||||
response = SampleTestHelper.post_json(
|
||||
f"{self.agent_url}/run",
|
||||
{"message": "What's the weather like?", "wait_for_response": False},
|
||||
)
|
||||
assert response.status_code == 202
|
||||
data = response.json()
|
||||
thread_id = data.get("thread_id")
|
||||
|
||||
# Wait for agent to start writing
|
||||
time.sleep(2)
|
||||
|
||||
# Stream with SSE format
|
||||
stream_response = requests.get(
|
||||
f"{self.stream_url}/{thread_id}",
|
||||
headers={"Accept": "text/event-stream"},
|
||||
timeout=30, # Shorter timeout
|
||||
)
|
||||
assert stream_response.status_code == 200
|
||||
content_type = stream_response.headers.get("content-type", "")
|
||||
assert "text/event-stream" in content_type
|
||||
|
||||
# Check for SSE event markers if we got content
|
||||
content = stream_response.text
|
||||
if content:
|
||||
assert "event:" in content or "data:" in content
|
||||
|
||||
def test_stream_nonexistent_conversation(self) -> None:
|
||||
"""Test streaming from a non-existent conversation.
|
||||
|
||||
The endpoint will wait for data in Redis, but since the conversation
|
||||
doesn't exist, it will timeout. This is expected behavior.
|
||||
"""
|
||||
fake_id = "nonexistent-conversation-12345"
|
||||
|
||||
# Should timeout since the conversation doesn't exist
|
||||
with pytest.raises(requests.exceptions.ReadTimeout):
|
||||
requests.get(
|
||||
f"{self.stream_url}/{fake_id}",
|
||||
headers={"Accept": "text/plain"},
|
||||
timeout=10, # Short timeout for non-existent ID
|
||||
)
|
||||
|
||||
def test_health_endpoint(self) -> None:
|
||||
"""Test health check endpoint."""
|
||||
response = SampleTestHelper.get(f"{self.base_url}/api/health")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "healthy"
|
||||
assert "agents" in data
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user