test(infra): integration phantom column gate (#197)

Add pytest integration tests that run EXPLAIN against real Postgres schema
to detect phantom column/table references before PR merge. 12 EXPLAIN test
cases covering domrf_kn_objects, ekburg_construction_permits,
mv_layout_velocity, domrf_kn_flats, objective_* and cad geo tables.
Tests skip without TEST_DATABASE_URL; CI integration is a follow-up.

Pre-empts phantom column bugs caught manually on PR #196 (geom_3857)
and PR #213 (units_count).

Closes #197

backend/pyproject.toml

@@ -83,4 +83,5 @@ addopts = ["-m", "not prod_smoke"]
 markers = [
     "slow: marks tests as slow (need real network, deselect with -m 'not slow')",
     "prod_smoke: production smoke tests against live https://gendsgn.ru (run only post-deploy with -m prod_smoke)",
+    "integration: phantom column gate tests requiring TEST_DATABASE_URL (SSH tunnel to prod Postgres)",
 ]

backend/tests/integration/conftest.py

@@ -0,0 +1,153 @@
+"""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()

backend/tests/integration/test_phantom_columns.py

@@ -0,0 +1,379 @@
+"""Phantom column gate — EXPLAIN-тесты против реальной Postgres schema.
+
+Каждый тест выполняет EXPLAIN (не EXECUTE!) на SQL из production services.
+Если колонка не существует — Postgres бросит ошибку при планировании → тест fail.
+Если таблица не существует — аналогично.
+
+Запуск:
+    # Без TEST_DATABASE_URL — все тесты skip
+    uv run pytest tests/integration/ -v
+
+    # С SSH-туннелем (ssh -N gendesign → localhost:15432):
+    export TEST_DATABASE_URL="postgresql+psycopg://user:pass@localhost:15432/db"
+    uv run pytest tests/integration/test_phantom_columns.py -v -m integration
+
+Детектируемые классы багов:
+    - Ссылка на несуществующую колонку (domrf_kn_objects.geom_3857 — PR #196)
+    - Ссылка на несуществующую колонку (ekburg_construction_permits.units_count — PR #213)
+    - Ссылка на несуществующую таблицу / view
+    - Опечатки в именах колонок и таблиц
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from tests.integration.conftest import requires_test_db
+
+# ── Маркеры ───────────────────────────────────────────────────────────────────
+
+pytestmark = [
+    requires_test_db,
+    pytest.mark.integration,
+]
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+
+def _explain(db: Session, query: str, params: dict | None = None) -> None:
+    """Выполнить EXPLAIN для query. Не выполняет DML — только план запроса.
+
+    Бросает исключение (тест fail) если:
+    - колонка не существует
+    - таблица/view не существует
+    - синтаксическая ошибка SQL
+    """
+    explain_sql = f"EXPLAIN {query}"
+    db.execute(text(explain_sql), params or {})
+
+
+# ── 1. domrf_kn_objects — реальные колонки ────────────────────────────────────
+
+
+class TestDomrfKnObjects:
+    """Проверяем что SQL в services используют реальные колонки domrf_kn_objects.
+
+    Исторический баг PR #196: geom_3857 не существует.
+    Реальная schema: latitude, longitude (float), snapshot_date, obj_class, etc.
+    """
+
+    def test_competitors_radius_query(self, phantom_check_session: Session) -> None:
+        """best_layouts._COMPETITORS_IN_RADIUS_SQL — latitude/longitude/snapshot_date."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT DISTINCT ON (obj_id) obj_id
+            FROM domrf_kn_objects
+            WHERE latitude IS NOT NULL AND longitude IS NOT NULL
+              AND ST_DWithin(
+                  ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)::geography,
+                  ST_SetSRID(
+                      ST_MakePoint(CAST(60.6 AS float), CAST(56.8 AS float)), 4326
+                  )::geography,
+                  CAST(1000 AS float)
+              )
+            ORDER BY obj_id, snapshot_date DESC NULLS LAST
+            """,
+        )
+
+    def test_competitors_full_cte_query(self, phantom_check_session: Session) -> None:
+        """competitors._COMPETITORS_SQL — полный CTE с latitude/longitude/flat_count."""
+        _explain(
+            phantom_check_session,
+            """
+            WITH latest_obj AS (
+                SELECT DISTINCT ON (obj_id)
+                    obj_id,
+                    comm_name,
+                    dev_name,
+                    obj_class,
+                    latitude,
+                    longitude,
+                    flat_count,
+                    site_status,
+                    snapshot_date
+                FROM domrf_kn_objects
+                WHERE latitude IS NOT NULL
+                  AND longitude IS NOT NULL
+                ORDER BY obj_id, snapshot_date DESC NULLS LAST
+            ),
+            distances AS (
+                SELECT
+                    o.obj_id,
+                    o.comm_name,
+                    o.dev_name,
+                    o.obj_class,
+                    o.latitude,
+                    o.longitude,
+                    o.flat_count,
+                    o.site_status,
+                    ST_Distance(
+                        ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
+                        ST_SetSRID(
+                            ST_MakePoint(CAST(60.6 AS float), CAST(56.8 AS float)),
+                            4326
+                        )::geography
+                    ) AS distance_m
+                FROM latest_obj o
+            )
+            SELECT
+                d.obj_id,
+                d.comm_name,
+                d.dev_name,
+                d.obj_class,
+                d.latitude,
+                d.longitude,
+                d.flat_count,
+                d.site_status,
+                d.distance_m
+            FROM distances d
+            WHERE d.distance_m <= CAST(1000 AS float)
+            ORDER BY d.distance_m ASC
+            """,
+        )
+
+    def test_velocity_competitor_query_columns(self, phantom_check_session: Session) -> None:
+        """velocity._competitors query — district_name, region_cd, snapshot_date."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT DISTINCT ON (obj_id)
+                   obj_id,
+                   comm_name,
+                   dev_name,
+                   obj_class,
+                   latitude,
+                   longitude,
+                   district_name
+              FROM domrf_kn_objects
+             WHERE latitude IS NOT NULL
+               AND longitude IS NOT NULL
+               AND region_cd = 66
+             ORDER BY obj_id, snapshot_date DESC NULLS LAST
+            """,
+        )
+
+
+# ── 2. ekburg_construction_permits — реальные колонки ────────────────────────
+
+
+class TestEkburgConstructionPermits:
+    """Проверяем что SQL используют реальные колонки ekburg_construction_permits.
+
+    Исторический баг PR #213: units_count не существует.
+    Реальная schema: total_area_sqm, living_area_sqm, living_area_fact_sqm, etc.
+    """
+
+    def test_recent_permits_query(self, phantom_check_session: Session) -> None:
+        """parcels.py recent_permits query — total_area_sqm (не units_count!)."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                permit_type, permit_number, issue_date,
+                developer_name, developer_inn, object_name, object_type,
+                construction_address, total_area_sqm
+              FROM ekburg_construction_permits
+             WHERE LEFT(cadastral_number, LENGTH(CAST('66:41:0303161' AS text)))
+                   = CAST('66:41:0303161' AS text)
+               AND issue_date > NOW() - INTERVAL '24 months'
+             ORDER BY issue_date DESC
+             LIMIT 50
+            """,
+        )
+
+    def test_permits_rns_columns(self, phantom_check_session: Session) -> None:
+        """Все колонки RNS-специфичные: living_area_sqm, permit_type."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                permit_type,
+                permit_number,
+                issue_date,
+                expiry_date,
+                developer_name,
+                developer_inn,
+                object_name,
+                object_type,
+                construction_address,
+                cadastral_number,
+                total_area_sqm,
+                living_area_sqm
+              FROM ekburg_construction_permits
+             WHERE permit_type = 'RNS'
+             LIMIT 1
+            """,
+        )
+
+    def test_permits_rve_columns(self, phantom_check_session: Session) -> None:
+        """Все колонки RVE-специфичные: living_area_fact_sqm, rve_number, rve_date."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                permit_type,
+                permit_number,
+                living_area_fact_sqm,
+                rve_number,
+                rve_date
+              FROM ekburg_construction_permits
+             WHERE permit_type = 'RVE'
+             LIMIT 1
+            """,
+        )
+
+
+# ── 3. mv_layout_velocity — materialized view ────────────────────────────────
+
+
+class TestMvLayoutVelocity:
+    """Проверяем materialized view mv_layout_velocity и её колонки."""
+
+    def test_velocity_by_room_bucket(self, phantom_check_session: Session) -> None:
+        """best_layouts._VELOCITY_BY_ROOM_SQL — room_bucket, total_deals_24mo, etc."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                room_bucket,
+                SUM(total_deals_24mo)                               AS sum_deals,
+                AVG(avg_area_m2)                                    AS avg_area_m2,
+                AVG(avg_price_thousand_rub_per_m2) * 1000.0         AS avg_price_per_m2_rub,
+                array_agg(DISTINCT obj_id)                          AS competitor_obj_ids,
+                COUNT(DISTINCT obj_id)                              AS competitor_count,
+                MIN(window_start)                                   AS window_start,
+                MAX(window_end)                                     AS window_end
+            FROM mv_layout_velocity
+            WHERE obj_id = ANY(ARRAY[1, 2, 3])
+            GROUP BY room_bucket
+            """,
+        )
+
+
+# ── 4. domrf_kn_flats ─────────────────────────────────────────────────────────
+
+
+class TestDomrfKnFlats:
+    """Проверяем таблицу domrf_kn_flats и её колонки."""
+
+    def test_supply_batch_query(self, phantom_check_session: Session) -> None:
+        """best_layouts._SUPPLY_BATCH_SQL — total_area, rooms, is_studio, flat_type."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                CASE
+                    WHEN f.is_studio = TRUE OR f.flat_type = 'Квартира-студия' THEN 'studio'
+                    WHEN f.rooms = 0 THEN 'studio'
+                    WHEN f.rooms IN (1, 2, 3) THEN f.rooms::text
+                    WHEN f.rooms >= 4 THEN '4+'
+                    ELSE '1'
+                END AS rb,
+                CASE
+                    WHEN f.total_area < 25  THEN '<25'
+                    WHEN f.total_area < 40  THEN '25-40'
+                    WHEN f.total_area < 60  THEN '40-60'
+                    WHEN f.total_area < 80  THEN '60-80'
+                    WHEN f.total_area < 100 THEN '80-100'
+                    ELSE '100+'
+                END AS ab,
+                COUNT(*) AS units
+            FROM domrf_kn_flats f
+            JOIN domrf_kn_objects o ON f.obj_id = o.obj_id
+            WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
+              AND f.snapshot_date = CAST('2026-01-01' AS date)
+            GROUP BY rb, ab
+            """,
+        )
+
+    def test_avg_price_sold_query(self, phantom_check_session: Session) -> None:
+        """competitors._AVG_PRICE_SQL — price_per_m2, status='sold'."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                f.obj_id,
+                AVG(f.price_per_m2) AS avg_price_per_m2
+            FROM domrf_kn_flats f
+            WHERE f.obj_id = ANY(ARRAY[1, 2, 3])
+              AND f.price_per_m2 IS NOT NULL
+              AND f.status = 'sold'
+            GROUP BY f.obj_id
+            """,
+        )
+
+
+# ── 5. objective_complex_mapping / objective_corpus_room_month ────────────────
+
+
+class TestObjectiveTables:
+    """Проверяем objective_* таблицы и их колонки."""
+
+    def test_objective_mapping_columns(self, phantom_check_session: Session) -> None:
+        """competitors / velocity: domrf_obj_id, objective_complex_name в mapping."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT cm.domrf_obj_id AS obj_id,
+                   cm.objective_complex_name
+              FROM objective_complex_mapping cm
+             LIMIT 1
+            """,
+        )
+
+    def test_objective_corpus_room_month_columns(self, phantom_check_session: Session) -> None:
+        """velocity._VELOCITY_BY_ROOM_SQL — deals_total_count, deals_total_vol_m2."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                m.obj_id,
+                SUM(COALESCE(crm.deals_total_vol_m2,
+                             crm.deals_total_count * 45.0)) AS total_sqm,
+                COUNT(DISTINCT crm.report_month) AS months_with_data,
+                MIN(crm.report_month)            AS period_start,
+                MAX(crm.report_month)            AS period_end
+              FROM objective_corpus_room_month crm
+              JOIN (
+                  SELECT cm.domrf_obj_id AS obj_id,
+                         cm.objective_complex_name
+                    FROM objective_complex_mapping cm
+                   WHERE cm.domrf_obj_id = ANY(ARRAY[1, 2, 3])
+              ) m
+                ON m.objective_complex_name = crm.project_name
+             WHERE crm.report_month >= (CURRENT_DATE - CAST('12 months' AS interval))
+               AND crm.deals_total_count > 0
+             GROUP BY m.obj_id
+            """,
+        )
+
+
+# ── 6. cad geo таблицы ────────────────────────────────────────────────────────
+
+
+class TestCadGeoTables:
+    """Проверяем cad_parcels_geom, cad_quarters_geom и их колонки."""
+
+    def test_parcel_centroid_query(self, phantom_check_session: Session) -> None:
+        """best_layouts._PARCEL_CENTROID_SQL — cad_num, cad_number, geom."""
+        _explain(
+            phantom_check_session,
+            """
+            SELECT ST_X(pt) AS center_lon,
+                   ST_Y(pt) AS center_lat
+            FROM (
+                SELECT ST_Centroid(geom) AS pt
+                  FROM cad_parcels_geom
+                 WHERE cad_num = '66:41:0303161:123' AND geom IS NOT NULL
+                UNION ALL
+                SELECT ST_Centroid(geom) AS pt
+                  FROM cad_quarters_geom
+                 WHERE cad_number = '66:41:0303161' AND geom IS NOT NULL
+            ) sub
+            LIMIT 1
+            """,
+        )