Accueil›Blog›Test technique : tester ses pipelines data avec pytest

Guide recrutement data

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

import pytest
import pandas as pd
from pandera import DataFrameSchema, Column, Check

# Pandera : schema validation pour DataFrames
orders_schema = DataFrameSchema({
    "order_id": Column(str, nullable=False, unique=True),
    "amount": Column(float, checks=[Check.greater_than(0), Check.less_than(1_000_000)]),
    "status": Column(str, checks=Check.isin(["pending", "completed", "cancelled"])),
    "order_date": Column("datetime64[ns]", nullable=False)
})

@pytest.fixture
def sample_df():
    return pd.DataFrame({
        "order_id": ["ORD-001", "ORD-002"],
        "amount": [100.0, 250.0],
        "status": ["completed", "pending"],
        "order_date": pd.to_datetime(["2025-01-01", "2025-01-02"])
    })

def test_transform_schema(sample_df):
    result = transform_orders(sample_df)
    orders_schema.validate(result)

@pytest.mark.parametrize("amount,valid", [
    (100.0, True), (-50.0, False), (0.0, False), (1e7, False)
])
def test_amount_validation(amount, valid):
    assert validate_amount(amount) == valid

Pandera - validation de schema pour DataFrames pandas/polars. Definir le schema comme code, valider a l entree/sortie des fonctions de transformation
pytest fixtures scope - scope="function" (defaut) : recree a chaque test. scope="session" : partage toute la session. Utiliser session pour les connexions DB couteuses
conftest.py - fixtures partagees entre plusieurs fichiers de tests. DRY : DataFrames de test definis une seule fois
Mocking - unittest.mock.patch pour remplacer les appels DB, API ou fichiers. Isoler la logique de transformation des dependances externes
Coverage - pytest --cov=src --cov-report=html. Viser 80% sur la logique metier. Les tests d integration couvrent ce que les unitaires ne peuvent pas

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

1pytest: the standard for testing data pipelines

Discriminating question

Why is pytest preferred over unittest for data pipelines?

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

# Simple unit test
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]

# Edge case tests
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

# Null value tests
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: less boilerplate, powerful fixtures, rich plugins (pytest-mock, pytest-cov)
Parametrization — @pytest.mark.parametrize to test multiple cases in a single function

2Fixtures: reusable test data

Discriminating question

How do you use pytest fixtures for your data tests?

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():
    # In-memory SQLite database for 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 (default), class, module, session. Session fixtures are created only once
conftest.py — special pytest file where shared fixtures are declared across multiple test files

3Mocking: isolating external dependencies

Discriminating question

How do you test code that calls a database or an external API?

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: cleaner syntax
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

4Testing SQL transformations

Discriminating question

How do you test a dbt SQL transformation or a Snowflake view?

# Testing a SQL transformation with DuckDB (fast, in-memory)
import duckdb
import pytest

def test_revenue_aggregation():
    conn = duckdb.connect()
    
    # Create test data in memory
    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)
    """)
    
    # Test the 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

5Advanced dbt tests with pytest-dbt

Discriminating question

How do you go beyond dbt YAML tests with pytest?

dbt YAML tests — not_null, unique, accepted_values: simple column-level tests
Singular tests — SQL files in tests/ that return failing rows
pytest-dbt — pytest plugin to run dbt tests within larger pytest suites
Regression tests — compare outputs of a refactored model against the old one via dbt-audit-helper

# Singular dbt test: test that must return no rows
-- tests/orders_no_negative_amount.sql
SELECT order_id, amount
FROM {{ ref('fct_orders') }}
WHERE amount < 0

6End-to-end integration tests

Discriminating question

How do you test an entire pipeline from ingestion to transformation?

Integration test — test the full pipeline on a real data sample
Test environment — dedicated test database (SQLite in-memory, Snowflake sandbox, BigQuery test dataset)
Test data — minimal but representative set: normal cases + edge cases + corrupted data
Output assertions — verify that the schema is correct, aggregations are consistent, keys are unique

import pytest
import pandas as pd
from pandera import DataFrameSchema, Column, Check

# Pandera: schema validation for DataFrames
orders_schema = DataFrameSchema({
    "order_id": Column(str, nullable=False, unique=True),
    "amount": Column(float, checks=[Check.greater_than(0), Check.less_than(1_000_000)]),
    "status": Column(str, checks=Check.isin(["pending", "completed", "cancelled"])),
    "order_date": Column("datetime64[ns]", nullable=False)
})

@pytest.fixture
def sample_df():
    return pd.DataFrame({
        "order_id": ["ORD-001", "ORD-002"],
        "amount": [100.0, 250.0],
        "status": ["completed", "pending"],
        "order_date": pd.to_datetime(["2025-01-01", "2025-01-02"])
    })

def test_transform_schema(sample_df):
    result = transform_orders(sample_df)
    orders_schema.validate(result)

@pytest.mark.parametrize("amount,valid", [
    (100.0, True), (-50.0, False), (0.0, False), (1e7, False)
])
def test_amount_validation(amount, valid):
    assert validate_amount(amount) == valid

Pandera - schema validation for pandas/polars DataFrames. Define the schema as code, validate at the input/output of transformation functions
pytest fixtures scope - scope="function" (default): recreated at each test. scope="session": shared across the entire session. Use session for expensive DB connections
conftest.py - fixtures shared across multiple test files. DRY: test DataFrames defined only once
Mocking - unittest.mock.patch to replace DB, API or file calls. Isolate transformation logic from external dependencies
Coverage - pytest --cov=src --cov-report=html. Aim for 80% on business logic. Integration tests cover what unit tests cannot

7Level grid

Level	Mastery	GO signal	NO-GO
Mid-level	pytest unit tests, fixtures, basic mocks	Has written tests for transformation functions, uses fixtures	Has no tests in their projects
Senior	SQL tests with DuckDB, integration tests, automated CI	Tests SQL transformations with DuckDB, has a CI that runs the tests	Thinks dbt YAML tests are sufficient for everything

Vous recrutez un Data Engineer rigoureux ?

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

Tester gratuitement Reserver un appel