From 76b1f8673ffb082c751c9958b44bd9f3381304d3 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 16 May 2026 13:07:13 +0000 Subject: [PATCH] test(infra): integration phantom column gate (#197) (#216) --- backend/pyproject.toml | 1 + backend/tests/integration/__init__.py | 0 backend/tests/integration/conftest.py | 153 +++++++ .../tests/integration/test_phantom_columns.py | 379 ++++++++++++++++++ 4 files changed, 533 insertions(+) create mode 100644 backend/tests/integration/__init__.py create mode 100644 backend/tests/integration/conftest.py create mode 100644 backend/tests/integration/test_phantom_columns.py diff --git a/backend/pyproject.toml b/backend/pyproject.toml index dbbbb3b0..3a42d9ed 100644 --- a/backend/pyproject.toml +++ b/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)", ] diff --git a/backend/tests/integration/__init__.py b/backend/tests/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/integration/conftest.py b/backend/tests/integration/conftest.py new file mode 100644 index 00000000..1ad9c920 --- /dev/null +++ b/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_ + 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=. + + Возвращает 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() diff --git a/backend/tests/integration/test_phantom_columns.py b/backend/tests/integration/test_phantom_columns.py new file mode 100644 index 00000000..0f395661 --- /dev/null +++ b/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 + """, + )