gendesign/backend/tests/integration/conftest.py

"""Integration test fixtures для phantom column gate.

Использует реальный PostgreSQL через SSH-туннель (localhost:15432).
Тесты пропускаются если TEST_DATABASE_URL не задан.

Подход (Option B): временная schema в prod-Postgres (не трогаем данные).
  1. CREATE SCHEMA test_phantom_<random>
  2. Копируем DDL public → временная schema через pg_dump --schema-only
  3. Выполняем EXPLAIN-запросы (не EXECUTE — readonly)
  4. DROP SCHEMA CASCADE после теста

Credentials: TEST_DATABASE_URL (env var, только через SSH-туннель).
Пример: postgresql+psycopg://user:pass@localhost:15432/gendesign
"""

from __future__ import annotations

import logging
import os
import random
import string
import subprocess
from collections.abc import Generator

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker

logger = logging.getLogger(__name__)

_TEST_DATABASE_URL = os.environ.get("TEST_DATABASE_URL", "")

# Используется во всех тестах этого пакета: пропускаем без TEST_DATABASE_URL
_SKIP_REASON = "TEST_DATABASE_URL не задан — интеграционные тесты требуют SSH-туннель"

requires_test_db = pytest.mark.skipif(
    not _TEST_DATABASE_URL,
    reason=_SKIP_REASON,
)


def _random_suffix(n: int = 8) -> str:
    return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))


def _pg_dump_schema_only(db_url: str, schema: str = "public") -> str:
    """Получить DDL через pg_dump --schema-only --schema=<schema>.

    Возвращает SQL как строку. Требует pg_dump в PATH.
    Credentials извлекаются из db_url.

    Fallback: если pg_dump недоступен — возвращает пустую строку,
    и тогда тест работает без создания отдельной schema (тестирует прямо в public).
    """
    try:
        result = subprocess.run(
            [
                "pg_dump",
                "--schema-only",
                f"--schema={schema}",
                "--no-owner",
                "--no-acl",
                db_url.replace("+psycopg", ""),  # pg_dump не понимает +psycopg dialect
            ],
            capture_output=True,
            text=True,
            timeout=30,
        )
        if result.returncode == 0:
            return result.stdout
        logger.warning("pg_dump failed (rc=%d): %s", result.returncode, result.stderr[:200])
    except FileNotFoundError:
        logger.warning("pg_dump не найден в PATH — phantom gate будет тестировать в public schema")
    except subprocess.TimeoutExpired:
        logger.warning("pg_dump timeout — phantom gate будет тестировать в public schema")
    return ""


@pytest.fixture(scope="session")
def phantom_schema_name() -> str:
    """Имя временной schema для phantom column тестов."""
    return f"test_phantom_{_random_suffix()}"


@pytest.fixture(scope="session")
def phantom_check_session(phantom_schema_name: str) -> Generator[Session, None, None]:
    """SQLAlchemy Session, настроенная на временную schema.

    Если pg_dump доступен — создаёт временную schema с копией DDL из public.
    Если нет — работает напрямую с public schema (EXPLAIN не требует изменения данных).
    После завершения сессии — DROP SCHEMA CASCADE.

    Пропускается если TEST_DATABASE_URL не задан.
    """
    if not _TEST_DATABASE_URL:
        pytest.skip(_SKIP_REASON)

    engine = create_engine(
        _TEST_DATABASE_URL,
        # Без пула — это одноразовая тестовая сессия
        pool_pre_ping=True,
        pool_size=1,
        max_overflow=0,
        echo=False,
    )
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)

    # Попробуем создать временную schema с копией DDL
    ddl_sql = _pg_dump_schema_only(_TEST_DATABASE_URL)
    use_temp_schema = bool(ddl_sql)
    schema = phantom_schema_name if use_temp_schema else "public"
    # Track schema creation independently — needed for teardown even if DDL apply fails
    schema_created = False

    if use_temp_schema:
        with engine.connect() as conn:
            conn.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"'))
            conn.commit()
            schema_created = True
            ddl_for_schema = ddl_sql.replace(
                "SET search_path = public", f'SET search_path = "{schema}"'
            ).replace("search_path TO public", f'search_path TO "{schema}"')
            prefixed_ddl = f'SET search_path TO "{schema}", public;\n' + ddl_for_schema
            try:
                conn.execute(text(prefixed_ddl))
                conn.commit()
            except Exception as e:
                logger.warning("DDL apply to temp schema failed: %s — falling back to public", e)
                use_temp_schema = False
                schema = "public"
                conn.rollback()

    db = session_factory()
    # Устанавливаем search_path для session: temp schema первая, public как fallback
    # (pg_dump DDL содержит объекты только под этой schema)
    if use_temp_schema:
        db.execute(text(f'SET search_path TO "{schema}", public'))
    else:
        db.execute(text("SET search_path TO public"))

    try:
        yield db
    finally:
        db.close()
        # Drop temp schema if it was created — even if DDL apply later failed.
        # Otherwise schemas leak when CREATE SCHEMA succeeds but DDL apply raises.
        if schema_created:
            drop_target = phantom_schema_name
            with engine.connect() as conn:
                conn.execute(text(f'DROP SCHEMA IF EXISTS "{drop_target}" CASCADE'))
                conn.commit()
                logger.info("phantom gate: dropped temp schema %s", drop_target)
        engine.dispose()