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

"""Тесты для snapshot_writer.py и интеграции с save_detail_enrichment / save_listings.

Цель: доказать что:
1. upsert_listing_snapshot строит корректный INSERT INTO listings_snapshots.
2. Повторный вызов с теми же listing_id + snapshot_date НЕ дублирует строку
   (ON CONFLICT DO UPDATE обновляет observed_at).
3. save_detail_enrichment вызывает snapshot-write после SELECT price из listings.
4. save_listings вызывает snapshot-write для каждого сохранённого listing.
5. Ошибка snapshot не прерывает основной INSERT в save_listings.

Используем unittest.mock — без реальной БД.

`app.services.scrapers.cian_detail` удалён (#2397 Part E2, 0 runtime-импортёров) —
`DetailEnrichment`/`save_detail_enrichment` ретаргетированы на
`scraper_kit.providers.cian.detail` (byte-identical strangler-копия).

`app.services.scrapers.base`/`snapshot_writer` удалены (#2397 финальный шаг E, 0
runtime-импортёров) — `upsert_listing_snapshot`/`ScrapedLot`/`save_listings`
переведены на `scraper_kit.snapshot_writer`/`scraper_kit.base`. Kit `save_listings`
принимает инжектируемый `matcher: HouseMatcher` (mandatory kwarg, strangler DI
#2133) вместо прямого импорта `match_or_create_house`/`upsert_listing_source` —
тесты используют `MagicMock()` matcher (тот же паттерн, что
`test_scraped_at_bump_2206.py::_kit_matcher`), плюс mandatory `region_code=66`.
"""

from __future__ import annotations

from contextlib import contextmanager
from datetime import date
from unittest.mock import MagicMock, patch

from scraper_kit.base import ScrapedLot, save_listings
from scraper_kit.providers.cian.detail import DetailEnrichment, save_detail_enrichment
from scraper_kit.snapshot_writer import upsert_listing_snapshot


def _mock_matcher() -> MagicMock:
    """MagicMock HouseMatcher — mirror test_scraped_at_bump_2206.py::_kit_matcher."""
    matcher = MagicMock()
    matcher.match_or_create_house.return_value = (1, 1.0, "new")
    matcher.upsert_listing_source.return_value = None
    return matcher


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


def _mock_db_simple() -> MagicMock:
    """Минимальный mock db.execute — scalar None для всех вызовов."""
    db = MagicMock()
    db.execute.return_value = MagicMock()
    db.execute.return_value.fetchone.return_value = None
    db.execute.return_value.scalar_one_or_none.return_value = None

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested
    return db


def _get_all_sqls(db: MagicMock) -> list[str]:
    """Собрать все SQL-строки из всех db.execute() вызовов."""
    sqls = []
    for call_args in db.execute.call_args_list:
        if call_args.args:
            sqls.append(str(call_args.args[0]))
    return sqls


def _get_all_params(db: MagicMock) -> list[dict]:
    """Собрать все params dict из db.execute() вызовов."""
    params_list = []
    for call_args in db.execute.call_args_list:
        if len(call_args.args) >= 2:
            params_list.append(call_args.args[1])
    return params_list


def _find_snapshot_call(db: MagicMock) -> tuple[str, dict] | None:
    """Найти первый db.execute вызов содержащий 'listings_snapshots'."""
    for call_args in db.execute.call_args_list:
        if not call_args.args:
            continue
        sql = str(call_args.args[0])
        if "listings_snapshots" in sql:
            params = call_args.args[1] if len(call_args.args) >= 2 else {}
            return sql, params
    return None


# ── upsert_listing_snapshot — SQL shape ──────────────────────────────────────


def test_upsert_snapshot_sql_contains_insert_and_on_conflict():
    """SQL содержит INSERT INTO listings_snapshots и ON CONFLICT (listing_id, snapshot_date)."""
    db = _mock_db_simple()
    upsert_listing_snapshot(db, listing_id=42, price_rub=5_000_000)

    sqls = _get_all_sqls(db)
    assert len(sqls) == 1, "ровно один db.execute вызов"
    sql = sqls[0]
    assert "listings_snapshots" in sql
    assert "ON CONFLICT" in sql
    assert "DO UPDATE" in sql


def test_upsert_snapshot_params_minimal():
    """Минимальный вызов: только listing_id + price_rub — остальные None."""
    db = _mock_db_simple()
    upsert_listing_snapshot(db, listing_id=7, price_rub=3_000_000)

    params = _get_all_params(db)
    assert len(params) == 1
    p = params[0]
    assert p["lid"] == 7
    assert p["price"] == 3_000_000
    assert p["run_id"] is None
    assert p["snap_date"] is None
    assert p["ppm2"] is None
    assert "pos" not in p  # #2674: position_in_serp больше не пишется
    assert p["status"] == "active"


