Test technique : tester ses pipelines data avec pytest

Un pipeline data sans tests est une bombe à retardement. En entretien Senior, on évalue la capacité à écrire des tests automatisés rigoureux pour les pipelines de données.

Data Builder·Juin 2025·6 min de lecture·Data Engineer · Analytics Engineer

Sommaire

pytest pour la data
Fixtures et données de test
Mocking des connexions
Tester les transformations SQL
Tests dbt avancés
Tests d intégration
Grille

1pytest : le standard pour tester les pipelines data

Question discriminante

Pourquoi pytest est-il préféré à unittest pour les pipelines data ?

import pytest
import pandas as pd
from my_pipeline.transforms import calculate_revenue

# Test unitaire simple
def test_calculate_revenue_basic():
    df = pd.DataFrame({
        'quantity': [1, 2, 3],
        'unit_price': [10.0, 5.0, 8.0]
    })
    result = calculate_revenue(df)
    assert list(result['revenue']) == [10.0, 10.0, 24.0]

# Test des cas limites
def test_calculate_revenue_zero_quantity():
    df = pd.DataFrame({'quantity': [0], 'unit_price': [10.0]})
    result = calculate_revenue(df)
    assert result['revenue'].iloc[0] == 0.0

# Test des valeurs nulles
def test_calculate_revenue_null_handling():
    df = pd.DataFrame({'quantity': [None, 1], 'unit_price': [10.0, 5.0]})
    result = calculate_revenue(df)
    assert result['revenue'].isna().sum() == 1

pytest vs unittest — pytest : moins de boilerplate, fixtures puissantes, plugins riches (pytest-mock, pytest-cov)
Paramétrisation — @pytest.mark.parametrize pour tester plusieurs cas en une seule fonction

2Fixtures : données de test réutilisables

Question discriminante

Comment utilisez-vous les fixtures pytest pour vos tests data ?

import pytest
import pandas as pd
from sqlalchemy import create_engine

@pytest.fixture
def sample_orders():
    return pd.DataFrame({
        'order_id': ['O001', 'O002', 'O003'],
        'customer_id': ['C1', 'C1', 'C2'],
        'amount': [100.0, 50.0, 200.0],
        'status': ['completed', 'cancelled', 'completed'],
        'order_date': pd.to_datetime(['2024-01-01', '2024-01-02', '2024-01-03'])
    })

@pytest.fixture(scope='session')
def test_db():
    # Base de données SQLite en mémoire pour les tests
    engine = create_engine('sqlite:///:memory:')
    yield engine
    engine.dispose()

def test_revenue_by_customer(sample_orders):
    result = compute_revenue(sample_orders)
    assert result.loc[result['customer_id'] == 'C1', 'revenue'].iloc[0] == 100.0

scope — function (par défaut), class, module, session. Les fixtures de session ne sont créées qu une fois
conftest.py — fichier spécial pytest où déclarer les fixtures partagées entre plusieurs fichiers de tests

3Mocking : isoler les dépendances externes

Question discriminante

Comment testez-vous du code qui appelle une base de données ou une API externe ?

from unittest.mock import patch, MagicMock
import pytest

def test_extract_from_api():
    with patch('my_pipeline.extract.requests.get') as mock_get:
        mock_get.return_value.json.return_value = [
            {'id': 1, 'amount': 100},
            {'id': 2, 'amount': 200}
        ]
        mock_get.return_value.status_code = 200
        
        result = extract_from_api('https://api.exemple.com/orders')
        
        assert len(result) == 2
        assert result[0]['amount'] == 100
        mock_get.assert_called_once_with('https://api.exemple.com/orders',
                                         timeout=10)

# pytest-mock : syntaxe plus propre
def test_with_mocker(mocker):
    mock_conn = mocker.patch('my_pipeline.db.get_connection')
    mock_conn.return_value.execute.return_value.fetchdf.return_value = pd.DataFrame({'id': [1]})
    result = load_from_db('SELECT * FROM orders')
    assert len(result) == 1

4Tester les transformations SQL

Question discriminante

Comment testez-vous une transformation SQL dbt ou une vue Snowflake ?

# Tester une transformation SQL avec DuckDB (rapide, en mémoire)
import duckdb
import pytest

def test_revenue_aggregation():
    conn = duckdb.connect()
    
    # Créer les données de test en mémoire
    conn.execute("""
        CREATE TABLE orders AS
        SELECT * FROM (VALUES
            ('O1', 'C1', 100.0, 'completed'),
            ('O2', 'C1',  50.0, 'cancelled'),
            ('O3', 'C2', 200.0, 'completed')
        ) t(order_id, customer_id, amount, status)
    """)
    
    # Tester la transformation
    result = conn.execute("""
        SELECT customer_id, SUM(amount) as revenue
        FROM orders
        WHERE status = 'completed'
        GROUP BY customer_id
    """).fetchdf()
    
    assert result.loc[result['customer_id'] == 'C1', 'revenue'].iloc[0] == 100.0
    assert result.loc[result['customer_id'] == 'C2', 'revenue'].iloc[0] == 200.0

5Tests dbt avancés avec pytest-dbt

Question discriminante

Comment allez-vous au-delà des tests YAML dbt avec pytest ?

Tests YAML dbt — not_null, unique, accepted_values : tests simples sur les colonnes
Singular tests — fichiers SQL dans tests/ qui retournent les lignes qui échouent
pytest-dbt — plugin pytest pour lancer les tests dbt dans des suites pytest plus larges
Tests de régression — comparer les outputs d un modèle refactorisé avec l ancien via dbt-audit-helper

# Singular test dbt : test qui ne doit pas retourner de lignes
-- tests/orders_no_negative_amount.sql
SELECT order_id, amount
FROM {{ ref('fct_orders') }}
WHERE amount < 0

6Tests d intégration end-to-end

Question discriminante

Comment testez-vous un pipeline entier de l ingestion à la transformation ?

Test d intégration — tester le pipeline complet sur un échantillon réel de données
Environnement de test — base de données dédiée aux tests (SQLite in-memory, Snowflake sandbox, BigQuery dataset de test)
Données de test — jeu minimal mais représentatif : cas normaux + cas limites + données corrompues
Assertion sur les outputs — vérifier que le schéma est correct, les agrégations cohérentes, les clés uniques

7Grille par niveau

Niveau	Maitrise	Signal GO	NO-GO
Confirmé	Tests unitaires pytest, fixtures, mocks basiques	A écrit des tests pour ses fonctions de transformation, utilise des fixtures	N a aucun test dans ses projets
Senior	Tests SQL avec DuckDB, tests d intégration, CI automatisée	Teste ses transformations SQL avec DuckDB, a une CI qui lance les tests	Pense que les tests dbt YAML suffisent pour tout

Vous recrutez un Data Engineer rigoureux ?

Premier entretien gratuit. Rapport GO/NO-GO sous 48h.

Tester gratuitement Reserver un appel