How do I test LLM calls in a data pipeline?

Use a deterministic stub that replaces the classify function via monkeypatch. That keeps tests reproducible and fast, while the real model is validated separately in a nightly smoke test.

How large should an RSS feed test fixture collection be?

For an indie project, 15 to 25 hand-picked snapshots with deliberate edge cases is plenty. More gets unwieldy, and fewer doesn't cover the real-world cases you'll hit.

What is this article about?

The article walks through a real-world use case around pytest strategies for data pipelines. You'll see where we started, a working solution with Python code, and the trade-offs we hit along the way.

7 August 2026 · 6 Min Lesezeit pytestdata pipelineSQLite testing

pytest strategies for data pipelines in our Watch-&-Alert

You've built yourself a small monitoring system that scrapes RSS feeds, stores them in SQLite, and classifies them via an LLM? Then you know the problem: the pipeline has lots of stages, external dependencies make tests flaky, and eventually you stop trusting your tests. That's exactly what happened to us with our Watch-&-Alert setup, before we tore down the test setup and rebuilt it from scratch. In this article, I'll show you how we restructured our test suite: SQLite fixtures, LLM mocking, property-based tests for feed parsers, and markers for slow integration tests. Chapter 16 of the book covers exactly these test strategies for data pipelines, using our system as the example. You'll get concrete code snippets, honest trade-offs, and a few numbers from the running setup on our VPS.

Where we started

Our Watch-&-Alert system has been running for nearly two years on a Hetzner VPS. It pulls in roughly 40 RSS and Atom feeds, a few webhooks, and two mirror boards. About 150 alerts go out per day, depending on severity via Telegram, ntfy.sh, or email. The core is a SQLite database with three tables: feeds, items, and classifications. The pipeline has five stages: Fetch, Parse, Persist, Classify, Notify.

At first, we wrote only three tests. One of them used a real HTTP connection to an RSS feed. That worked fine until the feed operator activated a new SSL certificate and our test turned red. A second test hit the real Ollama instance on the server and turned flaky under full load. The third test was an end-to-end smoke test that flickered on every small change to the LLM prompt file.

The problem was clear: we had tests, but no test strategy. External dependencies were hardcoded all over the test code. The SQLite database was shared between tests. LLM responses weren't handled deterministically. We couldn't reproduce a test run when something failed. During refactors, we spent more time debugging tests than working on actual code.

So we tore down the test setup and rebuilt it completely. Today the unit test suite runs in about 12 seconds, integration tests need around 90 seconds. We've measured zero flaky tests in the last three months. Coverage on the data layer sits at 94 percent. Those numbers didn't appear by magic — they came out of five concrete changes I'll show you in a moment.

What we needed

Before comparing solutions, we had to be clear about what the tests actually need to deliver. We defined four requirements that make sense for any pipeline test suite.

First: fast feedback. Unit tests must run in under 100 milliseconds per test. Anyone waiting longer than a second on a single test starts skipping or disabling it. Both behaviors are poison for a suite.

Second: reproducibility. A test that's green today and red tomorrow isn't a test, it's a coin flip. That means: no real HTTP calls in unit tests, no real LLM responses, no dependency on system time or filesystem ordering.

Third: realistic data. If the tests only work with dummy strings, they won't cover real edge cases. We hit a case where a feed returned published_parsed as None while published was populated. You only find issues like that when you test against actual feed snapshots.

Fourth: clear separation. Unit tests test one function in isolation. Integration tests test the pipeline as a whole with real SQLite but mocked external services. Both have different runtimes and should be runnable separately.

From these four requirements, the tools we need almost fall out automatically: pytest with tmp_path fixtures, pytest-httpx for HTTP mocking, Hypothesis for property tests, and pytest markers for separation.

Three options compared

We seriously evaluated three approaches before deciding.

Option A was the pure-mock approach: mock everything, even the SQLite database. Advantage: tests are super fast and fully isolated. Disadvantage: we'd be mocking what we actually want to test. In a data pipeline, the database is the core. If we replace sqlite3 with a mock, we no longer test whether our queries hold up under load or whether our indexes are chosen well. We dropped this variant after a prototype.

Option B was the realistic-everything approach: we used real SQLite files, real HTTP responses from cache, real Ollama responses from snapshots. Advantage: maximum realism. Disadvantage: the setup was complex. We had to maintain hundreds of HTTP response files. Every small change to the parser meant updating all the snapshots. On top of that, the snapshots were large and unwieldy.

Option C was the hybrid approach, and it's what we run today. The idea: unit tests use SQLite :memory: and mocked HTTP responses. Integration tests use tmp_path for a real SQLite file on disk and recorded HTTP responses for exactly the endpoints we need. LLM calls are replaced in both test types with a deterministic stub.

Why Option C won: it gives us Option A's speed for day-to-day development and Option B's realism for pipeline verification. The maintenance overhead is noticeably lower than Option B, because we only need to record HTTP responses for the truly critical paths.

One important trade-off: we don't test Ollama responses against the real model version. That's a conscious decision. When we deploy a new model, we have a separate smoke test that hits the real model and checks the distribution of classification outputs. But that test only runs nightly, not on every commit.

The solution

Concretely, it looks like this. We have four building blocks that work together. Each one addresses a specific weakness from our old suite.

Building block 1: SQLite fixture. In our conftest.py we define a fixture that creates a fresh in-memory database for each test and applies the schema. That takes around 3 milliseconds per test, which is negligible. The trick: we use sqlite3.Row as the row_factory so we can access columns like a dict inside the test. Production code uses the same library, so there's no mapping mismatch between tests and reality.

import pytest
import sqlite3
from watchalert.db.schema import apply_schema

@pytest.fixture
def db() -> sqlite3.Connection:
    """Frische SQLite-DB pro Test."""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    apply_schema(conn)
    yield conn
    conn.close()

For integration tests there's a second fixture that creates a file-based database in tmp_path. That lets us test multi-process scenarios too, for example when the Watch-&-Alert pipeline and a parallel heartbeat checker both write at the same time. The file gets cleaned up automatically after the test.

Building block 2: feed snapshots as fixture data. We keep roughly 20 hand-picked RSS and Atom snapshots in the repository under tests/fixtures/feeds/. Each file is a realistic slice of a real feed, often with deliberately embedded edge cases. We have a snapshot with broken date fields, one with HTML entities in the title, one with an empty body. A fixture loads these files on demand.

from pathlib import Path

@pytest.fixture
def feed_snapshot(request: pytest.FixtureRequest) -> str:
    """Lädt einen Feed-Snapshot aus tests/fixtures/feeds/."""
    path = (
        Path(__file__).parent
        / "fixtures" / "feeds" / request.param
    )
    return path.read_text(encoding="utf-8")

With pytest.mark.parametrize we can