gendesign/backend/tests/ops/test_2950_deploy_concurrency_group.py

"""Гейт: оба прод-деплоя обязаны лежать в ОДНОЙ группе concurrency.

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

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

20.08 так и вышло: run 8083 упал за 5 секунд, прод остался на старом коде. Разводка
групп обратно ничего не сломает в CI и не даст ни одного сигнала — отказ проявится
только в следующем совпадении окон двух мержей. Поэтому инвариант зафиксирован здесь.

`cancel-in-progress: false` частью инварианта тоже является: прод-деплои обязаны
достаиваться в очереди, а не отменять друг друга — отменённый деплой оставляет прод
на старом коде ровно так же, как упавший.
"""

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")


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


@pytest.mark.parametrize("name", PROD_DEPLOYS)
def test_prod_deploy_declares_concurrency(name: str) -> None:
    """У каждого прод-деплоя concurrency вообще объявлена."""
    assert _concurrency(name).get(
        "group"
    ), f"{name}: пропала секция concurrency — деплои снова могут пойти параллельно"


def test_both_prod_deploys_share_one_group() -> None:
    """Группа у обоих одна и та же — иначе прун одного убьёт pull другого."""
    groups = {name: _concurrency(name).get("group") for name in PROD_DEPLOYS}
    assert len(set(groups.values())) == 1, (
        f"прод-деплои разведены по разным группам concurrency: {groups}. "
        "Они ходят в ОДИН докер-демон; `docker image prune -af` одного обрывает "
        "`compose pull` другого (#2950)."
    )


@pytest.mark.parametrize("name", PROD_DEPLOYS)
def test_prod_deploy_queues_instead_of_cancelling(name: str) -> None:
    """Раны достаиваются в очереди: отменённый деплой = прод на старом коде."""
    assert _concurrency(name).get("cancel-in-progress") is False, (
        f"{name}: cancel-in-progress должен быть false — иначе более новый мерж "
        "отменит выкатку предыдущего и та молча не доедет"
    )