77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
import os
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import make_url
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from wisegrade.config import get_settings
|
|
from wisegrade.db import engine_for
|
|
from wisegrade.seed import seed_initial
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def test_database_url():
|
|
value = os.environ.get("TEST_DATABASE_URL")
|
|
if not value:
|
|
pytest.fail("Set TEST_DATABASE_URL to a dedicated PostgreSQL database named wisegrade_tests")
|
|
url = make_url(value)
|
|
if url.drivername != "postgresql+psycopg" or url.database != "wisegrade_tests":
|
|
pytest.fail("TEST_DATABASE_URL must use postgresql+psycopg and the wisegrade_tests database")
|
|
return url
|
|
|
|
|
|
@pytest.fixture
|
|
def environment(tmp_path, monkeypatch):
|
|
inbox = tmp_path / "inbox"
|
|
inbox.mkdir()
|
|
monkeypatch.setenv("INBOX_PATH", str(inbox))
|
|
monkeypatch.setenv("PROCESSED_PATH", str(tmp_path / "processed"))
|
|
monkeypatch.setenv("EXPORTS_PATH", str(tmp_path / "exports"))
|
|
monkeypatch.setenv("WISEGRADE_SECRET_KEY", "test-only-secret-key-of-at-least-32-characters")
|
|
get_settings.cache_clear()
|
|
yield tmp_path
|
|
get_settings.cache_clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def session_factory(environment, monkeypatch, test_database_url):
|
|
admin_engine = create_engine(test_database_url, isolation_level="AUTOCOMMIT")
|
|
name = f"wisegrade_test_{uuid4().hex}"
|
|
engine = None
|
|
created = False
|
|
try:
|
|
with admin_engine.connect() as connection:
|
|
connection.exec_driver_sql(f'CREATE DATABASE "{name}"')
|
|
created = True
|
|
url = test_database_url.set(database=name).render_as_string(hide_password=False)
|
|
monkeypatch.setenv("DATABASE_URL", url)
|
|
get_settings.cache_clear()
|
|
backend_root = Path(__file__).resolve().parents[1]
|
|
migration_config = Config(str(backend_root / "alembic.ini"))
|
|
migration_config.set_main_option("script_location", str(backend_root / "migrations"))
|
|
command.upgrade(migration_config, "head")
|
|
engine = engine_for(url)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
with factory() as db:
|
|
seed_initial(db)
|
|
yield factory
|
|
finally:
|
|
if engine is not None:
|
|
engine.dispose()
|
|
get_settings.cache_clear()
|
|
if created:
|
|
with admin_engine.connect() as connection:
|
|
connection.exec_driver_sql(f'DROP DATABASE "{name}" WITH (FORCE)')
|
|
admin_engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def db(session_factory):
|
|
with session_factory() as session:
|
|
yield session
|