hf_space/tests/conftest.py

from __future__ import annotations

import importlib
import os
from pathlib import Path
from typing import Generator

import pandas as pd
import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.engine import make_url
from sqlalchemy.exc import OperationalError

import scripts.init_db as init_db
from projet_05.settings import load_settings


@pytest.fixture(scope="session")
def test_db_url() -> str:
    """URL de connexion PostgreSQL utilisée pendant les tests."""

    return os.getenv(
        "PROJET05_TEST_DATABASE_URL",
        "postgresql+psycopg://postgres:postgres@localhost:5432/projet05_test",
    )


@pytest.fixture(scope="session")
def initialized_db(test_db_url: str) -> Generator[str, None, None]:
    """Crée une base dédiée aux tests, exécute scripts.init_db puis la met à disposition."""

    url = make_url(test_db_url)
    admin_url = url.set(database="postgres")
    try:
        admin_engine = create_engine(admin_url, future=True)
        with admin_engine.execution_options(isolation_level="AUTOCOMMIT").connect() as conn:
            conn.execute(text(f"DROP DATABASE IF EXISTS {url.database}"))
            conn.execute(text(f"CREATE DATABASE {url.database}"))
    except OperationalError as exc:
        pytest.skip(f"Base PostgreSQL de test indisponible: {exc}")
    os.environ["PROJET05_DATABASE_URL"] = test_db_url
    load_settings.cache_clear()
    init_db.main(settings_path=None)
    yield test_db_url
    load_settings.cache_clear()
    with admin_engine.execution_options(isolation_level="AUTOCOMMIT").connect() as conn:
        conn.execute(
            text(
                "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
                "WHERE datname = :db AND pid <> pg_backend_pid()"
            ),
            {"db": url.database},
        )
        conn.execute(text(f"DROP DATABASE IF EXISTS {url.database}"))


@pytest.fixture(scope="session")
def raw_rowcount() -> int:
    """Nombre de lignes attendues dans les CSV bruts."""

    source = Path("data/raw/extrait_sirh.csv")
    return pd.read_csv(source).shape[0]