"""Offline smoke for scheduler logic."""

import asyncio
import os
from unittest.mock import AsyncMock, MagicMock

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

from datetime import UTC, datetime

import pytest

import app.services.scheduler as sched
from app.core import shutdown as sd
from app.services.scheduler import ZOMBIE_THRESHOLD_HOURS, compute_next_run_at


@pytest.fixture(autouse=True)
def _reset_drain_state() -> None:
    """#1182 P2: shutdown-флаг и registry детей — module-global; чистим вокруг каждого
    теста (изоляция от detached-задач, что триггеры спавнят в _inflight_tasks)."""
    sd.reset_shutdown()
    sched._inflight_tasks.clear()
    yield
    sched._inflight_tasks.clear()
    sd.reset_shutdown()


def test_compute_next_run_in_normal_window() -> None:
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    next_at = compute_next_run_at(2, 5, now=now)
    assert next_at.date() == datetime(2026, 5, 24).date()
    assert 2 <= next_at.hour <= 4  # window [2, 5) UTC


def test_compute_next_run_cross_midnight() -> None:
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    # Window 22..3 = 22:00-23:59 + 00:00-02:59
    next_at = compute_next_run_at(22, 3, now=now)
    # Должен попасть либо в [22,23] либо в [0,3) range
    hour = next_at.hour
    assert hour >= 22 or hour < 3


def test_compute_next_run_window_1_hour() -> None:
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    # Минимальное окно [3, 4) → ровно 3-й час
    next_at = compute_next_run_at(3, 4, now=now)
    assert next_at.hour == 3


def test_zombie_threshold_constant() -> None:
    assert ZOMBIE_THRESHOLD_HOURS > 0


def test_compute_next_run_is_future() -> None:
    """next_run_at всегда в будущем (завтра) — не в прошлом."""
    now = datetime(2026, 5, 23, 3, 30, tzinfo=UTC)
    next_at = compute_next_run_at(2, 5, now=now)
    assert next_at > now


def test_compute_next_run_timezone_aware() -> None:
    """Результат всегда timezone-aware (UTC)."""
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    next_at = compute_next_run_at(2, 5, now=now)
    assert next_at.tzinfo is not None


# ── per-source interval_days cadence (weekly exhaustive avito) ──────────────


def test_compute_next_run_default_interval_is_daily() -> None:
    """Без interval_days → +1 сутки (back-compat, как раньше)."""
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    next_at = compute_next_run_at(13, 15, now=now)
    assert next_at.date() == datetime(2026, 5, 24).date()
    assert 13 <= next_at.hour <= 14  # window [13, 15) UTC


def test_compute_next_run_interval_days_1_explicit() -> None:
    """interval_days=1 явно → +1 сутки (идентично default)."""
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    next_at = compute_next_run_at(13, 15, now=now, interval_days=1)
    assert next_at.date() == datetime(2026, 5, 24).date()
    assert 13 <= next_at.hour <= 14


def test_compute_next_run_interval_days_7_weekly() -> None:
    """interval_days=7 → ~7 суток вперёд, всё ещё в окне."""
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    next_at = compute_next_run_at(13, 15, now=now, interval_days=7)
    assert next_at.date() == datetime(2026, 5, 30).date()  # 23 + 7
    assert 13 <= next_at.hour <= 14  # всё ещё внутри окна [13, 15) UTC
    assert next_at > now


def test_compute_next_run_interval_days_cross_midnight_weekly() -> None:
    """interval_days с cross-midnight окном → целевой день +N, час в окне."""
    now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC)
    next_at = compute_next_run_at(22, 3, now=now, interval_days=7)
    assert next_at > now
    assert next_at.hour >= 22 or next_at.hour < 3


# ── #750: атомарный _claim_run (advisory lock — anti-double-sweep) ──────────

from collections import deque  # noqa: E402

from app.services import scheduler as _sched  # noqa: E402


class _Scalar:
    def __init__(self, v: object) -> None:
        self._v = v

    def scalar(self) -> object:
        return self._v


class _Fetchone:
    def __init__(self, row: object) -> None:
        self._row = row

    def fetchone(self) -> object:
        return self._row


