Guidelines for writing reliable integration and end-to-end tests that verify VoxAgent's modules work together correctly. Covers test architecture, API testing with httpx, pipeline E2E flows, fixture design, and test data management.
-
itest-layer-separation — Clearly separate test layers. Each layer has different scope and speed:
Unit tests: Single function/class, no I/O, < 10ms each
Integration tests: Multiple modules, mocked external APIs, < 500ms each
API tests: HTTP endpoints via TestClient, mocked providers, < 1s each
E2E tests: Full pipeline, mocked audio/TTS, < 5s each
-
itest-naming — Name test files and functions to reflect scope and intent:
def test_intent_classifier_returns_open_app():
async def test_pipeline_routes_to_correct_skill():
async def test_chat_endpoint_returns_streaming_response():
async def test_voice_command_opens_application():
-
itest-mock-at-boundary — Mock at module boundaries, not inside implementations. Replace the provider, not the HTTP call:
mock_llm = AsyncMock(spec=LLMProvider)
mock_llm.chat.return_value = "Opening Spotify"
pipeline = Pipeline(llm=mock_llm)
with patch("httpx.AsyncClient.post"):
...
-
itest-no-sleep — Never use time.sleep() or asyncio.sleep() in tests. Use events, conditions, or asyncio.wait_for():
result_ready = asyncio.Event()
async def on_complete(result):
result_ready.set()
await asyncio.wait_for(result_ready.wait(), timeout=5.0)
await asyncio.sleep(2)
assert result is not None
-
itest-testclient — Use httpx AsyncClient with ASGITransport for async FastAPI testing:
from httpx import ASGITransport, AsyncClient
from api.main import app
@pytest.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield client
async def test_health_endpoint(client: AsyncClient):
response = await client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
-
itest-auth-fixtures — Create reusable auth fixtures for authenticated endpoints:
@pytest.fixture
def auth_headers() -> dict[str, str]:
token = "test-token-for-ci"
return {"Authorization": f"Bearer {token}"}
async def test_execute_requires_auth(client: AsyncClient):
response = await client.post("/execute", json={"skill": "test"})
assert response.status_code == 401
async def test_execute_with_auth(client: AsyncClient, auth_headers):
response = await client.post(
"/execute",
json={"skill": "open_app", "params": {"name": "spotify"}},
headers=auth_headers,
)
assert response.status_code == 200
-
itest-error-scenarios — Test error paths explicitly. Every endpoint needs tests for: 400 (bad input), 401 (no auth), 404 (not found), 503 (provider down):
async def test_chat_with_invalid_model(client: AsyncClient, auth_headers):
response = await client.post(
"/chat",
json={"messages": [{"role": "user", "content": "hi"}], "model": "nonexistent"},
headers=auth_headers,
)
assert response.status_code == 404
assert "model" in response.json()["detail"].lower()
-
itest-response-schema — Validate response shape, not just status code. Use Pydantic for assertion:
from api.schemas import ChatResponse
async def test_chat_response_schema(client: AsyncClient, auth_headers):
response = await client.post("/chat", json=payload, headers=auth_headers)
data = ChatResponse.model_validate(response.json())
assert data.model_used in VALID_MODELS
assert data.latency_ms > 0
-
itest-pipeline-flow — Test the full pipeline from transcript to skill execution:
async def test_full_pipeline_open_app(mock_providers):
pipeline = VoicePipeline(
stt=mock_providers.stt,
intent=mock_providers.intent,
llm=mock_providers.llm,
executor=mock_providers.executor,
tts=mock_providers.tts,
)
mock_providers.stt.transcribe.return_value = "open spotify"
mock_providers.intent.classify.return_value = Intent(action="open_app", params={"app": "spotify"})
result = await pipeline.process(audio=MOCK_AUDIO)
assert result.skill_executed == "open_app"
mock_providers.tts.speak.assert_called_once()
-
itest-provider-fallback-e2e — Test that provider fallback chains work end-to-end:
async def test_llm_fallback_on_primary_failure():
primary = AsyncMock(spec=LLMProvider)
primary.chat.side_effect = ProviderUnavailableError("rate limited")
fallback = AsyncMock(spec=LLMProvider)
fallback.chat.return_value = "I opened Spotify"
chain = ProviderChain([primary, fallback])
result = await chain.chat(messages)
assert result == "I opened Spotify"
primary.chat.assert_called_once()
fallback.chat.assert_called_once()
-
itest-concurrent-sessions — Verify multiple voice sessions don't interfere:
async def test_concurrent_sessions_isolated():
session_a = await create_session("user_a")
session_b = await create_session("user_b")
await session_a.process("open spotify")
await session_b.process("what's the weather")
assert session_a.last_skill != session_b.last_skill
assert session_a.context.history != session_b.context.history
-
itest-conftest-layered — Layer conftest.py files. Each test directory has its own with appropriate scope:
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture
def mock_keyring(monkeypatch):
monkeypatch.setattr("keyring.get_password", lambda *a: "test-key")
@pytest.fixture
async def client(mock_keyring):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest.fixture
def mock_audio() -> bytes:
return b"\x00" * 16000 * 2
-
itest-fixture-scope — Use the narrowest fixture scope that makes sense:
function (default): New instance per test — most isolated
class: Shared within a test class
module: Shared within a file — for expensive setup
session: Shared across all tests — for event loop, DB schema
-
itest-fixture-composition — Compose fixtures from smaller, focused fixtures:
@pytest.fixture
def mock_stt():
stt = AsyncMock(spec=STTProvider)
stt.transcribe.return_value = "hello world"
return stt
@pytest.fixture
def mock_llm():
llm = AsyncMock(spec=LLMProvider)
llm.chat.return_value = "Hi there!"
return llm
@pytest.fixture
def mock_providers(mock_stt, mock_llm):
return MockProviders(stt=mock_stt, llm=mock_llm)
-
itest-fixture-cleanup — Always clean up resources in fixtures. Use yield, not return, when cleanup is needed:
@pytest.fixture
async def db():
conn = await aiosqlite.connect(":memory:")
await init_schema(conn)
yield conn
await conn.close()
-
itest-factory-functions — Use factory functions for test data. Never duplicate test object creation:
def make_message(role: str = "user", content: str = "test") -> Message:
return Message(role=role, content=content)
def make_intent(action: str = "open_app", **params) -> Intent:
return Intent(action=action, parameters=params, confidence=0.95)
messages = [make_message("user", "open spotify"), make_message("assistant", "ok")]
-
itest-mock-specificity — Mock at the right level of specificity. Too broad hides bugs, too narrow is brittle:
mock_llm = AsyncMock(spec=LLMProvider)
mock_registry = AsyncMock(spec=ProviderRegistry)
with patch("httpx.AsyncClient.post"):
-
itest-deterministic — Make tests deterministic. Seed random values, freeze time, control async ordering:
from unittest.mock import patch
import time
@pytest.fixture
def frozen_time():
with patch("time.time", return_value=1700000000.0):
yield
async def test_session_expiry(frozen_time):
session = Session(created_at=time.time())
assert not session.is_expired
-
itest-markers — Use pytest markers to categorize tests. Run fast tests in CI, slow tests in nightly:
[tool.pytest.ini_options]
markers = [
"unit: Fast isolated tests (< 10ms)",
"integration: Multi-module tests (< 500ms)",
"api: HTTP endpoint tests (< 1s)",
"e2e: Full pipeline tests (< 5s)",
"benchmark: Performance benchmarks",
]
@pytest.mark.integration
async def test_pipeline_routes_correctly():
...
