105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
# #region Test.PytestEarly.IntegrationDatabase [C:5] [TYPE Module] [SEMANTICS test,pytest,early-hook,postgres,testcontainers,alembic]
|
|
# @defgroup Test.PytestEarly Prepare the integration database before application imports.
|
|
# @PRE pytest is invoked with --run-integration; Docker and Testcontainers are available.
|
|
# @POST DATABASE_URL points at a temporary PostgreSQL database migrated to head before conftest loading.
|
|
# @INVARIANT Application modules are not imported until the integration DATABASE_URL is installed.
|
|
# @RATIONALE pytest_load_initial_conftests is the earliest supported hook for a command-line plugin.
|
|
# It runs before root conftest.py, whose pytest_plugins import the integration fixture modules.
|
|
# @REJECTED Starting PostgreSQL from a session fixture was rejected because tests/conftest.py and
|
|
# plugin imports can import src.core.database before session fixtures are evaluated.
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import uuid
|
|
|
|
from sqlalchemy.engine import make_url
|
|
|
|
_container = None
|
|
_database_name: str | None = None
|
|
_admin_url: str | None = None
|
|
|
|
|
|
def pytest_load_initial_conftests(early_config, parser, args): # noqa: ARG001
|
|
"""Install the migrated PostgreSQL URL before any project conftest loads."""
|
|
if "--run-integration" not in args:
|
|
return
|
|
|
|
from testcontainers.postgres import PostgresContainer
|
|
|
|
global _admin_url, _container, _database_name
|
|
_container = PostgresContainer(
|
|
image="postgres:16-alpine",
|
|
username="test",
|
|
password="test",
|
|
dbname="test_translate",
|
|
)
|
|
_container.start()
|
|
|
|
_admin_url = _container.get_connection_url()
|
|
admin = make_url(_admin_url)
|
|
_database_name = f"ss_test_global_{uuid.uuid4().hex[:12]}"
|
|
|
|
import psycopg2
|
|
|
|
connection = psycopg2.connect(
|
|
host=_container.get_container_host_ip(),
|
|
port=_container.get_exposed_port(5432),
|
|
user=admin.username,
|
|
password=admin.password,
|
|
dbname=admin.database,
|
|
)
|
|
connection.autocommit = True
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(f'CREATE DATABASE "{_database_name}"')
|
|
finally:
|
|
connection.close()
|
|
|
|
database_url = admin.set(database=_database_name).render_as_string(hide_password=False)
|
|
os.environ["DATABASE_URL"] = database_url
|
|
|
|
backend_dir = Path(__file__).resolve().parent
|
|
subprocess.run(
|
|
[sys.executable, "-m", "alembic", "upgrade", "head"],
|
|
cwd=backend_dir,
|
|
env=os.environ.copy(),
|
|
check=True,
|
|
)
|
|
|
|
|
|
def pytest_unconfigure(config): # noqa: ARG001
|
|
"""Drop the temporary database and stop the container after the test session."""
|
|
if _container is None:
|
|
return
|
|
|
|
import psycopg2
|
|
|
|
if _database_name and _admin_url:
|
|
admin = make_url(_admin_url)
|
|
connection = psycopg2.connect(
|
|
host=_container.get_container_host_ip(),
|
|
port=_container.get_exposed_port(5432),
|
|
user=admin.username,
|
|
password=admin.password,
|
|
dbname=admin.database,
|
|
)
|
|
connection.autocommit = True
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
|
|
"WHERE datname = %s AND pid <> pg_backend_pid()",
|
|
(_database_name,),
|
|
)
|
|
cursor.execute(f'DROP DATABASE IF EXISTS "{_database_name}"')
|
|
finally:
|
|
connection.close()
|
|
_container.stop()
|
|
|
|
|
|
# #endregion Test.PytestEarly.IntegrationDatabase
|