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
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
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. Els exemples d'aquesta secció assumeixen un fitxer dades/vendes.csv; podeu descarregar-ne un d'exemple amb l'estructura esperada (vendes.csv) i desar-lo a dades/vendes.csv en local:
# 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%}"
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.
Configuració de pytest
Els exemples anteriors executen pytest amb el comportament per defecte, però en un projecte real cal fixar la configuració explícitament en un fitxer: així tothom a l'equip (i el pipeline de CI/CD) executa els tests exactament de la mateixa manera, sense dependre d'opcions que cadascú recordi (o no) escriure a la línia d'ordres.
pytest.ini o pyproject.toml
pytest busca la configuració, per ordre de prioritat, a pytest.ini, a la secció [tool.pytest.ini_options] de pyproject.toml, o a tox.ini/setup.cfg. Avui es prefereix pyproject.toml perquè centralitza en un sol fitxer tota la configuració del projecte (dependències, linters, tests...):
testpaths: limita a quin directori busca pytest els tests, en lloc d'escanejar tot el repositori (entorns virtuals inclosos).python_files/python_functions: permeten adaptar les convencions de descoberta si el projecte no segueixtest_*.py/test_*(poc habitual, però configurable).addopts: opcions que s'apliquen sempre, sense haver de recordar-les a cada execució.-ramostra al final un resum de tots els tests que no han passat (fallats, saltats, amb error);--strict-markersfa fallar l'execució si s'usa un marcador (@pytest.mark.X) no declarat amarkers, en lloc de només avisar-ho.markers: cal registrar explícitament qualsevol marcador personalitzat que es faci servir per classificar tests (vegeu el punt següent); si no es registra, pytest mostra un avísPytestUnknownMarkWarninga cada execució.
Marcadors per classificar i seleccionar tests
Un cop registrat un marcador a la configuració, s'aplica a qualsevol test amb un decorador:
import pytest
@pytest.mark.slow
def test_validacio_dataset_complet(df_vendes_complet):
... # test que triga diversos segons a executar-se
@pytest.mark.integracio
def test_connexio_real_a_postgres():
... # test que necessita una base de dades disponible
Els marcadors permeten executar només un subconjunt de tests sense tocar el codi:
pytest -m "not slow" # tots els tests excepte els marcats com a lents
pytest -m integracio # només els tests d'integració
pytest -m "not integracio and not slow"
Aquesta separació és clau en CI/CD: els tests ràpids (not slow) es poden executar a cada commit, mentre que els lents o d'integració es reserven per a una execució nocturna o abans d'una fusió a main.
conftest.py: fixtures compartides entre fitxers
Quan diverses fitxes de test necessiten la mateixa fixture (per exemple, df_vendes), no cal repetir-la a cada fitxer test_*.py: es defineix un cop a conftest.py, i pytest la fa disponible automàticament a tots els tests del mateix directori i subdirectoris, sense necessitat d'importar-la.
# conftest.py
import pandas as pd
import pytest
@pytest.fixture(scope="session")
def df_vendes():
"""Es carrega una sola vegada per a tota la sessió de tests, no un cop per test."""
return pd.read_csv("dades/vendes.csv")
El paràmetre scope controla quantes vegades s'executa la fixture: "function" (per defecte, un cop per test), "module" (un cop per fitxer), o "session" (un cop per tota l'execució de pytest). Fixar scope="session" en una fixture cara de calcular (llegir un CSV gran, obrir una connexió) estalvia temps d'execució, sempre que els tests no modifiquin les dades que la fixture retorna.
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:#ef4444Testejar 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ó.
Cobertura de codi amb pytest-cov
Una pregunta habitual quan es configuren els tests d'un projecte és: quina part del codi real s'executa quan corren els tests? El plugin pytest-cov respon exactament això:
Name Stmts Miss Cover Missing
--------------------------------------------------
validacions.py 42 6 86% 58-63
--------------------------------------------------
TOTAL 42 6 86%
La columna Missing indica exactament quines línies del mòdul validacions.py no s'han executat mai durant els tests: és a dir, codi sense cap test que el cobreixi. Un percentatge de cobertura alt (per exemple, >90%) no garanteix que els tests siguin bons (es pot cobrir una línia sense comprovar-ne realment el resultat), però un percentatge baix sí que indica amb certesa que hi ha codi que ningú testeja.
La cobertura és un mínim, no un objectiu
Perseguir un 100% de cobertura pot portar a escriure tests sense cap assert significatiu, només per "tocar" la línia. La cobertura serveix per detectar forats evidents (funcions senceres sense cap test), no per substituir el criteri de què val la pena testejar.
AC5074/05/07 — Miniactivitat (ampliació)
Agafa el fitxer test_qualitat_vendes.py d'aquesta pàgina (o els tests que hagis escrit a AC5074/05/03) i converteix el projecte en un projecte de tests correctament configurat:
- Crea un
pyproject.toml(opytest.ini) ambtestpaths,addopts = "-ra --strict-markers"i almenys dosmarkerspropis. - Mou la fixture
df_vendesa unconftest.pyperquè quedi disponible per a qualsevol fitxer de test del directori, sense haver-la d'importar. - Marca almenys un test com a
slowi un altre com aintegracio, i comprova ambpytest -m "not slow"que el primer no s'executa. - Instal·la
pytest-covi genera un informe de cobertura (--cov-report=term-missing) del mòdul que conté les funcions de validació. Identifica almenys una línia sense cobrir i escriu el test que hi faltava.
Lliurament: el directori del projecte (conftest.py, pytest.ini/pyproject.toml, els fitxers test_*.py i una captura de l'informe de cobertura), amb un breu README explicant les decisions de configuració preses.
Mòdul M5074 Sistemes de Big Data | Institut Sa Palomera (Blanes) | Curs CEIABD 2026-2027