File size: 2,100 Bytes
f84949e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
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]
|