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

"""#2674 — писатели наконец пишут то, что обещает схема.

Три находки одного класса: колонка есть, писатель есть, тест на писателя зелёный,
а данные не появляются. Обычный юнит-тест такое не ловит по построению — он
проверяет то, что автор себе представлял. Ловится это сверкой «что схема обещает»
с «что писатель реально перечисляет», поэтому тесты ниже читают миграции и
сравнивают их с SQL писателя, а не повторяют его же список колонок.

Числа с прода на 2026-08-05/06 (до фикса):
  1. house_suggestions — 25 055 строк, image_link/area_m2/rooms/floor/total_floors
     заполнены у 0 из них (колонки с миграции 064, ~74 дня).
  2. listings_snapshots.status — 'active' у всех 394 704 строк при 55 448 реально
     неактивных объявлений; ни 'closed', ни 'stale' не писал никто и никогда.
  3. listing_source_events — 8288 строк, все price_change; edited/first_seen —
     ноль за всё время.

Отдельный класс тестов — гейты на то, что писатель НЕ пишет: delisted/relisted схема
разрешает, но при покрытии обхода 10-35% они неотличимы от «скрейпер снова дошёл»
(контроль — домклик со 100% покрытием: 0 возвратов за 5 суток), а TTL-путь не имеет
права называть протухание снятием. Журнал и история из догадок хуже пустых.

БД и сеть замоканы — реального Postgres не нужно.
"""

from __future__ import annotations

import inspect
import json
import os
import re
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from scraper_kit.providers.avito.imv import _parse_suggestion

from app.services import house_imv_backfill as hib
from app.tasks import deactivate_stale_avito as deact_mod
from app.tasks import listing_source_snapshot as snap_mod

_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
_FIXTURES = Path(__file__).resolve().parent / "fixtures"


# ── Общие хелперы: схема vs writer ────────────────────────────────────────────


def _declared_columns(migration: str, table: str) -> set[str]:
    """Имена колонок из CREATE TABLE IF NOT EXISTS <table> ( ... ); в миграции."""
    body = migration.split(f"CREATE TABLE IF NOT EXISTS {table} (")[1].split("\n);")[0]
    cols: set[str] = set()
    for line in body.splitlines():
        m = re.match(r"\s+([a-z_]+)\s+[a-z]", line)
        if m:
            cols.add(m.group(1))
    return cols


def _insert_columns(sql: str, table: str) -> set[str]:
    """Имена колонок из INSERT INTO <table> ( ... ) VALUES."""
    m = re.search(rf"INSERT INTO {table}\s*\(([^)]*)\)", sql, re.S)
    assert m is not None, f"не найден INSERT INTO {table}"
    return {c.strip() for c in m.group(1).split(",") if c.strip()}


# ══ 1. Фотографии подсказок Avito IMV ═════════════════════════════════════════


def test_suggestion_parser_keeps_image_link() -> None:
    """imageLink из ответа площадки доезжает до модели, а не выбрасывается."""
    sugg = _parse_suggestion(
        {
            "id": 8000753763,
            "title": "3-к. квартира, 61,6 м², 1/5 эт.",
            "price": 7400000,
            "imageLink": "https://80.img.avito.st/image/1/abc",
        }
    )
    assert sugg.image_link == "https://80.img.avito.st/image/1/abc"
    # В raw_payload ссылку по-прежнему не дублируем — у неё теперь своя колонка.
    assert "imageLink" not in (sugg.raw_payload or {})


def test_suggestion_parser_derives_metrics_from_title() -> None:
    """rooms/area_m2/floor/total_floors парсятся из title тем же путём, что у
    placementHistory (колонки house_suggestions существуют с миграции 064)."""
    sugg = _parse_suggestion(
        {"id": 1, "title": "3-к. квартира, 61,6 м², 1/5 эт.", "price": 7400000}
    )
    assert (sugg.rooms, sugg.area_m2, sugg.floor, sugg.total_floors) == (3, 61.6, 1, 5)


