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

"""Smoke/shape tests for `scraper_kit.orchestration.scheduler` (#2136, post-strangler).

Kit — единственный scheduler-путь (#2397 Part C убрал legacy
`app.services.scheduler.scheduler_loop` + trigger_*/get_due_schedules/_claim_run/
reap_zombies machinery; сравнивать «golden-parity» больше не с чем). Этот файл проверяет
форму и инварианты kit-реестра самого по себе:

  1. REGISTRY: реальный `build_product_handlers()` (НЕ stub) + `build_registry` покрывает
     каждый canonical scheduled source (kit-native + продуктовые + deactivate_stale_*
     wildcard-члены); неизвестный source → None.
  2. CLAIM: `_claim_run` — happy-path, already-running skip, lock-busy skip, appeared-under-
     lock rollback. Мокаем БД + ctx.runs.
  3. ZOMBIE: `reap_zombies` возвращает число + commit.
  4. SUB-HOURLY: `reschedule_after_minutes` post_claim UPDATE + commit (proxy_healthcheck).
  5. DISPATCH e2e: `_dispatch` для kit-native sweep (мок pipeline) и продуктового handler —
     claim → свежая сессия → job вызван → close.

Без сети, без БД.
"""

from __future__ import annotations

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

import pytest

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

from scraper_kit.orchestration import scheduler as kit_sched
from scraper_kit.orchestration.pipeline import CITY_ANCHORS, DEFAULT_REGION_CODE
from scraper_kit.orchestration.scheduler import (
    Handler,
    SchedulerContext,
    _claim_run,
    _dispatch,
    _job_avito_city_sweep,
    _job_avito_full_load,
    _job_avito_full_load_exhaustive,
    _job_cian_city_sweep,
    _job_cian_full_load,
    _job_domclick_city_sweep,
    _job_yandex_city_sweep,
    _resolve_region_code,
    build_registry,
    reap_zombies,
    reschedule_after_minutes,
    resolve_handler,
)

# ── продуктовые source'ы (НЕ kit-native) — CANONICAL, поддерживается ВРУЧНУЮ ─────────
# Источник истины: app.services.product_handlers.build_product_handlers() keys (минус
# wildcard "deactivate_stale_*", раскрытый в 3 конкретных member-source'а) + семейство
# scrape_schedules-сидов (data/sql/*scrape_schedules*seed*.sql, 158_seed_proxy_healthcheck_
# schedule.sql, 162_seed_deals_freshness_monitor.sql). Этот список НЕ выводится из кода —
# добавил/убрал источник в product_handlers.py? Обнови и здесь, иначе
# test_real_build_product_handlers_covers_all_scheduled_sources ничего не поймает.
# (2026-07-04 deep-review #2397 Part C: список был устаревшим — не хватало
# cian_history_backfill/deals_freshness_monitor/osm_poi_ekb_refresh; исправлено.)
_PRODUCT_SOURCES: set[str] = {
    "cian_history_backfill",
    "rosreestr_dkp_import",
    # #3051 п.3: member-source семейства "rosreestr_dkp_import_*" (Москва, миграция 288
    # — то же раскрытие wildcard в конкретный member, что deactivate_stale_avito/yandex/
    # cian ниже, а не сам wildcard "rosreestr_dkp_import_*").
    "rosreestr_dkp_import_77",
    "listing_source_snapshot",
    "asking_to_sold_ratio_refresh",
    "deal_city_price_bands_refresh",
    "refresh_search_matview",
    "yandex_address_backfill",
    "deactivate_stale_avito",
    "deactivate_stale_yandex",
    "deactivate_stale_cian",
    "sber_index_pull",
    "rosreestr_quarter_poll",
    "deals_freshness_monitor",
    "landing_stats_refresh",
    "newbuilding_enrich",
    "yandex_newbuilding_sweep",
    "geoportal_coords_backfill",
    "house_coords_from_listings",
    "geocode_missing_listings",
    "avito_detail_backfill",
    "yandex_detail_backfill",
    "cadastral_geo_match",
    "osm_poi_ekb_refresh",
    "house_imv_backfill",
    "house_dedup_merge",
    "proxy_healthcheck",
}

