Tests en Python¶

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

from database import Base, get_db
from main import app

# Base de donnees SQLite en memoire pour les tests
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = async_sessionmaker(bind=test_engine, class_=AsyncSession)


async def override_get_db():
    """Remplace la session de production par une session de test."""
    async with TestSessionLocal() as session:
        yield session


@pytest.fixture(autouse=True)
async def setup_database():
    """Cree et detruit les tables autour de chaque test."""
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture
async def client():
    """Client HTTP de test avec injection de la base de test."""
    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac
    app.dependency_overrides.clear()

# tests/test_items.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_lister_items_vide(client: AsyncClient):
    """La liste est vide au demarrage."""
    reponse = await client.get("/items")
    assert reponse.status_code == 200
    assert reponse.json() == []


@pytest.mark.asyncio
async def test_creer_item(client: AsyncClient):
    """La creation retourne 201 et l'item cree."""
    payload = {"nom": "Widget", "prix": 9.99, "disponible": True}
    reponse = await client.post("/items", json=payload)
    assert reponse.status_code == 201
    data = reponse.json()
    assert data["nom"] == "Widget"
    assert data["prix"] == 9.99
    assert "id" in data


@pytest.mark.asyncio
async def test_lire_item_inexistant(client: AsyncClient):
    """La lecture d'un item inexistant retourne 404."""
    reponse = await client.get("/items/999")
    assert reponse.status_code == 404
    assert "non trouve" in reponse.json()["detail"]


@pytest.mark.asyncio
async def test_mettre_a_jour_item(client: AsyncClient):
    """La mise a jour partielle modifie uniquement les champs fournis."""
    # Creation
    creation = await client.post("/items", json={"nom": "Original", "prix": 5.0})
    item_id = creation.json()["id"]

    # Mise a jour du prix uniquement
    maj = await client.put(f"/items/{item_id}", json={"prix": 12.50})
    assert maj.status_code == 200
    assert maj.json()["prix"] == 12.50
    assert maj.json()["nom"] == "Original"  # Inchange


@pytest.mark.asyncio
async def test_supprimer_item(client: AsyncClient):
    """La suppression retourne 204 et l'item n'est plus accessible."""
    creation = await client.post("/items", json={"nom": "A supprimer", "prix": 1.0})
    item_id = creation.json()["id"]

    suppression = await client.delete(f"/items/{item_id}")
    assert suppression.status_code == 204

    lecture = await client.get(f"/items/{item_id}")
    assert lecture.status_code == 404


@pytest.mark.asyncio
@pytest.mark.parametrize("prix_invalide", [-1, 0, -100.5])
async def test_creer_item_prix_invalide(client: AsyncClient, prix_invalide: float):
    """Un prix non positif est rejete par la validation Pydantic."""
    reponse = await client.post(
        "/items", json={"nom": "Item", "prix": prix_invalide}
    )
    assert reponse.status_code == 422

# pip install pytest-mock

# tests/test_pipeline.py
from unittest.mock import MagicMock, patch
import polars as pl
import pytest


def test_pipeline_appelle_enrichir(mocker):
    """Verifie que le pipeline appelle les etapes dans le bon ordre."""
    # Arrange
    df_mock = pl.DataFrame({"quantite": [1], "prix_unitaire": [10.0]})
    mocker.patch("pipeline.charger_ventes", return_value=df_mock)
    enrichir_spy = mocker.patch(
        "pipeline.enrichir_ventes",
        wraps=lambda df: df.with_columns(
            (pl.col("quantite") * pl.col("prix_unitaire")).alias("ca"),
            pl.lit(1).alias("mois"),
        ),
    )

    from pipeline import executer_pipeline
    executer_pipeline("faux_chemin.csv")

    enrichir_spy.assert_called_once()


def test_envoi_email_retente_en_cas_erreur():
    """Simule une erreur reseau et verifie le mecanisme de retry."""
    service_email = MagicMock()
    service_email.envoyer.side_effect = [
        ConnectionError("Timeout"),  # Premier appel echoue
        {"statut": "ok"},            # Deuxieme appel reussit
    ]

    tentatives = 0
    for _ in range(2):
        try:
            service_email.envoyer("test@example.com")
            tentatives += 1
            break
        except ConnectionError:
            tentatives += 1

    assert tentatives == 2
    assert service_email.envoyer.call_count == 2

# pip install hypothesis

from hypothesis import given, settings, strategies as st
from hypothesis.strategies import floats, text
import pytest


def calculer_tva(prix_ht: float, taux: float = 0.20) -> float:
    """Calcule le prix TTC."""
    if prix_ht < 0:
        raise ValueError("Le prix HT ne peut pas etre negatif")
    return round(prix_ht * (1 + taux), 2)


@given(prix=floats(min_value=0, max_value=1_000_000, allow_nan=False))
def test_tva_toujours_superieure_au_ht(prix: float):
    """Le prix TTC est toujours >= au prix HT pour un taux positif."""
    resultat = calculer_tva(prix)
    assert resultat >= prix


@given(prix=floats(min_value=0, max_value=1_000_000, allow_nan=False))
def test_tva_zero_retourne_prix_ht(prix: float):
    """Avec un taux de 0, le TTC egal le HT."""
    assert calculer_tva(prix, taux=0.0) == round(prix, 2)


@given(prix=floats(max_value=-0.01, allow_nan=False))
def test_tva_rejette_prix_negatif(prix: float):
    """Un prix negatif leve une ValueError."""
    with pytest.raises(ValueError):
        calculer_tva(prix)


@given(
    nom=text(min_size=1, max_size=200, alphabet=st.characters(whitelist_categories=("L",)))
)
def test_nom_item_round_trip(nom: str):
    """Verifie que le nom est conserve intact apres serialisation Pydantic."""
    from schemas import ItemCreate, ItemResponse
    item = ItemCreate(nom=nom, prix=1.0)
    assert item.nom == nom