class _FakeSchedulerDB:
    """Stateful fake моделирующий advisory-lock claim (#750), без реальной БД.

    - SELECT pg_try_advisory_xact_lock → lock_available.
    - SELECT 1 FROM scrape_runs (has_running_run) → из running_results (если задано),
      иначе по флагу self.running[source].
    - UPDATE scrape_schedules → no-op.
    create_run (монкипатчится) выставляет self.running[source]=True.
    """

    def __init__(self, *, lock_available: bool = True, running_results: list[bool] | None = None):
        self.lock_available = lock_available
        self.running: dict[str, bool] = {}
        self._running_results: deque[bool] = deque(running_results or [])
        self.commits = 0
        self.rollbacks = 0

    def execute(self, stmt: object, params: dict | None = None):
        sql = str(stmt)
        if "pg_try_advisory_xact_lock" in sql:
            return _Scalar(self.lock_available)
        if "SELECT 1 FROM scrape_runs" in sql:
            if self._running_results:
                return _Fetchone(object() if self._running_results.popleft() else None)
            src = (params or {}).get("source")
            return _Fetchone(object() if self.running.get(src) else None)
        return _Fetchone(None)  # UPDATE scrape_schedules etc.

    def commit(self) -> None:
        self.commits += 1

    def rollback(self) -> None:
        self.rollbacks += 1


_SCHED_ROW = {
    "source": "avito_city_sweep",
    "window_start_hour": 2,
    "window_end_hour": 5,
    "default_params": {},
}


def test_claim_run_double_call_creates_one_run(monkeypatch: pytest.MonkeyPatch) -> None:
    """#750 DoD: два последовательных _claim_run одного source (нет running) →
    первый run_id, второй None (create_run вызван РОВНО один раз)."""
    db = _FakeSchedulerDB()
    calls = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        db.running[source] = True
        return 100 + calls["n"]

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    rid1 = _sched._claim_run(db, _SCHED_ROW)
    rid2 = _sched._claim_run(db, _SCHED_ROW)

    assert rid1 == 101
    assert rid2 is None
    assert calls["n"] == 1  # второй НЕ создал run


def test_claim_run_skips_when_advisory_lock_unavailable(monkeypatch: pytest.MonkeyPatch) -> None:
    """#750: lock занят конкурентным тиком → return None, create_run НЕ вызван."""
    db = _FakeSchedulerDB(lock_available=False)
    calls = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        return 999

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    assert _sched._claim_run(db, _SCHED_ROW) is None
    assert calls["n"] == 0


def test_claim_run_rechecks_running_under_lock(monkeypatch: pytest.MonkeyPatch) -> None:
    """#750: running появился МЕЖДУ pre-check и взятием лока (double-checked) →
    return None + rollback (освобождение лока), create_run НЕ вызван."""
    # pre-check → False (нет running), re-check под локом → True (появился).
    db = _FakeSchedulerDB(lock_available=True, running_results=[False, True])
    calls = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        return 999

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    assert _sched._claim_run(db, _SCHED_ROW) is None
    assert calls["n"] == 0
    assert db.rollbacks == 1  # лок освобождён


# ── #974: trigger_yandex_newbuilding_sweep_run + dispatch ────────────────────


_YANDEX_NB_SWEEP_ROW = {
    "source": "yandex_newbuilding_sweep",
    "window_start_hour": 2,
    "window_end_hour": 5,
    "default_params": {
        "limit": 5,
        "request_delay_sec": 8.0,
        "city": "ekaterinburg",
    },
}


