Salta el contingut

Testing automatitzat amb pytest

Per què testejar dades com si fossin codi

A qualsevol assignatura de programació s'aprèn que el codi es testeja: test_suma(), test_login_incorrecte()... Un pipeline de dades és codi que produeix un resultat (un dataset), i per tant es pot —i s'ha de— testejar amb la mateixa disciplina. La diferència és que, a més de testejar la lògica del pipeline (la funció fa el que hauria de fer), cal testejar les dades que produeix (el resultat compleix les expectatives de qualitat).

pytest és el framework de testing més utilitzat en l'ecosistema Python, tant per testejar codi d'aplicació com, en aquest bloc, per testejar dades i pipelines.

Instal·lació i primer test

pip install pytest

pytest descobreix automàticament qualsevol fitxer anomenat test_*.py o *_test.py, i dins d'aquests, qualsevol funció que comenci per test_:

# test_basic.py
def suma(a, b):
    return a + b

def test_suma_positius():
    assert suma(2, 3) == 5

def test_suma_amb_negatiu():
    assert suma(5, -2) == 3

pytest test_basic.py -v

test_basic.py::test_suma_positius PASSED
test_basic.py::test_suma_amb_negatiu PASSED

No cal cap classe ni cap import unittest: una funció amb un assert a dins ja és un test vàlid per a pytest.

Tests sobre un DataFrame

L'aplicació directa a qualitat de dades és escriure un test per a cada regla de qualitat que el dataset ha de complir:

# test_qualitat_vendes.py
import pandas as pd
import pytest

@pytest.fixture
def df_vendes():
    """Fixture: carrega el dataset una sola vegada i el reutilitza a tots els tests."""
    return pd.read_csv("dades/vendes.csv")

def test_no_hi_ha_ids_nuls(df_vendes):
    assert df_vendes["id_venda"].isnull().sum() == 0

def test_ids_son_unics(df_vendes):
    assert df_vendes["id_venda"].is_unique

def test_imports_son_positius(df_vendes):
    assert (df_vendes["import"] > 0).all()

def test_pais_es_un_codi_valid(df_vendes):
    paisos_valids = {"ES", "FR", "DE", "IT", "PT"}
    assert set(df_vendes["pais"].unique()).issubset(paisos_valids)

def test_no_hi_ha_mes_dun_5pct_de_nuls_en_email(df_vendes):
    pct_nuls = df_vendes["email_client"].isnull().mean()
    assert pct_nuls <= 0.05, f"Massa emails nuls: {pct_nuls:.1%}"

pytest test_qualitat_vendes.py -v

Una fixture (@pytest.fixture) és una funció que prepara dades o recursos reutilitzables entre tests: pytest la crida automàticament i injecta el resultat com a paràmetre a qualsevol test que la declari (df_vendes al paràmetre de cada funció).

Tests parametritzats

Quan la mateixa comprovació s'ha de repetir amb diferents valors (per exemple, validar el rang esperat de diverses columnes numèriques), @pytest.mark.parametrize evita duplicar codi:

@pytest.mark.parametrize("columna,minim,maxim", [
    ("edat", 0, 120),
    ("import", 0, 100_000),
    ("puntuacio_satisfaccio", 1, 5),
])
def test_columna_dins_de_rang(df_vendes, columna, minim, maxim):
    fora_de_rang = ~df_vendes[columna].between(minim, maxim)
    assert fora_de_rang.sum() == 0, (
        f"{fora_de_rang.sum()} valors de '{columna}' fora del rang [{minim}, {maxim}]"
    )

pytest executa aquesta funció tres vegades, una per cada tupla de la llista, i informa de quina combinació concreta ha fallat si n'hi ha alguna.

Integració amb el pipeline: fallar l'execució si un test falla

El valor real d'automatitzar els tests no és executar-los manualment de tant en tant, sinó fer-los part del pipeline: si les dades no compleixen les expectatives, el pipeline no ha de continuar carregant-les.

import subprocess
import sys

def valida_qualitat_abans_de_carregar(ruta_dataset: str) -> bool:
    """Executa la suite de tests de qualitat i retorna si ha passat."""
    resultat = subprocess.run(
        ["pytest", "test_qualitat_vendes.py", "-v", "--tb=short"],
        capture_output=True, text=True,
    )
    print(resultat.stdout)
    if resultat.returncode != 0:
        print("Validació de qualitat FALLIDA. Aturant la càrrega.", file=sys.stderr)
        return False
    return True

pytest retorna un codi de sortida diferent de zero quan algun test falla, cosa que permet encadenar-lo amb qualsevol eina d'orquestació (Airflow, un script bash, o un pipeline de CI/CD) i aturar el flux automàticament.

flowchart LR
    A["Extreu dades"] --> B["Executa\npytest"]
    B -->|"exit code 0\ntot correcte"| C["Carrega al\nData Warehouse"]
    B -->|"exit code != 0\nalgun test ha fallat"| D["Atura el pipeline\ni alerta"]

    style A fill:#1d4ed8,color:#ffffff,stroke:#3b82f6
    style B fill:#7c3aed,color:#ffffff,stroke:#a78bfa
    style C fill:#166534,color:#ffffff,stroke:#22c55e
    style D fill:#7f1d1d,color:#ffffff,stroke:#ef4444

Testejar dades a la integració contínua (CI)

Igual que un projecte de software executa la seva suite de tests a cada push (vegeu el mòdul de CI/CD si l'heu cursat), un pipeline de dades madur pot executar pytest sobre una mostra representativa de dades a cada canvi del codi del pipeline. Això detecta si un canvi al codi de transformació ha introduït sense voler una regressió de qualitat, abans que arribi a producció.

Mòdul M5074 Sistemes de Big Data | Institut Sa Palomera (Blanes) | Curs CEIABD 2026-2027