SQL sous contrôle : tests unitaires et CI/CD pour des requêtes fiables

Tout le monde se concentre sur l’écriture de SQL qui « fonctionne », mais très peu vérifient s’il continuera à fonctionner demain. Une seule nouvelle ligne, une hypothèse modifiée ou un refactoring peut casser une requête silencieusement. Cet article présente un flux de travail complet, montrant comment traiter le SQL comme un logiciel : versionné, testé et automatisé. Nous partirons d’une véritable question d’entretien d’Amazon sur l’identification des clients ayant les dépenses quotidiennes les plus élevées, puis nous convertirons la requête en un composant testable, définirons les résultats attendus et automatiserons les tests avec l’intégration continue et le déploiement continu (CI/CD).

LA REQUÊTE SQL D’AMAZON

Un entretien chez Amazon demande souvent de trouver les clients avec la dépense quotidienne la plus élevée sur une période donnée, à partir de deux tables : customers et orders. Ce problème illustre parfaitement comment le SQL peut être traité comme un logiciel : la requête doit être correcte, stable et résister aux régressions. Voici la solution finale en PostgreSQL :

WITH customerdailytotals AS (
  SELECT
    o.cust_id,
    o.order_date,
    SUM(o.totalordercost) AS totaldailycost
  FROM orders o
  WHERE o.order_date BETWEEN '2019-02-01' AND '2019-05-01'
  GROUP BY o.custid, o.orderdate
),
rankeddailytotals AS (
  SELECT
    cust_id,
    order_date,
    totaldailycost,
    RANK() OVER (
      PARTITION BY order_date
      ORDER BY totaldailycost DESC
    ) AS rnk
  FROM customerdailytotals
)
SELECT
  c.first_name,
  rdt.order_date,
  rdt.totaldailycost AS max_cost
FROM rankeddailytotals rdt
JOIN customers c ON rdt.cust_id = c.id
WHERE rdt.rnk = 1
ORDER BY rdt.order_date;

POURQUOI TESTER SON SQL ?

Le SQL casse plus facilement qu’on ne le croit. Une valeur par défaut modifiée, une colonne renommée ou une nouvelle source de données peut introduire des erreurs silencieuses. Les tests vous protègent de ces problèmes. Trois étapes de test seront couvertes : convertir la logique en une fonction, définir la sortie attendue et écrire une suite de tests unitaires.

PRÉPARER LES TESTS : DES DONNÉES CONTRÔLÉES

Pour tester le code SQL, on commence par l’envelopper dans une fonction Python à l’aide d’un framework léger comme unittest. On définit d’abord la requête à tester :

query = """
        WITH customerdailytotals AS (
          SELECT
            o.cust_id,
            o.order_date,
            SUM(o.totalordercost) AS totaldailycost
          FROM orders o
          WHERE o.order_date BETWEEN '2019-02-01' AND '2019-05-01'
          GROUP BY o.custid, o.orderdate
        ),
        rankeddailytotals AS (
          SELECT
            cust_id,
            order_date,
            totaldailycost,
            RANK() OVER (
              PARTITION BY order_date
              ORDER BY totaldailycost DESC
            ) AS rnk
          FROM customerdailytotals
        )
        SELECT
          c.first_name,
          rdt.order_date,
          rdt.totaldailycost AS max_cost
        FROM rankeddailytotals rdt
        JOIN customers c ON rdt.cust_id = c.id
        WHERE rdt.rnk = 1
        ORDER BY rdt.order_date;
    """

Ensuite, on crée un échantillon de données contrôlé pour les tests :

test_customers = [
    (15, "Mia"),
    (7, "Jill"),
    (3, "Farida")
]

test_orders = [
    (1, 3, "2019-03-04", 100),
    (2, 3, "2019-03-01", 80),
    (4, 7, "2019-02-01", 25),
    (6, 15, "2019-02-01", 100)
]

Et on définit la sortie attendue, qui sert de référence de comparaison :

expected = [
    ("Mia", "2019-02-01", 100),
    ("Farida", "2019-03-01", 80),
    ("Farida", "2019-03-04", 100)
]

ÉCRIRE LE TEST UNITAIRE

Maintenant que la requête, les entrées et les sorties attendues sont définies, on peut écrire un test unitaire. Le framework unittest de Python garde les dépendances minimales tout en offrant structure et répétabilité. On commence par créer une base de données SQLite en mémoire :

conn = sqlite3.connect(":memory:")
cursor = conn.cursor()

On recrée uniquement les tables nécessaires à la requête, mais avec un schéma réaliste (même si seules quelques colonnes sont utilisées) pour éviter une fausse confiance liée à une simplification excessive :

CREATE TABLE customers (.)
CREATE TABLE orders (.)

On insère ensuite les données de test contrôlées :

cursor.executemany("INSERT INTO customers VALUES (?, ?, ?, ?, ?, ?)", test_customers)
cursor.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", test_orders)
conn.commit()

Avant d’exécuter la requête, on peut charger et afficher les tables de test avec Pandas (étape optionnelle mais utile pour le débogage) :

customersdf = pd.readsql("SELECT id, firstname, lastname, city FROM customers", conn)
ordersdf = pd.readsql("SELECT * FROM orders", conn)

On lance la requête à tester :

result = pd.read_sql(query, conn)

La vérification se fait ligne par ligne : on compare d’abord la taille du résultat avec celle attendue :

all_correct = True

if len(result) .= len(expected):
    all_correct = False

