Module 6: Mocking, Fixtures and Validation Loops

Module Project: A Validation Pipeline

Next →

Module Project: A Validation Pipeline

Project overview

You've learned mocking to replace external services, advanced fixtures to organize test data, and validation loops to automate the correction cycle with Claude Code. Now you're going to integrate it all by building a text processing pipeline that depends on an AI API and a database — fully tested with mocks, fixtures, and at least one validation loop.

The pipeline receives text, sends it to an AI API for sentiment analysis, stores the result in the database, and returns a summary. In production it would depend on real services. In tests, everything is mocked: the API returns controlled responses, the database is an in-memory dict, and the tests are deterministic and fast.

Project Objective

Build a pipeline with mocked external services, organized fixtures, and validation loops to iterate automatically.

By completing this project:

  • ✅ You'll have mocks for 2 external services (the AI API + a database)
  • ✅ You'll have fixtures organized in conftest.py (global + per directory)
  • ✅ You'll have run at least 1 complete validation loop
  • ✅ The tests will be deterministic, fast, and independent
  • ✅ You'll have ≥85% coverage on the pipeline

Technical Specifications

Technology Stack

  • Language: Python 3.10+
  • Testing: pytest, pytest-mock, pytest-cov
  • AI: Claude Code
  • Dependencies: pytest, pytest-mock, pytest-cov

Initial Setup

mkdir validation-pipeline
cd validation-pipeline
python -m venv venv
source venv/bin/activate
pip install pytest pytest-mock pytest-cov

Project Structure

validation-pipeline/
├── pipeline/
│   ├── __init__.py
│   ├── analyzer.py       ← Calls the AI API (you create it with TDD + mocks)
│   ├── storage.py         ← Interacts with the database (you create it with TDD + mocks)
│   ├── processor.py       ← Orchestrates the pipeline (you create it with a validation loop)
│   └── models.py          ← Data models (given)
├── tests/
│   ├── __init__.py
│   ├── conftest.py            ← Global fixtures
│   ├── unit/
│   │   ├── __init__.py
│   │   ├── conftest.py        ← Unit fixtures
│   │   ├── test_analyzer.py
│   │   └── test_storage.py
│   └── integration/
│       ├── __init__.py
│       ├── conftest.py        ← Integration fixtures
│       └── test_processor.py
├── pyproject.toml
└── requirements.txt

The Base Code

models.py (given)

"""Data models for the pipeline."""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional


@dataclass
class AnalysisRequest:
    text: str
    language: str = "es"
    max_length: int = 1000

    def __post_init__(self):
        if not self.text.strip():
            raise ValueError("Text cannot be empty")
        if len(self.text) > self.max_length:
            raise ValueError(f"Text exceeds max length of {self.max_length}")


@dataclass
class SentimentResult:
    sentiment: str  # "positive", "negative", "neutral"
    confidence: float  # 0.0 - 1.0
    keywords: list[str] = field(default_factory=list)
    
    def __post_init__(self):
        if self.sentiment not in ("positive", "negative", "neutral"):
            raise ValueError(f"Invalid sentiment: {self.sentiment}")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"Confidence must be 0.0-1.0, got {self.confidence}")


@dataclass
class AnalysisRecord:
    id: Optional[int] = None
    text: str = ""
    sentiment: str = ""
    confidence: float = 0.0
    keywords: list[str] = field(default_factory=list)
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())

What to Build

1. analyzer.py — The AI service (with mocks)

class SentimentAnalyzer:
    def __init__(self, api_key: str, base_url: str = "https://api.ai-service.com"):
        self.api_key = api_key
        self.base_url = base_url
    
    def analyze(self, request: AnalysisRequest) -> SentimentResult:
        """Calls the AI API for sentiment analysis."""
        # In production: does an HTTP POST to self.base_url
        # In tests: mocked
        ...
    
    def batch_analyze(self, requests: list[AnalysisRequest]) -> list[SentimentResult]:
        """Analyzes multiple texts."""
        ...

Tests with mocks:

  • A mock of the HTTP API (success, error, timeout)
  • Mocks of the AI's responses (positive, negative, neutral)
  • A test of the retry logic when the API fails

2. storage.py — The database (with mocks)

class AnalysisStorage:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
    
    def save(self, record: AnalysisRecord) -> int:
        """Saves a result in the database. Returns the ID."""
        ...
    
    def get(self, record_id: int) -> Optional[AnalysisRecord]:
        """Gets a result by ID."""
        ...
    
    def list_by_sentiment(self, sentiment: str) -> list[AnalysisRecord]:
        """Lists results filtered by sentiment."""
        ...

Tests with mocks:

  • Mocks of the database operations (insert, query, error)
  • A test of connection errors
  • A test of data integrity

3. processor.py — The pipeline (with a validation loop)