# kit-native (тело в scraper_kit.orchestration.pipeline) — регистрируются встроенно.
# "*_city_sweep_*" — wildcard-семья oblast B1 rollout (region 66, dormant, #179):
# per-city source (avito_city_sweep_nizhniy_tagil и т.п.) резолвится на тот же
# handler, что EKB source, через resolve_handler-префикс (см. _default_kit_handlers).
_KIT_NATIVE_SOURCES = {
    "avito_city_sweep",
    "avito_city_sweep_*",
    "avito_full_load",
    "avito_full_load_exhaustive",
    "avito_newbuilding_sweep",
    "yandex_city_sweep",
    "yandex_city_sweep_*",
    "cian_city_sweep",
    "cian_city_sweep_*",
    "cian_full_load",
    "domclick_city_sweep",
    "domclick_city_sweep_*",
}


def _make_sched(source: str) -> dict[str, Any]:
    return {
        "id": 1,
        "source": source,
        "enabled": True,
        "window_start_hour": 2,
        "window_end_hour": 5,
        "default_params": {},
        "last_run_id": None,
        "last_run_at": None,
        "next_run_at": None,
    }


# ── kit-реестр с продуктовыми handler'ами (recording stubs) ──────────────────


def _build_recording_registry() -> tuple[dict[str, Handler], dict[str, MagicMock]]:
    """Kit-native sweeps + продуктовые recording-handler'ы. Возвращает (registry, fired-map)."""
    fired: dict[str, MagicMock] = {}

    def _make_job(name: str) -> Any:
        rec = MagicMock(name=name)
        fired[name] = rec

        async def _job(db: Any, run_id: int, params: dict[str, Any], ctx: Any) -> None:
            rec(db, run_id, params)

        return _job

    product: dict[str, Handler] = {}
    # продуктовые exact-match источники
    for src in _PRODUCT_SOURCES:
        if src in _KIT_NATIVE_SOURCES or src.startswith("deactivate_stale_"):
            continue
        product[src] = Handler(_make_job(src), src)
    # deactivate_stale_* — одна wildcard-запись (как боевой startswith)
    product["deactivate_stale_*"] = Handler(_make_job("deactivate_stale"), "deactivate_stale")

    return build_registry(product), fired


# ── 1. REGISTRY shape ────────────────────────────────────────────────────────


def test_real_build_product_handlers_covers_all_scheduled_sources() -> None:
    """Реальный (не stub!) build_product_handlers покрывает КАЖДЫЙ canonical source.

    2026-07-04 deep-review #2397 Part C: старая версия этого теста строила registry ИЗ
    `_PRODUCT_SOURCES` через recording-stub (`_build_recording_registry`) и проверяла, что
    тот же набор резолвится — тавтология, ничего не гоняющая через реальный код. Эта версия
    зовёт настоящий `product_handlers.build_product_handlers(ctx)` — если кто-то уронит
    handler-entry в product_handlers.py (или source добавили в scrape_schedules seed, но
    забыли завести handler), тест падает.

    ctx=None безопасен: build_product_handlers его не замыкает при сборке dict (сами job'ы
    получают ctx только во время dispatch) — см. её докстринг + test_deals_freshness_monitor.
    """
    from app.services.product_handlers import build_product_handlers

    real_registry = build_registry(build_product_handlers(ctx=None))  # type: ignore[arg-type]

    for source in _PRODUCT_SOURCES | _KIT_NATIVE_SOURCES:
        assert resolve_handler(source, real_registry) is not None, (
            f"real build_product_handlers()/build_registry() misses source={source}"
        )