def test_suggestion_parser_survives_unparsable_title() -> None:
    """Нераспознанный заголовок → None'ы, а не исключение (строка всё равно пишется)."""
    sugg = _parse_suggestion({"id": 2, "title": "Апартаменты", "price": 1})
    assert (sugg.rooms, sugg.area_m2, sugg.floor, sugg.total_floors) == (None, None, None, None)


def test_studio_title_keeps_area_and_floors() -> None:
    """Студия не роняет разбор целиком: 1991 заголовок из 25 055 (7.9%) — без комнатности.

    Обязательная группа комнатности обнуляла ВСЕ ЧЕТЫРЕ поля, хотя площадь и этажность
    в заголовке есть. rooms=0 — конвенция kit'а («0 = студия»), а не «неизвестно».
    """
    sugg = _parse_suggestion(
        {"id": 3, "title": "Квартира-студия, 34,2 м², 9/10 эт.", "price": 3_500_000}
    )
    assert (sugg.rooms, sugg.area_m2, sugg.floor, sugg.total_floors) == (0, 34.2, 9, 10)


def test_placement_history_gets_same_title_fix() -> None:
    """Тот же регексп чинит второго писателя — house_placement_history (8.8% без площади)."""
    from scraper_kit.providers.avito.imv import _parse_placement_item

    item = _parse_placement_item({"id": 9, "title": "Квартира-студия, 28 м², 2/17 эт."})
    assert (item.rooms, item.area_m2, item.floor, item.total_floors) == (0, 28.0, 2, 17)


def test_live_fixture_suggestions_carry_image_link() -> None:
    """Живой capture avito_imv_getdata.json: у подсказок реально есть imageLink."""
    data = json.loads((_FIXTURES / "avito_imv_getdata.json").read_text("utf-8"))
    items = data["suggestions"]["items"]
    parsed = [_parse_suggestion(raw) for raw in items]
    assert parsed, "фикстура без подсказок — тест бессмыслен"
    assert all(s.image_link for s in parsed)


def test_house_suggestions_insert_covers_every_declared_column() -> None:
    """Regression-гейт на весь класс бага: INSERT обязан перечислять КАЖДУЮ колонку
    house_suggestions из миграции 064 (кроме автоинкрементного id).

    Именно этот тест покраснел бы 74 дня назад: image_link (и заодно area_m2/rooms/
    floor/total_floors) объявлены схемой, но в запрос вставки не входили — 25 055
    строк с NULL. Тест не дублирует список колонок писателя, а сверяет его со схемой,
    поэтому ловит и следующую забытую колонку.
    """
    declared = _declared_columns(
        (_SQL_DIR / "064_house_imv_phase_c.sql").read_text("utf-8"), "house_suggestions"
    )
    written = _insert_columns(inspect.getsource(hib.save_imv_result), "house_suggestions")
    assert declared - {"id"} <= written, (
        f"колонки без писателя: {sorted(declared - {'id'} - written)}"
    )


def test_save_imv_result_binds_image_link_and_metrics() -> None:
    """save_imv_result передаёт значения подсказки в bind-параметры (не только в SQL)."""
    sugg = _parse_suggestion(
        {
            "id": 777,
            "title": "2-к. квартира, 42 м², 4/5 эт.",
            "price": 6300000,
            "imageLink": "https://img/x.jpg",
        }
    )
    result = MagicMock(
        cache_key="k",
        recommended_price=1,
        lower_price=1,
        higher_price=1,
        market_count=1,
        raw_response=None,
        placement_history=[],
        suggestions=[sugg],
    )
    params = {
        "rooms": 2,
        "area_m2": 42.0,
        "floor": 4,
        "floor_at_home": 5,
        "house_type": "panel",
        "renovation_type": "cosmetic",
        "has_balcony": True,
        "has_loggia": False,
    }
    db = MagicMock()
    hib.save_imv_result(db, house_id=1, params=params, result=result)

    sugg_calls = [
        c for c in db.execute.call_args_list if "INSERT INTO house_suggestions" in str(c.args[0])
    ]
    assert len(sugg_calls) == 1
    bound = sugg_calls[0].args[1]
    assert bound["img"] == "https://img/x.jpg"
    assert (bound["rooms"], bound["area"], bound["floor"], bound["total_floors"]) == (2, 42.0, 4, 5)


