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

"""Разбор мёртвого кода #2674: подключить / удалить / задокументировать.

Каждая правка эпика — тест, который краснеет без неё:

  подключено:
    - houses.has_panorama пишется из yandex_valuation (и НЕ пишется, когда страница
      не подтверждена — иначе false «не смотрели» выдаётся за false «посмотрели»);
    - domrf_kapremont_load зарегистрирован Handler'ом И засеян в scrape_schedules —
      именно отсутствие этой пары держало загрузчик ДОМ.РФ невызванным;
    - filters_hash читается с estimation.sale.filtersHash, а не .data.filtersHash.

  удалено (гейт против возврата):
    - _phys_dedup_key / _extract_street_token — обёртки без прод-вызовов;
    - asking_to_sold_ratios_tiered / asking_to_sold_tier_bounds — таблицы без
      читателя и писателя;
    - listings.merged_into, house_sources.raw_payload — колонки без писателя;
    - v_data_quality.price_disagreements_count — показатель, который не мог быть
      ненулевым.

  задокументировано:
    - BROWSER_BLOCK_RESOURCES: код его не читает с #1812, но прод его задаёт —
      сервис обязан сказать об этом вслух на старте.

Без БД и сети: сессия замокана, SQL-миграции читаются как текст.
"""

from __future__ import annotations

import os
import re
from pathlib import Path
from unittest.mock import MagicMock, patch

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

from scraper_kit.providers.cian.valuation import _parse_valuation_state
from scraper_kit.providers.yandex.valuation import (
    ValuationHistoryItem,
    ValuationHouseMeta,
    YandexValuationResult,
)

from app.services import estimator
from app.services.estimator import _save_yandex_history_items

REPO_ROOT = Path(__file__).resolve().parents[3]
TRADEIN = REPO_ROOT / "tradein-mvp"
SQL_DIR = TRADEIN / "backend" / "data" / "sql"
MIGRATION = SQL_DIR / "216_dead_code_sweep.sql"


# ─────────────────────────────────────────────────────────────────────────────
# Подключено 1/3: houses.has_panorama
# ─────────────────────────────────────────────────────────────────────────────


def _result_with_meta(meta: ValuationHouseMeta) -> YandexValuationResult:
    return YandexValuationResult(
        address="Екатеринбург, ул. Куйбышева, 106",
        offer_category="APARTMENT",
        offer_type="SELL",
        page=1,
        source_url="https://realty.yandex.ru/otsenka-kvartiry-po-adresu-onlayn/?address=test",
        house=meta,
        history_items=[ValuationHistoryItem(area_m2=50.0, rooms=2, floor=5, start_price=9_000_000)],
    )


def _panorama_updates(db: MagicMock) -> list[dict]:
    """Параметры всех db.execute, которые обновляли houses.has_panorama."""
    found = []
    for call in db.execute.call_args_list:
        sql = str(call.args[0])
        if "has_panorama" in sql and "UPDATE houses" in sql:
            found.append(call.args[1])
    return found


def test_has_panorama_written_when_page_rendered() -> None:
    """Разобранный флаг доезжает до houses — до #2674 он не доезжал ни одной строкой."""
    db = MagicMock()
    result = _result_with_meta(
        ValuationHouseMeta(year_built=2010, total_floors=16, has_panorama=True)
    )

    with patch(
        "app.services.estimator.match_or_create_house",
        return_value=(99, 0.9, "fp"),
    ):
        _save_yandex_history_items(db, result)

    updates = _panorama_updates(db)
    assert updates, "houses.has_panorama не записан — вернулась исходная болячка #2674"
    assert updates[0] == {"hid": 99, "panorama": True}


def test_has_panorama_false_written_when_page_rendered() -> None:
    """Отсутствие метки на ОТРИСОВАННОЙ странице — тоже наблюдение, пишем false."""
    db = MagicMock()
    result = _result_with_meta(
        ValuationHouseMeta(year_built=1998, total_floors=9, has_panorama=False)
    )

    with patch(
        "app.services.estimator.match_or_create_house",
        return_value=(7, 0.9, "fp"),
    ):
        _save_yandex_history_items(db, result)

    assert _panorama_updates(db) == [{"hid": 7, "panorama": False}]


def test_has_panorama_not_written_when_page_unconfirmed() -> None:
    """Пустая мета (капча/редизайн) → NULL, а не сфабрикованный false."""
    db = MagicMock()
    result = _result_with_meta(ValuationHouseMeta(has_panorama=False))

    with patch(
        "app.services.estimator.match_or_create_house",
        return_value=(5, 0.9, "fp"),
    ):
        _save_yandex_history_items(db, result)

    assert _panorama_updates(db) == [], "false записан там, где мы ничего не наблюдали"


