95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
"""Give every pytest-xdist worker its own database, so a suite that shares one test database can run in parallel.
|
|
|
|
Generic (source of truth: the memory repo's tools/pytest-parallel/db_per_worker.py); a project carries a verbatim copy
|
|
next to its conftest — tests often run in a container that cannot see this repo — and
|
|
`tools/pytest-parallel/sync-check.sh <copy>` reports drift. Keep the copy byte-identical; change it here.
|
|
|
|
Use, as a pytest plugin (it then runs before any conftest, i.e. before the app reads its settings):
|
|
pytest.ini: addopts = -p tests.db_per_worker (the copy's import path; `python -m pytest` puts cwd on sys.path)
|
|
db_per_worker_env = DATABASE_URL (the environment variable that holds the test database)
|
|
or call `apply("<ENV VAR>")` yourself before the app is imported. The variable may hold:
|
|
- a Postgres URL (`postgresql://…/name`, `postgresql+psycopg://…`): the worker gets `name_gw<N>`, created from the
|
|
base database's server if missing (empty — the project's own schema fixture migrates it, as for one process);
|
|
- a SQLite URL (`sqlite:///path.db`) or a plain file path: the worker gets `path-gw<N>.db`.
|
|
Without xdist (no PYTEST_XDIST_WORKER) or with the variable unset it changes nothing, so a serial run is unaffected.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
import pytest
|
|
|
|
_PG = re.compile(r"^postgres(?:ql)?(?:\+\w+)?://", re.I)
|
|
|
|
|
|
def pytest_addoption(parser: pytest.Parser) -> None:
|
|
parser.addini(
|
|
"db_per_worker_env",
|
|
"env var holding the test database (a Postgres URL, SQLite URL or file path)",
|
|
)
|
|
|
|
|
|
@pytest.hookimpl(tryfirst=True)
|
|
def pytest_load_initial_conftests(early_config: pytest.Config) -> None:
|
|
var = early_config.getini("db_per_worker_env")
|
|
if var:
|
|
apply(str(var))
|
|
|
|
|
|
def apply(var: str) -> str | None:
|
|
worker = os.environ.get("PYTEST_XDIST_WORKER")
|
|
value = os.environ.get(var)
|
|
if not worker or not value:
|
|
return value
|
|
if _PG.match(value):
|
|
new = _postgres(value, worker)
|
|
else:
|
|
new = _file(value, worker)
|
|
os.environ[var] = new
|
|
return new
|
|
|
|
|
|
def _postgres(url: str, worker: str) -> str:
|
|
parts = urlsplit(url)
|
|
base = parts.path.lstrip("/")
|
|
name = f"{base}_{worker}"
|
|
if not re.fullmatch(r"\w+", name):
|
|
raise ValueError(f"db_per_worker: refusing an unusual database name {name!r}")
|
|
# The server, through the BASE database (it exists; `postgres` may not be reachable for this role).
|
|
plain = urlunsplit(
|
|
(re.sub(r"\+\w+$", "", parts.scheme), parts.netloc, parts.path, parts.query, "")
|
|
)
|
|
_create_if_missing(plain, name)
|
|
return urlunsplit((parts.scheme, parts.netloc, f"/{name}", parts.query, parts.fragment))
|
|
|
|
|
|
def _create_if_missing(dsn: str, name: str) -> None:
|
|
try:
|
|
import psycopg # psycopg 3
|
|
|
|
conn = psycopg.connect(dsn, autocommit=True)
|
|
except ImportError:
|
|
import psycopg2 # fallback for projects on psycopg2
|
|
|
|
conn = psycopg2.connect(dsn)
|
|
conn.autocommit = True
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (name,))
|
|
if cur.fetchone() is None:
|
|
cur.execute(f'CREATE DATABASE "{name}"') # name checked to be \w+ above
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _file(value: str, worker: str) -> str:
|
|
prefix = ""
|
|
m = re.match(r"^(sqlite(?:\+\w+)?:///)(.*)$", value, re.I)
|
|
if m:
|
|
prefix, value = m.group(1), m.group(2)
|
|
root, ext = os.path.splitext(value)
|
|
return f"{prefix}{root}-{worker}{ext}"
|