98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
"""Postgres test engine on the existing Dev instance.
|
|
|
|
Two Compose Postgres instances exist: Dev and Prod. Tests use the Dev
|
|
instance, database `kansho_test` beside `kansho_dev`. That is not a third
|
|
server. The suite drops and recreates `public` on the test database only,
|
|
so personal Dev/Prod journals are not wiped.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
|
|
LIVE_DB_NAMES = frozenset(
|
|
{
|
|
"kansho",
|
|
"kansho_dev",
|
|
"postgres",
|
|
"template0",
|
|
"template1",
|
|
}
|
|
)
|
|
|
|
|
|
def _backend() -> str:
|
|
return (os.environ.get("KANSHO_DB_BACKEND") or "").strip().lower()
|
|
|
|
|
|
def _db_name() -> str:
|
|
return (os.environ.get("DB_NAME") or "").strip()
|
|
|
|
|
|
def is_postgres_test_database() -> bool:
|
|
if _backend() not in {"postgres", "postgresql", "pg"}:
|
|
return False
|
|
name = _db_name().lower()
|
|
if not name or name in LIVE_DB_NAMES:
|
|
return False
|
|
return name.endswith("_test")
|
|
|
|
|
|
def prepare_test_env() -> None:
|
|
os.environ.setdefault("KANSHO_PROVIDER_KEY", "")
|
|
os.environ.setdefault("KANSHO_DETECT_PROVIDER_KEY", "")
|
|
os.environ.pop("KANSHO_DB_PATH", None)
|
|
|
|
|
|
def _require_test_database() -> None:
|
|
if is_postgres_test_database():
|
|
return
|
|
backend = _backend()
|
|
if backend not in {"postgres", "postgresql", "pg"}:
|
|
raise SystemExit(
|
|
"Backend-Tests laufen gegen PostgreSQL (KANSHO_DB_BACKEND=postgres), "
|
|
"dieselbe Engine wie Dev und Prod. SQLite-Testdateien sind abgelöst. "
|
|
"Gitea nutzt auf der Dev-Postgres die Datenbank kansho_test "
|
|
"(neben kansho_dev, keine dritte Instanz)."
|
|
)
|
|
name = _db_name().lower()
|
|
if not name:
|
|
raise SystemExit(
|
|
"DB_NAME fehlt. Tests nutzen auf der Dev-Postgres die Datenbank kansho_test, "
|
|
"nicht kansho_dev."
|
|
)
|
|
raise SystemExit(
|
|
f"DB_NAME={_db_name()!r} ist für Tests gesperrt. "
|
|
"Die Suite setzt das Schema zurück und darf daher nicht kansho oder kansho_dev treffen."
|
|
)
|
|
|
|
|
|
def reset_postgres_schema() -> None:
|
|
import psycopg
|
|
|
|
from sql_compat import postgres_connect_kwargs
|
|
|
|
kwargs = postgres_connect_kwargs()
|
|
if "conninfo" in kwargs:
|
|
conn = psycopg.connect(kwargs["conninfo"], autocommit=True)
|
|
else:
|
|
conn = psycopg.connect(autocommit=True, **kwargs)
|
|
try:
|
|
conn.execute("DROP SCHEMA IF EXISTS public CASCADE")
|
|
conn.execute("CREATE SCHEMA public")
|
|
conn.execute("GRANT ALL ON SCHEMA public TO PUBLIC")
|
|
conn.execute("GRANT ALL ON SCHEMA public TO CURRENT_USER")
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def configure_test_engine() -> None:
|
|
"""Reset kansho_test. pytest uses the fixture in tests/conftest.py instead."""
|
|
prepare_test_env()
|
|
_require_test_database()
|
|
try:
|
|
reset_postgres_schema()
|
|
except Exception as exc: # noqa: BLE001 — fail before the suite touches a live DB
|
|
print(f"Postgres-Testschema konnte nicht zurückgesetzt werden: {exc}", file=sys.stderr)
|
|
raise SystemExit(1) from exc
|