def test_kit_native_handler_set() -> None:
    """Ровно 12 kit-native sweep-обработчиков зарегистрированы встроенно (8 + 4 wildcard).

    "*_city_sweep_*" — per-city/per-region source'ы вне ЕКБ: avito/cian/yandex завела
    oblast B1 rollout (region 66, #179), domclick — регулярный сбор 77/50 (миграция 306),
    у него суффикс несёт регион, а не город.
    """
    registry = build_registry()
    assert set(registry) == _KIT_NATIVE_SOURCES
    for src in _KIT_NATIVE_SOURCES:
        assert resolve_handler(src, registry).log_name == src


def test_domclick_region_wildcard_resolves_to_product_handler() -> None:
    """'domclick_city_sweep_moskva' резолвится на ПРОДУКТОВЫЙ handler, не на kit-native.

    Развилка не косметическая: kit-native тело ходит на BFF без кук сессии Sber ID, и
    московская строка, севшая на него, упёрлась бы в QRATOR ровно так, как было до #3264.
    Продуктовый ключ обязан перекрывать kit-native и на wildcard, а не только на точном
    'domclick_city_sweep' — иначе перекрытие есть у ЕКБ и молча отсутствует у регионов.
    """
    from app.services.product_handlers import (
        _job_domclick_city_sweep as product_job,
    )
    from app.services.product_handlers import build_product_handlers

    # ctx=None безопасен — job'ы его не замыкают при сборке dict (см. тест выше).
    registry = build_registry(build_product_handlers(ctx=None))  # type: ignore[arg-type]
    for src in (
        "domclick_city_sweep",
        "domclick_city_sweep_moskva",
        "domclick_city_sweep_moskovskaya_oblast",
    ):
        handler = resolve_handler(src, registry)
        assert handler is not None, src
        assert handler.job is product_job, src


def test_unknown_source_resolves_none() -> None:
    registry, _ = _build_recording_registry()
    assert resolve_handler("totally_unknown_source", registry) is None


def test_kit_scheduler_has_no_app_imports() -> None:
    """kit scheduler развязан от app.* — ни одного `import app` / `from app` (AST, не substring)."""
    import ast
    import inspect

    source = inspect.getsource(kit_sched)
    tree = ast.parse(source)
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                assert not alias.name.startswith("app"), f"illegal import app: {alias.name}"
        elif isinstance(node, ast.ImportFrom):
            assert not (node.module or "").startswith("app"), f"illegal from app: {node.module}"


def test_deactivate_stale_wildcard_prefix() -> None:
    """deactivate_stale_{avito,yandex,cian} → одна wildcard-запись (как боевой startswith)."""
    registry, _ = _build_recording_registry()
    h_avito = resolve_handler("deactivate_stale_avito", registry)
    h_yandex = resolve_handler("deactivate_stale_yandex", registry)
    h_cian = resolve_handler("deactivate_stale_cian", registry)
    assert h_avito is not None
    assert h_avito is h_yandex is h_cian  # один и тот же handler на всё семейство


def test_rosreestr_dkp_wildcard_resolves() -> None:
    """#3051 п.3: 'rosreestr_dkp_import_77' (Москва) резолвится через wildcard-запись

    'rosreestr_dkp_import_*' — в реестре НЕТ отдельного exact-match ключа под
    каждый region_code-суффикс (тот же механизм, что deactivate_stale_*/
    avito_city_sweep_*). Точное имя БЕЗ суффикса ('rosreestr_dkp_import', регион 66)
    резолвится СВОЕЙ exact-match записью, не wildcard'ом.
    """
    from app.services.product_handlers import _job_rosreestr_dkp, build_product_handlers

    real_registry = build_registry(build_product_handlers(ctx=None))  # type: ignore[arg-type]

    h77 = resolve_handler("rosreestr_dkp_import_77", real_registry)
    assert h77 is real_registry["rosreestr_dkp_import_*"]
    assert h77 is not None and h77.job is _job_rosreestr_dkp

    h_exact = resolve_handler("rosreestr_dkp_import", real_registry)
    assert h_exact is real_registry["rosreestr_dkp_import"]
    assert h_exact is not real_registry["rosreestr_dkp_import_*"]


