gendesign/tradein-mvp/backend/tests/test_base_save_listings.py

"""Тесты для base.py save_listings — проверяем что Cian-специфичные колонки
передаются в INSERT и ON CONFLICT DO UPDATE SET.

Используем unittest.mock для замены db.execute — без реальной БД.
"""

import json
from contextlib import contextmanager
from unittest.mock import MagicMock, patch

import pytest

from app.services.scrapers.base import ScrapedLot, save_listings

# ── Autouse fixture: patch matching service so all save_listings tests are deterministic ──


@pytest.fixture(autouse=True)
def _patch_matching():
    """Stub matching service for all tests in this module.

    save_listings() now calls match_or_create_house + upsert_listing_source after
    each INSERT. Without patching, those would issue extra db.execute() calls with
    real SQL (and MagicMock returns truthy mappings() that would simulate Tier 0
    matches with fake house_ids). Tests assert via call_args_list[0] (the listings
    INSERT) and call_count on the patched stubs to verify the hook fires.
    """
    with (
        patch("app.services.scrapers.base.match_or_create_house") as m_house,
        patch("app.services.scrapers.base.upsert_listing_source") as m_link,
    ):
        m_house.return_value = (101, 1.0, "new")  # (house_id, conf, method)
        m_link.return_value = None
        yield {"house": m_house, "link": m_link}


# ── Helpers ──────────────────────────────────────────────────────────────────


def _mock_db(inserted: bool = True, listing_id: int = 42) -> MagicMock:
    """Mock Session: INSERT listings RETURNING id, inserted=<bool>; SAVEPOINTs no-op."""
    row = MagicMock()
    row.id = listing_id
    row.inserted = inserted

    result = MagicMock()
    result.fetchone.return_value = row

    db = MagicMock()
    db.execute.return_value = result

    # begin_nested() returns a context manager; SAVEPOINT semantics — no-op here
    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested
    return db


def _base_lot(**overrides) -> ScrapedLot:
    """Минимальный валидный ScrapedLot — только обязательные поля."""
    defaults = {
        "source": "avito",
        "source_url": "https://www.avito.ru/ekaterinburg/kvartiry/1-k_987654",
        "source_id": "987654",
        "price_rub": 4_500_000,
    }
    defaults.update(overrides)
    return ScrapedLot(**defaults)


def _cian_lot(**overrides) -> ScrapedLot:
    """ScrapedLot с полным набором Cian-специфичных полей."""
    base = {
        "source": "cian",
        "source_url": "https://ekb.cian.ru/sale/flat/123456/",
        "source_id": "123456",
        "price_rub": 6_200_000,
        "address": "Екатеринбург, улица Постовского, 17А",
        "lat": 56.8385,
        "lon": 60.5940,
        "rooms": 2,
        "area_m2": 52.0,
        "floor": 7,
        "total_floors": 16,
        # Cian-specific
        "living_area_m2": 35.0,
        "bedrooms_count": 2,
        "balconies_count": 1,
        "loggias_count": 0,
        "description_minhash": "abc123def456abc1",
        "cadastral_number": "66:41:0204016:1234",
        "building_cadastral_number": "66:41:0204016:100",
        "phones": [{"countryCode": "7", "number": "9012345678"}],
        "is_homeowner": True,
        "is_pro_seller": False,
        "bargain_allowed": True,
        "sale_type": "free",
        "metro_stations": [{"name": "Геологическая", "time": 10, "mode": "walk"}],
        "house_source": "cian",
        "house_ext_id": None,
        "listing_segment": "vtorichka",
    }
    base.update(overrides)
    return ScrapedLot(**base)


# ── Utility: extract params from mock db call ────────────────────────────────


def _find_listings_insert_call(db: MagicMock):
    """Найти (sql, params) вызова INSERT INTO listings.

    save_listings теперь делает лёгкий pre-read `SELECT card_hash FROM listings`
    ПЕРЕД upsert'ом, поэтому INSERT больше не обязательно call_args_list[0].
    Ищем по SQL-тексту, а не по фиксированному индексу.
    """
    assert db.execute.called, "db.execute was not called"
    for call in db.execute.call_args_list:
        sql = str(call.args[0])
        if "INSERT INTO listings (" in sql:
            return sql, call.args[1]
    raise AssertionError("INSERT INTO listings call not found")


def _get_execute_params(db: MagicMock) -> dict:
    """Получить params dict из вызова INSERT listings."""
    _sql, params = _find_listings_insert_call(db)
    return params


def _get_listings_insert_sql(db: MagicMock) -> str:
    """Получить SQL-текст INSERT INTO listings."""
    sql, _params = _find_listings_insert_call(db)
    return sql


# ── Test 1: Cian lot → все новые колонки попадают в params ──────────────────


def test_save_listings_persists_cian_specific_fields():
    """Cian lot с заполненными полями → все Cian колонки в params INSERT."""
    db = _mock_db(inserted=True)
    lot = _cian_lot()

    inserted, updated = save_listings(db, [lot])

    assert inserted == 1
    assert updated == 0

    params = _get_execute_params(db)

    assert params["living_area_m2"] == 35.0
    assert params["bedrooms_count"] == 2
    assert params["balconies_count"] == 1
    assert params["loggias_count"] == 0
    assert params["description_minhash"] == "abc123def456abc1"
    assert params["cadastral_number"] == "66:41:0204016:1234"
    assert params["building_cadastral_number"] == "66:41:0204016:100"
    assert params["is_homeowner"] is True
    assert params["is_pro_seller"] is False
    assert params["bargain_allowed"] is True
    assert params["sale_type"] == "free"

    # phones и metro_stations — JSON-строки, проверяем десериализацией
    phones = json.loads(params["phones"])
    assert phones == [{"countryCode": "7", "number": "9012345678"}]

    metro = json.loads(params["metro_stations"])
    assert metro[0]["name"] == "Геологическая"
    assert metro[0]["time"] == 10


