gendesign/backend/tests/services/test_newbuilding_crossload.py

"""Тесты cross-load ETL tradein→gendesign newbuilding_listings (#976).

Тесты без реальной БД:
  (a) disabled-режим: пустой tradein_database_url → {"disabled": True}, psycopg не вызывается.
  (b) маппинг строки source → insert-параметры (_row_to_params).
  (c) UPSERT SQL содержит ON CONFLICT (source, ext_house_id).
"""

from __future__ import annotations

import json
from unittest.mock import patch

from app.services.etl.newbuilding_crossload import (
    _UPSERT_SQL,
    _row_to_params,
    run_crossload,
)

# ─── (a) Disabled-режим ──────────────────────────────────────────────────────


def test_run_crossload_disabled_when_url_empty():
    """Если tradein_database_url пусто — возвращает {"disabled": True}, коннекта нет."""
    with patch("app.services.etl.newbuilding_crossload.settings") as mock_settings:
        mock_settings.tradein_database_url = ""
        with patch("psycopg.connect") as mock_connect:
            result = run_crossload()

    assert result == {"disabled": True}
    mock_connect.assert_not_called()


# ─── (b) Маппинг _row_to_params ──────────────────────────────────────────────


_SAMPLE_ROW: dict = {
    "source": "cian",
    "ext_house_id": "123456",
    "url": "https://cian.ru/novostrojki/123456/",
    "slug": "zhk-test",
    "address": "Екатеринбург, ул. Тестовая, 1",
    "full_address": "Свердловская область, Екатеринбург, ул. Тестовая, 1",
    "lat": 56.838_926,
    "lon": 60.605_702,
    "house_class": "comfort",
    "developer_name": "ТестСтрой",
    "developer_key": "teststr_001",
    "year_built": 2025,
    "total_floors": 18,
    "total_units": 360,
    "rating": 4.7,
    "reviews_count": 12,
    "houses_by_turn": [{"turn": 1, "corps": ["1А"]}],
    "cian_internal_house_id": 9876543,
    "raw_payload": {"foo": "bar"},
    "last_scraped_at": None,
    "yandex_jk_id": "yandex-42",
}


def test_row_to_params_basic_fields():
    """Прямые поля копируются без изменений."""
    p = _row_to_params(_SAMPLE_ROW)

    assert p["source"] == "cian"
    assert p["ext_house_id"] == "123456"
    assert p["url"] == "https://cian.ru/novostrojki/123456/"
    assert p["house_class"] == "comfort"
    assert p["developer_name"] == "ТестСтрой"
    assert p["developer_key"] == "teststr_001"
    assert p["year_built"] == 2025
    assert p["total_floors"] == 18
    assert p["total_units"] == 360
    assert p["reviews_count"] == 12
    assert p["yandex_jk_id"] == "yandex-42"
    assert p["cian_internal_house_id"] == 9876543


def test_row_to_params_geo_duplicated():
    """lat/lon продублированы в lat2/lon2 для CASE-ветки geom."""
    p = _row_to_params(_SAMPLE_ROW)
    assert p["lat"] == _SAMPLE_ROW["lat"]
    assert p["lon"] == _SAMPLE_ROW["lon"]
    assert p["lat2"] == _SAMPLE_ROW["lat"]
    assert p["lon2"] == _SAMPLE_ROW["lon"]


def test_row_to_params_jsonb_serialized():
    """houses_by_turn и raw_payload сериализованы в JSON-строки."""
    p = _row_to_params(_SAMPLE_ROW)
    # houses_by_turn
    hbt = json.loads(p["houses_by_turn"])
    assert hbt == [{"turn": 1, "corps": ["1А"]}]
    # raw_payload
    rp = json.loads(p["raw_payload"])
    assert rp == {"foo": "bar"}


def test_row_to_params_none_jsonb():
    """Если houses_by_turn/raw_payload равны None — параметр None, не строка."""
    row = dict(_SAMPLE_ROW)
    row["houses_by_turn"] = None
    row["raw_payload"] = None
    p = _row_to_params(row)
    assert p["houses_by_turn"] is None
    assert p["raw_payload"] is None


def test_row_to_params_rating_as_string():
    """rating приводится к строке для CAST(:rating AS numeric)."""
    p = _row_to_params(_SAMPLE_ROW)
    assert p["rating"] == "4.7"


def test_row_to_params_rating_none():
    """Если rating = None — параметр тоже None."""
    row = dict(_SAMPLE_ROW)
    row["rating"] = None
    p = _row_to_params(row)
    assert p["rating"] is None


def test_row_to_params_source_last_scraped_at():
    """last_scraped_at из источника → source_last_scraped_at в параметрах."""
    p = _row_to_params(_SAMPLE_ROW)
    assert "source_last_scraped_at" in p
    assert p["source_last_scraped_at"] is None  # SAMPLE_ROW имеет None


# ─── (c) UPSERT SQL содержит ON CONFLICT ──────────────────────────────────────


def test_upsert_sql_contains_on_conflict():
    """UPSERT-запрос содержит ON CONFLICT (source, ext_house_id)."""
    sql_text = str(_UPSERT_SQL)
    assert "ON CONFLICT" in sql_text.upper()
    assert "source" in sql_text
    assert "ext_house_id" in sql_text


def test_upsert_sql_contains_do_update():
    """UPSERT-запрос обновляет строку при конфликте (DO UPDATE SET)."""
    sql_text = str(_UPSERT_SQL)
    assert "DO UPDATE SET" in sql_text.upper()


def test_upsert_sql_coalesce_external_ids():
    """yandex_jk_id и cian_internal_house_id используют COALESCE в DO UPDATE.

    Источники cian/yandex пишут попеременно — важно не затирать чужой id NULL'ом.
    Проверяем: в секции DO UPDATE каждая из двух колонок встречается после COALESCE,
    а не как голое EXCLUDED.<col> (что затёрло бы чужой id NULL'ом).
    """
    sql_text = str(_UPSERT_SQL)
    sql_upper = sql_text.upper()

    do_update_idx = sql_upper.index("DO UPDATE SET")
    do_update_section = sql_upper[do_update_idx:]

    # COALESCE должен присутствовать в секции DO UPDATE
    assert "COALESCE" in do_update_section, "DO UPDATE должен содержать COALESCE"

    # yandex_jk_id — строка вида "YANDEX_JK_ID = COALESCE("
    assert (
        "YANDEX_JK_ID           = COALESCE(" in do_update_section
    ), "yandex_jk_id в DO UPDATE должен использовать COALESCE чтобы не затирать NULL'ом"
    # cian_internal_house_id — строка вида "CIAN_INTERNAL_HOUSE_ID = COALESCE("
    assert (
        "CIAN_INTERNAL_HOUSE_ID = COALESCE(" in do_update_section
    ), "cian_internal_house_id в DO UPDATE должен использовать COALESCE чтобы не затирать NULL'ом"