Puis on itère sur chaque élément attendu pour vérifier les valeurs :

for i, (fname, lname, date, cost) in enumerate(expected):
    if i < len(result):
        actual = result.iloc[i]
        if not (
            actual["first_name"] == fname
            and actual["last_name"] == lname
            and actual["order_date"] == date
            and actual["max_cost"] == cost
        ):
            all_correct = False

Enfin, un message clair résume le résultat :

if all_correct and len(result) == len(expected):
    print("ALL TESTS PASSED")
else:
    print("SOME TESTS FAILED")

AUTOMATISATION AVEC CI/CD

Une suite de tests n’est utile que si elle s’exécute de manière cohérente à chaque modification du code. On utilise GitHub Actions pour définir un workflow d’intégration continue qui lance les tests SQL à chaque push ou pull request sur la branche principale.

On crée le fichier .github/workflows/test_sql.yml avec le contenu suivant :

name: Run SQL Tests

on:
  push:
    branches: [ "main" ]
  pull_request:
    branches: [ "main" ]

Cela permet de détecter les régressions SQL avant qu’elles ne soient fusionnées.

On définit ensuite un job test qui s’exécute sur un environnement Ubuntu propre :

jobs:
  test:
    runs-on: ubuntu-latest

Les étapes du job :

- name: Checkout repository
  uses: actions/checkout@v4

Cette étape télécharge le code du dépôt pour que le runner puisse accéder aux fichiers SQL et aux tests.

- name: Set up Python
  uses: actions/setup-python@v5
  with:
    python-version: "3.10"

Elle installe Python 3.10 pour garantir un runtime cohérent.

- name: Install dependencies
  run: |
    python -m pip install --upgrade pip
    pip install -r requirements.txt

Cette commande installe toutes les bibliothèques Python nécessaires (comme Pandas) listées dans requirements.txt.

- name: Run unit tests
  run: python -m unittest discover

Enfin, on lance les tests unitaires. Le workflow s’exécute automatiquement à chaque commit ou pull request, et les résultats sont visibles dans l’onglet Actions du dépôt GitHub.

LA QUALITÉ DES DONNÉES, MAILLON FAIBLE DES PIPELINES

Les tests unitaires confirment que la logique produit toujours la sortie attendue, et la CI garantit leur exécution automatique. Mais dans les environnements de données réels, les données elles-mêmes peuvent causer des échecs : lignes arrivant en retard, dates mal formées, clés manquantes, doublons inattendus. C’est là qu’intervient l’automatisation de la qualité des données. Sans contrôles automatisés, ces problèmes faussent silencieusement les résultats car SQL ne lève pas d’exceptions évidentes.

LES RÈGLES DE QUALITÉ EN SQL

Chaque requête SQL repose sur des hypothèses implicites. Pour notre requête sur les meilleurs dépensiers quotidiens, plusieurs hypothèses critiques doivent être vérifiées.

Les prénoms doivent être uniques : la jointure se fait par ID, mais l’affichage utilise le prénom. Si les prénoms ne sont plus uniques, le résultat devient ambigu :

SELECT first_name, COUNT(*)
FROM customers
GROUP BY first_name
HAVING COUNT(*) > 1;

Les coûts des commandes ne doivent pas être négatifs : des valeurs négatives indiquent souvent des erreurs d’ingestion :

SELECT *
FROM orders
WHERE totalordercost < 0;

Les dates de commande doivent être valides et dans une plage attendue : des dates nulles ou aberrantes trahissent des problèmes de synchronisation :

SELECT *
FROM orders
WHERE order_date IS NULL
   OR orderdate > CURRENTDATE;

Chaque commande doit référencer un client valide : une commande orpheline serait silencieusement exclue des jointures :

SELECT o.*
FROM orders o
LEFT JOIN customers c ON c.id = o.cust_id
WHERE c.id IS NULL;

AUTOMATISER LES CONTRÔLES QUALITÉ

Plutôt que d’exécuter ces vérifications manuellement, on les encapsule dans une fonction Python qui lève une exception si une règle est violée :

import pandas as pd

def rundataquality_checks(conn):
    checks = {
        "Duplicate first names": """
            SELECT first_name
            FROM customers
            GROUP BY first_name
            HAVING COUNT(*) > 1;
        """,
        "Negative order costs": """
            SELECT *
            FROM orders
            WHERE totalordercost < 0;
        """,
        "Invalid order dates": """
            SELECT *
            FROM orders
            WHERE order_date IS NULL
               OR order_date > CURRENT_DATE;
        """,
        "Orders without customers": """
            SELECT o.*
            FROM orders o
            LEFT JOIN customers c ON c.id = o.cust_id
            WHERE c.id IS NULL;
        """
    }

    for rule_name, query in checks.items():
        result = pd.read_sql(query, conn)
        if not result.empty:
            raise ValueError(f"Data quality check failed: {rule_name}")

    print("All data quality checks passed.")

Ces contrôles qualité se comportent comme des tests unitaires : succès ou échec. Comme ils sont en Python, ils sont automatiquement exécutés par le workflow GitHub Actions existant, via la commande python -m unittest discover.

CONCLUSION : VERS UN SQL FIABLE

La plupart des gens s’arrêtent une fois que la requête produit une réponse correcte. Mais les environnements de données réels récompensent ceux qui rendent leurs requêtes stables, testables et versionnées. Combiner les tests unitaires, l’intégration continue et les contrôles de qualité des données garantit que la requête continuera de fournir des résultats fiables, même lorsque les données évoluent.