def test_upsert_snapshot_params_full():
    """Полный вызов — все поля правильно попадают в params."""
    db = _mock_db_simple()
    snap_date = date(2026, 5, 1)
    upsert_listing_snapshot(
        db,
        listing_id=100,
        price_rub=7_200_000,
        price_per_m2=130_000,
        run_id=99,
        snapshot_date=snap_date,
        status="active",
    )

    params = _get_all_params(db)[0]
    assert params["lid"] == 100
    assert params["price"] == 7_200_000
    assert params["ppm2"] == 130_000
    assert params["run_id"] == 99
    assert params["snap_date"] == snap_date
    assert params["status"] == "active"


def test_upsert_snapshot_does_not_write_position_in_serp():
    """#2674: колонка не упоминается ни в сигнатуре, ни в SQL.

    Диагноз — «механизм невыразим»: PK (listing_id, snapshot_date) держит одну строку
    на объявление в СУТКИ, а позиция есть свойство конкретного прогона выдачи с
    конкретными фильтрами (в одни сутки по ключу пишут до 13 прогонов и 4 разных
    SERP-источника). Значение осело бы от последнего писателя дня и читалось бы как
    факт. Тест ловит попытку «подключить проводку» обратно.
    """
    import inspect

    from scraper_kit.snapshot_writer import upsert_listing_snapshot as fn

    assert "position_in_serp" not in inspect.signature(fn).parameters

    db = _mock_db_simple()
    upsert_listing_snapshot(db, listing_id=1, price_rub=1_000_000)
    sql = str(db.execute.call_args_list[0].args[0])
    assert "position_in_serp" not in sql


def test_upsert_snapshot_no_commit():
    """upsert_listing_snapshot НЕ вызывает db.commit() — commit делает caller."""
    db = _mock_db_simple()
    upsert_listing_snapshot(db, listing_id=1, price_rub=1_000_000)
    db.commit.assert_not_called()


# ── save_detail_enrichment интеграция ────────────────────────────────────────


def _mock_db_detail(price_rub: int = 5_500_000, price_per_m2: int | None = 110_000) -> MagicMock:
    """Mock для save_detail_enrichment: SELECT listings возвращает price_rub/price_per_m2."""
    db = MagicMock()

    # SELECT price_rub, price_per_m2 FROM listings
    price_row = MagicMock()
    price_row.price_rub = price_rub
    price_row.price_per_m2 = price_per_m2

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    # Разные вызовы db.execute() возвращают разное:
    # 1. UPDATE listings → result.fetchone() = None (не используется)
    # 2. SELECT price → result.fetchone() = price_row
    # 3+. INSERT offer_price_history / agents → не важно
    execute_results = []

    def _execute_side_effect(sql, params=None):
        sql_str = str(sql)
        mock_result = MagicMock()
        mock_result.scalar_one_or_none.return_value = None
        if "SELECT price_rub" in sql_str:
            mock_result.fetchone.return_value = price_row
        else:
            mock_result.fetchone.return_value = None
        execute_results.append((sql_str, params, mock_result))
        return mock_result

    db.execute.side_effect = _execute_side_effect
    db._execute_log = execute_results
    return db


def test_save_detail_enrichment_writes_snapshot():
    """save_detail_enrichment вызывает upsert_listing_snapshot через SELECT listings."""
    db = _mock_db_detail(price_rub=6_000_000, price_per_m2=120_000)
    enrichment = DetailEnrichment(ceiling_height=2.7)

    save_detail_enrichment(db, 55, enrichment)

    snap = _find_snapshot_call(db)
    assert snap is not None, "должен быть INSERT INTO listings_snapshots"
    _sql, params = snap
    assert params["lid"] == 55
    assert params["price"] == 6_000_000
    assert params["ppm2"] == 120_000
    assert params["status"] == "active"


def test_save_detail_enrichment_snapshot_skipped_when_no_price_row():
    """Если listings строка не найдена (None) — snapshot не записывается, ошибки нет."""
    db = MagicMock()

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested

    def _exec(sql, params=None):
        mock_result = MagicMock()
        mock_result.fetchone.return_value = None
        mock_result.scalar_one_or_none.return_value = None
        return mock_result

    db.execute.side_effect = _exec

    enrichment = DetailEnrichment()
    # Должно завершиться без исключений
    save_detail_enrichment(db, 99, enrichment)

    # Убедимся что listings_snapshots не упоминается
    sqls = _get_all_sqls(db)
    assert not any("listings_snapshots" in s for s in sqls)


