Kansho/backend/tests/harness.py

"""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)
    os.environ.pop("KANSHO_ALLOW_REMOTE_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