pytest-Strategien für Daten-Pipelines

Worum es geht

Pipelines aus Feeds, Datenbank und Sprachmodell wirken schwer testbar: Das Netz ist unzuverlässig, das Modell antwortet nie zweimal gleich, und die Datenbank hat Zustand. Mit ein paar Entscheidungen wird daraus eine schnelle, deterministische Testsuite, die bei jeder Änderung in Sekunden läuft.

Das Grundprinzip: Trenne Logik von Ein- und Ausgabe, und ersetze alles Externe an einer klar definierten Stelle.

SQLite im Speicher als Fixture

Statt einer Datei nutzt du eine In-Memory-Datenbank. Jeder Test bekommt eine frische, leere Instanz, sodass sich Tests nicht gegenseitig beeinflussen.

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()

Wichtig: Das Schema kommt aus derselben Quelle wie im Produktivcode, sonst testest du eine andere Datenbank als die, die läuft.

Das Sprachmodell durch einen Stub ersetzen

Ein Test soll nie ein echtes Modell aufrufen. Übergib die Klassifikationsfunktion als Parameter und setze im Test eine einfache Fake-Funktion ein.

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"

Dieses Muster nennt sich Dependency Injection und ist die wichtigste Voraussetzung für testbaren Code. Zusätzlich lohnt sich ein Stub, der Fehler simuliert, etwa Timeouts, damit du den Fehlerpfad prüfst.

Parametrisierte Tests für kaputte Feeds

Für Parser sind Randfälle wichtiger als der Normalfall. Mit parametrize prüfst du viele Eingaben mit einer Testfunktion.

@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)

Sammle jeden echten Fehlerfall aus dem Betrieb als weiteren Eintrag. So wächst die Suite mit den Problemen, die tatsächlich auftreten.

Marker für langsame Tests

Manche Tests brauchen das Netz oder ein echtes Modell. Kennzeichne sie mit einem Marker und schließe sie im Standardlauf aus.

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

In der Konfiguration legst du fest, dass pytest -m "not slow" der Standardaufruf ist. Die langsamen Tests laufen bei Bedarf oder in einem separaten Lauf.

Was du nicht testen solltest

• Die Qualität der Modellantworten. Sie ist nicht deterministisch, prüfe stattdessen Format und Fehlerbehandlung.
• Fremde Bibliotheken. Du testest deinen Code, nicht sqlite3 oder requests.
• Implementierungsdetails, die sich beim Refactoring ändern. Teste Verhalten und Ergebnisse.

Eigenschaftstests kurz erklärt

Bibliotheken wie Hypothesis erzeugen zufällige Eingaben und prüfen eine Eigenschaft, die immer gelten muss, zum Beispiel: Ein Parser wirft bei beliebigem Text keine Ausnahme. Das findet Randfälle, an die man nicht gedacht hat, ist aber ein zusätzliches Werkzeug und für den Einstieg nicht nötig.

Wann sich das lohnt (und wann nicht)

Die Strategie lohnt sich, sobald die Pipeline dauerhaft läuft und du sie weiterentwickelst. Jede Änderung am Parser oder an der Klassifikation kann dann ohne Angst getestet werden.

Für ein einmaliges Skript, das du nach einem Lauf wegwirfst, ist der Aufwand zu hoch. Ein kurzer manueller Test genügt.