def test_save_detail_enrichment_snapshot_failure_does_not_abort():
    """Если upsert_listing_snapshot падает — save_detail_enrichment всё равно commit'ит."""
    db = _mock_db_detail(price_rub=4_000_000)

    with patch(
        "scraper_kit.providers.cian.detail.upsert_listing_snapshot",
        side_effect=RuntimeError("db snapshot blip"),
    ):
        enrichment = DetailEnrichment(ceiling_height=2.5)
        # Должно завершиться без исключений — snapshot error логируется, не пробрасывается
        save_detail_enrichment(db, 10, enrichment)

    db.commit.assert_called_once()


def test_save_detail_enrichment_writes_offer_price_history_rows():
    """price_changes из enrichment записываются в offer_price_history."""
    db = _mock_db_detail(price_rub=5_000_000)
    enrichment = DetailEnrichment(
        price_changes=[
            {"change_time": "2026-04-01T00:00:00Z", "price_rub": 5_200_000, "diff_percent": None},
            {"change_time": "2026-04-10T00:00:00Z", "price_rub": 5_000_000, "diff_percent": -3.8},
        ]
    )

    save_detail_enrichment(db, 20, enrichment)

    sqls = _get_all_sqls(db)
    oph_inserts = [s for s in sqls if "offer_price_history" in s]
    assert len(oph_inserts) == 2, "по одному INSERT для каждой price_change"


def test_save_detail_enrichment_oph_on_conflict_constraint():
    """INSERT offer_price_history использует именованный ON CONFLICT ON CONSTRAINT."""
    db = _mock_db_detail(price_rub=5_000_000)
    enrichment = DetailEnrichment(
        price_changes=[
            {"change_time": "2026-05-01T12:00:00Z", "price_rub": 5_000_000, "diff_percent": 0},
        ]
    )

    save_detail_enrichment(db, 21, enrichment)

    sqls = _get_all_sqls(db)
    oph_sqls = [s for s in sqls if "offer_price_history" in s]
    assert oph_sqls, "INSERT offer_price_history не найден"
    assert "offer_price_history_listing_change_uq" in oph_sqls[0], (
        "ON CONFLICT должен ссылаться на UNIQUE constraint"
    )


def test_save_detail_enrichment_recomputes_diff_after_decimal_slip():
    """cian: после выброса сдвига разряда соседу пересчитывается diff_percent (#3376).

    377 000 → 3 770 000 → 377 000 при цене листинга 377 000: средняя точка ×10 с
    возвратом к базе — выбрасывается. У последней точки парсер посчитал −90% от
    исчезнувшей базы 3 770 000; от настоящей базы 377 000 это 0%. Без пересчёта в
    колонку уехал бы процент от цены, которой в истории больше нет (у domclick это
    уже чинилось, cian отставал).
    """
    db = _mock_db_detail(price_rub=377_000)
    enrichment = DetailEnrichment(
        price_changes=[
            {"change_time": "2026-04-01T00:00:00Z", "price_rub": 377_000, "diff_percent": None},
            {"change_time": "2026-04-05T00:00:00Z", "price_rub": 3_770_000, "diff_percent": 900.0},
            {"change_time": "2026-04-10T00:00:00Z", "price_rub": 377_000, "diff_percent": -90.0},
        ]
    )

    save_detail_enrichment(db, 23, enrichment)

    oph = [
        (params["price"], params["diff"])
        for sql, params, _ in db._execute_log
        if "offer_price_history" in sql
    ]
    assert oph == [(377_000, None), (377_000, 0.0)]


def test_save_detail_enrichment_skips_price_change_without_change_time():
    """price_changes без change_time пропускаются, не вызывают INSERT."""
    db = _mock_db_detail(price_rub=5_000_000)
    enrichment = DetailEnrichment(
        price_changes=[
            {"change_time": None, "price_rub": 5_000_000},  # пропустить
            {"change_time": "2026-04-01T00:00:00Z", "price_rub": None},  # пропустить
            {"change_time": "2026-04-01T00:00:00Z", "price_rub": 5_000_000},  # записать
        ]
    )

    save_detail_enrichment(db, 22, enrichment)

    sqls = _get_all_sqls(db)
    oph_inserts = [s for s in sqls if "offer_price_history" in s]
    assert len(oph_inserts) == 1


# ── save_listings интеграция ─────────────────────────────────────────────────


def _mock_db_for_save_listings(
    inserted: bool = True,
    listing_id: int = 42,
) -> MagicMock:
    """Mock db для save_listings с правильным RETURNING + begin_nested."""
    row = MagicMock()
    row.id = listing_id
    row.inserted = inserted

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

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

    @contextmanager
    def _nested():
        yield MagicMock()

    db.begin_nested.side_effect = _nested
    return db


