Aller au contenu

Tests miroir

Le scaffold cookiecutter livre déjà la structure. Voici les conventions pour la maintenir.

Fixtures partagées (conftest.py)

import pytest
from fastapi.testclient import TestClient
from mic.datastorage import SqlDataStorage
from sqlalchemy.pool import StaticPool
from sqlmodel import create_engine

from app.bootstrap.config.settings import Settings
from app.bootstrap.startup.http_runtime import build_app


@pytest.fixture
def settings():
    return Settings(
        database_url="sqlite:///:memory:",
        log_level="warning",
        log_format="text",
    )


@pytest.fixture
def database(settings):
    """SQLite :memory: + StaticPool pour traverser les threads du TestClient."""
    engine = create_engine(
        settings.database_url,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    db = SqlDataStorage(url=settings.database_url, engine=engine)
    db.prepare()
    db.create_all()    # SQLite mémoire : crée le schéma (en prod = Alembic)
    yield db
    db.dispose()


@pytest.fixture
def client(settings, database):
    app = build_app(settings=settings, database=database)
    with TestClient(app) as test_client:
        yield test_client

Test miroir d'un controller (+ repository)

import pytest
from mic.datastorage import SqlDataStorage
from mic.model.errors import DomainError

from app.modules.mail.controller.enqueue_mail import EnqueueMailController
from app.modules.mail.model.repositories.mail_repository import MailRepository


def test_enqueue_creates_pending_job(database: SqlDataStorage) -> None:
    controller = EnqueueMailController(
        repository=MailRepository(session_factory=database.session),
    )
    result = controller.execute(recipient="a@b.c", subject="hi", body="hello")
    assert result.status == "pending"


def test_enqueue_rejects_blank_recipient(database: SqlDataStorage) -> None:
    controller = EnqueueMailController(
        repository=MailRepository(session_factory=database.session),
    )
    with pytest.raises(DomainError) as exc:
        controller.execute(recipient="   ", subject="s", body="b")
    assert exc.value.code == "mail.invalid_recipient"

Motif : on construit le repository sur la SqlDataStorage de test puis on exerce le controller — la chaîne controller → repository → persistence sans TestClient. Un test par chemin (heureux + chaque chemin d'erreur) ; le code d'erreur est vérifié explicitement — si quelqu'un le renomme, le test casse.

Test miroir d'une route via TestClient

def test_enqueue_message_returns_201(client) -> None:
    response = client.post(
        "/api/v1/messages",
        json={"recipient": "u@v.w", "subject": "S", "body": "B"},
    )
    assert response.status_code == 201
    payload = response.json()
    assert payload["success"] is True
    assert payload["data"]["status"] == "pending"


def test_enqueue_invalid_recipient_returns_400(client) -> None:
    response = client.post(
        "/api/v1/messages",
        json={"recipient": "not-an-email", "subject": "S", "body": "B"},
    )
    assert response.status_code == 400
    assert response.json()["code"] == "mail.invalid_recipient"


def test_enqueue_rejects_extra_field(client) -> None:
    """StrictSchema → extra=forbid → 400 http.validation_error."""
    response = client.post(
        "/api/v1/messages",
        json={"recipient": "u@v.w", "subject": "S", "body": "B", "secret": "boom"},
    )
    assert response.status_code == 400
    assert response.json()["code"] == "http.validation_error"

Test miroir d'un worker (sans temps réel)

from app.workers.mail import drain_once, run_worker_loop


def test_drain_once_processes_pending_via_dryrun(database):
    _enqueue(database, recipient="x@y.z")
    transport = DryRunTransport()
    processed = drain_once(
        session_factory=database.session,
        transport=transport,
    )
    assert processed == 1


def test_run_worker_loop_iterates_max_iterations_times(database):
    sleep_calls = []

    transport = DryRunTransport()
    total = run_worker_loop(
        session_factory=database.session,
        transport=transport,
        max_iterations=3,
        sleep_fn=sleep_calls.append,    # ← injection
    )
    assert sleep_calls == [5.0, 5.0]    # 2 sleeps entre 3 ticks

Couverture 100 % — diagnostic

Si pytest --cov-fail-under=100 plante :

app/modules/mail/controller/enqueue_mail.py     50      2     12      1   95%   42-44, 87

→ lignes 42-44 et 87 ne sont jamais exécutées. Trois options :

  1. Ajouter le test manquant (souvent un cas limite oublié).
  2. Refactoriser si le code est intestable (souvent il est aussi mal conçu).
  3. # pragma: no cover justifié en dernier recours.

Le make gate est l'arbitre

make gate
# → format-check + lint + typecheck + test-cov (--cov-fail-under=100)

La CI reflète le local. Pas de surprise à la fusion.