# ══ 2. «Снято» и «протухло» в дневной истории объявлений ══════════════════════

_DEACT_SQL_BUILDERS = (deact_mod._build_all_segments_sql, deact_mod._build_segments_sql)


@pytest.mark.parametrize("build", _DEACT_SQL_BUILDERS)
@pytest.mark.parametrize("column", sorted(deact_mod._ALLOWED_STALENESS_COLUMNS))
def test_deactivation_writes_stale_snapshot_in_same_statement(build: Any, column: str) -> None:
    """Деактивация и снимок 'stale' — один statement, значит одна транзакция.

    До #2674 задача только двигала флаг: в listings_snapshots не появлялось ничего,
    и дата снятия объявления (лучший сигнал «скорее всего продано») восстанавливалась
    на глаз из последнего показа + предполагаемого срока жизни.
    """
    sql = str(build(column).text)
    assert "SET is_active = false" in sql
    assert "RETURNING id, price_rub" in sql
    assert "INSERT INTO listings_snapshots" in sql
    assert "'stale'" in sql
    # Снимок пишется по строкам, которые вернул сам UPDATE, — не отдельной выборкой.
    assert "FROM stale" in sql
    # Идемпотентность: повторный прогон в те же сутки не падает на PK.
    assert "ON CONFLICT (listing_id, snapshot_date) DO UPDATE" in sql
    # psycopg v3: никаких :param::type.
    assert not re.search(r":\w+::", sql)


@pytest.mark.parametrize("build", _DEACT_SQL_BUILDERS)
def test_ttl_path_never_claims_closed(build: Any) -> None:
    """TTL-путь НЕ имеет права писать 'closed' — он не знает, что объявление снято.

    Замер: прогон по домклику 02.08 деактивировал 6131 объявление за раз (TTL 14 суток
    против 12 суток простоя обхода). Под статусом 'closed' это 6131 фальшивая «дата
    продажи» одной датой. 'closed' остаётся только за 404 — там ответила площадка.
    """
    assert "'closed'" not in str(build("last_seen_at").text)


class _FakeResult:
    def __init__(self, rowcount: int) -> None:
        self.rowcount = rowcount


class _FakeDB:
    def __init__(self, rowcount: int = 0) -> None:
        self._rowcount = rowcount
        self.executed: list[tuple[Any, Any]] = []
        self.committed = False

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
        self.executed.append((stmt, params))
        return _FakeResult(self._rowcount)

    def commit(self) -> None:
        self.committed = True

    def rollback(self) -> None:  # pragma: no cover — путь ошибки тут не проверяется
        pass