# ── Test 2: Avito lot → Cian-специфичные params = None (NULL) ───────────────


def test_save_listings_avito_compat_cian_cols_are_null():
    """Avito lot не имеет Cian-полей → все Cian params = None (→ SQL NULL).

    Проверяет backward compat — Avito / DomRF scrapers не ломаются.
    """
    db = _mock_db(inserted=True)
    lot = _base_lot(source="avito", source_id="987654")

    inserted, _updated = save_listings(db, [lot])

    assert inserted == 1

    params = _get_execute_params(db)

    # ScrapedLot.living_area_m2 defaults to None
    assert params["living_area_m2"] is None
    assert params["bedrooms_count"] is None
    assert params["balconies_count"] is None
    assert params["loggias_count"] is None
    assert params["description_minhash"] is None
    assert params["cadastral_number"] is None
    assert params["building_cadastral_number"] is None
    assert params["is_homeowner"] is None
    assert params["is_pro_seller"] is None
    assert params["bargain_allowed"] is None
    assert params["sale_type"] is None
    # phones=[] (default_factory=list) → None (пустой список → None в helper)
    assert params["phones"] is None
    # metro_stations=[] → None
    assert params["metro_stations"] is None


# ── Test 3: ON CONFLICT — при updated=False все новые поля в UPDATE SET ──────


def test_save_listings_on_conflict_updates_cian_cols():
    """ON CONFLICT path: xmax != 0 → updated counter. Params верны для UPDATE.

    Проверяем что SQL-текст содержит все новые колонки в DO UPDATE SET.
    """
    db = _mock_db(inserted=False)  # xmax != 0 → updated
    lot = _cian_lot(
        source_id="111",
        cadastral_number="66:41:0000000:NEW",
        description_minhash="newhash",
    )

    inserted, updated = save_listings(db, [lot])

    assert inserted == 0
    assert updated == 1

    # Проверяем SQL-текст на наличие DO UPDATE SET с новыми колонками
    sql_text = _get_listings_insert_sql(db)

    assert "cadastral_number = EXCLUDED.cadastral_number" in sql_text
    assert "description_minhash = EXCLUDED.description_minhash" in sql_text
    assert "building_cadastral_number = EXCLUDED.building_cadastral_number" in sql_text
    assert "phones = EXCLUDED.phones" in sql_text
    assert "is_homeowner = EXCLUDED.is_homeowner" in sql_text
    assert "is_pro_seller = EXCLUDED.is_pro_seller" in sql_text
    assert "bargain_allowed = EXCLUDED.bargain_allowed" in sql_text
    assert "sale_type = EXCLUDED.sale_type" in sql_text
    assert "metro_stations = EXCLUDED.metro_stations" in sql_text
    assert "living_area_m2 = EXCLUDED.living_area_m2" in sql_text
    assert "bedrooms_count = EXCLUDED.bedrooms_count" in sql_text
    assert "balconies_count = EXCLUDED.balconies_count" in sql_text
    assert "loggias_count = EXCLUDED.loggias_count" in sql_text

    # Params с новым значением cadastral_number
    params = _get_execute_params(db)
    assert params["cadastral_number"] == "66:41:0000000:NEW"
    assert params["description_minhash"] == "newhash"


# ── Test 3b: kitchen/ceiling promote (#2007) ─────────────────────────────────


def test_save_listings_promotes_kitchen_ceiling_to_columns():
    """yandex lot с kitchen_area_m2/ceiling_height_m → params + обе ceiling-колонки.

    ceiling пишется ОДНИМ param :ceiling_height_m в ДВЕ колонки: ceiling_height
    (019, coverage-дашборд + yandex_detail) и ceiling_height_m (111, avito-колонка).
    """
    db = _mock_db(inserted=True)
    lot = _base_lot(
        source="yandex",
        source_id=None,
        source_url="https://realty.yandex.ru/offer/123/",
        kitchen_area_m2=12.7,
        ceiling_height_m=2.7,
    )

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["kitchen_area_m2"] == 12.7
    assert params["ceiling_height_m"] == 2.7

    sql_text = _get_listings_insert_sql(db)
    # обе колонки в списке INSERT
    assert "kitchen_area_m2" in sql_text
    assert "ceiling_height" in sql_text
    assert "ceiling_height_m" in sql_text
    # один param :ceiling_height_m питает обе ceiling-колонки (дважды в VALUES)
    assert sql_text.count(":ceiling_height_m") >= 2
    # ON CONFLICT — COALESCE, чтобы re-scrape без полей не затирал detail-enrichment
    assert "kitchen_area_m2 = COALESCE(" in sql_text
    assert "ceiling_height = COALESCE(" in sql_text
    assert "ceiling_height_m = COALESCE(" in sql_text


def test_save_listings_kitchen_ceiling_none_for_serp_without_them():
    """Avito SERP lot без kitchen/ceiling → params None (COALESCE сохранит detail-значение)."""
    db = _mock_db(inserted=True)
    lot = _base_lot(source="avito", source_id="987654")

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["kitchen_area_m2"] is None
    assert params["ceiling_height_m"] is None


# ── Test 4: empty list → early return, no db.execute call ───────────────────


def test_save_listings_empty_lots():
    """Пустой список → (0, 0), db.execute не вызывается."""
    db = _mock_db()

    inserted, updated = save_listings(db, [])

    assert inserted == 0
    assert updated == 0
    db.execute.assert_not_called()


