1445043649
31361e2dcf
新建 backend/tests/fixtures/postgres.py，提供两层 fixture：
- postgres_container (session 级)：用 PostgresContainer 起 postgres:16-alpine
- postgres_url (function 级)：每 test 一个 ephemeral DB，teardown 时
pg_terminate_backend 清掉残连后 DROP DATABASE
为什么 per-DB 而不是 per-schema：asyncpg + SQLAlchemy 不通过 URL 传 search_path，
per-DB 一次 ~50ms 开销可接受，让 test 代码不感知 schema。
Docker 不可用时 fixture 自动 pytest.skip 而非 error，dev 环境无 Docker
仍能跑其余 3086 个测试。
注册 pytest mark `postgres`、把 tests/ 加 sys.path 让 pytest_plugins
按 `fixtures.postgres` 路径解析（不加 tests/__init__.py 避免干扰
现有 pytest 发现行为）。
Stage 0 PR1 T1.3.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
81 lines
3.3 KiB
Python
81 lines
3.3 KiB
Python
"""Postgres testcontainer fixtures for Stage 0 PR1.
|
|
|
|
Provides per-test ephemeral database isolation atop a single session-scoped
|
|
container. Tests marked ``@pytest.mark.postgres`` request the ``postgres_url``
|
|
fixture, which yields an asyncpg connection URL pointing at a freshly-created
|
|
database. The database is force-dropped after the test (any leaked connections
|
|
get pg_terminate_backend'd first).
|
|
|
|
Why per-database rather than per-schema:
|
|
asyncpg (the SQLAlchemy async driver we use) doesn't honor URL-embedded
|
|
search_path the way psycopg does. Per-database isolation is one extra
|
|
CREATE/DROP per test (~50ms), but lets test app code use its full schema
|
|
unchanged.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from collections.abc import Iterator
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def postgres_container():
|
|
"""Session-scoped Postgres 16 container shared across all postgres tests.
|
|
|
|
Started once per pytest session. Subsequent tests piggy-back on the same
|
|
container; each gets its own database via the ``postgres_url`` fixture.
|
|
|
|
Skipped (and the test marked skip) if Docker is unavailable on the host —
|
|
testcontainers raises ``DockerException`` when it can't reach the daemon.
|
|
"""
|
|
try:
|
|
from testcontainers.postgres import PostgresContainer
|
|
except ImportError as exc: # pragma: no cover - install boundary
|
|
pytest.skip(f"testcontainers[postgres] not installed: {exc}")
|
|
|
|
try:
|
|
with PostgresContainer("postgres:16-alpine") as pg:
|
|
yield pg
|
|
except Exception as exc: # pragma: no cover - environment-dependent
|
|
# DockerException, ConnectionError, etc. — surface a skip rather than
|
|
# an error so devs without Docker can still run the rest of the suite.
|
|
pytest.skip(f"could not start Postgres container ({exc})")
|
|
|
|
|
|
@pytest.fixture
|
|
def postgres_url(postgres_container) -> Iterator[str]:
|
|
"""Per-test ephemeral database URL (asyncpg dialect).
|
|
|
|
Each invocation creates a unique database on the shared container and
|
|
yields its URL. Teardown force-drops the database, terminating any
|
|
backend connections the test forgot to close.
|
|
"""
|
|
import psycopg
|
|
from psycopg import sql
|
|
|
|
db_name = f"test_{secrets.token_hex(8)}"
|
|
raw = postgres_container.get_connection_url() # postgresql+psycopg2://...
|
|
# Strip the SQLAlchemy dialect prefix so plain psycopg can connect.
|
|
base = raw.replace("postgresql+psycopg2://", "postgresql://")
|
|
parent_url = base.rsplit("/", 1)[0] + "/postgres"
|
|
|
|
# CREATE DATABASE must run outside a transaction; psycopg autocommit=True.
|
|
with psycopg.connect(parent_url, autocommit=True) as conn:
|
|
conn.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(db_name)))
|
|
|
|
asyncpg_url = raw.replace("postgresql+psycopg2://", "postgresql+asyncpg://").rsplit("/", 1)[0] + f"/{db_name}"
|
|
|
|
try:
|
|
yield asyncpg_url
|
|
finally:
|
|
with psycopg.connect(parent_url, autocommit=True) as conn:
|
|
# Kick any leaked connections so DROP DATABASE doesn't block.
|
|
conn.execute(
|
|
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = %s AND pid <> pg_backend_pid()",
|
|
(db_name,),
|
|
)
|
|
conn.execute(sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(db_name)))
|