Testovanie ETL v Pythone s pytest (2026)

Aktualizované: 1. augusta 2026

Testovanie ETL pipeline v Pythone s pytest znamená písanie automatizovaných testov, ktoré overia, že vaše transformácie dát produkujú správne výstupy pre známe vstupy: od unit testov jednotlivých funkcií cez schema validáciu s Pandera až po integračné testy proti reálnej databáze v Dockeri. V roku 2026 už testovanie nie je voliteľné. Každá netestovaná transformácia sa zvykne premeniť na hodiny spätných opráv o 3:00 ráno (osobne odhlasované). V tomto sprievodcovi ukážem presné vzory, ktoré držia moje produkčné pipelinie funkčné, vrátane pytest fixtures, Pandera kontraktov, Great Expectations a dbt integrácií.

pytest 8.3 (2026) je de facto štandard pre testovanie Python ETL pipeline vďaka fixture systému a bohatému ekosystému pluginov ako pytest-mock, pytest-xdist a pytest-postgresql.
Testovacia pyramída pre dáta: 70 % unit testov (čisté transformačné funkcie), 20 % integračných testov (proti reálnej DB v testcontainers), 10 % end-to-end testov na dennej vzorke.
Pandera 0.22 poskytuje typovo bezpečné DataFrame schémy s runtime validáciou; Great Expectations 1.5 exceluje v produkčnej data quality vrstve s reportingom.
Property-based testovanie s Hypothesis odhaľuje edge cases (NaN, prázdne skupiny, timezone drift), na ktoré by ste v ručne písaných testoch neprišli.
dbt Core 1.9 modely testujte cez dbt test plus pytest-dbt-core pre unit testy SQL logiky pred deploymentom.
CI/CD musí bežať plnú testovaciu suite pri každom PR, inak zbytočne investujete čas do testov, ktoré nikdy nechytia regresiu.

Prečo je testovanie ETL pipeline v Pythone kľúčové

Osobne som počas kariéry data engineerky zažila dva druhy pipelinov: tie, ktoré padnú hlasno a v CI, a tie, ktoré tichýkajú v produkcii. Druhé sú horšie. Netestovaný ETL vám nedá error, dá vám zlé dáta, ktoré silently kontaminujú dashboardy, ML modely a rozhodnutia biznisu. Kým si niekto všimne, že MRR chart vypadá "nejako divne", pretečú tri sprinty a analytici stratia dôveru v celý data warehouse.

Testy ETL pipeline riešia tri konkrétne problémy. Po prvé, regresie v transformáciách: refaktor calculate_arr() vyzerá nevinne, ale drobná zmena v znamienku spôsobí, že revenue klesne o 8 %. Po druhé, schema drift: upstream systém pridá novú NULL hodnotu do stĺpca, ktorý vaša pipeline používa ako join kľúč, a downstream tabuľka sa vyprázdni. Po tretie, edge cases v dátach (prázdne skupiny, duplicity, mimoriadne dlhé stringy, timestampy v cudzej timezone). Všetko toto sa v laptope vývojára nikdy neobjaví, ale v produkcii áno.

V roku 2026 podľa najnovšieho Data Council prieskumu prevažná väčšina zrelých data teamov považuje pipeline testy za povinné pred deploymentom. To nie je bureaucracy, to je poistenie proti nočným pageom. Ak vaša pipeline nemá aspoň smoke test, ktorý overí, že output má očakávaný počet stĺpcov a nenulové kľúčové polia, ste jeden refaktor od incidentu.

Nastavenie pytest pre dátový projekt v roku 2026

Pytest 8.3 je aktuálny stable release z júna 2026 a pridal natívnu podporu pre async fixtures bez pluginu (predtým vyžadovalo pytest-asyncio). Pre dátový projekt inštalujem tento základný set:

# pyproject.toml
[project.optional-dependencies]
test = [
    "pytest==8.3.*",
    "pytest-cov==5.0.*",
    "pytest-mock==3.14.*",
    "pytest-xdist==3.7.*",         # paralelné testy
    "pytest-postgresql==6.1.*",    # in-process Postgres
    "testcontainers==4.9.*",       # Docker-based DB fixtures
    "pandera[polars,pandas]==0.22.*",
    "great-expectations==1.5.*",
    "hypothesis==6.115.*",
    "faker==30.0.*",
]

Štruktúra projektu, ktorú konzistentne používam:

my_pipeline/
├── src/pipeline/
│   ├── extract.py
│   ├── transform.py
│   └── load.py
├── tests/
│   ├── conftest.py           # zdieľané fixtures
│   ├── unit/
│   │   └── test_transform.py
│   ├── integration/
│   │   └── test_load_postgres.py
│   └── data/                 # malé fixture CSVka pod 100 riadkov
│       ├── orders_sample.csv
│       └── expected_output.parquet
├── pyproject.toml
└── pytest.ini

Kľúčová konfigurácia v pytest.ini:

[pytest]
testpaths = tests
markers =
    unit: rýchle testy bez externých závislostí (default v pre-commite)
    integration: testy vyžadujúce Docker/DB (bežia v CI)
    slow: pomalé testy, spúšťajú sa nightly
addopts =
    --strict-markers
    --cov=src/pipeline
    --cov-report=term-missing
    --cov-fail-under=80

Marker system je nenápadný, ale spasí vám hodiny čakania. Pri vývoji spúšťam pytest -m unit a mám feedback do 3 sekúnd. Plná suite (pytest -m "unit or integration") beží až v CI. Coverage prah 80 % nie je vestige, je to hranica, pod ktorou začínajú testy byť dekoratívne.

Fixtures pre DataFrame a databázové pripojenia

Fixtures sú srdcom pytest a v dátovom kontexte oddelujú testovacie dáta od testovacej logiky. Nikdy nehardcodujem DataFrame priamo v teste, vždy v conftest.py. Toto je fixture pattern, ktorý používam vo všetkých svojich pipelinách:

# tests/conftest.py
import pytest
import pandas as pd
from pathlib import Path

FIXTURE_DIR = Path(__file__).parent / "data"


@pytest.fixture(scope="session")
def raw_orders_df() -> pd.DataFrame:
    """Malá vzorka raw dát z upstream API."""
    return pd.read_csv(FIXTURE_DIR / "orders_sample.csv", parse_dates=["created_at"])


@pytest.fixture
def orders_with_nulls(raw_orders_df: pd.DataFrame) -> pd.DataFrame:
    """Variant s injektovanými NULL hodnotami pre testovanie odolnosti."""
    df = raw_orders_df.copy()
    df.loc[df.sample(frac=0.1, random_state=42).index, "customer_id"] = None
    return df


@pytest.fixture
def expected_daily_revenue() -> pd.DataFrame:
    return pd.read_parquet(FIXTURE_DIR / "expected_output.parquet")

scope="session" je dôležitý pre statické fixture súbory. Načíta CSV raz na celú testovaciu suite namiesto pri každom teste. Pre modifikovateľné fixtures (ako orders_with_nulls) nechajte default function scope, aby každý test dostal čerstvú kópiu.

Pre databázové testy používam testcontainers, ktoré spustia skutočný Postgres v Dockeri. Je to oveľa spoľahlivejšie než mock, pretože chytí SQL syntax chyby, chýbajúce indexy a constraint violations, ktoré mock nikdy nezaznamená:

# tests/conftest.py
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text


@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres


@pytest.fixture
def db_engine(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    # Vytvor schému
    with engine.begin() as conn:
        conn.execute(text(Path("sql/schema.sql").read_text()))
    yield engine
    # Cleanup medzi testami
    with engine.begin() as conn:
        conn.execute(text("TRUNCATE TABLE orders, customers CASCADE"))

Unit testy transformačných funkcií

Unit test v ETL kontexte znamená testovanie čistej transformačnej funkcie. Takej, ktorá zoberie DataFrame, urobí niečo deterministické a vráti nový DataFrame. Ak vaša transformačná funkcia sama otvára databázu alebo číta zo S3, refaktorujte ju: extract oddeľte od transform, aby ste transform mohli testovať bez I/O. Toto je najdôležitejší architektonický vzor pre testovateľné pipeline.

# src/pipeline/transform.py
import pandas as pd


def calculate_daily_revenue(orders: pd.DataFrame) -> pd.DataFrame:
    """Agreguje objednávky na denné revenue, exkludujúc refundované."""
    valid = orders[orders["status"] != "refunded"].copy()
    valid["date"] = valid["created_at"].dt.tz_convert("UTC").dt.date
    return (
        valid.groupby("date", as_index=False)
        .agg(revenue=("amount", "sum"), order_count=("order_id", "nunique"))
        .sort_values("date")
        .reset_index(drop=True)
    )

A prislúchajúci unit test:

# tests/unit/test_transform.py
import pandas as pd
from pandas.testing import assert_frame_equal
from pipeline.transform import calculate_daily_revenue


def test_daily_revenue_excludes_refunded(raw_orders_df):
    result = calculate_daily_revenue(raw_orders_df)
    # Nesmie obsahovať revenue z refundovaných objednávok
    refunded_amount = raw_orders_df.loc[
        raw_orders_df["status"] == "refunded", "amount"
    ].sum()
    assert result["revenue"].sum() == raw_orders_df["amount"].sum() - refunded_amount


def test_daily_revenue_handles_empty_input():
    empty = pd.DataFrame(
        columns=["order_id", "amount", "status", "created_at"]
    ).astype({"amount": "float64", "created_at": "datetime64[ns, UTC]"})
    result = calculate_daily_revenue(empty)
    assert result.empty
    assert list(result.columns) == ["date", "revenue", "order_count"]


def test_daily_revenue_deterministic_output(raw_orders_df, expected_daily_revenue):
    result = calculate_daily_revenue(raw_orders_df)
    assert_frame_equal(result, expected_daily_revenue, check_like=True)

Všimnite si tri vzory. Po prvé, testujte behavior, nie implementáciu. Kontrolujem, že refundované sú vylúčené, nie ako. Po druhé, prázdny vstup má vlastný test, pretože prázdne DataFrames sú najčastejší zdroj AttributeError v pipelinách (na toto som narazila trikrát za posledný rok). Po tretie, assert_frame_equal s check_like=True ignoruje poradie stĺpcov, ale trvá na typoch, čo je presne to, čo chcete. Ak potrebujete tolerovať drobnú numerickú odchýlku (napr. pri float agregáciách), pridajte rtol=1e-6. Pre viac vzorov pri práci s NULL hodnotami odporúčam môj skorší článok o spracovaní chýbajúcich hodnôt v Pandas 2026.

Integračné testy s testcontainers a Postgres

Unit testy neochránia pred SQL chybami. Ak vaša load() funkcia píše do Postgres, potrebujete integračný test, ktorý overí, že SQL beží, constraintny platia a upsert správa sa ako očakávate. Toto je test pattern, ktorý používam:

# tests/integration/test_load_postgres.py
import pytest
import pandas as pd
from sqlalchemy import text
from pipeline.load import upsert_daily_revenue

pytestmark = pytest.mark.integration


def test_upsert_inserts_new_rows(db_engine, expected_daily_revenue):
    upsert_daily_revenue(db_engine, expected_daily_revenue)

    with db_engine.connect() as conn:
        rows = conn.execute(text("SELECT COUNT(*) FROM daily_revenue")).scalar()
    assert rows == len(expected_daily_revenue)


def test_upsert_updates_existing_rows(db_engine, expected_daily_revenue):
    # Prvý beh
    upsert_daily_revenue(db_engine, expected_daily_revenue)
    # Druhý beh s modifikovanou revenue pre jeden deň
    modified = expected_daily_revenue.copy()
    modified.loc[0, "revenue"] = 999_999.99
    upsert_daily_revenue(db_engine, modified)

    with db_engine.connect() as conn:
        result = conn.execute(
            text("SELECT revenue FROM daily_revenue WHERE date = :d"),
            {"d": modified.loc[0, "date"]},
        ).scalar()
    assert result == pytest.approx(999_999.99)


def test_upsert_rejects_null_date(db_engine):
    bad = pd.DataFrame({"date": [None], "revenue": [100.0], "order_count": [1]})
    with pytest.raises(Exception, match="null value in column"):
        upsert_daily_revenue(db_engine, bad)

Tieto testy trvajú 2 až 5 sekúnd každý (kvôli Docker spinup), preto ich mám za markerom integration a bežia iba v CI. Test číslo tri je špecificky pre defenzívne správanie: chcem, aby pipeline hlasno padla, keď dostane NULL date, nie aby ticho vložila poškodený riadok.

Validácia schém s Pandera 0.22

Pandera prináša do sveta DataFrame to, čo pydantic priniesol JSON API: runtime typovú kontrolu s deklaratívnymi schémami. Verzia 0.22 (júl 2026) pridala natívnu podporu pre Polars vedľa Pandas a výrazne zlepšila error messages. Používam ju na hraniciach pipeline, hneď za extract a hneď pred load.

# src/pipeline/schemas.py
import pandera.pandas as pa
from pandera.typing import Series, DataFrame


class RawOrderSchema(pa.DataFrameModel):
    order_id: Series[str] = pa.Field(unique=True, nullable=False)
    customer_id: Series[str] = pa.Field(nullable=True)
    amount: Series[float] = pa.Field(ge=0, lt=1_000_000)
    status: Series[str] = pa.Field(isin=["pending", "paid", "refunded", "cancelled"])
    created_at: Series[pa.DateTime] = pa.Field(nullable=False)

    class Config:
        strict = True   # padne, ak dorazí neočakávaný stĺpec
        coerce = True


class DailyRevenueSchema(pa.DataFrameModel):
    date: Series[pa.Date] = pa.Field(unique=True)
    revenue: Series[float] = pa.Field(ge=0)
    order_count: Series[int] = pa.Field(gt=0)

Použitie ako dekorátor priamo na transformačnej funkcii:

from pandera.typing import DataFrame
from pipeline.schemas import RawOrderSchema, DailyRevenueSchema


@pa.check_types
def calculate_daily_revenue(
    orders: DataFrame[RawOrderSchema],
) -> DataFrame[DailyRevenueSchema]:
    # ... implementácia
    ...

V testoch pridám špecifický test, ktorý validuje, že schema chytí zámerne pokazené dáta:

def test_schema_rejects_negative_amount(raw_orders_df):
    bad = raw_orders_df.copy()
    bad.loc[0, "amount"] = -10
    with pytest.raises(pa.errors.SchemaError, match="greater_than_or_equal_to"):
        RawOrderSchema.validate(bad)

Detailná dokumentácia je na pandera.readthedocs.io. Kľúčový insight: Pandera schémy sú testovacie kontrakty a zároveň runtime bariéry. Ten istý kód, čo padne v CI, padne aj v produkcii, čo je presne to, čo chcete.

Great Expectations pre data quality

Great Expectations 1.5 (máj 2026) prešla značným zjednodušením v novom Fluent API a hodí sa tam, kde Pandera prestáva stačiť: produkčné data quality monitoring s reportingom a alertingom. Zatiaľ čo Pandera je fantastická pre in-code kontrakty, GE exceluje v generovaní data docs, ktoré vidia analytici a stakeholderi.

import great_expectations as gx

context = gx.get_context(mode="ephemeral")   # v testoch bez perzistencie

batch = context.data_sources.pandas_default.read_dataframe(daily_revenue_df)

suite = context.suites.add(gx.ExpectationSuite(name="daily_revenue_suite"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="date"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="revenue", min_value=0, max_value=10_000_000
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="date"))

result = batch.validate(suite)
assert result.success, f"Data quality checks failed: {result}"

V praxi mám dvojité pokrytie: Pandera vo vnútri transformácií (rýchle, blokujúce) a Great Expectations na výstupe do warehouse (bohatý report, non-blocking upozorňuje data teamov). Pre viac o práci s dátami veľkých objemov si pozrite môj sprievodca DuckDB s Pandas pre SQL analytiku.

Property-based testovanie s Hypothesis

Toto je technika, ktorú väčšina data teamov vynecháva, a je to škoda. Property-based testing generuje tisíce náhodných vstupov, ktoré overia, že určitá vlastnosť platí, namiesto testovania jedného konkrétneho vstupu. Pre ETL pipeline sú vlastnosti veci ako "súčet skupín sa rovná súčtu vstupu" alebo "aggregovaný output má menej alebo rovnaký počet riadkov ako input".

from hypothesis import given, strategies as st, settings
from hypothesis.extra.pandas import data_frames, column
import pandas as pd
from pipeline.transform import calculate_daily_revenue


@given(
    orders=data_frames(
        columns=[
            column("order_id", elements=st.text(min_size=1, max_size=20), unique=True),
            column("amount", elements=st.floats(min_value=0, max_value=1e6, allow_nan=False)),
            column("status", elements=st.sampled_from(["paid", "refunded", "cancelled"])),
            column("created_at", elements=st.datetimes(
                min_value=pd.Timestamp("2020-01-01"),
                max_value=pd.Timestamp("2030-01-01"),
            ).map(lambda d: d.tz_localize("UTC"))),
        ],
        rows=st.tuples()
    )
)
@settings(max_examples=200, deadline=None)
def test_revenue_never_exceeds_input_sum(orders):
    result = calculate_daily_revenue(orders)
    assert result["revenue"].sum() <= orders["amount"].sum() + 1e-6

Hypothesis mi už niekoľkokrát našiel bug, ktorý by som ručne nevymyslela. Napríklad DataFrame s jednou skupinou, kde groupby vrátil Series namiesto DataFrame, alebo timezone-naive timestamp v inak tz-aware kolone. Investícia 30 minút do jedného property testu často odhalí problém, ktorý by inak spálil hodiny debuggingu v produkcii.

Ako testovať dbt modely v pytest

Ak používate dbt Core 1.9 (august 2026) pre transformačnú vrstvu v warehouse, máte dve úrovne testovania: dbt native tests (schema.yml, generic tests, singular tests) pre production data assertions a pytest-dbt-core pre unit testy SQL logiky pred deploymentom.

# models/marts/schema.yml
version: 2
models:
  - name: daily_revenue
    columns:
      - name: date
        tests:
          - unique
          - not_null
      - name: revenue
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 10000000

Pre unit test SQL logiky (ktorý beží bez zásahu do warehouse):

# tests/unit/test_dbt_daily_revenue.py
from dbt.tests.util import run_dbt

def test_daily_revenue_excludes_refunded(project):
    run_dbt(["seed", "--select", "orders_sample"])
    run_dbt(["run", "--select", "daily_revenue"])

    result = project.run_sql(
        "SELECT SUM(revenue) FROM {{ ref('daily_revenue') }}", fetch="one"
    )
    # Vzorka má 100.00 v refundovaných, nesmú byť v output
    assert result[0] == 950.00

V mojich pipelinách bežia dbt testy dvakrát: raz v CI na PR proti DuckDB backend (rýchle, izolované) a raz v post-deploy hook na Snowflake proti reálnym dátam. Prvé chytí regresie v logike, druhé chytí drift v upstream zdrojoch. Kompletnú referenciu nájdete v oficiálnej dbt dokumentácii k data testom.

CI/CD pipeline pre dátové testy

Testy, ktoré nebežia v CI, sú testy, ktoré nechytajú regresie. GitHub Actions konfiguráciu, ktorú používam pre všetky dátové projekty:

# .github/workflows/test.yml
name: Test pipeline
on:
  pull_request:
  push:
    branches: [main]

jobs:
  unit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv sync --extra test
      - run: uv run pytest -m unit -n auto

  integration:
    runs-on: ubuntu-latest
    services:
      docker:
        image: docker:26-dind
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv sync --extra test
      - run: uv run pytest -m integration
        timeout-minutes: 15

  data-quality:
    runs-on: ubuntu-latest
    if: github.event_name == 'schedule'
    steps:
      - uses: actions/checkout@v4
      - run: uv run python scripts/run_great_expectations.py

-n auto aktivuje pytest-xdist, ktorý paralelizuje testy naprieč CPU jadrami. U nás to skrátilo unit suite z 45 s na 8 s. Integračné testy nezvládnu paralelizáciu (zdieľajú Docker volume), takže bežia serializovane. Detaily o oficiálnych GitHub Actions runneroch nájdete v GitHub Actions dokumentácii.

Pre podobný deployment pattern pre ML modely si prečítajte môj sprievodca nasadzovanie scikit-learn modelov cez FastAPI. Testy sa aplikujú rovnako, len s inou schémou.

Často kladené otázky

Aký je rozdiel medzi Great Expectations a Pandera?

Pandera je knižnica typovej schémy pre DataFrame v Pythone, bežná v transformačnej vrstve ako in-code kontrakt s minimálnym overheadom. Great Expectations je komplexnejšia data quality platforma s HTML reportmi, checkpoint konceptom a integráciou do orchestrátorov ako Airflow. V praxi ich používam spolu: Pandera vo vnútri transformácií, Great Expectations na výstupoch do warehouse.

Prečo je testovanie dátových pipeline dôležité?

Bez testov silne opreté dátové rozhodnutia stoja na netestovanej logike. Regresia v transformácii môže silently korumpovať dashboardy a ML modely bez akéhokoľvek upozornenia. Testy chytia tri najčastejšie zdroje incidentov: zmeny v transformačnej logike (unit testy), schema drift v upstream systémoch (Pandera/GE) a edge cases v dátach (Hypothesis).

Ako testovať pipeline, ktorá číta z externého API?

Extract logiku (volania API) oddeľte od transform logiky (spracovanie dát). Transform testujte cez fixtures s uloženými JSON/CSV vzorkami. Extract testujte s pytest-mock alebo responses knižnicou, ktorá zachytáva HTTP volania. Nikdy netestujte proti live API v CI, bude flaky a zbytočne konzumuje rate limity.

Koľko testov je "dosť" pre ETL pipeline?

Cieľom je aspoň 80 % coverage transformačnej logiky, s prioritou na kritické biznis pravidlá (revenue, deduplikácia, filtrovacie predikáty). Každá netriviálna transformačná funkcia by mala mať aspoň tri testy: happy path, prázdny vstup a edge case (NaN, duplicity, extrémne hodnoty). Nad rámec: property-based test pre invariants ako "output row count ≤ input row count".

Môžem používať pytest fixtures pre Polars DataFrames rovnako ako pre Pandas?

Áno, mechanizmus fixtures je identický. Jediný rozdiel je, že namiesto assert_frame_equal z pandas.testing použijete polars.testing.assert_frame_equal. Pandera 0.22 podporuje obe knižnice s rovnakým schema modelom. Ak pipeline používa Polars, získate typicky 3 až 5-krát rýchlejšie testy vďaka rýchlejšiemu I/O.