Module 8: Unified AI Client — Final integrating project
Testing and validation
Your client has five big features: adapters, factory, fallback, routing, metrics. Without tests, a refactor breaks something and you don't notice until a user reports the bug. This capsule teaches you to write the suite that lets you modify with confidence.
By the end you'll be able to:
- Structure tests, separating unit (with mocks) from integration (against a real API)
- Mock adapters to test
UnifiedClientwithout paying for tokens - Verify fallback by simulating provider failures
- Validate contracts (Pydantic) of inputs/outputs
- Run the suite with
pytestand understand the output
Mental model: three levels of test
| Level | What it tests | Speed | When to run |
|---|---|---|---|
| Unit | Internal logic (factory, routing, metrics) with adapter mocks | <1s each | Every commit |
| Integration | Real adapters against an API (OpenAI, local Ollama) | 5-30s each | Before merge |
| Contract | That the provider's response meets the expected shape (ChatResponse) | 5-10s | Before release |
Most of your suite is unit tests with mocks. Integration tests are few and selective (they can't run in CI without API keys).
pytest setup
Create tests/conftest.py:
# tests/conftest.py
"""Shared fixtures for tests."""
import pytest
from unittest.mock import MagicMock
from unified_ai_client.models import ProviderConfig, ChatResponse, Message
from unified_ai_client.adapters.base import BaseAdapter
@pytest.fixture
def basic_provider_config():
return ProviderConfig(
name="test-provider",
type="openai",
model="test-model",
api_key_env="TEST_KEY",
price_input_per_1m=0.10,
price_output_per_1m=0.20,
)
class FakeAdapter(BaseAdapter):
"""Fake adapter that returns a predictable response. Useful in tests."""
def __init__(self, config: ProviderConfig, response_text: str = "OK",
raises: Exception | None = None):
super().__init__(config)
self.response_text = response_text
self.raises = raises
self.calls = 0 # counter to verify usage
def chat(self, messages, max_tokens=256, temperature=0.7) -> ChatResponse:
self.calls += 1
if self.raises:
raise self.raises
return ChatResponse(
text=self.response_text,
model=self.config.model or "test",
provider=self.name,
tokens_input=10,
tokens_output=20,
duration_ms=100,
cost_usd=0.000005,
)
@pytest.fixture
def fake_adapter_factory():
"""FakeAdapter factory to use in tests."""
def create(name: str, response: str = "OK", raises: Exception | None = None):
config = ProviderConfig(name=name, type="openai", model="test", api_key_env="X")
return FakeAdapter(config, response_text=response, raises=raises)
return create
Test 1 — Models (Pydantic validation)
tests/test_models.py:
# tests/test_models.py
import pytest
from pydantic import ValidationError
from unified_ai_client.models import Message, ChatResponse, ProviderConfig
def test_message_accepts_valid_roles():
Message(role="system", content="hi")
Message(role="user", content="hi")
Message(role="assistant", content="hi")
def test_message_rejects_invalid_role():
with pytest.raises(ValidationError):
Message(role="invalid", content="hi")
def test_chat_response_defaults():
r = ChatResponse(text="hi", model="m", provider="p")
assert r.tokens_input == 0
assert r.cost_usd is None
def test_provider_config_accepts_valid_types():
ProviderConfig(name="x", type="openai")
ProviderConfig(name="x", type="openai_compatible")
ProviderConfig(name="x", type="modal_custom")
def test_provider_config_rejects_invalid_type():
with pytest.raises(ValidationError):
ProviderConfig(name="x", type="nonexistent")
Test 2 — Factory (type → adapter mapping)
tests/test_factory.py:
# tests/test_factory.py
import pytest
from unified_ai_client.factory import ProviderFactory
from unified_ai_client.models import ProviderConfig
from unified_ai_client.adapters.openai_adapter import OpenAIAdapter
from unified_ai_client.exceptions import ConfigError
def test_factory_creates_openai_adapter(monkeypatch):
monkeypatch.setenv("FAKE_KEY", "x")
config = ProviderConfig(name="x", type="openai", model="gpt-4o-mini", api_key_env="FAKE_KEY")
adapter = ProviderFactory.create(config)
assert isinstance(adapter, OpenAIAdapter)
assert adapter.name == "x"
def test_factory_raises_for_unknown_type():
config = ProviderConfig(name="x", type="modal_custom")
with pytest.raises(ConfigError):
ProviderFactory.create(config)
def test_factory_allows_registering_custom(monkeypatch):
from unified_ai_client.adapters.base import BaseAdapter
class MyAdapter(BaseAdapter):
def chat(self, messages, max_tokens=256, temperature=0.7):
...
ProviderFactory.register("my_type", MyAdapter)
config = ProviderConfig(name="x", type="openai") # note: the Literal accepts types, so for real registry tests adjust
# Here the test would demonstrate that registration works if the type is allowed
Test 3 — UnifiedClient with FakeAdapter
tests/test_client.py:
# tests/test_client.py
import pytest
from unittest.mock import patch
from unified_ai_client import UnifiedClient
from unified_ai_client.models import ClientConfig, ProviderConfig
from unified_ai_client.exceptions import (
AllProvidersFailedError,
RateLimitError,
AuthError,
TimeoutError,
ConfigError,
)
from tests.conftest import FakeAdapter
def create_client_with_fakes(primary_fake, fallback_fakes=None):
"""Helper: creates a real UnifiedClient but with FakeAdapters injected."""
fallback_fakes = fallback_fakes or []
config = ClientConfig(
primary=primary_fake.name,
fallback=[f.name for f in fallback_fakes],
providers={
primary_fake.name: primary_fake.config,
**{f.name: f.config for f in fallback_fakes},
},
)
# Create the client and replace the real adapters with fakes
with patch("unified_ai_client.factory.ProviderFactory.create",
side_effect=lambda cfg: next(
a for a in [primary_fake, *fallback_fakes] if a.name == cfg.name
)):
client = UnifiedClient(config)
return client
def test_chat_returns_primary_response(fake_adapter_factory):
primary = fake_adapter_factory("primary", response="from primary")
client = create_client_with_fakes(primary)
r = client.chat("hi")
assert r.text == "from primary"
assert r.provider == "primary"
assert primary.calls == 1
def test_fallback_when_primary_fails(fake_adapter_factory):
primary = fake_adapter_factory(
"primary",
raises=TimeoutError("primary", "timeout"),
)
fb = fake_adapter_factory("fallback", response="rescue")
client = create_client_with_fakes(primary, [fb])
r = client.chat("hi")
assert r.text == "rescue"
assert r.provider == "fallback"
assert primary.calls == 1
assert fb.calls == 1
def test_raises_error_if_all_fail(fake_adapter_factory):
p = fake_adapter_factory("p", raises=TimeoutError("p", "t"))
fb = fake_adapter_factory("fb", raises=AuthError("fb", "bad key"))
client = create_client_with_fakes(p, [fb])
with pytest.raises(AllProvidersFailedError) as exc:
client.chat("hi")
# Must contain detail of both errors
assert "p" in str(exc.value)
assert "fb" in str(exc.value)
def test_circuit_breaker_opens_after_n_failures(fake_adapter_factory):
primary = fake_adapter_factory("primary", raises=TimeoutError("primary", "t"))
fb = fake_adapter_factory("fb", response="ok")
client = create_client_with_fakes(primary, [fb])
# Do 3 requests; each one fails primary and goes to the fallback
for _ in range(3):
client.chat("hi")
assert primary.calls == 3
# After 3 consecutive failures, the circuit opens. Next request doesn't touch primary.
primary.calls = 0
client.chat("hi")
assert primary.calls == 0 # circuit open, primary skipped
assert fb.calls == 4
def test_use_fallback_false_doesnt_try_others(fake_adapter_factory):
p = fake_adapter_factory("p", raises=TimeoutError("p", "t"))
fb = fake_adapter_factory("fb", response="ok")
client = create_client_with_fakes(p, [fb])
with pytest.raises(AllProvidersFailedError):
client.chat("hi", use_fallback=False)
assert p.calls == 1
assert fb.calls == 0
def test_metrics_record_successes_and_errors(fake_adapter_factory):
p = fake_adapter_factory("primary", raises=TimeoutError("primary", "t"))
fb = fake_adapter_factory("fb", response="ok")
client = create_client_with_fakes(p, [fb])
client.chat("one")
client.chat("two")
metrics = client.get_metrics()
assert metrics.total_requests == 4 # 2 primary errors + 2 fb successes
assert metrics.total_successes == 2
assert metrics.total_errors == 2
assert metrics.by_provider["primary"]["errors"] == 2
assert metrics.by_provider["fb"]["successes"] == 2
def test_routing_cost_first(fake_adapter_factory):
expensive = fake_adapter_factory("expensive", response="expensive")
expensive.config.cost_tier = "high"
cheap = fake_adapter_factory("cheap", response="cheap")
cheap.config.cost_tier = "low"
client = create_client_with_fakes(expensive, [cheap])
r = client.chat("hi", priority="cost-first")
# cost-first should go to the cheap one (low) first before the expensive one (high)
assert r.provider == "cheap"
Test 4 — Contract testing of the OpenAIAdapter
When you hit the real API (integration test), verify that the response meets your contract:
tests/test_integration_openai.py:
# tests/test_integration_openai.py
"""Tests that hit the real API. Only run with OPENAI_API_KEY set."""
import os
import pytest
from unified_ai_client import UnifiedClient
from unified_ai_client.models import ChatResponse
skip_if_no_key = pytest.mark.skipif(
not os.environ.get("OPENAI_API_KEY"),
reason="OPENAI_API_KEY not set"
)
@skip_if_no_key
def test_openai_real_returns_valid_chat_response():
CONFIG = {
"primary": "openai-mini",
"providers": {
"openai-mini": {
"name": "openai-mini",
"type": "openai",
"model": "gpt-4o-mini",
"api_key_env": "OPENAI_API_KEY",
"price_input_per_1m": 0.15,
"price_output_per_1m": 0.60,
}
},
}
client = UnifiedClient.from_dict(CONFIG)
r = client.chat("Say 'hi' and nothing else", max_tokens=10)
# Contract assertions
assert isinstance(r, ChatResponse)
assert len(r.text) > 0
assert r.provider == "openai-mini"
assert r.model == "gpt-4o-mini"
assert r.tokens_input > 0
assert r.tokens_output > 0
assert r.duration_ms > 0
assert r.cost_usd is not None and r.cost_usd > 0
assert r.raw_response is not None
Running the tests
# All tests
pytest tests/ -v
# Only unit tests (without touching real APIs)
pytest tests/ -v -k "not integration"
# Only integration tests
pytest tests/test_integration_openai.py -v
# With coverage
pip install pytest-cov
pytest tests/ --cov=unified_ai_client --cov-report=term-missing
Expected output of unit tests:
tests/test_models.py::test_message_accepts_valid_roles PASSED
tests/test_models.py::test_message_rejects_invalid_role PASSED
tests/test_models.py::test_chat_response_defaults PASSED
tests/test_models.py::test_provider_config_accepts_valid_types PASSED
tests/test_models.py::test_provider_config_rejects_invalid_type PASSED
tests/test_factory.py::test_factory_creates_openai_adapter PASSED
tests/test_factory.py::test_factory_raises_for_unknown_type PASSED
tests/test_client.py::test_chat_returns_primary_response PASSED
tests/test_client.py::test_fallback_when_primary_fails PASSED
tests/test_client.py::test_raises_error_if_all_fail PASSED
tests/test_client.py::test_circuit_breaker_opens_after_n_failures PASSED
tests/test_client.py::test_use_fallback_false_doesnt_try_others PASSED
tests/test_client.py::test_metrics_record_successes_and_errors PASSED
tests/test_client.py::test_routing_cost_first PASSED
14 passed in 0.42s
Under half a second, perfect for CI.
CI: GitHub Actions
.github/workflows/test.yml:
name: Tests
on: [push, pull_request]
jobs:
unit-tests:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.11"
- run: pip install -e ".[dev]"
- run: pytest tests/ -v -k "not integration"
integration-tests:
runs-on: ubuntu-latest
needs: unit-tests
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.11"
- run: pip install -e ".[dev]"
- env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: pytest tests/test_integration_openai.py -v
Unit tests run on every PR (fast, no API keys). Integration tests only on main (with GitHub secrets).
Advanced patterns
Pattern 1 — Property-based testing with Hypothesis
To escape "I tested the cases that occurred to me":
from hypothesis import given, strategies as st
@given(st.text(min_size=1, max_size=1000))
def test_chat_accepts_any_prompt(prompt):
client = create_client_with_fakes(fake_adapter("p", response="ok"))
r = client.chat(prompt)
assert r.text == "ok"
Pattern 2 — Snapshot testing
If you're worried that accidental changes modify the output format:
import json
def test_metrics_format_stable(snapshot, ...):
metrics_json = client.metrics.to_json()
snapshot.assert_match(metrics_json, "metrics.json")
If the format changes, the test fails and you must confirm the change.
Pattern 3 — Regression test per bug fix
Every time you fix a bug, write a test that fails without the fix. That guarantees the bug never comes back.
Common traps
Trap 1 — "Tests that depend on order."
If one test resets shared state and the next one assumes it's reset, your suite is fragile. Use fixtures with scope="function" (default) to isolate.
Trap 2 — "I mock internal things and my test passes but the real one fails."
Mock edges (HTTP calls, system clock), not the internals of your own library. If you mock OpenAIAdapter.chat, your test checks "that UnifiedClient calls chat", not "that the integration with OpenAI works".
Trap 3 — "No integration tests." Unit tests with mocks don't detect: breaking changes in the OpenAI SDK, auth errors, real-response format problems. You need some integration tests running periodically.
Trap 4 — "Integration tests on every CI run."
$0.10 per test × 50 tests × 30 PRs/month = $150/month just in CI. Filter integration tests to main or a nightly cron.
Trap 5 — "The suite takes 5 minutes." Suspect missing mocks. A decent unit suite for a library this size should run in <2 seconds.
Exercise
Write tests for:
- That
quality-firstrouting ranks correctly when there are 3 providers with differentcost_tier - That
reset_metricseffectively clears the collector and the next summary shows 0 requests - That an
AuthErrorin primary doesn't stop fallback (the client should keep trying OpenRouter) - That an empty prompt fails with
ValidationErrorbefore reaching an adapter
See solutions
def test_routing_quality_first_goes_expensive_first(fake_adapter_factory):
cheap = fake_adapter_factory("cheap", raises=TimeoutError("b", "t"))
cheap.config.cost_tier = "low"
medium = fake_adapter_factory("medium", raises=TimeoutError("m", "t"))
medium.config.cost_tier = "medium"
expensive = fake_adapter_factory("expensive", response="expensive_ok")
expensive.config.cost_tier = "high"
# Any primary; priority changes the order
client = create_client_with_fakes(medium, [cheap, expensive])
r = client.chat("hi", priority="quality-first")
assert r.provider == "expensive"
def test_reset_metrics(fake_adapter_factory):
p = fake_adapter_factory("p", response="x")
client = create_client_with_fakes(p)
client.chat("one")
assert client.get_metrics().total_requests == 1
client.reset_metrics()
assert client.get_metrics().total_requests == 0
def test_auth_error_in_primary_continues_to_fallback(fake_adapter_factory):
p = fake_adapter_factory("p", raises=AuthError("p", "bad key"))
fb = fake_adapter_factory("fb", response="rescue")
client = create_client_with_fakes(p, [fb])
r = client.chat("hi")
assert r.provider == "fb"
def test_empty_prompt_fails_validation():
# Test that UnifiedClient.chat rejects an empty prompt before calling the adapter
# Requires adding validation in chat(): if not prompt: raise ValueError
p = fake_adapter("p", response="x")
client = create_client_with_fakes(p)
with pytest.raises((ValueError, ValidationError)):
client.chat("")
Summary
You learned:
- ✅ Three levels of test: unit (mocks), integration (real API), contract (shape validation)
- ✅ A reusable
FakeAdapterto testUnifiedClientwithout paying for tokens - ✅ How to verify fallback, circuit breaker, routing and metrics with deterministic mocks
- ✅ Skipping integration tests when there's no API key
- ✅ CI with GitHub Actions: unit on every PR, integration only on
main - ✅ Patterns: property-based, snapshot, regression test
Checkpoint: if pytest tests/ -v passes all tests green and pytest --cov shows >80% coverage, you're ready.
Next capsule
08 — Final project consolidates everything. A complete version of the package with all providers (including Modal custom), final documentation, real usage examples, and optional publishing to PyPI. It's the deliverable you take to your portfolio.
Resources
- pytest documentation — the source of truth.
- pytest-mock —
mockerfixture, cleaner thanunittest.mock. - Hypothesis — property-based testing.
- VCR.py — records HTTP responses for deterministic integration tests.
- Coverage.py — measure what % of your code the tests cover.