148 lines
4.7 KiB
Python
148 lines
4.7 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:
|
|
"""Fail-closed defaults. Do not keep Compose provider keys or leftover fakes."""
|
|
os.environ["KANSHO_PROVIDER_KEY"] = ""
|
|
os.environ["KANSHO_DETECT_PROVIDER_KEY"] = ""
|
|
os.environ.pop("KANSHO_DB_PATH", None)
|
|
os.environ.pop("KANSHO_FAKE_PROVIDER", None)
|
|
os.environ.pop("KANSHO_FAKE_DETECT", None)
|
|
|
|
|
|
def apply_module_env(source) -> None:
|
|
"""Restore per-file env so pytest does not leak fakes/keys across scripts."""
|
|
from pathlib import Path
|
|
|
|
prepare_test_env()
|
|
text = Path(source).read_text(encoding="utf-8")
|
|
if 'os.environ["KANSHO_FAKE_PROVIDER"] = "1"' in text or "os.environ['KANSHO_FAKE_PROVIDER'] = '1'" in text:
|
|
os.environ["KANSHO_FAKE_PROVIDER"] = "1"
|
|
if (
|
|
'os.environ["KANSHO_FAKE_DETECT"] = "1"' in text
|
|
or "os.environ['KANSHO_FAKE_DETECT'] = '1'" in text
|
|
or 'os.environ.setdefault("KANSHO_FAKE_DETECT", "1")' in text
|
|
):
|
|
os.environ["KANSHO_FAKE_DETECT"] = "1"
|
|
|
|
|
|
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 db_init import mark_schema_dirty
|
|
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()
|
|
mark_schema_dirty()
|
|
|
|
|
|
def reset_postgres_data() -> None:
|
|
"""Clear rows on kansho_test without reloading schema.sql."""
|
|
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:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT tablename FROM pg_tables
|
|
WHERE schemaname = 'public' AND tablename <> 'schema_migrations'
|
|
"""
|
|
).fetchall()
|
|
names = [row[0] for row in rows]
|
|
if names:
|
|
joined = ", ".join('"' + name.replace('"', '""') + '"' for name in names)
|
|
conn.execute(f"TRUNCATE TABLE {joined} RESTART IDENTITY CASCADE")
|
|
finally:
|
|
conn.close()
|
|
from db import refresh_runtime_seed
|
|
|
|
refresh_runtime_seed()
|
|
|
|
|
|
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
|