# ── Test 5: phones=[] и metro_stations=[] → None в params ───────────────────


def test_save_listings_empty_phones_and_metro_become_null():
    """Если phones=[] / metro_stations=[] — передаём None (не '[]' строку)."""
    db = _mock_db(inserted=True)
    lot = _cian_lot(phones=[], metro_stations=[])

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["phones"] is None
    assert params["metro_stations"] is None


# ── Test 6: INSERT SQL содержит все новые колонки ────────────────────────────


def test_save_listings_sql_contains_all_new_columns():
    """SQL INSERT перечисляет все Cian-специфичные колонки."""
    db = _mock_db(inserted=True)
    lot = _base_lot()

    save_listings(db, [lot])

    sql_text = _get_listings_insert_sql(db)

    for col in (
        "living_area_m2",
        "bedrooms_count",
        "balconies_count",
        "loggias_count",
        "description_minhash",
        "cadastral_number",
        "building_cadastral_number",
        "phones",
        "is_homeowner",
        "is_pro_seller",
        "bargain_allowed",
        "sale_type",
        "metro_stations",
    ):
        assert col in sql_text, f"Column '{col}' missing from INSERT SQL"


# ── Test 7: INSERT RETURNING id — used downstream by matching hook ──────────


def test_save_listings_returning_id_for_matching_hook():
    """INSERT … RETURNING id, (xmax = 0) — needs both columns for the hook."""
    db = _mock_db(inserted=True, listing_id=12345)
    lot = _base_lot()

    save_listings(db, [lot])

    sql_text = _get_listings_insert_sql(db)
    assert "RETURNING id, (xmax = 0) AS inserted" in sql_text


# ── Test 8: matching hook fires exactly once per saved listing ──────────────


def test_save_listings_invokes_matching_hook_once_per_lot(_patch_matching):
    """Each saved lot triggers exactly one upsert_listing_source call.

    match_or_create_house may or may not be called (depends on address/lat/lon
    presence). upsert_listing_source MUST always be called.
    """
    db = _mock_db(inserted=True, listing_id=777)
    lots = [
        _base_lot(source_id="a1"),
        _base_lot(source_id="a2", address="Екатеринбург, Ленина 1", lat=56.84, lon=60.6),
        _cian_lot(source_id="c1"),
    ]

    save_listings(db, lots)

    # Hook fires once per lot
    assert _patch_matching["link"].call_count == 3

    # House resolution attempted only when address/coords present (2 of 3)
    assert _patch_matching["house"].call_count == 2

    # All upsert_listing_source calls use ext_source = listing.source
    for idx, call_args in enumerate(_patch_matching["link"].call_args_list):
        kwargs = call_args.kwargs
        assert kwargs["listing_id"] == 777
        assert kwargs["ext_source"] == lots[idx].source
        assert kwargs["ext_id"] == lots[idx].source_id
        assert kwargs["method"] == "source_link"
        assert kwargs["confidence"] == 1.0


# ── Test 9: matching hook receives full listing context ─────────────────────


def test_save_listings_matching_hook_passes_listing_context(_patch_matching):
    """upsert_listing_source receives price/area/floor/rooms/url from the lot."""
    db = _mock_db(inserted=True, listing_id=555)
    lot = _cian_lot(
        source_id="cian-999",
        price_rub=7_500_000,
        area_m2=54.5,
        floor=3,
        rooms=2,
    )

    save_listings(db, [lot])

    kwargs = _patch_matching["link"].call_args.kwargs
    assert kwargs["listing_id"] == 555
    assert kwargs["ext_source"] == "cian"
    assert kwargs["ext_id"] == "cian-999"
    assert kwargs["price_rub"] == 7_500_000
    assert kwargs["area_m2"] == 54.5
    assert kwargs["floor"] == 3
    assert kwargs["rooms_count"] == 2
    assert kwargs["source_url"] == lot.source_url


# ── Test 10: matching hook idempotent on re-scrape (ON CONFLICT updated path) ─


def test_save_listings_matching_hook_fires_on_update_path(_patch_matching):
    """ON CONFLICT DO UPDATE (re-scrape) — hook still fires so last_seen_at refreshes."""
    db = _mock_db(inserted=False, listing_id=12)  # updated, not new
    lot = _base_lot()

    save_listings(db, [lot])

    assert _patch_matching["link"].call_count == 1


# ── Test 11: matching failure logs warning but does not abort the batch ─────


def test_save_listings_matching_failure_does_not_abort_batch(_patch_matching, caplog):
    """When upsert_listing_source raises, the loop logs and continues."""
    import logging

    _patch_matching["link"].side_effect = RuntimeError("simulated DB blip")

    db = _mock_db(inserted=True, listing_id=99)
    lots = [_base_lot(source_id="a1"), _base_lot(source_id="a2")]

    with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"):
        inserted, updated = save_listings(db, lots)

    # Both listings still INSERTed; only the hook failed
    assert inserted == 2
    assert updated == 0

    # 2 warnings — one per lot
    warns = [r for r in caplog.records if "match_failed" in r.getMessage()]
    assert len(warns) == 2

    # Hook attempted on both
    assert _patch_matching["link"].call_count == 2


# ── Test 12: lot without source_id falls back to dedup_hash as ext_id ───────


