"""honest-run-status (2026-08-15): статус прогона не должен рапортовать 'done' поверх
провала или нуля. Три прод-факта закрыты этой правкой:

  (a) avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,"blocked":1}
      -> status='done' — 89% отказов, статус зелёный. mark_backfill_finished звал
      mark_done, потому что produced=6 (>0); ни _sweep_run_did_nothing (нет
      anchors_total/errors_count у backfill'ов), ни _phase_totally_failed (ключи
      "attempted"/"failed" без фазового префикса) эту форму counters не ловили.
      Фикс: _failed_ratio_too_high внутри mark_done.

  (b) yandex_newbuilding_sweep 26.07-10.08: десять прогонов подряд 'done' при
      processed=5, succeeded=0, rows_inserted=0, failed_resolve=4-5 — сторож нулевого
      результата (_alert_if_consecutive_zero_results) слеп, т.к. _RESULT_COUNTER_KEYS
      не знал ни одного ключа этого sweep'а (total_seen/lots_fetched/unique_fetched).
      Фикс: _RESULT_COUNTER_KEYS дополнен 'succeeded'. Первая версия правки добавляла
      голые 'rows_inserted'/'processed' — ревью нашло, что 'rows_inserted' пишет ЕЩЁ
      rosreestr_dkp_import (66/67 прод-прогонов, здоровый ноль догнавшего импорта, а не
      отказ) и завёл бы непрерываемый ложный zero-стрик, а 'processed' — счётчик
      попыток (==limit даже при частичном провале у newbuilding_enrich) и маскирует
      реальные отказы. 'succeeded' пишут только yandex_newbuilding_sweep и
      newbuilding_enrich, численно совпадает с прежним 'rows_inserted' на всех
      прод-прогонах sweep'а — см. test_rosreestr_dkp_import_healthy_zero_stays_unmeasured
      и test_newbuilding_enrich_partial_failure_not_masked_by_processed ниже.

  (c) admin-витрина показывала new_count=0 у трёх подряд cian_full_load, хотя реально
      сохранено saved_inserted=482/214/239 — full-load'ы не пишут ни 'new_count', ни
      'lots_inserted'. Фикс: _column_counts дополнен saved_inserted/rows_inserted.

Проверяем на обоих модулях (kit-копия и app-копия — байт-эквивалентны по докстрингу
runs.py), тем же паттерном, что test_2625_run_that_did_nothing.py.
"""

from __future__ import annotations

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

import pytest

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

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 _capture_status(mod: Any, counters: dict[str, Any]) -> list[str]:
    """Прогнать mark_done на фейковой сессии, вернуть статусы всех UPDATE'ов.

    Тот же helper, что в test_2625_run_that_did_nothing.py — читаем СТАТУС В SQL, а не
    имя вызванной функции.
    """
    statuses: list[str] = []

    def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
        sql = str(stmt)
        for status in ("done", "failed", "banned"):
            if f"status = '{status}'" in sql:
                statuses.append(status)
        return MagicMock()

    db = MagicMock()
    db.execute.side_effect = _execute
    with patch.object(mod, "sentry_sdk", MagicMock()):
        mod.mark_done(db, 1, dict(counters))
    return statuses


def _capture_backfill_status(
    counters: dict[str, Any], *, source: str = "avito_detail_backfill", aborted: bool = False
) -> list[str]:
    """Прогнать app_runs.mark_backfill_finished на фейковой сессии (mark_done НЕ мокан —
    в отличие от test_backfill_honest_status.py, здесь важно именно его РЕАЛЬНОЕ
    поведение: mark_backfill_finished решает вызвать mark_done, а решает ли mark_done
    остаться 'done' или сам себя переквалифицировать в 'failed' — предмет этого теста).

    mark_backfill_finished есть только в app_runs (kit-копия его не держит — см.
    docstring модуля runs.py, "mark_skipped есть только здесь" — тот же принцип
    относится к продуктовым финализаторам detail-backfill'ов).
    """
    statuses: list[str] = []

    def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
        sql = str(stmt)
        for status in ("done", "failed", "banned"):
            if f"status = '{status}'" in sql:
                statuses.append(status)
        return MagicMock()

    db = MagicMock()
    db.execute.side_effect = _execute
    with patch.object(app_runs, "sentry_sdk", MagicMock()):
        app_runs.mark_backfill_finished(
            db, 1, dict(counters), source=source, aborted_by_blocks=aborted
        )
    return statuses