@pytest.mark.asyncio
async def test_trigger_yandex_newbuilding_sweep_creates_run(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """trigger_yandex_newbuilding_sweep_run claims run + launches create_task (#974)."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"create_run": [], "sweep": []}

    def fake_create_run(_d, *, source, params):
        calls["create_run"].append(source)
        db.running[source] = True
        return 301

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    # Patch enrich_yandex_newbuilding_sweep to a coroutine that records call
    sweep_result = MagicMock()
    sweep_result.to_dict.return_value = {}

    async def fake_sweep(run_db, *, limit, **kwargs):
        calls["sweep"].append(limit)
        return sweep_result

    # Patch inside the _run closure (lazy import)
    monkeypatch.setattr(
        "app.tasks.yandex_newbuilding_sweep.enrich_yandex_newbuilding_sweep",
        fake_sweep,
    )
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    # mark_done is called inside _run; mock it so it doesn't fail on fake db
    monkeypatch.setattr(_sched.runs_mod, "mark_done", lambda *a, **kw: None)
    monkeypatch.setattr(_sched.runs_mod, "mark_failed", lambda *a, **kw: None)

    run_id = await _sched.trigger_yandex_newbuilding_sweep_run(db, _YANDEX_NB_SWEEP_ROW)

    assert run_id == 301
    assert calls["create_run"] == ["yandex_newbuilding_sweep"]
    # Drain event-loop so the asyncio.create_task runs
    await asyncio.sleep(0)
    assert calls["sweep"] == [5]


@pytest.mark.asyncio
async def test_trigger_yandex_newbuilding_sweep_skips_when_running(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """trigger_yandex_newbuilding_sweep_run returns None when source already running."""
    db = _FakeSchedulerDB()
    db.running["yandex_newbuilding_sweep"] = True

    calls: dict[str, int] = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        return 999

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    result = await _sched.trigger_yandex_newbuilding_sweep_run(db, _YANDEX_NB_SWEEP_ROW)

    assert result is None
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_scheduler_dispatch_routes_yandex_newbuilding_sweep(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dispatch block в scheduler_loop роутит yandex_newbuilding_sweep → trigger."""
    triggered: list[str] = []

    async def fake_trigger(db, sch):
        triggered.append(sch["source"])
        return 1

    monkeypatch.setattr(_sched, "trigger_yandex_newbuilding_sweep_run", fake_trigger)

    db = object()
    sch = _YANDEX_NB_SWEEP_ROW
    source = sch["source"]
    # Replicate the elif chain from scheduler_loop for yandex_newbuilding_sweep
    if source == "yandex_newbuilding_sweep":
        await _sched.trigger_yandex_newbuilding_sweep_run(db, sch)

    assert "yandex_newbuilding_sweep" in triggered


# ── #973: trigger_cian_city_sweep_run + dispatch ─────────────────────────────


_CIAN_SWEEP_ROW = {
    "source": "cian_city_sweep",
    "window_start_hour": 2,
    "window_end_hour": 5,
    "default_params": {
        "pages_per_anchor": 3,
        "request_delay_sec": 5.0,
        "radius_m": 1500,
        "detail_top_n": 10,
        "enrich_houses": True,
    },
}


@pytest.mark.asyncio
async def test_trigger_cian_city_sweep_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
    """trigger_cian_city_sweep_run claims run + launches create_task (#973)."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"create_run": [], "sweep": []}

    def fake_create_run(_d, *, source, params):
        calls["create_run"].append(source)
        db.running[source] = True
        return 201

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    # Patch run_cian_city_sweep to a coroutine that records call
    async def fake_sweep(run_db, *, run_id, **kwargs):
        calls["sweep"].append(run_id)

    monkeypatch.setattr(_sched, "run_cian_city_sweep", fake_sweep)
    # Patch SessionLocal to return the same db
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    run_id = await _sched.trigger_cian_city_sweep_run(db, _CIAN_SWEEP_ROW)

    assert run_id == 201
    assert calls["create_run"] == ["cian_city_sweep"]
    # Drain event-loop so the asyncio.create_task runs
    await asyncio.sleep(0)
    assert calls["sweep"] == [201]


@pytest.mark.asyncio
async def test_trigger_cian_city_sweep_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
    """trigger_cian_city_sweep_run returns None when source already running."""
    db = _FakeSchedulerDB()
    db.running["cian_city_sweep"] = True

    calls: dict[str, int] = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        return 999

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    result = await _sched.trigger_cian_city_sweep_run(db, _CIAN_SWEEP_ROW)

    assert result is None
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_scheduler_dispatch_routes_cian_city_sweep(monkeypatch: pytest.MonkeyPatch) -> None:
    """Dispatch block в scheduler_loop роутит cian_city_sweep → trigger_cian_city_sweep_run."""
    triggered: list[str] = []

    async def fake_trigger(db, sch):
        triggered.append(sch["source"])
        return 1

    monkeypatch.setattr(_sched, "trigger_cian_city_sweep_run", fake_trigger)

    # Simulate the dispatch block for one due schedule
    db = object()
    sch = _CIAN_SWEEP_ROW
    source = sch["source"]
    # Replicate the elif chain from scheduler_loop for cian_city_sweep
    if source == "cian_city_sweep":
        await _sched.trigger_cian_city_sweep_run(db, sch)

    assert "cian_city_sweep" in triggered


# ── cian_full_load: trigger_cian_full_load_run + dispatch ────────────────────


_CIAN_FULL_LOAD_ROW = {
    "source": "cian_full_load",
    "window_start_hour": 20,
    "window_end_hour": 22,
    "default_params": {
        "price_cap_per_bucket": 1400,
        "concurrency": 5,
        "request_delay_sec": 4.0,
        "enrich_detail": False,
        "detail_top_n": 0,
    },
}


@pytest.mark.asyncio
async def test_trigger_cian_full_load_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
    """trigger_cian_full_load_run claims run + launches run_cian_full_load via create_task."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"create_run": [], "load": []}

    def fake_create_run(_d, *, source, params):
        calls["create_run"].append(source)
        db.running[source] = True
        return 301

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    async def fake_full_load(run_db, *, run_id, **kwargs):
        calls["load"].append((run_id, kwargs))

    monkeypatch.setattr(_sched, "run_cian_full_load", fake_full_load)
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    run_id = await _sched.trigger_cian_full_load_run(db, _CIAN_FULL_LOAD_ROW)

    assert run_id == 301
    assert calls["create_run"] == ["cian_full_load"]
    # Drain event-loop so the asyncio.create_task runs
    await asyncio.sleep(0)
    assert len(calls["load"]) == 1
    loaded_run_id, kwargs = calls["load"][0]
    assert loaded_run_id == 301
    # Params propagated from default_params; resume_run_id forced to None
    assert kwargs["price_cap_per_bucket"] == 1400
    assert kwargs["concurrency"] == 5
    assert kwargs["request_delay_sec"] == 4.0
    assert kwargs["enrich_detail"] is False
    assert kwargs["detail_top_n"] == 0
    assert kwargs["resume_run_id"] is None


@pytest.mark.asyncio
async def test_trigger_cian_full_load_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
    """trigger_cian_full_load_run returns None when source already running."""
    db = _FakeSchedulerDB()
    db.running["cian_full_load"] = True

    calls: dict[str, int] = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        return 999

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    result = await _sched.trigger_cian_full_load_run(db, _CIAN_FULL_LOAD_ROW)

    assert result is None
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_scheduler_dispatch_routes_cian_full_load(monkeypatch: pytest.MonkeyPatch) -> None:
    """Dispatch block в scheduler_loop роутит cian_full_load → trigger_cian_full_load_run."""
    triggered: list[str] = []

    async def fake_trigger(db, sch):
        triggered.append(sch["source"])
        return 1

    monkeypatch.setattr(_sched, "trigger_cian_full_load_run", fake_trigger)

    db = object()
    sch = _CIAN_FULL_LOAD_ROW
    source = sch["source"]
    # Replicate the elif chain from scheduler_loop for cian_full_load
    if source == "cian_full_load":
        await _sched.trigger_cian_full_load_run(db, sch)

    assert "cian_full_load" in triggered


# ── avito_full_load: trigger_avito_full_load_run + dispatch ──────────────────


_AVITO_FULL_LOAD_ROW = {
    "source": "avito_full_load",
    "window_start_hour": 13,
    "window_end_hour": 15,
    "default_params": {
        "price_cap_per_bucket": 1400,
        "concurrency": 5,
        "request_delay_sec": 7.0,
        "secondary_only": True,
    },
}


@pytest.mark.asyncio
async def test_trigger_avito_full_load_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
    """trigger_avito_full_load_run claims run + launches run_avito_full_load via create_task."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"create_run": [], "load": []}

    def fake_create_run(_d, *, source, params):
        calls["create_run"].append(source)
        db.running[source] = True
        return 401

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    async def fake_full_load(run_db, *, run_id, **kwargs):
        calls["load"].append((run_id, kwargs))

    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    run_id = await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)

    assert run_id == 401
    assert calls["create_run"] == ["avito_full_load"]
    # Drain event-loop so the asyncio.create_task runs
    await asyncio.sleep(0)
    assert len(calls["load"]) == 1
    loaded_run_id, kwargs = calls["load"][0]
    assert loaded_run_id == 401
    # Params propagated from default_params; resume_run_id forced to None
    assert kwargs["price_cap_per_bucket"] == 1400
    assert kwargs["concurrency"] == 5
    assert kwargs["request_delay_sec"] == 7.0
    assert kwargs["secondary_only"] is True
    assert kwargs["resume_run_id"] is None


@pytest.mark.asyncio
async def test_trigger_avito_full_load_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
    """trigger_avito_full_load_run returns None when source already running."""
    db = _FakeSchedulerDB()
    db.running["avito_full_load"] = True

    calls: dict[str, int] = {"n": 0}

    def fake_create_run(_d, *, source, params):
        calls["n"] += 1
        return 999

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    result = await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)

    assert result is None
    assert calls["n"] == 0


@pytest.mark.asyncio
async def test_scheduler_dispatch_routes_avito_full_load(monkeypatch: pytest.MonkeyPatch) -> None:
    """Dispatch block в scheduler_loop роутит avito_full_load → trigger_avito_full_load_run."""
    triggered: list[str] = []

    async def fake_trigger(db, sch):
        triggered.append(sch["source"])
        return 1

    monkeypatch.setattr(_sched, "trigger_avito_full_load_run", fake_trigger)

    db = object()
    sch = _AVITO_FULL_LOAD_ROW
    source = sch["source"]
    # Replicate the elif chain from scheduler_loop for avito_full_load
    if source == "avito_city_sweep":
        pass
    elif source == "avito_full_load":
        await _sched.trigger_avito_full_load_run(db, sch)

    assert "avito_full_load" in triggered


# ── avito_full_load incremental_days + avito_full_load_exhaustive split ───────


@pytest.mark.asyncio
async def test_trigger_avito_full_load_reads_incremental_days(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """incremental_days из default_params прокидывается в run_avito_full_load."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"load": []}

    def fake_create_run(_d, *, source, params):
        db.running[source] = True
        return 410

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    async def fake_full_load(run_db, *, run_id, **kwargs):
        calls["load"].append(kwargs)

    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    row = {
        **_AVITO_FULL_LOAD_ROW,
        "default_params": {**_AVITO_FULL_LOAD_ROW["default_params"], "incremental_days": 2},
    }
    await _sched.trigger_avito_full_load_run(db, row)
    await asyncio.sleep(0)

    assert len(calls["load"]) == 1
    assert calls["load"][0]["incremental_days"] == 2


@pytest.mark.asyncio
async def test_trigger_avito_full_load_no_incremental_days_is_none(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Без incremental_days в params → run_avito_full_load(incremental_days=None) (exhaustive)."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"load": []}

    def fake_create_run(_d, *, source, params):
        db.running[source] = True
        return 411

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    async def fake_full_load(run_db, *, run_id, **kwargs):
        calls["load"].append(kwargs)

    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)
    await asyncio.sleep(0)

    assert len(calls["load"]) == 1
    assert calls["load"][0]["incremental_days"] is None


_AVITO_FULL_LOAD_EXHAUSTIVE_ROW = {
    "source": "avito_full_load_exhaustive",
    "window_start_hour": 13,
    "window_end_hour": 15,
    "default_params": {
        "price_cap_per_bucket": 1400,
        "concurrency": 1,
        "request_delay_sec": 7.0,
        "secondary_only": True,
    },
}


@pytest.mark.asyncio
async def test_trigger_avito_full_load_exhaustive_forces_none(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """avito_full_load_exhaustive форсит incremental_days=None даже если в params задан."""
    db = _FakeSchedulerDB()
    calls: dict[str, list] = {"load": []}

    def fake_create_run(_d, *, source, params):
        db.running[source] = True
        return 420

    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)

    async def fake_full_load(run_db, *, run_id, **kwargs):
        calls["load"].append((run_id, kwargs))

    monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load)
    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)

    # Даже если оператор случайно положит incremental_days в params — exhaustive игнорит.
    row = {
        **_AVITO_FULL_LOAD_EXHAUSTIVE_ROW,
        "default_params": {
            **_AVITO_FULL_LOAD_EXHAUSTIVE_ROW["default_params"],
            "incremental_days": 5,
        },
    }
    run_id = await _sched.trigger_avito_full_load_exhaustive_run(db, row)
    await asyncio.sleep(0)

    assert run_id == 420
    assert len(calls["load"]) == 1
    loaded_run_id, kwargs = calls["load"][0]
    assert loaded_run_id == 420
    assert kwargs["incremental_days"] is None
    assert kwargs["concurrency"] == 1
    assert kwargs["price_cap_per_bucket"] == 1400
    assert kwargs["resume_run_id"] is None