def test_has_panorama_not_written_without_house_id() -> None:
    """Дом не сматчился → писать некуда, но и падать нельзя."""
    db = MagicMock()
    result = _result_with_meta(ValuationHouseMeta(year_built=2010, total_floors=16))

    with patch(
        "app.services.estimator.match_or_create_house",
        side_effect=RuntimeError("no house"),
    ):
        _save_yandex_history_items(db, result)

    assert _panorama_updates(db) == []


# ─────────────────────────────────────────────────────────────────────────────
# Подключено 2/3: загрузчик ДОМ.РФ — оборванная проводка
# ─────────────────────────────────────────────────────────────────────────────


def test_domrf_loader_has_a_scheduler_handler() -> None:
    """Без Handler'а загрузчик ДОМ.РФ был невызываем — этого и не хватало."""
    from app.services.product_handlers import build_product_handlers

    handlers = build_product_handlers(MagicMock())
    assert "domrf_kapremont_load" in handlers


def test_domrf_loader_is_seeded_into_schedules() -> None:
    """Handler без строки расписания так же нем, как расписание без Handler'а."""
    sql = MIGRATION.read_text(encoding="utf-8")
    assert "'domrf_kapremont_load'" in sql
    assert "INSERT INTO scrape_schedules" in sql
    # Недельный такт: реестр капремонта не меняется ежедневно, а прогон качает два zip.
    assert '"interval_days": 7' in sql


def test_domrf_handler_reuses_loader_functions() -> None:
    """Дизайн-инвариант product_handlers: job переиспользует боевое тело, не копирует."""
    src = (TRADEIN / "backend" / "app" / "services" / "product_handlers.py").read_text(
        encoding="utf-8"
    )
    body = src.split("_job_domrf_kapremont_load", 1)[1].split("# ──", 1)[0]
    for fn in (
        "load_domrf_kapremont",
        "backfill_houses_from_domrf",
        "propagate_listings_year_from_houses",
    ):
        assert fn in body, f"{fn} не вызывается — загрузчик подключён лишь наполовину"


# ─────────────────────────────────────────────────────────────────────────────
# Подключено 3/3: filters_hash лежит на уровень выше, чем его читали
# ─────────────────────────────────────────────────────────────────────────────


def test_filters_hash_read_from_sale_wrapper() -> None:
    """Прод-форма ответа: ключи estimation.sale = {isError, filtersHash, data, isFetching}."""
    state = {
        "user": {"isAuthenticated": True, "userId": 1},
        "estimation": {
            "sale": {
                "isError": False,
                "isFetching": False,
                "filtersHash": "96bba2876162e2b822f80eec",
                "data": {"price": 9_000_000, "accuracy": 12},
            },
            "rent": {"data": {}},
        },
    }
    assert _parse_valuation_state(state).filters_hash == "96bba2876162e2b822f80eec"


def test_filters_hash_absent_stays_none() -> None:
    """Нет ключа → None. Со старым (вложенным) путём тест бы прошёл — он не про фикс."""
    state = {"estimation": {"sale": {"data": {"price": 1}}, "rent": {"data": {}}}}
    assert _parse_valuation_state(state).filters_hash is None


def test_filters_hash_backfill_uses_the_same_path() -> None:
    """Миграция достаёт хеш ровно оттуда же, откуда его теперь читает парсер."""
    sql = MIGRATION.read_text(encoding="utf-8")
    assert "'{estimation,sale,filtersHash}'" in sql
    assert "WHERE filters_hash IS NULL" in sql


# ─────────────────────────────────────────────────────────────────────────────
# Удалено: гейты против возврата
# ─────────────────────────────────────────────────────────────────────────────


def _live_python_sources() -> list[Path]:
    """Боевой Python trade-in: app + scraper-kit + browser. Без тестов."""
    roots = [
        TRADEIN / "backend" / "app",
        TRADEIN / "packages" / "scraper-kit" / "src",
        TRADEIN / "browser",
    ]
    return [p for root in roots for p in root.rglob("*.py") if not p.name.startswith("test_")]


def test_dead_dedup_wrappers_are_gone() -> None:
    """Обёртки без прод-вызовов (25 ссылок, все из тестов) не должны вернуться."""
    for name in ("_phys_dedup_key", "_extract_street_token"):
        assert not hasattr(estimator, name), (
            f"{name} снова в estimator — эта обёртка описывала правило, "
            "которого в боевом дедупе (_union_find_phys_dedup) нет"
        )


