"""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.

        Зеркалит objects-first MATERIALIZED CTE (latest-snapshot dedup, anti-fan-out)
        с хардкод-значениями вместо bind-параметров — EXPLAIN-гейт на phantom-колонки
        остаётся осмысленным после перехода с flats-first на objects-first.
        """
        _explain(
            phantom_check_session,
            """
            WITH latest_obj AS (
                SELECT DISTINCT ON (o.obj_id)
                    o.obj_id, o.latitude, o.longitude
                FROM domrf_kn_objects o
                WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
                ORDER BY o.obj_id, o.snapshot_date DESC
            ),
            nearby AS MATERIALIZED (
                SELECT lo.obj_id
                FROM latest_obj lo
                WHERE ST_DWithin(
                    ST_SetSRID(ST_MakePoint(lo.longitude, lo.latitude), 4326)::geography,
                    ST_SetSRID(
                        ST_MakePoint(CAST(60.6 AS float), CAST(56.84 AS float)), 4326
                    )::geography,
                    CAST(1500 AS float)
                )
            )
            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 nearby n
            JOIN domrf_kn_flats f
                ON f.obj_id = n.obj_id
               AND f.snapshot_date = CAST('2026-01-01' AS date)
            GROUP BY rb, ab
            """,
        )

    def test_avg_price_query(self, phantom_check_session: Session) -> None:
        """competitors._AVG_PRICE_SQL — price_per_m2 на последнем snapshot.

        #1926: status больше НЕ используется (был ~0.2%-заполнен). avg_price
        считается по всем квартирам с known price_per_m2 на latest snapshot;
        sold-count вынесен в _SOLD_COUNT_SQL (objective_lots.contract_date).
        """
        _explain(
            phantom_check_session,
            """
            SELECT
                f.obj_id,
                AVG(f.price_per_m2) FILTER (WHERE f.price_per_m2 IS NOT NULL)
                    AS avg_price_per_m2
            FROM domrf_kn_flats f
            WHERE f.obj_id = ANY(ARRAY[1, 2, 3])
              AND f.snapshot_date = (
                  SELECT MAX(snapshot_date) FROM domrf_kn_flats
              )
            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_sold_count_query(self, phantom_check_session: Session) -> None:
        """competitors._SOLD_COUNT_SQL (#1926, code-review #1927) — mapping → lots.

        flats_sold выровнен с MarketMetrics._STOCK_SQL: предикат
        is_sold/contract_date/status='продан' + premise_kind='квартира'.
        COUNT(DISTINCT objective_lot_id) — fan-out-safe (mapping НЕ unique по
        domrf_obj_id). Мост objective_complex_mapping.domrf_obj_id →
        objective_complex_name == objective_lots.project_name.
        """
        _explain(
            phantom_check_session,
            """
            SELECT
                cm.domrf_obj_id AS obj_id,
                COUNT(DISTINCT ol.objective_lot_id) FILTER (
                    WHERE ol.is_sold IS TRUE
                       OR ol.contract_date IS NOT NULL
                       OR LOWER(COALESCE(ol.status, '')) = 'продан'
                ) AS flats_sold
            FROM objective_complex_mapping cm
            JOIN objective_lots ol
                ON ol.project_name = cm.objective_complex_name
            WHERE cm.domrf_obj_id = ANY(ARRAY[1, 2, 3])
              AND ol.premise_kind = 'квартира'
            GROUP BY cm.domrf_obj_id
            """,
        )

    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
            """,
        )