@pytest.mark.asyncio
async def test_scheduler_dispatch_routes_avito_full_load_exhaustive(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dispatch роутит avito_full_load_exhaustive → trigger_avito_full_load_exhaustive_run."""
    triggered: list[str] = []

    async def fake_trigger(db, sch):
        triggered.append(sch["source"])
        return 1

    monkeypatch.setattr(_sched, "trigger_avito_full_load_exhaustive_run", fake_trigger)

    db = object()
    sch = _AVITO_FULL_LOAD_EXHAUSTIVE_ROW
    source = sch["source"]
    # Replicate the elif chain from scheduler_loop
    if source == "avito_full_load":
        pass
    elif source == "avito_full_load_exhaustive":
        await _sched.trigger_avito_full_load_exhaustive_run(db, sch)

    assert "avito_full_load_exhaustive" in triggered


# ---------------------------------------------------------------------------
# #1182 Phase 2 — graceful SIGTERM-drain детей (detached run-задач)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_drain_inflight_awaits_detached_cooperative_child() -> None:
    """Гвоздь Phase 2: _drain_inflight ДОЖИДАЕТСЯ detached-ребёнка (как scrape-run,
    отвязанный через _spawn_tracked) до его clean-exit, а НЕ хард-кансельит его.

    Регрессия gap'а: coordinator раньше ждал только себя → asyncio.run teardown
    хард-кансельил живых детей mid-await (#1182 failure mode).
    """
    clean_exit = asyncio.Event()  # == mark_done-эквивалент ребёнка

    async def _child() -> None:
        # Кооперативный ребёнок: крутится, пока не запрошен shutdown, потом выходит чисто.
        while not sd.shutdown_requested():
            await asyncio.sleep(0.01)
        clean_exit.set()

    child = sched._spawn_tracked(_child())  # detached, попадает в _inflight_tasks
    await asyncio.sleep(0.02)
    assert not child.done(), "ребёнок ещё бежит (shutdown не запрошен)"

    sd.request_shutdown()
    await asyncio.wait_for(sched._drain_inflight(), timeout=2.0)

    assert clean_exit.is_set(), "drain дождался clean-exit ребёнка"
    assert child.done()
    assert not child.cancelled(), "ребёнок вышел сам, НЕ хард-cancel"


@pytest.mark.asyncio
async def test_drain_inflight_times_out_on_noncooperative_child(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Некооперативный ребёнок (игнорит shutdown) упирается в _CHILD_DRAIN_TIMEOUT_S и
    остаётся бежать — отдаётся внешнему hard-cancel fallback (scheduler_main/docker)."""
    monkeypatch.setattr(sched, "_CHILD_DRAIN_TIMEOUT_S", 0.05)

    async def _stuck() -> None:
        await asyncio.Event().wait()  # никогда не проверяет shutdown_requested()

    child = sched._spawn_tracked(_stuck())
    sd.request_shutdown()

    await asyncio.wait_for(sched._drain_inflight(), timeout=2.0)

    assert not child.done(), "drain НЕ повесил процесс — оставил ребёнка под hard-cancel"

    # cleanup
    child.cancel()
    with pytest.raises(asyncio.CancelledError):
        await child


@pytest.mark.asyncio
async def test_drain_inflight_noop_when_no_children() -> None:
    """Без in-flight детей _drain_inflight() — мгновенный no-op (без сна/ожидания)."""
    assert not sched._inflight_tasks
    await asyncio.wait_for(sched._drain_inflight(), timeout=1.0)


@pytest.mark.asyncio
async def test_scheduler_loop_calls_drain_on_shutdown(monkeypatch: pytest.MonkeyPatch) -> None:
    """scheduler_loop при выходе по SIGTERM-drain'у вызывает _drain_inflight (проводка
    coordinator → дренаж детей). Initial sleep(30) и tick-sleep занулены, чтобы тест
    был детерминированным и быстрым."""
    drain_spy = AsyncMock()
    monkeypatch.setattr(sched, "_drain_inflight", drain_spy)
    monkeypatch.setattr(sched.asyncio, "sleep", AsyncMock())  # без реальных 30s/60s

    sd.request_shutdown()  # первый же top-of-tick check → break

    await asyncio.wait_for(sched.scheduler_loop(), timeout=2.0)

    drain_spy.assert_awaited_once()