# ── 2. _claim_run advisory-lock parity ───────────────────────────────────────


class _FakeResult:
    def __init__(self, *, scalar: Any = None, fetchone: Any = None, fetchall: Any = None) -> None:
        self._scalar = scalar
        self._fetchone = fetchone
        self._fetchall = fetchall or []

    def scalar(self) -> Any:
        return self._scalar

    def fetchone(self) -> Any:
        return self._fetchone

    def fetchall(self) -> Any:
        return self._fetchall


class _FakeClaimDB:
    """Мок Session для _claim_run: сценарий has_running_run + advisory-lock."""

    def __init__(self, *, running_states: list[bool], lock: bool = True) -> None:
        # running_states: очередь ответов has_running_run (по вызовам SELECT 1 FROM scrape_runs)
        self._running = list(running_states)
        self._lock = lock
        self.committed = False
        self.rolled_back = False
        self.update_calls = 0
        # #2658: строки-пропуски (scrape_runs status='skipped'), которые пишет mark_skipped.
        self.skip_rows = 0

    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
        sql = str(stmt)
        if "SELECT 1 FROM scrape_runs" in sql:
            is_running = self._running.pop(0)
            return _FakeResult(fetchone=(1,) if is_running else None)
        if "pg_try_advisory_xact_lock" in sql:
            return _FakeResult(scalar=self._lock)
        if "UPDATE scrape_schedules" in sql:
            self.update_calls += 1
            return _FakeResult()
        # #2658 mark_skipped: сначала пробует схлопнуть последнюю skip-строку (UPDATE
        # ... RETURNING), не нашёл — вставляет новую.
        if "UPDATE scrape_runs r" in sql:
            return _FakeResult(fetchone=None)
        if "INSERT INTO scrape_runs" in sql:
            self.skip_rows += 1
            return _FakeResult(fetchone=MagicMock(id=999))
        return _FakeResult()

    def commit(self) -> None:
        self.committed = True

    def rollback(self) -> None:
        self.rolled_back = True


def _ctx_with_runs(create_run_ret: int = 42) -> SchedulerContext:
    runs = MagicMock()
    runs.create_run = MagicMock(return_value=create_run_ret)
    return SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(),
        runs=runs,
    )


def test_claim_run_happy_path() -> None:
    db = _FakeClaimDB(running_states=[False, False], lock=True)
    ctx = _ctx_with_runs(create_run_ret=42)
    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
    assert run_id == 42
    ctx.runs.create_run.assert_called_once()
    assert db.update_calls == 1
    assert db.committed is True


def test_claim_run_skip_already_running() -> None:
    db = _FakeClaimDB(running_states=[True], lock=True)
    ctx = _ctx_with_runs()
    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
    assert run_id is None
    ctx.runs.create_run.assert_not_called()
    assert db.skip_rows == 1  # #2658: пропуск оставляет строку, а не только лог


def test_claim_run_skip_lock_busy() -> None:
    # pre-check clean, но advisory-lock занят конкурентным тиком → skip
    db = _FakeClaimDB(running_states=[False], lock=False)
    ctx = _ctx_with_runs()
    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
    assert run_id is None
    ctx.runs.create_run.assert_not_called()
    assert db.skip_rows == 1  # #2658


def test_claim_run_running_appeared_under_lock() -> None:
    # pre-check clean, lock взят, но под локом появился running → rollback + skip (double-check)
    db = _FakeClaimDB(running_states=[False, True], lock=True)
    ctx = _ctx_with_runs()
    run_id = _claim_run(db, _make_sched("avito_city_sweep"), ctx)
    assert run_id is None
    assert db.rolled_back is True
    ctx.runs.create_run.assert_not_called()
    assert db.skip_rows == 1  # #2658