def test_save_listings_matching_hook_dedup_hash_fallback(_patch_matching):
    """Yandex-style lot without stable source_id → ext_id := dedup_hash."""
    db = _mock_db(inserted=True, listing_id=44)
    lot = _base_lot(
        source="yandex",
        source_url="https://realty.yandex.ru/offer/some-offer/",
        source_id=None,
        address="Екатеринбург, Малышева 51",
        lat=56.84,
        lon=60.6,
    )

    save_listings(db, [lot])

    kwargs = _patch_matching["link"].call_args.kwargs
    assert kwargs["ext_source"] == "yandex"
    # ext_id should be the dedup_hash (sha256 hex), not None
    assert kwargs["ext_id"]
    assert len(kwargs["ext_id"]) == 64  # sha256 hex
    assert kwargs["ext_id"] == lot.compute_dedup_hash()


# ── Test 13: house_match failure aborts the hook (single SAVEPOINT level) ─────


def test_save_listings_house_match_failure_aborts_hook(_patch_matching, caplog):
    """If match_or_create_house raises, the outer per-row SAVEPOINT rolls back the
    entire hook call (_link_listing_to_house). upsert_listing_source is NOT called
    because there is no inner SAVEPOINT to shelter it — a single SAVEPOINT level
    (per backend.md) means one failure = one hook roll-back. The listings INSERT
    that happened before the SAVEPOINT is unaffected (inserted counter still 1).

    This is a behaviour change from the old double-SAVEPOINT design where house
    failure was silently swallowed and listing_sources was still upserted. With a
    single SAVEPOINT the outer except at save_listings:386 logs match_failed and
    continues to the next lot. (#849 Part 3)
    """
    import logging

    _patch_matching["house"].side_effect = RuntimeError("address parse boom")

    db = _mock_db(inserted=True, listing_id=88)
    lot = _base_lot(
        source_id="x1",
        address="Bad addr",
        lat=56.84,
        lon=60.6,
    )

    with caplog.at_level(logging.WARNING, logger="app.services.scrapers.base"):
        inserted, _ = save_listings(db, [lot])

    assert inserted == 1
    # House match attempted — hook aborted, listing_source NOT recorded
    assert _patch_matching["house"].call_count == 1
    assert _patch_matching["link"].call_count == 0
    # Outer except logs the failure
    warns = [r for r in caplog.records if "match_failed" in r.getMessage()]
    assert len(warns) == 1


# ── Test 14: empty list → matching service not called ───────────────────────


def test_save_listings_empty_lots_skips_matching(_patch_matching):
    """Empty input → no hook calls (verifies early return preserves behaviour)."""
    db = _mock_db()
    save_listings(db, [])

    assert _patch_matching["house"].call_count == 0
    assert _patch_matching["link"].call_count == 0


# ── #753: dedup_hash стабильность (Avito ?context= + смена цены) ─────────────


def test_dedup_hash_stable_across_price_and_context_query():
    """#753: тот же Avito item (один source_id) с РАЗНОЙ ценой и РАЗНЫМ ?context=
    в URL → ОДИНАКОВЫЙ dedup_hash. Иначе смена цены / ре-скрейп = дубль."""
    lot_a = _base_lot(
        source_url="https://www.avito.ru/ekaterinburg/kvartiry/1-k_987654?context=AAAA",
        source_id="987654",
        price_rub=4_500_000,
    )
    lot_b = _base_lot(
        source_url="https://www.avito.ru/ekaterinburg/kvartiry/1-k_987654?context=ZZZZ",
        source_id="987654",
        price_rub=4_300_000,  # цена снижена
    )
    assert lot_a.compute_dedup_hash() == lot_b.compute_dedup_hash()


def test_dedup_hash_strips_query_when_no_source_id():
    """#753: при отсутствии source_id ключ = source_url без query-string —
    волатильный ?context= не должен влиять на hash."""
    lot_a = _base_lot(
        source_url="https://www.avito.ru/ekaterinburg/kvartiry/1-k_55?context=AAAA",
        source_id=None,
        price_rub=5_000_000,
    )
    lot_b = _base_lot(
        source_url="https://www.avito.ru/ekaterinburg/kvartiry/1-k_55?context=BBBB",
        source_id=None,
        price_rub=5_900_000,
    )
    assert lot_a.compute_dedup_hash() == lot_b.compute_dedup_hash()


def test_dedup_hash_distinguishes_different_listings():
    """#753 guard: разные объявления (разные source_id) → РАЗНЫЙ hash
    (фикс не должен схлопывать несвязанные listings)."""
    lot_a = _base_lot(source_id="111", source_url="https://www.avito.ru/a_111")
    lot_b = _base_lot(source_id="222", source_url="https://www.avito.ru/a_222")
    assert lot_a.compute_dedup_hash() != lot_b.compute_dedup_hash()


def test_dedup_hash_distinguishes_source_even_same_id():
    """#753 guard: одинаковый source_id у разных источников → РАЗНЫЙ hash
    (source — часть ключа)."""
    lot_avito = _base_lot(source="avito", source_id="500")
    lot_cian = _base_lot(source="cian", source_id="500")
    assert lot_avito.compute_dedup_hash() != lot_cian.compute_dedup_hash()


# ── Tests 15-17: #821 COALESCE backfill listing_date / area_m2 ──────────────


def test_save_listings_sql_contains_coalesce_listing_date_area_m2():
    """#821: ON CONFLICT SET block содержит COALESCE для listing_date и area_m2."""
    db = _mock_db(inserted=False)  # conflict path
    lot = _cian_lot()

    save_listings(db, [lot])

    sql_text = _get_listings_insert_sql(db)
    assert "COALESCE(EXCLUDED.listing_date, listings.listing_date)" in sql_text
    assert "COALESCE(EXCLUDED.area_m2, listings.area_m2)" in sql_text


