153 lines
6.2 KiB
Python
153 lines
6.2 KiB
Python
"""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()
|