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

"""#2674: «временная» ошибка домовой IMV-оценки снова означает временную.

Что было. Очередь backfill'а выбирает РОВНО ОДИН imv_status за прогон
(`only_status`, дефолт 'pending'). Возможность повтора существовала и была
задокументирована в admin-API, но за всю историю (41 прогон, 26.06–11.08.2026)
её не передал никто: все прогоны ушли с 'pending'. Дома, упавшие по причине,
которой больше нет (503/500 от браузерного сайдкара, #2698), лежали в
'transient_error' без единого шанса вернуться — 1390 штук на 12.08.2026.

Что проверяем — по существу, а не по настройке:
  * дом в 'transient_error' ПОПАДАЕТ в пакет прогона с дефолтными параметрами
    (тот же вызов, что делает расписание) — на старом коде он не попадает;
  * pending при этом из пакета НЕ вытесняется;
  * дом, исчерпавший лимит попыток, в автоматический повтор не берётся, а
    ЯВНЫЙ only_status='transient_error' (ручной запрос оператора) берёт и его;
  * дом без пригодных параметров помечается no_params БЕЗ запроса к площадке.

Живой Postgres обязателен: проверяется поведение SQL-выборки, а не то, какие
строки вернул мок. Без БД тест самоскипается (в CI БД есть — ci-tradein.yml
поднимает Postgres и накатывает схему).
"""

from __future__ import annotations

import os
from typing import Any
from unittest.mock import AsyncMock, patch

import pytest

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

from app.services import house_imv_backfill as hib

# Фикстуры живут в собственном диапазоне id (ср. tests/test_house_dedup_merge.py).
_H_STUCK = 900601  # transient_error, попыток 0 → обязан вернуться в очередь
_H_EXHAUSTED = 900602  # transient_error, попытки исчерпаны → в автоповтор не берём
_H_PENDING = 900603  # pending → не должен быть вытеснен retry-очередью
_H_NO_PARAMS = 900604  # pending без объявлений с rooms+area → no_params без запроса
_IDS = (_H_STUCK, _H_EXHAUSTED, _H_PENDING, _H_NO_PARAMS)