# ── (a) failed_ratio: прод-факт avito_detail_backfill 15.08 ─────────────────────────


def test_prod_fact_avito_15_08_no_longer_done() -> None:
    """{"attempted":64,"failed":57,"enriched":6,"blocked":1} — 89% отказов — 'failed',
    НЕ 'done'. Красный на старом коде (produced=6 != 0 -> mark_done -> 'done')."""
    counters = {"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
    assert _capture_backfill_status(counters) == ["failed"]


def test_prod_fact_avito_reason_names_the_ratio() -> None:
    reason = app_runs._failed_ratio_too_high(
        {"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
    )
    assert reason is not None
    assert "failed-ratio-honest-status" in reason
    assert "57 из 64" in reason
    assert "89%" in reason


@pytest.mark.parametrize("name", list(_MODULES))
@pytest.mark.parametrize(
    ("counters", "flagged", "why"),
    [
        ({"attempted": 64, "failed": 57}, True, "прод-факт: 89% отказов"),
        ({"attempted": 10, "failed": 5}, True, "ровно порог failed (0.5)"),
        ({"attempted": 20, "failed": 3}, True, "ровно порог degraded (0.15)"),
        ({"attempted": 20, "failed": 2}, False, "ниже порога degraded (0.10)"),
        ({"attempted": 2, "failed": 2}, False, "ratio=1.0, но < _FAILED_RATIO_MIN_ATTEMPTS"),
        ({"attempted": 0, "failed": 0}, False, "нет попыток вовсе"),
        ({"failed": 5}, False, "нет attempted — чужой словарь"),
        ({"attempted": 50}, False, "нет failed — чужой словарь"),
        ({}, False, "пустые counters"),
        (
            {"anchors_total": 5, "errors_count": 5, "lots_fetched": 0},
            False,
            "sweep-словарь (anchors_total), не detail-backfill",
        ),
    ],
)
def test_failed_ratio_classifier_boundaries(
    name: str, counters: dict[str, Any], flagged: bool, why: str
) -> None:
    reason = _MODULES[name]._failed_ratio_too_high(counters)
    assert (reason is not None) is flagged, why


# ── (5) не должен палить прогоны с малой/умеренной долей отказов ────────────────────


@pytest.mark.parametrize("name", list(_MODULES))
def test_low_failure_ratio_stays_done(name: str) -> None:
    """Штатный шум (10% отказов) не становится 'failed' — не каждый отказ диагноз."""
    counters = {"attempted": 50, "enriched": 45, "failed": 5}
    assert _capture_status(_MODULES[name], counters) == ["done"]


def test_tiny_batch_zero_produced_fails_via_old_rule_not_ratio() -> None:
    """2 попытки, обе отказали, produced=0 — доля тут не при чём (attempted < floor
    _FAILED_RATIO_MIN_ATTEMPTS, _failed_ratio_too_high вернул бы None); статус всё
    равно 'failed', но по СТАРОМУ правилу #2674 (produced==0), внутри
    mark_backfill_finished — mark_done/_failed_ratio_too_high тут не вызываются вовсе.
    Показывает, что новая проверка не дублирует и не подменяет старую."""
    counters = {"attempted": 2, "enriched": 0, "failed": 2}
    assert _capture_backfill_status(counters) == ["failed"]


def test_tiny_batch_with_partial_success_stays_done() -> None:
    """2 попытки, 1 успех, 1 отказ (ratio=0.5, но attempted < floor=3) — стрик слишком
    короткий, чтобы доля что-то значила -> остаётся 'done'."""
    counters = {"attempted": 2, "enriched": 1, "failed": 1}
    assert _capture_backfill_status(counters) == ["done"]


@pytest.mark.parametrize("name", list(_MODULES))
def test_honest_empty_sweep_unaffected_by_failed_ratio(name: str) -> None:
    """Сознательно спящее расписание (город без новостроек): sweep-словарь без
    attempted/failed вовсе -> failed_ratio не о чем судить, честная пустота остаётся
    'done' (см. также test_2625_run_that_did_nothing.py::test_honest_empty_stays_done)."""
    counters = {"anchors_total": 1, "errors_count": 0, "lots_fetched": 0}
    assert _capture_status(_MODULES[name], counters) == ["done"]


# ── (b) _RESULT_COUNTER_KEYS: прод-факт yandex_newbuilding_sweep 26.07-10.08 ─────────


def test_prod_fact_yandex_newbuilding_sweep_measured_as_zero() -> None:
    """processed=5, succeeded=0, rows_inserted=0, failed_resolve=4 — раньше
    _run_result_count возвращал None ("не измерено"); теперь — измеренный 0 (через
    'succeeded', не 'rows_inserted' — см. ниже, почему ключ переигран ревью)."""
    counters = {
        "total": 309,
        "fetchable": 200,
        "pending": 50,
        "processed": 5,
        "skipped_already_enriched": 0,
        "succeeded": 0,
        "resolved_slug": 1,
        "failed_resolve": 4,
        "failed_fetch": 0,
        "rows_inserted": 0,
        "duration_sec": 42.0,
    }
    assert app_runs._run_result_count(counters) == 0
    assert kit_runs._run_result_count(counters) == 0


def test_succeeded_is_the_measured_key_not_rows_inserted_or_processed() -> None:
    """'succeeded' читается как результат; голые 'rows_inserted'/'processed' в
    _RESULT_COUNTER_KEYS больше не участвуют (были в первой версии правки, снято
    ревью — см. test_rosreestr_dkp_import_healthy_zero_stays_unmeasured и
    test_newbuilding_enrich_partial_failure_not_masked_by_processed ниже)."""
    counters = {"processed": 5, "rows_inserted": 0}
    assert app_runs._run_result_count(counters) is None
    assert kit_runs._run_result_count(counters) is None


def test_rosreestr_dkp_import_healthy_zero_stays_unmeasured() -> None:
    """Прод-факт rosreestr_dkp_import (2026-08-15, 66 из 67 прогонов за 90д): инкрементальный
    импорт догнал источник — rows_fetched==rows_skipped, rows_inserted=0. Это ЗДОРОВЫЙ
    ответ (нечего вставлять), а не отказ; словарь не содержит 'succeeded' вовсе.

    Первая версия правки добавляла голый 'rows_inserted' в _RESULT_COUNTER_KEYS — тогда
    этот прод-факт читался бы как "измеренный провал" и копил бы практически
    непрерываемый zero-стрик (rosreestr_dkp_import не прерывается другим статусом:
    он либо 'done' с этим же нулём, либо не бежал). Ревью поймало это до деплоя —
    правильный ответ: "не измерено" (None), стрик не копится."""
    counters = {
        "last_id": 6829903,
        "batches_done": 49,
        "rows_errored": 0,
        "rows_fetched": 96974,
        "rows_skipped": 96974,
        "rows_updated": 0,
        "rows_inserted": 0,
    }
    assert app_runs._run_result_count(counters) is None
    assert kit_runs._run_result_count(counters) is None


def test_newbuilding_enrich_partial_failure_not_masked_by_processed() -> None:
    """Прод-факт newbuilding_enrich (09.08): processed=25 (счётчик ПОПЫТОК, ==limit),
    succeeded=14 — 44% отказов. Если бы сторож читал 'processed' как результат, партиальный
    провал замаскировался бы под measured-25 (сторож нулевого результата промолчал бы
    ровно там, где должен был сработать при полном провале). 'succeeded' даёт честные 14."""
    counters = {
        "failed": 11,
        "enriched": 14,
        "attempted": 25,
        "processed": 25,
        "succeeded": 14,
        "failed_fetch": 11,
    }
    assert app_runs._run_result_count(counters) == 14
    assert kit_runs._run_result_count(counters) == 14


@pytest.mark.parametrize("name", list(_MODULES))
def test_zero_result_watchdog_now_fires_for_newbuilding_sweep_streak(name: str) -> None:
    """(b) integration: 3 подряд yandex_newbuilding_sweep-подобных 'done' с succeeded=0
    -> алерт срабатывает. До фикса _RESULT_COUNTER_KEYS сторож считал результат "не
    измеренным" и молчал бы вечно (см. #2703 в docstring модуля)."""
    mod = _MODULES[name]
    row = MagicMock()
    row.status = "done"
    row.counters = {"processed": 5, "succeeded": 0, "rows_inserted": 0, "failed_resolve": 4}
    db = MagicMock()
    result = MagicMock()
    result.fetchall.return_value = [row, row, row]
    db.execute.return_value = result
    with patch.object(mod, "sentry_sdk") as mock_sentry:
        mod._alert_if_consecutive_zero_results(db, "yandex_newbuilding_sweep")
    mock_sentry.capture_message.assert_called_once()


@pytest.mark.parametrize("name", list(_MODULES))
def test_zero_result_watchdog_silent_on_rosreestr_dkp_import_streak(name: str) -> None:
    """Негативный аналог теста выше: та же лестница из 3 подряд 'done', но словарь
    rosreestr_dkp_import (нет 'succeeded') -> сторож не считает результат измеренным
    и НЕ шлёт алерт — регрессионный тест на замечание ревью (HIGH #1)."""
    mod = _MODULES[name]
    row = MagicMock()
    row.status = "done"
    row.counters = {
        "last_id": 6829903,
        "rows_fetched": 96974,
        "rows_skipped": 96974,
        "rows_inserted": 0,
    }
    db = MagicMock()
    result = MagicMock()
    result.fetchall.return_value = [row, row, row]
    db.execute.return_value = result
    with patch.object(mod, "sentry_sdk") as mock_sentry:
        mod._alert_if_consecutive_zero_results(db, "rosreestr_dkp_import")
    mock_sentry.capture_message.assert_not_called()


# ── (c) _column_counts: прод-факт cian_full_load new_count=0 при saved_inserted>0 ───


@pytest.mark.parametrize("name", list(_MODULES))
def test_prod_fact_cian_full_load_saved_inserted_surfaces_as_new_count(name: str) -> None:
    """saved_inserted=482 (прод-факт: три подряд прогона 482/214/239) — new_count
    больше не 0, хотя ключей 'new_count'/'lots_inserted' в counters нет вовсе."""
    counters = {"unique_fetched": 1200, "saved_inserted": 482, "saved_updated": 30}
    total_seen, new_count = _MODULES[name]._column_counts(counters)
    assert total_seen == 1200
    assert new_count == 482


@pytest.mark.parametrize("name", list(_MODULES))
def test_yandex_newbuilding_rows_inserted_surfaces_as_new_count(name: str) -> None:
    counters = {"rows_inserted": 7}
    _, new_count = _MODULES[name]._column_counts(counters)
    assert new_count == 7


@pytest.mark.parametrize("name", list(_MODULES))
def test_new_count_priority_unchanged_by_new_keys(name: str) -> None:
    """'new_count' явный ключ всё ещё побеждает 'lots_inserted'/'saved_inserted' —
    расширение списка не меняет приоритет уже существующих ключей."""
    counters = {"new_count": 5, "lots_inserted": 99, "saved_inserted": 1}
    _, new_count = _MODULES[name]._column_counts(counters)
    assert new_count == 5


@pytest.mark.parametrize("name", list(_MODULES))
def test_lots_inserted_still_beats_saved_inserted(name: str) -> None:
    """Порядок пикулярно НЕ переставлен для уже существующей пары — 'lots_inserted'
    (city/newbuilding-sweep'ы) проверяется раньше 'saved_inserted' (full-load'ы),
    т.к. это разные, непересекающиеся семейства источников."""
    counters = {"lots_inserted": 12, "saved_inserted": 999}
    _, new_count = _MODULES[name]._column_counts(counters)
    assert new_count == 12