"""На worker_ready зомби помечается ЛЮБАЯ строка в 'running' (#2464).

Докстринг `_resume_zombie_runs` формулирует инвариант прямо:

    No time threshold: by definition, on worker_ready ANY 'running' row is a zombie
    because there is no active worker. Previously we required heartbeat … stayed in
    'running' status forever and required manual cancel/resume.

А запрос добавлял `AND objects_snapshot IS NOT NULL`. Строка без снапшота в выборку не
попадала и оставалась `'running'` НАВСЕГДА — ровно то состояние, ради устранения которого
функция и заводилась.

Снапшот нужен, но не для пометки, а для ВОЗОБНОВЛЕНИЯ: `resume_kn_run` восстанавливает
обход «using objects_snapshot». Поэтому зомби помечаются все, а resume ставится только тем,
кого есть чем возобновить; остальные получают честную причину.

Тесты смотрят на выполненный SQL и на поставленные resume-задачи — то есть на поведение.
"""

from __future__ import annotations

import os

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

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


class _Result:
    def __init__(self, rows: list[dict]) -> None:
        self._rows = rows

    def mappings(self) -> _Result:
        return self

    def all(self) -> list[dict]:
        return self._rows

    def scalar(self) -> Any:
        return None

    def first(self) -> Any:
        return None


class _Session:
    """Отдаёт заданные строки на SELECT и запоминает весь выполненный SQL."""

    def __init__(self, rows: list[dict]) -> None:
        self._rows = rows
        self.sql: list[tuple[str, dict]] = []
        self.commits = 0

    def execute(self, statement: Any = None, params: Any = None, *a: Any, **kw: Any) -> _Result:
        text_ = str(statement)
        self.sql.append((text_, params or {}))
        if "SELECT run_id" in text_ and "kn_scrape_runs" in text_:
            # Двойник ОБЯЗАН соблюдать ровно тот фильтр, о котором идёт спор.
            # Без этого тест краснел бы на origin/main по ложной причине: там
            # строка без снапшота отсекается запросом и до пометки не доходит,
            # а наивный двойник отдавал бы её всё равно — и главный тест
            # («строка не помечена») проходил бы на обеих сторонах.
            # Ищем именно условие в WHERE (с ведущим AND), а не подвыражение в
            # SELECT: правка выносит `(objects_snapshot IS NOT NULL) AS resumable`
            # в список полей, и совпадение по голой подстроке отсекало бы строки
            # и у исправленной версии тоже.
            if "AND objects_snapshot IS NOT NULL" in text_:
                return _Result([r for r in self._rows if r["resumable"]])
            return _Result(self._rows)
        return _Result([])

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

    def rollback(self) -> None:
        pass

    def close(self) -> None:
        pass


def _run(rows: list[dict]) -> tuple[_Session, list[int]]:
    from app.workers import lifecycle as mod

    db = _Session(rows)
    enqueued: list[int] = []

    resume_task = MagicMock()
    resume_task.apply_async.side_effect = lambda args=None, **_kw: enqueued.append(args[0])

    # SessionLocal импортируется ВНУТРИ функции (late import), поэтому подменяем
    # его в модуле-источнике, а не на lifecycle.
    import app.core.db as core_db

    with (
        patch.object(core_db, "SessionLocal", lambda: db),
        patch.dict(
            "sys.modules",
            {"app.workers.tasks.scrape_kn": MagicMock(resume_kn_run=resume_task)},
        ),
    ):
        mod._resume_zombie_runs()
    return db, enqueued


def _updated_ids(db: _Session) -> set[int]:
    out: set[int] = set()
    for sql, params in db.sql:
        if "UPDATE kn_scrape_runs" in sql and "zombie" in sql:
            for key in ("ids", "orphans"):
                out.update(params.get(key) or [])
    return out


def test_running_row_without_snapshot_is_marked_zombie() -> None:
    """Строка без objects_snapshot обязана быть помечена, а не остаться 'running'.

    На origin/main она вообще не попадает в выборку: фильтр отсекает её,
    и никакого UPDATE по ней не выполняется.
    """
    db, _ = _run([{"run_id": 7, "resumable": False}])

    assert 7 in _updated_ids(db), (
        "прогон без снапшота не помечен zombie — останется в 'running' навсегда; "
        f"выполненный SQL: {[s for s, _ in db.sql]}"
    )


def test_unresumable_row_gets_no_resume_task() -> None:
    """Контроль: возобновлять нечем — resume не ставим.

    Ловит «починку», которая ставила бы resume всем подряд: resume_kn_run
    восстанавливает обход ИЗ objects_snapshot и без него упадёт.
    """
    _, enqueued = _run([{"run_id": 7, "resumable": False}])
    assert enqueued == [], f"поставлен resume для невозобновляемого прогона: {enqueued}"


def test_resumable_row_still_gets_resume() -> None:
    """Контроль: прежнее поведение для строк со снапшотом не тронуто."""
    db, enqueued = _run([{"run_id": 3, "resumable": True}])
    assert 3 in _updated_ids(db)
    assert enqueued == [3]


def test_mixed_batch_splits_correctly() -> None:
    """Контроль: смешанная выборка — помечены все, resume только у пригодных."""
    db, enqueued = _run(
        [
            {"run_id": 1, "resumable": True},
            {"run_id": 2, "resumable": False},
            {"run_id": 3, "resumable": True},
        ]
    )
    assert _updated_ids(db) == {1, 2, 3}, f"помечены не все: {_updated_ids(db)}"
    assert sorted(enqueued) == [1, 3], f"resume поставлен не тем: {enqueued}"