ben.stull
9462423642
Forward-only, fail-stop, advisory-lock-guarded psycopg runner; migration 0001 lays the account/auth_code/storefront/membership schema. Tests prove migrate-from-empty and idempotent re-migrate (INV-1 partial) and the INV-4 many-capable shape. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
"""Test fixtures — a fresh, empty PostgreSQL database per test.
|
|
|
|
Each test gets its own database created on the dev/CI Postgres (every test runs the
|
|
real engine, §6.8). The admin DSN points at the `postgres` maintenance database;
|
|
ECOMM_TEST_ADMIN_URL overrides it (CI sets it to the service container).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
|
|
import psycopg
|
|
import pytest
|
|
|
|
ADMIN_URL = os.environ.get(
|
|
"ECOMM_TEST_ADMIN_URL", "postgresql://ecomm:ecomm@localhost:5432/postgres"
|
|
)
|
|
|
|
|
|
def _swap_dbname(dsn: str, dbname: str) -> str:
|
|
# ADMIN_URL ends in `/postgres`; point the same credentials at `dbname`.
|
|
base, _, _old = dsn.rpartition("/")
|
|
return f"{base}/{dbname}"
|
|
|
|
|
|
@pytest.fixture()
|
|
def fresh_db_url():
|
|
"""Create a uniquely-named empty database; drop it after the test."""
|
|
dbname = f"ecomm_test_{uuid.uuid4().hex}"
|
|
admin = psycopg.connect(ADMIN_URL, autocommit=True)
|
|
try:
|
|
admin.execute(f'CREATE DATABASE "{dbname}"')
|
|
finally:
|
|
admin.close()
|
|
try:
|
|
yield _swap_dbname(ADMIN_URL, dbname)
|
|
finally:
|
|
admin = psycopg.connect(ADMIN_URL, autocommit=True)
|
|
try:
|
|
admin.execute(
|
|
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
|
|
"WHERE datname = %s AND pid <> pg_backend_pid()",
|
|
(dbname,),
|
|
)
|
|
admin.execute(f'DROP DATABASE IF EXISTS "{dbname}"')
|
|
finally:
|
|
admin.close()
|