def test_dead_names_absent_from_live_code() -> None:
    """Имена удалённых таблиц/колонок/показателей не упоминаются в боевом коде.

    SQL-миграции сознательно НЕ проверяем: 098/028/029/046 — исторические файлы,
    переписывать их задним числом нельзя (пересборка с нуля идёт по ним).
    """
    dead = [
        "asking_to_sold_ratios_tiered",
        "asking_to_sold_tier_bounds",
        "price_disagreements_count",
    ]
    offenders: list[str] = []
    for path in _live_python_sources():
        text = path.read_text(encoding="utf-8")
        for name in dead:
            if name in text:
                offenders.append(f"{path.relative_to(REPO_ROOT)}: {name}")
    assert not offenders, "удалённое снова упоминается: " + "; ".join(offenders)


def test_dropped_columns_have_no_python_writer() -> None:
    """merged_into / house_sources.raw_payload: писателя не было и быть не должно."""
    offenders = [
        str(p.relative_to(REPO_ROOT))
        for p in _live_python_sources()
        if "merged_into" in p.read_text(encoding="utf-8")
    ]
    assert not offenders, f"listings.merged_into снова упомянут: {offenders}"

    hs_writers = [
        p
        for p in _live_python_sources()
        if "INSERT INTO house_sources" in p.read_text(encoding="utf-8")
    ]
    assert hs_writers, "писатели house_sources исчезли — тест потерял смысл, проверь grep"
    for path in hs_writers:
        stmt = path.read_text(encoding="utf-8").split("INSERT INTO house_sources", 1)[1]
        stmt = stmt.split("VALUES", 1)[0]
        assert "raw_payload" not in stmt, f"{path} снова пишет удалённую колонку"


def test_migration_drops_exactly_what_was_declared_dead() -> None:
    sql = MIGRATION.read_text(encoding="utf-8")
    for stmt in (
        "DROP TABLE IF EXISTS asking_to_sold_ratios_tiered",
        "DROP TABLE IF EXISTS asking_to_sold_tier_bounds",
        "ALTER TABLE IF EXISTS listings DROP COLUMN IF EXISTS merged_into",
        "ALTER TABLE IF EXISTS house_sources DROP COLUMN IF EXISTS raw_payload",
        "DROP INDEX IF EXISTS house_sources_raw_payload_gin_idx",
    ):
        assert stmt in sql, f"миграция не выполняет: {stmt}"
    # Показатель убран из KPI-снимка, но сам view-источник оставлен как задел.
    view_ddl = sql.split("CREATE OR REPLACE VIEW v_data_quality", 1)[1].split(";", 1)[0]
    assert "price_disagreements_count" not in view_ddl
    assert "COMMENT ON VIEW v_price_divergence" in sql


def test_price_divergence_is_documented_as_structurally_empty() -> None:
    """Оставленный задел обязан говорить, чем он НЕ является сегодня."""
    sql = MIGRATION.read_text(encoding="utf-8")
    comment = sql.split("COMMENT ON VIEW v_price_divergence IS", 1)[1].split(";", 1)[0]
    assert (
        "match_or_create_listing" in comment
    ), "комментарий не называет причину пустоты — без неё это просто «пока пусто»"


# ─────────────────────────────────────────────────────────────────────────────
# Задокументировано: мёртвая переменная окружения
# ─────────────────────────────────────────────────────────────────────────────


def test_browser_warns_about_retired_env() -> None:
    """BROWSER_BLOCK_RESOURCES выставлен в трёх прод-контейнерах и ни на что не влияет.

    Сам browser/server.py тянет aiohttp+camoufox и в backend-окружении не
    импортируется, поэтому проверяем исходник: переменная обязана быть в реестре
    отставных И должна логироваться предупреждением на старте.
    """
    src = (TRADEIN / "browser" / "server.py").read_text(encoding="utf-8")
    assert "_RETIRED_ENV" in src
    assert '"BROWSER_BLOCK_RESOURCES"' in src
    assert "_warn_retired_env()" in src, "предупреждение не вызывается со старта"
    # Переменная НЕ должна снова начать что-то менять — только предупреждать.
    read_sites = re.findall(r'os\.environ(?:\.get)?[\[(]"BROWSER_BLOCK_RESOURCES"', src)
    assert not read_sites, "BROWSER_BLOCK_RESOURCES снова читается как рабочий флаг"