"""Гейт: докер-секции обоих прод-деплоев исключают друг друга (#2950).

Что произошло. Деплой ПТИЦЫ и деплой Trade-In ходят по SSH в один и тот же
докер-демон — стеки gendesign-*, tradein-* и сам forgejo-runner живут на одной
VM. Каждый в конце делает `docker image prune -af`, и прун одного сносит leases
ещё не доехавшего `compose pull` другого:

    unable to lease content: lease does not exist: not found

20.08 деплой ПТИЦЫ упал за 5 секунд, прод остался на старом коде.

ЧЕМ ЭТО ЧИНИТСЯ СЕГОДНЯ — host-lock, а НЕ секцией `concurrency`. На Forgejo
10.0.3 (gitea-1.22) workflow-level concurrency не исполняется: после мержа
правки, сводившей обе группы к `deploy-prod`, обе цепочки стартовали на одном
коммите ОДНОВРЕМЕННО (`docker ps` на раннере показал их build-джобы бок о бок).
Секция оставлена как декларация на будущее — она заработает при обновлении
Forgejo, — но выдавать её за действующий механизм нельзя, поэтому проверки
разделены: про лок — обязательные, про группу — про декларацию.

Разводка групп обратно или пропажа `flock` не ломают ни один прогон CI и не
дают ни одного сигнала: отказ проявится только в следующем совпадении окон двух
мержей. Поэтому инвариант зафиксирован здесь.
"""

from pathlib import Path

import pytest
import yaml

# backend/tests/ops/<этот файл> → корень репозитория
REPO_ROOT = Path(__file__).resolve().parents[3]
WORKFLOWS = REPO_ROOT / ".forgejo" / "workflows"

PROD_DEPLOYS = ("deploy.yml", "deploy-tradein.yml")

# Один и тот же путь у обоих — иначе взаимного исключения не выйдет.
LOCK_PATH = "/var/lock/gendesign-docker-deploy.lock"


def _text(name: str) -> str:
    path = WORKFLOWS / name
    assert path.is_file(), f"нет {path} — переехал workflow, гейт ослеп"
    return path.read_text()


def _deploy_script(name: str) -> str:
    """Тело ssh-шага, который реально мутирует докер (в нём есть агрессивный прун).

    Именно этот блок обязан брать лок. Второй ssh-шаг (перезагрузка прокси)
    образов не тянет, пруну там нечего портить — от него лок не требуется.
    """
    text = _text(name)
    marker = "docker image prune -af"
    assert marker in text, f"{name}: пропал `{marker}` — гейт опирается на него, перепроверить"
    return text


@pytest.mark.parametrize("name", PROD_DEPLOYS)
def test_prod_deploy_takes_the_host_lock(name: str) -> None:
    """Деплой берёт лок на хосте перед докер-секцией."""
    script = _deploy_script(name)
    assert f"exec 9>{LOCK_PATH}" in script, (
        f"{name}: не берёт host-lock {LOCK_PATH}. Без него `docker image prune -af` "
        "соседнего деплоя оборвёт наш `compose pull` (#2950)."
    )
    assert "flock -w" in script, (
        f"{name}: лок открывается, но не захватывается через `flock -w` — "
        "открытый fd сам по себе ничего не исключает"
    )


@pytest.mark.parametrize("name", PROD_DEPLOYS)
def test_lock_wait_leaves_a_trace(name: str) -> None:
    """Ожидание лока видно в логе — иначе работающий лок ненаблюдаем.

    `flock` при успехе молчит. Если брать лок сразу блокирующим вызовом, отличить
    «второй деплой дождался первого» от «они просто разошлись по времени» нельзя —
    а это и есть критерий приёмки #2950. Поэтому сначала неблокирующая попытка, и
    при занятом локе в лог уходит и факт ожидания, и его длительность.
    """
    script = _deploy_script(name)
    assert "flock -n" in script, (
        f"{name}: лок берётся сразу блокирующим вызовом — ожидание не попадёт в лог, "
        "и проверить, что взаимное исключение сработало, будет нечем"
    )
    assert "жду" in script, f"{name}: нет строки об ожидании лока"
    assert "с ожидания" in script, (
        f"{name}: не печатается длительность ожидания — без неё непонятно, "
        "ждал деплой две секунды или четверть часа"
    )


@pytest.mark.parametrize("name", PROD_DEPLOYS)
def test_lock_wait_is_bounded_and_loud(name: str) -> None:
    """Ожидание ограничено и провал слышен: молча ждать вечно — не вариант."""
    script = _deploy_script(name)
    assert "не дождался лока докер-деплоя" in script, (
        f"{name}: нет внятного сообщения на исчерпание ожидания лока — "
        "деплой упадёт с голым кодом возврата flock"
    )


def test_both_prod_deploys_use_the_same_lock_path() -> None:
    """Путь лока общий. Разные пути = два независимых лока = нет исключения."""
    holders = {name: f"exec 9>{LOCK_PATH}" in _deploy_script(name) for name in PROD_DEPLOYS}
    assert all(holders.values()), (
        f"лок берут не все прод-деплои: {holders}. Взаимное исключение работает, "
        "только когда ОБА ходят через один и тот же файл."
    )


# ── Ниже — про секцию concurrency. Сегодня она НЕ исполняется (см. шапку), ────
# поэтому это проверки декларации, а не работающего механизма.


@pytest.mark.parametrize("name", PROD_DEPLOYS)
def test_prod_deploy_declares_shared_concurrency_group(name: str) -> None:
    """Группа объявлена и общая — заработает при обновлении Forgejo.

    Отдельно от лока: если однажды эта секция начнёт исполняться, разведённые
    группы снова разрешат параллельный запуск.
    """
    conc = yaml.safe_load(_text(name)).get("concurrency") or {}
    assert (
        conc.get("group") == "deploy-prod"
    ), f"{name}: группа concurrency = {conc.get('group')!r}, ожидалась общая 'deploy-prod'"
    assert conc.get("cancel-in-progress") is False, (
        f"{name}: cancel-in-progress должен быть false — отменённый деплой оставляет "
        "прод на старом коде ровно так же, как упавший"
    )