def _live_session() -> Any | None:
    """Session к достижимому не-заглушечному Postgres, иначе None."""
    try:
        from sqlalchemy import create_engine, text
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        conn = engine.connect()
        conn.execute(text("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


def _seed(db: Any) -> None:
    from sqlalchemy import text as _t

    # url/source/ext_house_id — NOT NULL в houses (002_core_tables).
    db.execute(
        _t(
            "INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon, "
            "                    imv_status, imv_transient_attempts, last_imv_attempt_at) VALUES "
            "(:s,  'avito', 'EXT-2674-S', 'u', 'тестдом 2674, 1', 56.84, 60.60, "
            " 'transient_error', 0, NOW() - INTERVAL '10 days'),"
            "(:x,  'avito', 'EXT-2674-X', 'u', 'тестдом 2674, 2', 56.84, 60.60, "
            " 'transient_error', :cap, NOW() - INTERVAL '10 days'),"
            "(:p,  'avito', 'EXT-2674-P', 'u', 'тестдом 2674, 3', 56.84, 60.60, "
            " 'pending', 0, NULL),"
            "(:n,  'avito', 'EXT-2674-N', 'u', 'тестдом 2674, 4', 56.84, 60.60, "
            " 'pending', 0, NULL)"
        ),
        {
            "s": _H_STUCK,
            "x": _H_EXHAUSTED,
            "p": _H_PENDING,
            "n": _H_NO_PARAMS,
            # Литерал, а не hib._MAX_TRANSIENT_ATTEMPTS: подготовка теста обязана
            # работать и на СТАРОМ коде, иначе красный прогон получится про
            # отсутствующий атрибут, а не про застрявший дом. Синхронность
            # литерала и константы стережёт test_transient_attempts_counter_*.
            "cap": 3,
        },
    )
    # Объявления с rooms+area+house_type — у всех, КРОМЕ _H_NO_PARAMS.
    # source_url / dedup_hash / price_rub — NOT NULL, dedup_hash UNIQUE.
    db.execute(
        _t(
            "INSERT INTO listings (id, source, source_url, source_id, dedup_hash, price_rub, "
            "                      house_id_fk, rooms, area_m2, floor, total_floors, house_type) "
            "VALUES "
            "(:l1, 'avito', 'http://t/2674/1', 'L-2674-1', 'dh-2674-1', 5000000, :s, "
            " 2, 52.0, 3, 9, 'brick'),"
            "(:l2, 'avito', 'http://t/2674/2', 'L-2674-2', 'dh-2674-2', 5000000, :x, "
            " 2, 52.0, 3, 9, 'brick'),"
            "(:l3, 'avito', 'http://t/2674/3', 'L-2674-3', 'dh-2674-3', 5000000, :p, "
            " 2, 52.0, 3, 9, 'brick')"
        ),
        {
            "l1": 910601,
            "l2": 910602,
            "l3": 910603,
            "s": _H_STUCK,
            "x": _H_EXHAUSTED,
            "p": _H_PENDING,
        },
    )
    db.commit()


def _cleanup(db: Any) -> None:
    from sqlalchemy import text as _t

    db.rollback()
    db.execute(_t("DELETE FROM listings WHERE id IN (910601, 910602, 910603)"))
    db.execute(_t("DELETE FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"), {"ids": list(_IDS)})
    db.commit()
    db.close()


async def _run(db: Any, **kwargs: Any) -> Any:
    """Прогнать backfill с замоканным вызовом площадки.

    Мокаем ровно сетевую границу: evaluate_via_imv поднимает «адрес не найден»,
    поэтому дом уходит в not_found и путь сохранения (три таблицы) не нужен —
    проверяем именно ВЫБОРКУ. Флаг браузерного фетчера гасим, чтобы не поднимать
    сайдкар.
    """
    err = hib.IMVAddressNotFoundError("address not found (test)")
    with (
        patch.object(hib.settings, "avito_imv_use_browser_fetcher", False),
        patch.object(hib, "evaluate_via_imv", AsyncMock(side_effect=err)),
    ):
        return await hib.backfill_house_imv(db, request_delay_sec=0.0, **kwargs)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
async def test_stuck_transient_house_returns_to_the_queue_by_itself() -> None:
    """Прогон с дефолтными параметрами (как у расписания) забирает застрявший дом.

    На старом коде выборка — `WHERE imv_status = :status` с единственным
    значением 'pending', поэтому дом остаётся в 'transient_error' навсегда.
    """
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    if db.execute(_t("SELECT count(*) FROM houses")).scalar():
        pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
    try:
        _seed(db)

        # Ровно то, что вызывает расписание: batch_size из default_params, статус по умолчанию.
        result = await _run(db, batch_size=50)

        picked = {
            r.id: r.imv_status
            for r in db.execute(
                _t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
                {"ids": list(_IDS)},
            )
        }
        assert picked[_H_STUCK] != "transient_error", (
            "дом с временной ошибкой и непотраченным лимитом обязан попасть в пакет "
            f"обычного прогона; фактически статус {picked[_H_STUCK]!r} "
            f"(в пакет прогона попало домов: {result.checked})"
        )
        # pending не вытеснен повтором.
        assert picked[_H_PENDING] != "pending", "pending-дом обязан остаться в том же пакете"
        # Исчерпавший лимит в автоповтор не берётся — иначе вечный цикл.
        assert picked[_H_EXHAUSTED] == "transient_error"
        assert result.retried == 1
        # Дом без объявлений с rooms+area помечен ДО пакета, слот не потрачен.
        assert picked[_H_NO_PARAMS] == "no_params"
        assert result.premarked == 1
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
async def test_explicit_only_status_still_takes_exhausted_houses() -> None:
    """Явный only_status — ручной запрос оператора: лимит попыток его не режет."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    if db.execute(_t("SELECT count(*) FROM houses")).scalar():
        pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
    try:
        _seed(db)

        result = await _run(db, batch_size=50, only_status="transient_error")

        statuses = {
            r.id: r.imv_status
            for r in db.execute(
                _t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
                {"ids": list(_IDS)},
            )
        }
        assert statuses[_H_EXHAUSTED] != "transient_error", (
            "явно запрошенный статус обрабатывается целиком, включая исчерпавшие лимит"
        )
        # Автоповтора поверх явного запроса нет: pending не тронут.
        assert statuses[_H_PENDING] == "pending"
        assert result.retried == 0
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
async def test_transient_attempts_counter_only_counts_transient() -> None:
    """Счётчик — это «сколько раз подряд упал по временной причине», и он обнуляем.

    Без верхней границы у повтора не было бы выхода; без обнуления успехом дом,
    однажды поймавший 3 сетевых сбоя, навсегда терял бы право на повтор.
    """
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    if db.execute(_t("SELECT count(*) FROM houses")).scalar():
        pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
    try:
        _seed(db)
        assert hib._MAX_TRANSIENT_ATTEMPTS == 3, "литерал 'cap' в _seed рассинхронизирован"

        def _attempts() -> int:
            return int(
                db.execute(
                    _t("SELECT imv_transient_attempts FROM houses WHERE id = :i"),
                    {"i": _H_STUCK},
                ).scalar()
            )

        hib._mark_status(db, _H_STUCK, "transient_error", "boom")
        assert _attempts() == 1
        hib._mark_status(db, _H_STUCK, "not_found", "адрес не найден")
        assert _attempts() == 1, "нетранзиентные исходы счётчик не двигают"
        db.execute(
            _t("UPDATE houses SET imv_status='ok', imv_transient_attempts=0 WHERE id = :i"),
            {"i": _H_STUCK},
        )
        db.commit()
        assert _attempts() == 0
    finally:
        _cleanup(db)


def test_retry_queue_is_not_bound_to_only_status() -> None:
    """Статика (бежит и без БД): retry-выборка не зависит от параметра only_status.

    Ровно этот дефект и застрял на проде: единственный путь к 'transient_error'
    шёл через значение, которое никто не передавал.
    """
    retry_sql = str(hib._RETRY_QUEUE_SQL.text)
    assert "imv_status = 'transient_error'" in retry_sql
    assert ":status" not in retry_sql, "retry-очередь обязана быть независимой от only_status"
    assert "imv_transient_attempts < :max_attempts" in retry_sql, "у повтора должен быть выход"
    # psycopg v3: только CAST(:x AS type), никогда :x::type.
    for sql in (retry_sql, str(hib._QUEUE_SQL.text), str(hib._PREMARK_UNUSABLE_SQL.text)):
        assert "::" not in sql