# ── 3. reap_zombies ──────────────────────────────────────────────────────────


def test_reap_zombies_counts_and_commits() -> None:
    db = MagicMock()
    db.execute.return_value = _FakeResult(fetchall=[MagicMock(id=1), MagicMock(id=2)])
    n = reap_zombies(db)
    assert n == 2
    db.commit.assert_called_once()
    # порог 6 часов передан в SQL-параметр
    _stmt, params = db.execute.call_args[0]
    assert params["interval"] == "6 hours"


def test_reap_zombies_threshold_constant() -> None:
    assert kit_sched.ZOMBIE_THRESHOLD_HOURS == 6


# ── 4. proxy_healthcheck sub-hourly post_claim ───────────────────────────────


def test_reschedule_after_minutes_updates_next_run() -> None:
    db = MagicMock()
    hook = reschedule_after_minutes(param="interval_minutes", default=30)
    ctx = _ctx_with_runs()
    hook(db, run_id=7, params={"interval_minutes": 15}, ctx=ctx)
    db.commit.assert_called_once()
    _stmt, params = db.execute.call_args[0]
    assert params["mins"] == 15
    assert params["rid"] == 7


def test_reschedule_after_minutes_default() -> None:
    db = MagicMock()
    hook = reschedule_after_minutes()
    hook(db, run_id=9, params={}, ctx=_ctx_with_runs())
    _stmt, params = db.execute.call_args[0]
    assert params["mins"] == 30


# ── 5. _dispatch end-to-end (claim → fresh session → job → close) ────────────


async def test_dispatch_kit_native_sweep_invokes_pipeline() -> None:
    """kit-native handler: _dispatch клеймит, открывает сессию и зовёт pipeline.run_*."""
    run_db = MagicMock()
    ctx = SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(return_value=run_db),
        runs=MagicMock(),
    )
    ctx.runs.create_run = MagicMock(return_value=100)

    db = _FakeClaimDB(running_states=[False, False], lock=True)
    registry = build_registry()
    handler = registry["avito_city_sweep"]

    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        run_id = await _dispatch(handler, db, _make_sched("avito_city_sweep"), ctx)
        assert run_id == 100
        # дождаться detached run-задачи
        await asyncio.gather(*list(ctx._inflight_tasks))

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs["run_id"] == 100
    assert kwargs["config"] is ctx.config
    assert kwargs["matcher"] is ctx.matcher
    run_db.close.assert_called_once()


async def test_dispatch_product_handler_invokes_job() -> None:
    """Продуктовый handler: _dispatch клеймит и зовёт инжектированный job на свежей сессии."""
    run_db = MagicMock()
    ctx = SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(return_value=run_db),
        runs=MagicMock(),
    )
    ctx.runs.create_run = MagicMock(return_value=200)

    db = _FakeClaimDB(running_states=[False, False], lock=True)
    registry, fired = _build_recording_registry()
    handler = resolve_handler("sber_index_pull", registry)

    run_id = await _dispatch(handler, db, _make_sched("sber_index_pull"), ctx)
    assert run_id == 200
    await asyncio.gather(*list(ctx._inflight_tasks))

    fired["sber_index_pull"].assert_called_once()
    called_args = fired["sber_index_pull"].call_args[0]
    assert called_args[0] is run_db  # свежая сессия
    assert called_args[1] == 200  # run_id
    run_db.close.assert_called_once()


async def test_dispatch_pre_claim_gate_skips() -> None:
    """pre_claim → False: _dispatch пропускает claim (cian cookie-gate семантика)."""
    ctx = _ctx_with_runs()

    async def _deny(db: Any, sch: dict[str, Any], c: Any) -> bool:
        return False

    job = AsyncMock()

    async def _job(db: Any, run_id: int, params: dict[str, Any], c: Any) -> None:
        job()

    handler = Handler(_job, "cian_history_backfill", pre_claim=_deny)
    db = _FakeClaimDB(running_states=[], lock=True)
    run_id = await _dispatch(handler, db, _make_sched("cian_history_backfill"), ctx)
    assert run_id is None
    ctx.runs.create_run.assert_not_called()
    job.assert_not_called()