def test_save_listings_params_listing_date_area_m2_passed_through():
    """#821: listing_date и area_m2 попадают в params INSERT (VALUES side)."""
    import datetime

    db = _mock_db(inserted=True)
    lot = _base_lot(
        source_id="ld_test",
        area_m2=47.5,
        listing_date=datetime.date(2024, 3, 15),
    )

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["area_m2"] == 47.5
    assert params["listing_date"] == datetime.date(2024, 3, 15)


def test_save_listings_null_area_and_date_pass_none_in_params():
    """#821: re-scrape без area_m2/listing_date → params содержат None.

    COALESCE в SQL сохранит существующее значение в БД — здесь проверяем
    только что params не ломают вызов (None передаётся без ошибки).
    """
    db = _mock_db(inserted=False)  # conflict/update path
    lot = _base_lot(
        source_id="null_test",
        area_m2=None,
        listing_date=None,
    )

    inserted, updated = save_listings(db, [lot])

    # Ожидаем updated (xmax != 0)
    assert inserted == 0
    assert updated == 1

    params = _get_execute_params(db)
    assert params["area_m2"] is None
    assert params["listing_date"] is None


# ── Tests: house_id_fk realtime link write (avito/cian unlinked fix) ─────────


def _find_house_fk_update(db: MagicMock):
    """Return (sql, params) of the UPDATE listings SET house_id_fk call, or None.

    match_or_create_house / upsert_listing_source are patched (no db.execute), so
    the only db.execute calls are the listings INSERT (call 0) and — when a house
    is resolved — the house_id_fk UPDATE added by the realtime link hook.
    """
    for call in db.execute.call_args_list:
        sql = str(call.args[0])
        if "house_id_fk" in sql and "UPDATE listings" in sql:
            return sql, call.args[1]
    return None


def test_save_listings_writes_house_id_fk_when_house_resolved(_patch_matching):
    """House resolved (address/coords present) → UPDATE listings.house_id_fk runs
    with the resolved house_id and the saved listing id."""
    _patch_matching["house"].return_value = (909, 1.0, "geo")

    db = _mock_db(inserted=True, listing_id=4242)
    lot = _cian_lot(source_id="c-fk-1")  # has address + lat/lon → house resolved

    save_listings(db, [lot])

    found = _find_house_fk_update(db)
    assert found is not None, "UPDATE listings SET house_id_fk was not issued"
    _sql, params = found
    assert params["hid"] == 909
    assert params["lid"] == 4242


def test_save_listings_no_house_fk_update_when_house_not_resolved(_patch_matching):
    """No address and no coords → match_or_create_house skipped → no FK UPDATE."""
    db = _mock_db(inserted=True, listing_id=4243)
    # base lot has no address / lat / lon → house resolution skipped
    lot = _base_lot(source="avito", source_id="a-fk-no", address=None, lat=None, lon=None)

    save_listings(db, [lot])

    assert _patch_matching["house"].call_count == 0
    assert _find_house_fk_update(db) is None


def test_save_listings_house_fk_update_guards_no_op_write(_patch_matching):
    """The UPDATE is idempotency-guarded (IS DISTINCT FROM) and casts to bigint
    (psycopg v3 CAST, never ::bigint)."""
    _patch_matching["house"].return_value = (55, 1.0, "cadastr")

    db = _mock_db(inserted=True, listing_id=1)
    lot = _cian_lot(source_id="c-fk-guard")

    save_listings(db, [lot])

    found = _find_house_fk_update(db)
    assert found is not None
    sql, _params = found
    assert "IS DISTINCT FROM" in sql
    assert "CAST(:hid AS bigint)" in sql
    assert "::bigint" not in sql


# ── Tests 18-20: Yandex rich fields (predictedPrice, trend, description, agency) ──


def _yandex_rich_lot(**overrides) -> ScrapedLot:
    """ScrapedLot с заполненными Yandex rich-полями."""
    import datetime

    base = {
        "source": "yandex",
        "source_url": "https://realty.yandex.ru/offer/5500000000000000001/",
        "source_id": "5500000000000000001",
        "price_rub": 7_200_000,
        "address": "Екатеринбург, улица Мира, 10",
        "rooms": 2,
        "area_m2": 48.0,
        "listing_date": datetime.date(2026, 5, 24),
        "publish_date": datetime.date(2026, 5, 24),
        "days_on_market": 24,
        "description": "Светлая квартира.",
        "is_homeowner": True,
        "agency_name": None,
        "metro_stations": [{"name": "Динамо", "min": 7}],
        "yandex_offer_id": "5500000000000000001",
        "predicted_price_rub": 7_554_000,
        "predicted_price_min": 7_100_000,
        "predicted_price_max": 8_000_000,
        "price_trend": "DECREASED",
        "price_previous_rub": 7_500_000,
    }
    base.update(overrides)
    return ScrapedLot(**base)


def test_save_listings_persists_yandex_rich_fields():
    """Yandex rich-поля попадают в params INSERT."""
    db = _mock_db(inserted=True)
    lot = _yandex_rich_lot()

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["description"] == "Светлая квартира."
    assert params["days_on_market"] == 24
    assert params["yandex_offer_id"] == "5500000000000000001"
    assert params["predicted_price_rub"] == 7_554_000
    assert params["predicted_price_min"] == 7_100_000
    assert params["predicted_price_max"] == 8_000_000
    assert params["price_trend"] == "DECREASED"
    assert params["price_previous_rub"] == 7_500_000
    assert params["agency_name"] is None
    metro = json.loads(params["metro_stations"])
    assert metro == [{"name": "Динамо", "min": 7}]


