pytest strategies for data pipelines

What this is about

Pipelines made of feeds, a database and a language model seem hard to test: the network is unreliable, the model never answers the same way twice, and the database has state. A few decisions turn this into a fast, deterministic test suite that runs in seconds after every change.

The basic principle: separate logic from input and output, and replace everything external at one clearly defined place.

SQLite in memory as a fixture

Instead of a file you use an in-memory database. Every test gets a fresh, empty instance, so tests do not influence each other.

import sqlite3
import pytest

SCHEMA = """
CREATE TABLE items (
    key TEXT PRIMARY KEY,
    title TEXT,
    label TEXT
);
"""

@pytest.fixture
def db():
    conn = sqlite3.connect(":memory:")
    conn.executescript(SCHEMA)
    yield conn
    conn.close()

Important: the schema comes from the same source as in the production code, otherwise you test a different database than the one that runs.

Replace the language model with a stub

A test must never call a real model. Pass the classification function in as a parameter and use a simple fake function in the test.

def fake_llm(prompt: str) -> str:
    return "relevant" if "Sicherheitslücke" in prompt else "routine"

def test_classify_marks_security_items(db):
    from pipeline import process_item
    process_item(db, key="a", title="Sicherheitslücke gefunden",
                 text="Details", ask=fake_llm)
    label = db.execute("SELECT label FROM items WHERE key='a'").fetchone()[0]
    assert label == "relevant"

This pattern is called dependency injection and is the most important prerequisite for testable code. A stub that simulates errors, such as timeouts, is also worthwhile so that you test the failure path.

Parametrized tests for broken feeds

For parsers, edge cases matter more than the normal case. With parametrize you check many inputs with one test function.

@pytest.mark.parametrize("xml", [
    "",                                  # leer
    "<rss><channel></channel></rss>",    # ohne Items
    "<html>Fehlerseite</html>",          # falscher Inhalt
    "<rss><channel><item></item></channel></rss>",  # Item ohne Felder
])
def test_parser_does_not_crash(xml):
    from pipeline import parse_feed
    assert isinstance(parse_feed(xml), list)

Collect every real failure from operation as another entry. That way the suite grows with the problems that actually occur.

Markers for slow tests

Some tests need the network or a real model. Tag them with a marker and exclude them from the default run.

@pytest.mark.slow
def test_real_ollama_roundtrip():
    ...

In the configuration you define that pytest -m "not slow" is the default call. The slow tests run on demand or in a separate run.

What you should not test

• The quality of model answers. It is not deterministic, so check format and error handling instead.
• Third-party libraries. You test your code, not sqlite3 or requests.
• Implementation details that change during refactoring. Test behavior and results.

Property tests in brief

Libraries such as Hypothesis generate random inputs and check a property that must always hold, for example: a parser raises no exception for arbitrary text. This finds edge cases you did not think of, but it is an additional tool and not necessary to get started.

When it is worth it (and when not)

The strategy is worth it as soon as the pipeline runs permanently and you keep developing it. Every change to the parser or the classification can then be tested without fear.

For a one-off script that you throw away after a single run, the effort is too high. A short manual test is enough.