def test_deactivate_stale_listings_threads_run_id_into_snapshot(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """run_id доезжает до снимка — провенанс «каким прогоном закрыто» не теряется."""
    monkeypatch.setattr(deact_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(deact_mod.runs_mod, "mark_failed", lambda *a, **k: None)
    db = _FakeDB(rowcount=7)

    out = deact_mod.deactivate_stale_listings(
        db,  # type: ignore[arg-type]
        run_id=555,
        listing_source="avito",
        ttl_days=10,
    )

    assert out == {"deactivated": 7}
    assert len(db.executed) == 1, "деактивация и снимок обязаны быть одним statement'ом"
    stmt, bound = db.executed[0]
    assert "INSERT INTO listings_snapshots" in str(stmt)
    assert bound is not None and bound["run_id"] == 555


@pytest.mark.asyncio
async def test_avito_404_records_closed_snapshot() -> None:
    """404 с площадки — самый достоверный сигнал снятия; он тоже попадает в историю."""
    from scraper_kit.avito_exceptions import AvitoListingGoneError

    from app.tasks.avito_detail_backfill import run_avito_detail_backfill

    db = MagicMock()
    sel = MagicMock()
    sel.mappings.return_value.all.return_value = [
        {"id": 42, "source_url": "/items/42", "price_rub": 5_000_000}
    ]
    db.execute.return_value = sel
    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)

    with (
        patch("app.tasks.avito_detail_backfill.settings", fake_settings),
        patch("app.tasks.avito_detail_backfill.AsyncSession", return_value=AsyncMock()),
        patch("app.tasks.avito_detail_backfill.AvitoScraper"),
        patch("app.tasks.avito_detail_backfill.runs_mod", MagicMock()),
        patch("app.tasks.avito_detail_backfill.asyncio.sleep", new_callable=AsyncMock),
        patch(
            "app.tasks.avito_detail_backfill.fetch_detail",
            AsyncMock(side_effect=AvitoListingGoneError("404 gone")),
        ),
        patch("app.tasks.avito_detail_backfill.upsert_listing_snapshot") as snap,
    ):
        result = await run_avito_detail_backfill(db, run_id=3, params={"budget_sec": 60})

    assert result.gone == 1
    snap.assert_called_once()
    assert snap.call_args.kwargs["status"] == "closed"
    assert snap.call_args.kwargs["listing_id"] == 42
    assert snap.call_args.kwargs["price_rub"] == 5_000_000


# ══ 3. Журнал событий объявлений ══════════════════════════════════════════════


def _schema_event_types() -> set[str]:
    """Пять типов из CHECK-констрейнта миграции 079 — источник правды."""
    sql = (_SQL_DIR / "079_listing_source_history.sql").read_text("utf-8")
    check = sql.split("event_type IN (")[1].split(")")[0]
    return set(re.findall(r"'([a-z_]+)'", check))


# Два типа схемы НЕ ВЫВОДИМЫ из наших данных и намеренно не пишутся (#2674).
# is_active в снимке значит «мы видели», а не «есть на площадке», поэтому переход
# рождается тем, что скрейпер снова дошёл до источника. Контрольная группа за 14-18.07:
# domklik при покрытии 99.9-100% дал возвратов РОВНО 0 и снятий 1-4 в сутки, yandex при
# 34-43% — снятий 343-433 в сутки. Тот же обход, тот же день, разница только в покрытии.
# Отсюда: avito 13.07 (остановка обхода) 3023 «снятия» за сутки против контрольных 1-4
# (точность ≈4%), и 4705 «возвратов» из 5493 за 12 дней — два дня после возобновления.
_NOT_DERIVABLE_EVENT_TYPES = {"delisted", "relisted"}


def test_event_writer_covers_every_derivable_schema_event_type() -> None:
    """Писатель обязан уметь каждый ВЫВОДИМЫЙ тип из CHECK схемы.

    До #2674 из пяти типов писался один (price_change, 8288 строк). Дописаны два
    выводимых; два оставшихся — сознательное решение, а не забытая ветка (см.
    _NOT_DERIVABLE_EVENT_TYPES). Тест сверяет со схемой, а не с копией списка,
    поэтому покраснеет и на шестом типе, добавленном в CHECK без писателя.
    """
    declared = _schema_event_types()
    assert len(declared) == 5, f"схема 079 изменилась: {sorted(declared)}"
    sql = str(snap_mod._EVENT_DIFF_SQL.text)
    expected = declared - _NOT_DERIVABLE_EVENT_TYPES
    missing = {t for t in expected if f"'{t}'" not in sql}
    assert not missing, f"выводимые типы без писателя: {sorted(missing)}"


def test_not_derivable_events_are_never_written() -> None:
    """delisted/relisted не пишутся: при покрытии обхода 10-35% они неотличимы от
    «скрейпер снова дошёл». Контроль — домклик со 100% покрытием: 0 возвратов за 5 суток.

    Гейт против «дописать для полноты»: журнал из догадок хуже пустого журнала.
    """
    sql = str(snap_mod._EVENT_DIFF_SQL.text)
    written = {t for t in _NOT_DERIVABLE_EVENT_TYPES if f"'{t}'" in sql}
    assert not written, f"невыводимые типы попали в писатель: {sorted(written)}"
    # is_active больше не читается вовсе — иначе ветка вернётся незаметно.
    assert "is_active" not in sql


def test_first_seen_requires_left_join_and_derivations_use_snapshot_fields() -> None:
    """Ветки выводятся из полей снимка, first_seen достижим только через LEFT JOIN.

    С обычным JOIN источник без предыдущего снимка отбрасывался джойном — событие
    «первое появление» было недостижимо по построению.
    """
    sql = str(snap_mod._EVENT_DIFF_SQL.text)
    assert "LEFT JOIN LATERAL" in sql
    assert "p.snapshot_date IS NULL" in sql  # first_seen
    assert "t.payload_hash IS DISTINCT FROM p.payload_hash" in sql  # edited
    # Снимок за сегодня обязан отдавать поля, из которых выводятся ветки.
    assert "SELECT listing_source_id, price_rub, payload_hash" in sql


def test_event_dedup_works_across_runs_not_only_within_one() -> None:
    """change_time усечён до суток: UNIQUE(source, change_time, type) должен гасить
    повторный прогон в те же сутки (2 августа их было два), а не только строки одного."""
    sql = str(snap_mod._EVENT_DIFF_SQL.text)
    assert "date_trunc('day', now())" in sql


def test_price_change_division_guarded_by_nullif() -> None:
    """VALUES вычисляется ДО фильтра e.fires → без NULLIF прогон падал бы на
    первом источнике с нулевой прошлой ценой (предикат p.price_rub <> 0 не спасает)."""
    sql = str(snap_mod._EVENT_DIFF_SQL.text)
    assert "NULLIF(p.price_rub, 0)" in sql


class _EventFakeDB:
    """Session-заглушка: SET LOCAL и snapshot дают rowcount, event-diff — пары счётчиков."""

    def __init__(self, event_rows: list[tuple[str, int]]) -> None:
        self._event_rows = event_rows
        self.committed = False

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> Any:
        result = MagicMock()
        if "listing_source_events" in str(stmt):
            result.fetchall.return_value = self._event_rows
        else:
            result.rowcount = 100
        return result

    def commit(self) -> None:
        self.committed = True

    def rollback(self) -> None:  # pragma: no cover — путь ошибки тут не проверяется
        pass


def test_counters_report_every_written_type_including_zeros(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Счётчики прогона показывают все пишущиеся типы; не сработавший честно равен 0.

    Ровно этого счётчика не хватало, чтобы заметить четыре нуля из пяти за 66 дней.
    Счётчиков НЕвыводимых типов быть не должно — иначе вечный 0 будет читаться как
    «событий не было», а не как «мы это сознательно не пишем».
    """
    monkeypatch.setattr(snap_mod.runs_mod, "mark_done", lambda *a, **k: None)
    monkeypatch.setattr(snap_mod.runs_mod, "mark_failed", lambda *a, **k: None)

    db = _EventFakeDB([("first_seen_events", 600), ("edited_events", 17)])
    out = snap_mod.snapshot_listing_sources(db, run_id=1)  # type: ignore[arg-type]

    assert out["first_seen_events"] == 600
    assert out["edited_events"] == 17
    assert out["price_change_events"] == 0
    # Ключ на каждый пишущийся тип — иначе «ноль» неотличим от «типа нет в counters».
    for event_type in _schema_event_types() - _NOT_DERIVABLE_EVENT_TYPES:
        assert f"{event_type}_events" in out
    for event_type in _NOT_DERIVABLE_EVENT_TYPES:
        assert f"{event_type}_events" not in out