def test_save_listings_yandex_rich_cols_in_sql():
    """INSERT + ON CONFLICT SET содержат новые Yandex rich колонки."""
    db = _mock_db(inserted=False)
    lot = _yandex_rich_lot(agency_name="Агентство «Диал»", is_pro_seller=True, is_homeowner=None)

    save_listings(db, [lot])

    sql_text = _get_listings_insert_sql(db)
    for col in (
        "description",
        "publish_date",
        "agency_name",
        "yandex_offer_id",
        "predicted_price_rub",
        "predicted_price_min",
        "predicted_price_max",
        "price_trend",
        "price_previous_rub",
    ):
        assert col in sql_text, f"Column '{col}' missing from INSERT SQL"

    # COALESCE backfill in DO UPDATE SET (never wipe a prior value)
    assert "description = COALESCE(EXCLUDED.description, listings.description)" in sql_text
    assert "agency_name = COALESCE(EXCLUDED.agency_name, listings.agency_name)" in sql_text
    assert "predicted_price_rub = COALESCE(" in sql_text
    assert "price_trend = COALESCE(EXCLUDED.price_trend, listings.price_trend)" in sql_text


def test_save_listings_non_yandex_rich_cols_are_null():
    """Avito lot не имеет Yandex rich-полей → params = None (backward compat)."""
    db = _mock_db(inserted=True)
    lot = _base_lot(source="avito", source_id="987654")

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["description"] is None
    assert params["publish_date"] is None
    assert params["agency_name"] is None
    assert params["yandex_offer_id"] is None
    assert params["predicted_price_rub"] is None
    assert params["predicted_price_min"] is None
    assert params["predicted_price_max"] is None
    assert params["price_trend"] is None
    assert params["price_previous_rub"] is None


# ── newbuilding (ЖК) id/url persistence ──────────────────────────────────────


def test_save_listings_persists_newbuilding_fields():
    """Lot с newbuilding_id/url → оба попадают в params INSERT + есть в SQL."""
    db = _mock_db(inserted=True)
    lot = _base_lot(
        source="avito",
        source_id="nb_test",
        newbuilding_id="federatsiya-ekaterinburg",
        newbuilding_url="https://zhk-federatsiya-ekaterinburg.avito.ru",
    )

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["newbuilding_id"] == "federatsiya-ekaterinburg"
    assert params["newbuilding_url"] == "https://zhk-federatsiya-ekaterinburg.avito.ru"

    sql_text = _get_listings_insert_sql(db)
    assert "newbuilding_id" in sql_text
    assert "newbuilding_url" in sql_text
    # COALESCE backfill in DO UPDATE SET (never wipe a prior value).
    # SQL wraps long COALESCE calls; assert on a whitespace-normalised copy.
    sql_norm = " ".join(sql_text.split())
    assert (
        "newbuilding_id = COALESCE( EXCLUDED.newbuilding_id, listings.newbuilding_id )" in sql_norm
    )
    assert (
        "newbuilding_url = COALESCE( EXCLUDED.newbuilding_url, listings.newbuilding_url )"
        in sql_norm
    )


def test_save_listings_newbuilding_fields_default_none():
    """Lot без ЖК-полей → params newbuilding_id/url = None (backward compat)."""
    db = _mock_db(inserted=True)
    lot = _base_lot(source="avito", source_id="987654")

    save_listings(db, [lot])

    params = _get_execute_params(db)
    assert params["newbuilding_id"] is None
    assert params["newbuilding_url"] is None


# ── card_hash: ScrapedLot.compute_card_hash determinism + sensitivity ────────


def test_compute_card_hash_deterministic():
    """Тот же lot → тот же hash (вызов дважды и два эквивалентных lot)."""
    lot = _cian_lot()
    assert lot.compute_card_hash() == lot.compute_card_hash()

    lot_b = _cian_lot()
    assert lot.compute_card_hash() == lot_b.compute_card_hash()
    # sha256 hex
    assert len(lot.compute_card_hash()) == 64


def test_compute_card_hash_changes_on_price():
    """Смена price_rub → другой hash."""
    a = _base_lot(price_rub=4_500_000)
    b = _base_lot(price_rub=4_300_000)
    assert a.compute_card_hash() != b.compute_card_hash()


def test_compute_card_hash_changes_on_area_floor():
    """Смена area_m2, floor, total_floors, rooms → другой hash."""
    base = _cian_lot()
    assert base.compute_card_hash() != _cian_lot(area_m2=53.0).compute_card_hash()
    assert base.compute_card_hash() != _cian_lot(floor=8).compute_card_hash()
    assert base.compute_card_hash() != _cian_lot(total_floors=17).compute_card_hash()
    assert base.compute_card_hash() != _cian_lot(rooms=3).compute_card_hash()


def test_compute_card_hash_changes_on_address_and_photos():
    """Смена address (после нормализации) и числа фото → другой hash."""
    base = _base_lot(address="Екатеринбург, Ленина 1", photo_urls=["a", "b"])
    assert (
        base.compute_card_hash()
        != _base_lot(address="Екатеринбург, Ленина 2", photo_urls=["a", "b"]).compute_card_hash()
    )
    assert (
        base.compute_card_hash()
        != _base_lot(
            address="Екатеринбург, Ленина 1", photo_urls=["a", "b", "c"]
        ).compute_card_hash()
    )


def test_compute_card_hash_address_normalized():
    """address нормализуется (strip + lower) — регистр/пробелы не меняют hash."""
    a = _base_lot(address="  Екатеринбург, Ленина 1  ")
    b = _base_lot(address="екатеринбург, ленина 1")
    assert a.compute_card_hash() == b.compute_card_hash()