async def test_dispatch_post_claim_hook_runs() -> None:
    """post_claim вызывается сразу после claim (proxy_healthcheck sub-hourly reschedule)."""
    run_db = MagicMock()
    ctx = SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(return_value=run_db),
        runs=MagicMock(),
    )
    ctx.runs.create_run = MagicMock(return_value=300)
    post = MagicMock()

    async def _job(db: Any, run_id: int, params: dict[str, Any], c: Any) -> None:
        pass

    handler = Handler(_job, "proxy_healthcheck", post_claim=post)
    db = _FakeClaimDB(running_states=[False, False], lock=True)
    run_id = await _dispatch(handler, db, _make_sched("proxy_healthcheck"), ctx)
    assert run_id == 300
    await asyncio.gather(*list(ctx._inflight_tasks))
    post.assert_called_once()
    assert post.call_args[0][1] == 300  # run_id


# ── 6. B1 oblast rollout — city param → city anchors, wildcard dispatch ──────────


def _oblast_ctx() -> SchedulerContext:
    return SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(),
        runs=MagicMock(),
    )


async def test_job_avito_city_sweep_with_city_param_uses_city_anchors() -> None:
    """params={"city": "nizhniy_tagil"} → Н.Тагил anchors переданы в run_avito_city_sweep."""
    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        await _job_avito_city_sweep(MagicMock(), 1, {"city": "nizhniy_tagil"}, _oblast_ctx())

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs["anchors"] == CITY_ANCHORS["nizhniy_tagil"]
    assert kwargs["anchors"] == [(57.910, 59.980, "Н.Тагил центр")]


async def test_job_cian_city_sweep_with_city_param_uses_city_anchors() -> None:
    """params={"city": "kamensk_uralskiy"} → anchors города переданы в run_cian_city_sweep."""
    with patch.object(kit_sched, "run_cian_city_sweep", AsyncMock()) as mock_run:
        await _job_cian_city_sweep(MagicMock(), 1, {"city": "kamensk_uralskiy"}, _oblast_ctx())

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs["anchors"] == CITY_ANCHORS["kamensk_uralskiy"]


async def test_job_yandex_city_sweep_with_city_param_uses_city_anchors() -> None:
    """params={"city": "serov"} → anchors города переданы в run_yandex_city_sweep."""
    with patch.object(kit_sched, "run_yandex_city_sweep", AsyncMock()) as mock_run:
        await _job_yandex_city_sweep(MagicMock(), 1, {"city": "serov"}, _oblast_ctx())

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs["anchors"] == CITY_ANCHORS["serov"]


async def test_job_avito_city_sweep_without_city_param_falls_back_to_none() -> None:
    """Без city (EKB-schedule, back-compat) — anchors=None передан (run_* сам берёт EKB_ANCHORS)."""
    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        await _job_avito_city_sweep(MagicMock(), 1, {}, _oblast_ctx())

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs["anchors"] is None


async def test_job_avito_city_sweep_unknown_city_raises_instead_of_sweeping() -> None:
    """Неизвестный city slug (опечатка в расписании) обязан упасть, а не собрать ЕКБ.

    Раньше `get_city_anchors` отдавал None, и свип уходил на EKB_ANCHORS: опечатка в
    `default_params.city` тихо давала прогон по Екатеринбургу, проштампованный чужим
    городом. Теперь реестр падает с ValueError; прогон крашится, а тик планировщика
    это ловит (`logger.exception` в `_run`, scheduler.py) — цикл не умирает, но и
    мусора в корпус не попадает.
    """
    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        with pytest.raises(ValueError, match="nonexistent_town"):
            await _job_avito_city_sweep(MagicMock(), 1, {"city": "nonexistent_town"}, _oblast_ctx())

    mock_run.assert_not_awaited()