class TextProcessor:
    def __init__(self, analyzer: SentimentAnalyzer, storage: AnalysisStorage):
        self.analyzer = analyzer
        self.storage = storage
    
    def process(self, text: str) -> AnalysisRecord:
        """The complete pipeline: validate → analyze → save → return."""
        ...
    
    def process_batch(self, texts: list[str]) -> list[AnalysisRecord]:
        """Processes multiple texts."""
        ...
    
    def get_stats(self) -> dict:
        """Returns statistics: the count per sentiment."""
        ...

Use a validation loop to implement this: Write the tests first, then ask Claude Code to implement TextProcessor using validation loops.

Step-by-Step Process

Step 1: analyzer tests with mocks

# tests/unit/test_analyzer.py

def test_analyze_positive_text(mocker):
    mock_post = mocker.patch("pipeline.analyzer.requests.post")
    mock_post.return_value.status_code = 200
    mock_post.return_value.json.return_value = {
        "sentiment": "positive",
        "confidence": 0.95,
        "keywords": ["excellent", "recommended"]
    }
    
    analyzer = SentimentAnalyzer(api_key="test-key")
    result = analyzer.analyze(AnalysisRequest(text="Excellent product"))
    
    assert result.sentiment == "positive"
    assert result.confidence == 0.95

Step 2: storage tests with mocks

Step 3: processor tests (integration) + a validation loop

Write the TextProcessor tests and use a validation loop:

Implement TextProcessor in pipeline/processor.py so the tests 
in tests/integration/test_processor.py pass.

After implementing, run: pytest tests/integration/ -v

If any test fails, read the error, fix it, and re-run.
Repeat until they all pass.

Constraints: use the SentimentAnalyzer and AnalysisStorage interfaces.

Step 4: Professional fixtures

# tests/conftest.py (global)
@pytest.fixture
def mock_analyzer(mocker):
    analyzer = mocker.MagicMock(spec=SentimentAnalyzer)
    analyzer.analyze.return_value = SentimentResult(
        sentiment="positive", confidence=0.9, keywords=["test"]
    )
    return analyzer

@pytest.fixture
def mock_storage(mocker):
    storage = mocker.MagicMock(spec=AnalysisStorage)
    storage.save.return_value = 1
    return storage

Step 5: Measure coverage and close the gaps

pytest --cov=pipeline --cov-report=term-missing --cov-branch tests/

Evaluation Rubric (100 points)

Mocks (30 points)

  • (15 pts) Working mocks for the AI API (success, error, timeout)
  • (10 pts) Working mocks for the database (CRUD, connection error)
  • (5 pts) The mocks verify that they're called correctly (assert_called)

Fixtures (25 points)

  • (10 pts) A global conftest.py with shared fixtures
  • (10 pts) A conftest.py per directory (unit/, integration/)
  • (5 pts) Factory fixtures for variable data

Validation Loop (25 points)

  • (15 pts) At least 1 validation loop run and documented
  • (10 pts) The loop resulted in a working implementation (tests green)

Coverage and Organization (20 points)

  • (10 pts) ≥85% coverage of the pipeline
  • (5 pts) Tests organized by level
  • (5 pts) Descriptive naming throughout the suite

Extra Credit (up to +10 points)

  • (+5 pts) A property-based test with hypothesis for the processor
  • (+5 pts) A mock of retry logic with exponential backoff

Common Mistakes

Mistake 1: Mocking the class you're testing

Cause: Mocking TextProcessor in test_processor.py.

Solution: Mock the dependencies (analyzer, storage), not the class under test. The processor is real; its dependencies are mocks.

Mistake 2: A validation loop without clear tests

Cause: Ambiguous tests that Claude Code can't resolve.

Solution: Each test must have a descriptive name, a clear assertion, and a single reason to fail.

Mistake 3: Fixtures with the wrong scope

Cause: Using scope="session" for fixtures that need a reset between tests.

Solution: Mocks must be scope="function" (the default) so each test gets fresh mocks.

Mistake 4: Not verifying that the mocks were used

Cause: The tests pass but the real code doesn't call the mock.

Solution: Use assert_called_once(), assert_called_with() to verify that the interaction happened.

Resources for the Project

  1. pytest-mock Documentation - A mocking plugin for pytest
  2. unittest.mock - The official reference
  3. pytest Fixtures - Advanced fixtures
  4. pytest-cov - Coverage with pytest
  5. Martin Fowler: Mocks Aren't Stubs - The philosophy of mocking

Connection with the Next Module

What you built today is the foundation of the final project:

  • Module 7 (Strategy): You'll define the testing strategy for a complete app
  • Module 8 (Final Project): You'll replicate this pattern of mocks + fixtures + validation loops at a larger scale

Phase 2 completed. You have the TDD workflow, coverage, edge cases, mocking, fixtures, and validation loops. Phase 3 is where you put it all together in a professional project.

Module 6, Capsule 06 — Testing with Claude Code Guide A professional pipeline with mocks, fixtures and validation loops

← PreviousMocking External Services in PracticeNext →Module 7: CI Integration with GitHub Actions