def test_compute_card_hash_none_address_stable():
    """address=None → "" — два lot без адреса дают одинаковый hash."""
    a = _base_lot(address=None)
    b = _base_lot(address=None)
    assert a.compute_card_hash() == b.compute_card_hash()


def test_compute_card_hash_stable_when_only_timestamps_or_detail_change():
    """Изменение только timestamp/detail-only полей (не входящих в карточку) →
    hash НЕ меняется."""
    import datetime

    base = _cian_lot()
    # detail-only / metadata поля — НЕ часть card_hash
    variant = _cian_lot(
        listing_date=datetime.date(2026, 1, 1),
        publish_date=datetime.date(2026, 1, 2),
        days_on_market=42,
        description="совсем другое описание",
        cadastral_number="66:41:9999999:9999",
        is_homeowner=False,
        agency_name="Другое агентство",
        raw_payload={"foo": "bar"},
    )
    assert base.compute_card_hash() == variant.compute_card_hash()


# ── save_listings: card_hash persisted on insert + update ────────────────────


def test_save_listings_card_hash_in_sql_and_params():
    """INSERT SQL содержит колонку card_hash + EXCLUDED в DO UPDATE SET;
    params несут вычисленный hash."""
    db = _mock_db(inserted=True)
    lot = _base_lot()

    save_listings(db, [lot])

    sql_text = _get_listings_insert_sql(db)
    assert "card_hash" in sql_text
    assert "card_hash = EXCLUDED.card_hash" in sql_text

    params = _get_execute_params(db)
    assert params["card_hash"] == lot.compute_card_hash()
    assert len(params["card_hash"]) == 64


def test_save_listings_card_hash_on_update_path():
    """ON CONFLICT (update) — card_hash тоже передаётся в params."""
    db = _mock_db(inserted=False)
    lot = _cian_lot(source_id="ch-upd")

    inserted, updated = save_listings(db, [lot])
    assert inserted == 0
    assert updated == 1

    params = _get_execute_params(db)
    assert params["card_hash"] == lot.compute_card_hash()


def test_save_listings_pre_reads_existing_card_hash():
    """Перед upsert'ом делается лёгкий SELECT card_hash, last_seen_at FROM listings."""
    db = _mock_db(inserted=True)
    lot = _base_lot()

    save_listings(db, [lot])

    pre_reads = [
        c
        for c in db.execute.call_args_list
        if "SELECT card_hash" in str(c.args[0]) and "WHERE dedup_hash" in str(c.args[0])
    ]
    assert len(pre_reads) == 1
    assert pre_reads[0].args[1]["dedup"] == lot.compute_dedup_hash()


def test_save_listings_skips_snapshot_when_card_unchanged():
    """Если prior card_hash == new card_hash И snapshot за сегодня уже есть —
    upsert_listing_snapshot НЕ вызывается (snapshot-dedup)."""
    lot = _base_lot()
    new_hash = lot.compute_card_hash()

    # row для INSERT RETURNING (id, inserted)
    insert_row = MagicMock()
    insert_row.id = 42
    insert_row.inserted = False  # update path

    # pre-read row с тем же card_hash → карточка не изменилась
    prior_row = MagicMock()
    prior_row.card_hash = new_hash
    prior_row.last_seen_at = None  # не сегодня → skip_seen_today не срабатывает

    # today-snapshot EXISTS row (любой truthy)
    today_row = MagicMock()

    db = MagicMock()

    def _execute(sql, params=None):
        s = str(sql)
        res = MagicMock()
        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
            res.fetchone.return_value = prior_row
        elif "FROM listings_snapshots" in s:
            res.fetchone.return_value = today_row
        elif "INSERT INTO listings (" in s:
            res.fetchone.return_value = insert_row
        else:
            res.fetchone.return_value = None
        return res

    db.execute.side_effect = _execute

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
        save_listings(db, [lot])
        m_snap.assert_not_called()


def test_save_listings_writes_snapshot_when_card_changed():
    """Если card_hash изменился — snapshot пишется даже если сегодня уже был
    (когда в сомнениях — пишем)."""
    lot = _base_lot(price_rub=4_300_000)  # новая цена → новый hash

    insert_row = MagicMock()
    insert_row.id = 42
    insert_row.inserted = False

    prior_row = MagicMock()
    prior_row.card_hash = "оld-different-hash"  # отличается

    db = MagicMock()

    def _execute(sql, params=None):
        s = str(sql)
        res = MagicMock()
        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
            res.fetchone.return_value = prior_row
        elif "INSERT INTO listings (" in s:
            res.fetchone.return_value = insert_row
        else:
            res.fetchone.return_value = None
        return res

    db.execute.side_effect = _execute

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
        save_listings(db, [lot])
        assert m_snap.call_count == 1


def test_save_listings_writes_snapshot_on_fresh_insert():
    """Свежий insert (prior_card_hash is None) → snapshot всегда пишется."""
    lot = _base_lot()

    insert_row = MagicMock()
    insert_row.id = 42
    insert_row.inserted = True

    db = MagicMock()

    def _execute(sql, params=None):
        s = str(sql)
        res = MagicMock()
        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
            res.fetchone.return_value = None  # row не существует
        elif "INSERT INTO listings (" in s:
            res.fetchone.return_value = insert_row
        else:
            res.fetchone.return_value = None
        return res

    db.execute.side_effect = _execute

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
        save_listings(db, [lot])
        assert m_snap.call_count == 1


