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

"""#2703: сторож «ноль результатов» перестаёт читать «не измерено» как ноль.

`_alert_if_consecutive_zero_results` смотрел колонку `scrape_runs.total_seen`, у
которой DEFAULT 0. Замер на проде (2026-08-06, 53 источника):

  * 21 источник / 455 прогонов — колонку заполняет `lots_fetched` (все city-sweep'ы);
  * 4 источника / 133 прогона — пишут `unique_fetched` (full-load'ы avito/cian/yandex),
    сторож их НЕ видел, хотя у `cian_full_load` 6 из 38 успешных прогонов дали ноль;
  * 28 источников / 2650 прогонов — не пишут НИ ОДНОГО результатного ключа
    (`proxy_healthcheck` 1650, три detail-backfill'а, `deactivate_stale_*`,
    `refresh_search_matview` с буквально пустыми `{}` во всех 55 строках …).

Для последних сторож читал 0 ВСЕГДА, в том числе у полностью успешного прогона:
стрик не прерывался никогда, а анти-спам «один раз на стрик» превращался в «один
раз навсегда». Ноль записей `total_seen` в counters на всей истории — проверено.

Правка: признак берётся из counters, «не измерено» (None) стрик ПРЕРЫВАЕТ.
"""

from __future__ import annotations

import logging
import os
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, patch

import pytest

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

from scraper_kit.orchestration import runs as kit_runs

from app.services import scrape_runs as app_runs

_MODULES = {"kit": kit_runs, "app": app_runs}


def _row(status: str, counters: dict[str, Any] | None) -> SimpleNamespace:
    """Строка scrape_runs.

    `total_seen` кладём НАРОЧНО: старый сторож читал именно её, поэтому на строках
    без результатного ключа он видел ноль и копил ложный стрик. Тесты ниже
    фальсифицируются ровно этим — на старом коде они дают другой ответ.
    """
    return SimpleNamespace(status=status, counters=counters, total_seen=0)


def _db(rows: list[SimpleNamespace]) -> MagicMock:
    db = MagicMock()
    db.execute.return_value.fetchall.return_value = rows
    return db


def _run_watchdog(mod: Any, rows: list[SimpleNamespace]) -> MagicMock:
    """Прогнать сторож на заданных строках, вернуть мок sentry_sdk."""
    sentry = MagicMock()
    with patch.object(mod, "sentry_sdk", sentry):
        mod._alert_if_consecutive_zero_results(_db(rows), "src")
    return sentry


# ── 1. Различие «не измерено» и «измеренный ноль» ────────────────────────────


@pytest.mark.parametrize("name", list(_MODULES))
@pytest.mark.parametrize(
    ("counters", "expected"),
    [
        ({"lots_fetched": 0}, 0),
        ({"lots_fetched": 12}, 12),
        ({"total_seen": 7}, 7),
        ({"unique_fetched": 0}, 0),  # full-load'ы — раньше сторож их не видел
        ({"unique_fetched": 340}, 340),
        ({"attempted": 1600, "enriched": 0, "failed": 1600}, None),  # backfill
        ({}, None),  # refresh_search_matview
        (None, None),
    ],
)
def test_run_result_count(name: str, counters: dict[str, Any] | None, expected: int | None) -> None:
    """1600 попыток без результатного ключа — это None, а не ноль и не 1600."""
    assert _MODULES[name]._run_result_count(counters) == expected


# ── 2. Слепая зона больше не копит ложный стрик ──────────────────────────────


@pytest.mark.parametrize("name", list(_MODULES))
def test_source_without_result_metric_never_alerts(name: str, caplog: Any) -> None:
    """Backfill: три успешных прогона подряд, результатной метрики нет → тишина + лог.

    Фальсификация: старый сторож читал total_seen=0 у всех трёх → отправлял алерт
    и запирал анти-спам навсегда. Здесь sentry не вызывается вовсе, а слепая зона
    названа вслух в логе (раньше её признаком был вечно молчащий сторож).
    """
    mod = _MODULES[name]
    mod._warn_source_has_no_result_metric.cache_clear()
    rows = [_row("done", {"attempted": 1600, "enriched": 0}) for _ in range(4)]

    with caplog.at_level(logging.WARNING):
        sentry = _run_watchdog(mod, rows)

    sentry.capture_message.assert_not_called()
    assert "zero-result watchdog неприменим" in caplog.text
    assert "attempted, enriched" in caplog.text


@pytest.mark.parametrize("name", list(_MODULES))
def test_unmeasured_run_breaks_the_streak(name: str) -> None:
    """Два измеренных нуля + прогон без метрики → стрик прерван, алерта нет.

    Фальсификация: на старом коде все три строки имели total_seen=0 → стрик
    считался полным и алерт уходил. Это и есть «условие прерывания недостижимо»:
    теперь оно достижимо по построению — любой немеряный прогон его выполняет.
    """
    mod = _MODULES[name]
    mod._warn_source_has_no_result_metric.cache_clear()
    rows = [
        _row("done", {"lots_fetched": 0}),
        _row("done", {"lots_fetched": 0}),
        _row("done", {"deactivated": 3}),  # чужой словарь → не измерено
        _row("done", {"lots_fetched": 5}),
    ]

    assert _run_watchdog(mod, rows).capture_message.call_count == 0


# ── 3. Настоящая деградация по-прежнему видна ────────────────────────────────


@pytest.mark.parametrize("name", list(_MODULES))
@pytest.mark.parametrize("key", ["lots_fetched", "unique_fetched"])
def test_real_zero_streak_still_alerts(name: str, key: str) -> None:
    """Три измеренных нуля подряд после нормального прогона → алерт.

    `unique_fetched` (full-load'ы) добавлен этой правкой: раньше эти 4 источника
    (133 прогона) сторож не видел вовсе. Фальсификация для обоих ключей: старый
    сторож читал r.total_seen, которого в SELECT больше нет — на новых строках он
    падал бы в свой best-effort except и молчал.
    """
    mod = _MODULES[name]
    mod._warn_source_has_no_result_metric.cache_clear()
    rows = [_row("done", {key: 0}) for _ in range(3)] + [_row("done", {key: 42})]

    assert _run_watchdog(mod, rows).capture_message.call_count == 1


@pytest.mark.parametrize("name", list(_MODULES))
def test_anti_spam_still_holds_within_a_streak(name: str) -> None:
    """Четвёртый ноль подряд повторного алерта не даёт — анти-спам не сломан."""
    mod = _MODULES[name]
    mod._warn_source_has_no_result_metric.cache_clear()
    rows = [_row("done", {"lots_fetched": 0}) for _ in range(4)]

    assert _run_watchdog(mod, rows).capture_message.call_count == 0


# ── 4. Колонка total_seen тоже начинает видеть full-load'ы ───────────────────


@pytest.mark.parametrize("name", list(_MODULES))
def test_column_counts_picks_unique_fetched(name: str) -> None:
    """`_column_counts` разделяет словарь со сторожем — одна функция, два потребителя.

    Побочная польза: admin/observability перестаёт показывать total_seen=0 у
    full-load'ов (avito 70 прогонов, cian 52, yandex 4, exhaustive 7).
    """
    assert _MODULES[name]._column_counts({"unique_fetched": 340})[0] == 340
    assert _MODULES[name]._column_counts({"attempted": 5})[0] is None