# ── #2616: _job_avito_full_load(_exhaustive) прокидывает ctx.proxy_provider ──────
#
# run_avito_full_load — единственный avito sweep, где было НЕ прокинуто proxy_provider
# вообще (ни в сигнатуре run_avito_full_load, ни на call-site здесь) — asymmetric
# с _job_avito_city_sweep/_job_avito_newbuilding_sweep (proxy_provider=ctx.proxy_provider
# уже были) и с _job_cian_full_load (уже пробрасывал). Root cause 13/14 banned avito_full_load
# прогонов за 14 дней (issue #2613/#2616).


def _proxy_ctx(sentinel: object) -> SchedulerContext:
    return SchedulerContext(
        config=MagicMock(),
        matcher=MagicMock(),
        enrichment=MagicMock(),
        session_factory=MagicMock(),
        runs=MagicMock(),
        proxy_provider=sentinel,
    )


async def test_job_avito_full_load_passes_proxy_provider() -> None:
    """_job_avito_full_load(ctx) → run_avito_full_load(..., proxy_provider=ctx.proxy_provider).

    Falsification: если scheduler.py перестанет прокидывать proxy_provider=ctx.proxy_provider
    в run_avito_full_load(...), kwargs['proxy_provider'] не будет `sentinel` (либо ключа не
    будет вовсе) — assert падает на VALUE, не на TypeError (AsyncMock не проверяет сигнатуру).
    """
    sentinel = object()
    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
        await _job_avito_full_load(MagicMock(), 1, {}, _proxy_ctx(sentinel))

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs.get("proxy_provider") is sentinel


async def test_job_avito_full_load_exhaustive_passes_proxy_provider() -> None:
    """_job_avito_full_load_exhaustive(ctx) →

    run_avito_full_load(..., proxy_provider=ctx.proxy_provider).
    """
    sentinel = object()
    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
        await _job_avito_full_load_exhaustive(MagicMock(), 1, {}, _proxy_ctx(sentinel))

    mock_run.assert_awaited_once()
    _args, kwargs = mock_run.call_args
    assert kwargs.get("proxy_provider") is sentinel


def test_wildcard_resolves_oblast_city_sweep_sources() -> None:
    """resolve_handler матчит per-city oblast source на wildcard "*_city_sweep_*"."""
    registry = build_registry()
    assert (
        resolve_handler("avito_city_sweep_nizhniy_tagil", registry)
        is registry["avito_city_sweep_*"]
    )
    assert resolve_handler("cian_city_sweep_serov", registry) is registry["cian_city_sweep_*"]
    assert (
        resolve_handler("yandex_city_sweep_pervouralsk", registry)
        is registry["yandex_city_sweep_*"]
    )
    # EKB source (no city) — по-прежнему резолвится на exact non-wildcard entry.
    assert resolve_handler("avito_city_sweep", registry) is registry["avito_city_sweep"]
    assert registry["avito_city_sweep"] is not registry["avito_city_sweep_*"]


# ── region_code: default_params → run_*-функции pipeline.py ─────────────────────────
#
# Планировщик кита раньше НЕ читал default_params["region_code"] ни разу — все sweep'ы
# писали листings регионом 66 (DEFAULT_REGION_CODE) независимо от того, что лежит в
# scrape_schedules.default_params. Это единственная причина, по которой московский обход
# нельзя было завести конфигом — pipeline.py уже принимает region_code параметром,
# scheduler.py его просто не прокидывал. Ниже: (1) отсутствие ключа — БЕЗ изменений
# поведения для всех существующих строк расписания (default 66); (2) валидный код (77)
# доезжает до run_*; (3) мусорное значение — явный отказ ДО сетевого похода, а не тихая
# запись в listings.region_code.