def test_save_listings_writes_snapshot_per_lot():
    """save_listings вызывает upsert_listing_snapshot для каждого сохранённого lot."""

    db = _mock_db_for_save_listings(inserted=True, listing_id=77)
    lots = [
        ScrapedLot(
            source="cian",
            source_url="https://ekb.cian.ru/sale/flat/111/",
            source_id="111",
            price_rub=5_000_000,
        ),
        ScrapedLot(
            source="cian",
            source_url="https://ekb.cian.ru/sale/flat/222/",
            source_id="222",
            price_rub=6_000_000,
        ),
    ]

    with patch("scraper_kit.base.upsert_listing_snapshot") as mock_snap:
        save_listings(db, lots, matcher=_mock_matcher(), region_code=66, run_id=10)

    assert mock_snap.call_count == 2, "snapshot должен вызваться для каждого lot"
    # Первый вызов
    kwargs0 = mock_snap.call_args_list[0].kwargs
    assert kwargs0["listing_id"] == 77
    assert kwargs0["price_rub"] == 5_000_000
    assert kwargs0["run_id"] == 10
    assert kwargs0["status"] == "active"


def test_save_listings_snapshot_receives_run_id():
    """Если передан run_id — он пробрасывается в upsert_listing_snapshot."""

    db = _mock_db_for_save_listings(listing_id=88)
    lot = ScrapedLot(
        source="avito",
        source_url="https://avito.ru/1",
        source_id="1",
        price_rub=3_000_000,
    )

    with patch("scraper_kit.base.upsert_listing_snapshot") as mock_snap:
        save_listings(db, [lot], matcher=_mock_matcher(), region_code=66, run_id=55)

    kwargs = mock_snap.call_args.kwargs
    assert kwargs["run_id"] == 55


def test_save_listings_snapshot_none_run_id_by_default():
    """Без run_id — upsert_listing_snapshot получает run_id=None."""

    db = _mock_db_for_save_listings(listing_id=9)
    lot = ScrapedLot(
        source="cian",
        source_url="https://ekb.cian.ru/sale/flat/9/",
        source_id="9",
        price_rub=2_000_000,
    )

    with patch("scraper_kit.base.upsert_listing_snapshot") as mock_snap:
        save_listings(db, [lot], matcher=_mock_matcher(), region_code=66)  # no run_id

    kwargs = mock_snap.call_args.kwargs
    assert kwargs["run_id"] is None


def test_save_listings_snapshot_failure_does_not_abort_batch():
    """Ошибка snapshot не прерывает основной INSERT; все lots сохраняются."""

    db = _mock_db_for_save_listings(inserted=True, listing_id=5)
    lots = [
        ScrapedLot(
            source="cian",
            source_url=f"https://ekb.cian.ru/sale/flat/{i}/",
            source_id=str(i),
            price_rub=1_000_000 * i,
        )
        for i in range(1, 4)
    ]

    with patch(
        "scraper_kit.base.upsert_listing_snapshot",
        side_effect=RuntimeError("snapshot fail"),
    ):
        inserted, _updated = save_listings(db, lots, matcher=_mock_matcher(), region_code=66)

    # INSERT listings всё равно прошёл для всех 3 lots
    assert inserted == 3


def test_save_listings_empty_no_snapshot_call():
    """Пустой список → upsert_listing_snapshot не вызывается."""

    db = _mock_db_for_save_listings()

    with patch("scraper_kit.base.upsert_listing_snapshot") as mock_snap:
        save_listings(db, [], matcher=_mock_matcher(), region_code=66)

    mock_snap.assert_not_called()


# ── Идемпотентность: повторный вызов тот же listing_id + дата ────────────────


def test_upsert_snapshot_idempotent_same_params_no_error():
    """Два одинаковых вызова (listing_id + date совпадают) — нет ошибок.

    Реальная де-дупликация — ON CONFLICT DO UPDATE в БД.
    Здесь проверяем что Python-код не падает при повторном вызове с теми же данными.
    """
    db = _mock_db_simple()
    snap_date = date(2026, 5, 28)

    upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=snap_date)
    upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=snap_date)

    # Два вызова db.execute — каждый попытается INSERT ON CONFLICT DO UPDATE
    assert db.execute.call_count == 2
    # Убедимся что оба SQL содержат ON CONFLICT DO UPDATE
    sqls = _get_all_sqls(db)
    assert all("ON CONFLICT" in s for s in sqls)


def test_upsert_snapshot_different_dates_no_conflict():
    """Разные даты для одного listing → два отдельных snapshot (нет конфликта по PK)."""
    db = _mock_db_simple()

    upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=date(2026, 5, 1))
    upsert_listing_snapshot(db, listing_id=1, price_rub=4_800_000, snapshot_date=date(2026, 5, 2))

    # Каждый вызов делает INSERT: разные snapshot_date → разные PRIMARY KEY → нет конфликта
    params_list = _get_all_params(db)
    assert params_list[0]["snap_date"] == date(2026, 5, 1)
    assert params_list[1]["snap_date"] == date(2026, 5, 2)
    assert params_list[0]["price"] == 5_000_000
    assert params_list[1]["price"] == 4_800_000