AI Testing Strategy

Testing AI applications requires a special approach - we need to verify not only the code, but also the quality of models and responses!

The Test Pyramid for AI

1                    ╱╲
2                   ╱  ╲
3                  ╱ E2E╲           ← 10% - End-to-end tests
4                 ╱──────╲
5                ╱        ╲
6               ╱Integration╲       ← 20% - Integration tests
7              ╱────────────╲
8             ╱              ╲
9            ╱     Unit       ╲     ← 40% - Unit tests
10           ╱──────────────────╲
11          ╱                    ╲
12         ╱    AI Evaluation     ╲  ← 30% - AI Evaluation
13        ╱────────────────────────╲

Unit Tests

1import pytest
2from unittest.mock import Mock, AsyncMock
3
4class TestDocumentService:
5    @pytest.fixture
6    def mock_repo(self):
7        return Mock(spec=DocumentRepository)
8
9    @pytest.fixture
10    def mock_embedder(self):
11        embedder = Mock()
12        embedder.embed.return_value = [0.1] * 1536
13        return embedder
14
15    @pytest.fixture
16    def service(self, mock_repo, mock_embedder):
17        return DocumentService(mock_repo, mock_embedder)
18
19    def test_chunk_document(self, service):
20        doc = Document(content="A" * 1000)
21        chunks = service.chunk_document(doc, chunk_size=200)
22
23        assert len(chunks) == 5
24        assert all(len(c.content) <= 200 for c in chunks)
25
26    @pytest.mark.asyncio
27    async def test_save_document(self, service, mock_repo):
28        doc = Document(title="Test", content="Content")
29        mock_repo.save = AsyncMock(return_value=doc)
30
31        result = await service.save(doc)
32
33        mock_repo.save.assert_called_once()
34        assert result.title == "Test"

Integration Tests

1import pytest
2from httpx import AsyncClient
3from testcontainers.qdrant import QdrantContainer
4
5@pytest.fixture(scope="module")
6def qdrant_container():
7    with QdrantContainer() as qdrant:
8        yield qdrant
9
10@pytest.fixture
11async def app_client(qdrant_container):
12    app = create_app(qdrant_url=qdrant_container.get_connection_url())
13    async with AsyncClient(app=app, base_url="http://test") as client:
14        yield client
15
16@pytest.mark.asyncio
17async def test_upload_and_query(app_client):
18    # Upload document
19    response = await app_client.post(
20        "/documents",
21        json={"title": "Test", "content": "Python is great"}
22    )
23    assert response.status_code == 201
24
25    # Query
26    response = await app_client.post(
27        "/query",
28        json={"question": "What is Python?"}
29    )
30    assert response.status_code == 200
31    assert "Python" in response.json()["answer"]

AI Evaluation

1from dataclasses import dataclass
2
3@dataclass
4class EvalCase:
5    question: str
6    expected_keywords: list[str]
7    context_required: bool = True
8
9eval_dataset = [
10    EvalCase(
11        question="What is RAG?",
12        expected_keywords=["retrieval", "generation", "augmented"],
13        context_required=True
14    ),
15    EvalCase(
16        question="How does embedding work?",
17        expected_keywords=["vector", "semantic", "representation"],
18        context_required=True
19    )
20]
21
22async def evaluate_rag_system(query_fn, dataset: list[EvalCase]) -> dict:
23    results = {"total": len(dataset), "passed": 0, "failed": []}
24
25    for case in dataset:
26        response = await query_fn(case.question)
27
28        # Check keywords
29        answer_lower = response.answer.lower()
30        keywords_found = sum(1 for k in case.expected_keywords if k in answer_lower)
31        keyword_score = keywords_found / len(case.expected_keywords)
32
33        # Check sources
34        has_sources = len(response.sources) > 0 if case.context_required else True
35
36        if keyword_score >= 0.5 and has_sources:
37            results["passed"] += 1
38        else:
39            results["failed"].append({
40                "question": case.question,
41                "keyword_score": keyword_score,
42                "has_sources": has_sources
43            })
44
45    results["accuracy"] = results["passed"] / results["total"]
46    return results

CI/CD Pipeline

1# .github/workflows/test.yml
2name: Test Suite
3
4on: [push, pull_request]
5
6jobs:
7  test:
8    runs-on: ubuntu-latest
9    services:
10      qdrant:
11        image: qdrant/qdrant
12        ports:
13          - 6333:6333
14
15    steps:
16      - uses: actions/checkout@v4
17      - uses: actions/setup-python@v5
18        with:
19          python-version: "3.11"
20
21      - name: Install dependencies
22        run: pip install -r requirements.txt
23
24      - name: Run unit tests
25        run: pytest tests/unit -v
26
27      - name: Run integration tests
28        run: pytest tests/integration -v
29        env:
30          QDRANT_URL: http://localhost:6333
31
32      - name: Run AI evaluation
33        run: python scripts/evaluate.py
34        env:
35          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

Tests give you confidence that the application works correctly. In the next lesson, we will cover documentation!