def test_save_listings_writes_snapshot_when_unchanged_but_no_today_snapshot():
    """card_hash не изменился, НО snapshot за сегодня ещё нет (первый скрейп дня) →
    snapshot всё равно пишется (не теряем сегодняшнее наблюдение)."""
    lot = _base_lot()
    new_hash = lot.compute_card_hash()

    insert_row = MagicMock()
    insert_row.id = 42
    insert_row.inserted = False

    prior_row = MagicMock()
    prior_row.card_hash = new_hash  # тот же

    db = MagicMock()

    def _execute(sql, params=None):
        s = str(sql)
        res = MagicMock()
        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
            res.fetchone.return_value = prior_row
        elif "FROM listings_snapshots" in s:
            res.fetchone.return_value = None  # сегодня snapshot'а ещё нет
        elif "INSERT INTO listings (" in s:
            res.fetchone.return_value = insert_row
        else:
            res.fetchone.return_value = None
        return res

    db.execute.side_effect = _execute

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
        save_listings(db, [lot])
        assert m_snap.call_count == 1


# ── Tests: skip_seen_today ────────────────────────────────────────────────────


def _mock_db_with_last_seen(
    last_seen_at,
    inserted: bool = False,
    listing_id: int = 42,
) -> MagicMock:
    """Mock Session где prior SELECT возвращает строку с card_hash и last_seen_at.

    last_seen_at: datetime с tzinfo (timezone-aware) или None.
    """
    insert_row = MagicMock()
    insert_row.id = listing_id
    insert_row.inserted = inserted

    prior_row = MagicMock()
    prior_row.card_hash = "somehash"
    prior_row.last_seen_at = last_seen_at

    db = MagicMock()

    def _execute(sql, params=None):
        s = str(sql)
        res = MagicMock()
        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
            res.fetchone.return_value = prior_row
        elif "FROM listings_snapshots" in s:
            res.fetchone.return_value = None
        elif "INSERT INTO listings (" in s:
            res.fetchone.return_value = insert_row
        else:
            res.fetchone.return_value = None
        return res

    db.execute.side_effect = _execute

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested
    return db


def test_skip_seen_today_skips_when_last_seen_is_today(_patch_matching):
    """skip_seen_today=True + last_seen_at=сегодня по МСК → лот пропускается (ins=upd=0)."""
    from datetime import datetime
    from zoneinfo import ZoneInfo

    msk_tz = ZoneInfo("Europe/Moscow")
    now_msk = datetime.now(msk_tz)

    db = _mock_db_with_last_seen(last_seen_at=now_msk, inserted=False)
    lot = _base_lot(source_id="skip-today")

    inserted, updated = save_listings(db, [lot], skip_seen_today=True)

    assert inserted == 0
    assert updated == 0
    # INSERT INTO listings не должен вызываться
    insert_calls = [
        c for c in db.execute.call_args_list if "INSERT INTO listings (" in str(c.args[0])
    ]
    assert len(insert_calls) == 0
    # matching hook не должен срабатывать
    assert _patch_matching["link"].call_count == 0


def test_skip_seen_today_processes_when_last_seen_is_yesterday(_patch_matching):
    """skip_seen_today=True + last_seen_at=вчера по МСК → лот обрабатывается."""
    from datetime import datetime, timedelta
    from zoneinfo import ZoneInfo

    msk_tz = ZoneInfo("Europe/Moscow")
    yesterday_msk = datetime.now(msk_tz) - timedelta(days=1)

    db = _mock_db_with_last_seen(last_seen_at=yesterday_msk, inserted=False)
    lot = _base_lot(source_id="process-yesterday")

    inserted, updated = save_listings(db, [lot], skip_seen_today=True)

    # updated=1 (upsert path)
    assert inserted == 0
    assert updated == 1


def test_skip_seen_today_new_lot_always_inserted(_patch_matching):
    """skip_seen_today=True + prior_row=None (новый листинг) → вставляется."""
    insert_row = MagicMock()
    insert_row.id = 10
    insert_row.inserted = True

    db = MagicMock()

    def _execute(sql, params=None):
        s = str(sql)
        res = MagicMock()
        if "SELECT card_hash" in s and "WHERE dedup_hash" in s:
            res.fetchone.return_value = None  # строки нет — новый листинг
        elif "INSERT INTO listings (" in s:
            res.fetchone.return_value = insert_row
        else:
            res.fetchone.return_value = None
        return res

    db.execute.side_effect = _execute

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    lot = _base_lot(source_id="brand-new")

    inserted, updated = save_listings(db, [lot], skip_seen_today=True)

    assert inserted == 1
    assert updated == 0


def test_skip_seen_today_false_always_updates(_patch_matching):
    """skip_seen_today=False (дефолт) + last_seen_at=сегодня → upsert как обычно."""
    from datetime import datetime
    from zoneinfo import ZoneInfo

    msk_tz = ZoneInfo("Europe/Moscow")
    now_msk = datetime.now(msk_tz)

    db = _mock_db_with_last_seen(last_seen_at=now_msk, inserted=False)
    lot = _base_lot(source_id="no-skip-flag")

    # skip_seen_today не передаём (дефолт False)
    inserted, updated = save_listings(db, [lot])

    assert inserted == 0
    assert updated == 1


def test_skip_seen_today_skips_snapshot_for_skipped_lot(_patch_matching):
    """Пропущенный лот (seen_today) не вызывает upsert_listing_snapshot."""
    from datetime import datetime
    from zoneinfo import ZoneInfo

    msk_tz = ZoneInfo("Europe/Moscow")
    now_msk = datetime.now(msk_tz)

    db = _mock_db_with_last_seen(last_seen_at=now_msk, inserted=False)
    lot = _base_lot(source_id="no-snap-skip")

    with patch("app.services.scrapers.base.upsert_listing_snapshot") as m_snap:
        save_listings(db, [lot], skip_seen_today=True)
        m_snap.assert_not_called()