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

"""Свежесть источника меряется данными, а не статусом прогона (#3172).

Дефект: `_STALE_SOURCES_SQL` считала свежесть возрастом последнего прогона со статусом
'done'. Прогон, поймавший блок, честно финализируется как 'banned' (#2657) и ПРИ ЭТОМ
вставляет строки — у domclick_city_sweep 886 строк 25.08 и 128 строк 23.08, оба
'banned'. Источник, регулярно ловящий блок и столь же регулярно приносящий данные, для
сводки был мёртв навсегда: отсюда ложный P1 #3118 «домклик не собирается с 5 августа».

Мок-строки несут ОДНУ И ТУ ЖЕ правду БД в двух кодировках: сырые поля прогона
(finished_at/status/counters), которые читает новая свёртка, и since/never_ok, которые
СТАРЫЙ код считал бы по правилу `status='done'` (`_old_rule`, а не руками). Поэтому
`git apply -R` даёт красное ПО ЗНАЧЕНИЮ, а не ImportError:
  - (а) домклик снова попадает в просроченные (старая мера смотрит на 05.08);
  - (б) источник с ежедневным 'done' и нулевой выдачей ПЕРЕСТАЁТ быть просроченным
        (старая мера засчитывает пустой 'done' как свежесть);
  - (в) источник с одними 'banned'-прогонами с данными снова «не собирал ни разу».
"""

from __future__ import annotations

import os
from datetime import UTC, datetime
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 scheduler as sched

NOW = datetime(2026, 8, 26, 8, 0, tzinfo=UTC)
CREATED = datetime(2026, 5, 1, 0, 0, tzinfo=UTC)


def _run(day: int, status: str, **counters: Any) -> dict[str, Any]:
    return {
        "finished_at": datetime(2026, 8, day, 6, 0, tzinfo=UTC),
        "status": status,
        "counters": counters,
    }


def _old_rule(runs: list[dict[str, Any]]) -> tuple[datetime, bool]:
    """Свежесть ДО правки: max(finished_at) по прогонам со статусом 'done'."""
    done = [r["finished_at"] for r in runs if r["status"] == "done"]
    return (max(done) if done else CREATED), not done


def _rows(source: str, runs: list[dict[str, Any]], interval_days: Any = None) -> list[Any]:
    """Строки запроса: прогон на строку + since/never_ok в кодировке старого правила."""
    since, never_ok = _old_rule(runs)
    return [
        SimpleNamespace(
            source=source,
            interval_days=interval_days,
            created_at=CREATED,
            since=since,
            never_ok=never_ok,
            **run,
        )
        for run in runs
    ] or [
        SimpleNamespace(
            source=source,
            interval_days=interval_days,
            created_at=CREATED,
            since=since,
            never_ok=never_ok,
            finished_at=None,
            status=None,
            counters=None,
        )
    ]


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


def _stale(rows: list[Any]) -> dict[str, sched.StaleSource]:
    sched._last_stale_digest_at = None  # сводка выходит раз в сутки — снимаем анти-спам
    with patch.object(sched, "sentry_sdk", MagicMock()):
        return {s.source: s for s in sched.emit_stale_digest(_db(rows), now=NOW)}


@pytest.fixture(autouse=True)
def _reset_digest_clock() -> Any:
    sched._last_stale_digest_at = None
    yield
    sched._last_stale_digest_at = None


# ── (а) блок с данными — это сбор, а не смерть источника ─────────────────────

DOMCLICK = [
    _run(5, "done", lots_fetched=1200, lots_inserted=430),
    _run(23, "banned", lots_fetched=310, lots_inserted=128),
    _run(25, "banned", lots_fetched=940, lots_inserted=886),
]


def test_banned_run_with_data_keeps_source_fresh() -> None:
    """Прод-случай #3118: последние прогоны 'banned', но строки собраны → НЕ просрочен."""
    assert "domclick_city_sweep" not in _stale(_rows("domclick_city_sweep", DOMCLICK))


def test_freshness_is_the_last_run_that_brought_data() -> None:
    """since = 25.08 (banned, 940 в выдаче), а не 05.08 (последний 'done')."""
    (row,) = sched.freshness_rows(_rows("domclick_city_sweep", DOMCLICK))
    assert row.since == datetime(2026, 8, 25, 6, 0, tzinfo=UTC)
    assert row.never_ok is False


# ── (б) пустой 'done' свежести не даёт ───────────────────────────────────────


def test_daily_done_runs_with_zero_results_are_stale() -> None:
    """Ежедневный 'done' с пустой выдачей: последние данные 01.08 → 25 суток просрочки."""
    runs = [_run(1, "done", lots_fetched=800, lots_inserted=210)]
    runs += [_run(day, "done", lots_fetched=0, lots_inserted=0) for day in range(2, 26)]

    stale = _stale(_rows("yandex_city_sweep", runs))

    assert "yandex_city_sweep" in stale
    assert stale["yandex_city_sweep"].age_days == pytest.approx(25.08, abs=0.1)


def test_source_without_result_metric_is_judged_by_status_as_before() -> None:
    """КОНТРОЛЬ (зелёный до и после): 'не измерено' ≠ 'ноль' — судим прежней мерой.

    refresh_search_matview и 27 других источников не пишут результатного ключа вовсе;
    строгое «данные > 0» разом объявило бы их всех просроченными навсегда.
    """
    fresh = _rows("refresh_search_matview", [_run(25, "done")])
    old = _rows("refresh_search_matview", [_run(1, "done")])
    assert "refresh_search_matview" not in _stale(fresh)
    assert "refresh_search_matview" in _stale(old)


# ── (в) never_ok согласован с той же мерой ───────────────────────────────────


def test_never_ok_counts_banned_runs_that_brought_data() -> None:
    """Источник, собиравший только сквозь баны, СОБИРАЛ — «ни разу» о нём соврало бы.

    Через сводку, а не только через `freshness_rows`: на откате правки это красное ПО
    ЗНАЧЕНИЮ (старая мера не видит ни одного 'done' → since=created_at, never_ok=True,
    просрочка 117 суток), а не «функции нет».
    """
    rows = _rows("avito_full_load", [_run(24, "banned", lots_fetched=17, lots_inserted=17)])

    assert "avito_full_load" not in _stale(rows)
    assert sched.freshness_rows(rows)[0].never_ok is False


def test_never_ok_stays_true_when_no_run_ever_brought_data() -> None:
    """Обратная сторона: прогоны есть, данных не было ни в одном → «ни разу», от created_at."""
    runs = [_run(2, "banned", lots_fetched=0), _run(9, "failed", lots_fetched=0)]

    (row,) = sched.freshness_rows(_rows("cian_history_backfill", runs))

    assert row.never_ok is True
    assert row.since == CREATED


def test_schedule_without_runs_is_never_ok() -> None:
    (row,) = sched.freshness_rows(_rows("brand_new_source", []))
    assert (row.never_ok, row.since) == (True, CREATED)


# ── дополнение к логическим тестам: сам запрос больше не фильтрует по статусу ─


def test_query_no_longer_selects_by_run_status() -> None:
    sql = str(sched._STALE_SOURCES_SQL)
    assert "status = 'done'" not in sql
    assert "LEFT JOIN scrape_runs" in sql