def test_resolve_region_code_missing_key_defaults_to_66() -> None:
    """Нет ключа в default_params → DEFAULT_REGION_CODE (66) — прежнее поведение всех
    123 существующих строк расписания не меняется."""
    assert _resolve_region_code({}) == DEFAULT_REGION_CODE == 66


def test_resolve_region_code_known_value_passes_through() -> None:
    assert _resolve_region_code({"region_code": 77}) == 77


def test_resolve_region_code_string_digit_coerces_to_int() -> None:
    """jsonb default_params может прислать региона строкой — тот же int(...), что и
    у остальных числовых параметров этого файла (params.get("pages", ...) и т.п.)."""
    assert _resolve_region_code({"region_code": "77"}) == 77


def test_resolve_region_code_unknown_value_raises() -> None:
    """Регион вне реестра — явный ValueError, а не тихая запись мусора в listings."""
    with pytest.raises(ValueError, match="реестре регионов"):
        _resolve_region_code({"region_code": 999})


def test_resolve_region_code_non_numeric_raises() -> None:
    with pytest.raises(ValueError, match="не приводится к int"):
        _resolve_region_code({"region_code": "москва"})


async def test_job_avito_city_sweep_region_code_absent_defaults() -> None:
    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        await _job_avito_city_sweep(MagicMock(), 1, {}, _oblast_ctx())

    _args, kwargs = mock_run.call_args
    assert kwargs["region_code"] == DEFAULT_REGION_CODE


async def test_job_avito_city_sweep_region_code_from_params() -> None:
    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        await _job_avito_city_sweep(MagicMock(), 1, {"region_code": 77}, _oblast_ctx())

    _args, kwargs = mock_run.call_args
    assert kwargs["region_code"] == 77


async def test_job_avito_city_sweep_region_code_garbage_raises_before_run() -> None:
    """Мусорный region_code роняет job ДО вызова run_* — сеть/скрапинг не трогаются."""
    with patch.object(kit_sched, "run_avito_city_sweep", AsyncMock()) as mock_run:
        with pytest.raises(ValueError):
            await _job_avito_city_sweep(MagicMock(), 1, {"region_code": 12345}, _oblast_ctx())

    mock_run.assert_not_awaited()


async def test_job_avito_full_load_region_code_from_params() -> None:
    with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run:
        await _job_avito_full_load(MagicMock(), 1, {"region_code": 77}, _oblast_ctx())

    _args, kwargs = mock_run.call_args
    assert kwargs["region_code"] == 77


async def test_job_domclick_city_sweep_region_code_absent_defaults() -> None:
    """Регионный проброс не должен ломать уже читаемый city_id (тот же params dict)."""
    with patch.object(kit_sched, "run_domclick_city_sweep", AsyncMock()) as mock_run:
        await _job_domclick_city_sweep(MagicMock(), 1, {"city_id": 9}, _oblast_ctx())

    _args, kwargs = mock_run.call_args
    assert kwargs["region_code"] == DEFAULT_REGION_CODE
    assert kwargs["city_id"] == 9


async def test_job_domclick_city_sweep_region_code_from_params() -> None:
    with patch.object(kit_sched, "run_domclick_city_sweep", AsyncMock()) as mock_run:
        await _job_domclick_city_sweep(MagicMock(), 1, {"region_code": 77}, _oblast_ctx())

    _args, kwargs = mock_run.call_args
    assert kwargs["region_code"] == 77


async def test_job_cian_full_load_region_code_absent_defaults() -> None:
    with patch.object(kit_sched, "run_cian_full_load", AsyncMock()) as mock_run:
        await _job_cian_full_load(MagicMock(), 1, {}, _oblast_ctx())

    _args, kwargs = mock_run.call_args
    assert kwargs["region_code"] == DEFAULT_REGION_CODE