gendesign/backend/tests/test_2950_latest_image_revision_gate.py

"""Гард свежести :latest перед деплоем (#2950) — коды выхода по значению.

Сценарий, ради которого гард существует: Forgejo отменил ещё не стартовавший deploy
run'а A, run B (билды пропущены) катит :latest, а :latest несёт ревизию СТАРШЕ
последнего коммита по компоненту — билд A не доехал. Гард обязан ждать и, не
дождавшись, падать; а при ревизии не старше последнего коммита — пропускать.

`docker` подменяется шимом на PATH: он печатает конфиг образа с меткой ревизии из
переменной окружения FAKE_REV (пусто — метки нет; FAKE_DOCKER_FAIL=1 — registry не
ответил). Временный git-репозиторий даёт настоящие коммиты для `git log`/merge-base.
"""

from __future__ import annotations

import os
import stat
import subprocess
import sys
from pathlib import Path

import pytest

# Каждый тест в файле запускает scripts/check-latest-image-revision.sh (POSIX shell,
# shebang #!/usr/bin/env bash) дочерним процессом. На Windows нет POSIX shell на PATH,
# чтобы это исполнить, — покрытие обеспечивается в CI на Linux.
pytestmark = pytest.mark.skipif(
    sys.platform == "win32",
    reason="запускает bash-скрипт (check-latest-image-revision.sh) дочерним процессом; "
    "POSIX-only, покрытие в CI (Linux)",
)

КОРЕНЬ = Path(__file__).resolve().parents[2]
СКРИПТ = КОРЕНЬ / "scripts" / "check-latest-image-revision.sh"

_SHIM = """#!/usr/bin/env bash
# подменный docker: только `buildx imagetools inspect <img> --format ...`
if [ "${FAKE_DOCKER_FAIL:-}" = "1" ]; then echo "unauthorized" >&2; exit 1; fi
FMT='{"linux/amd64": {"config": {"Labels": {"org.opencontainers.image.revision": "%s"}}}}'
if [ -n "${FAKE_REV:-}" ]; then
  printf "$FMT" "$FAKE_REV"
else
  printf '{"linux/amd64": {"config": {"Labels": {}}}}'
fi
"""


def _git(repo: Path, *args: str) -> str:
    return subprocess.run(
        ["git", *args],
        cwd=repo,
        check=True,
        capture_output=True,
        text=True,
        env={
            **os.environ,
            "GIT_AUTHOR_NAME": "t",
            "GIT_AUTHOR_EMAIL": "t@t",
            "GIT_COMMITTER_NAME": "t",
            "GIT_COMMITTER_EMAIL": "t@t",
        },
    ).stdout.strip()


def _commit(repo: Path, rel: str, msg: str) -> str:
    p = repo / rel
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(msg, encoding="utf-8")
    _git(repo, "add", rel)
    _git(repo, "commit", "-q", "-m", msg)
    return _git(repo, "rev-parse", "HEAD")


@pytest.fixture()
def repo(tmp_path: Path) -> dict[str, object]:
    r = tmp_path / "repo"
    r.mkdir()
    _git(r, "init", "-q")
    c1 = _commit(r, "backend/a.py", "c1 backend")
    c2 = _commit(r, "frontend/b.ts", "c2 frontend")
    c3 = _commit(r, "backend/c.py", "c3 backend")
    c4 = _commit(r, "frontend/d.ts", "c4 frontend")
    shim_dir = tmp_path / "bin"
    shim_dir.mkdir()
    shim = shim_dir / "docker"
    shim.write_text(_SHIM, encoding="utf-8")
    shim.chmod(shim.stat().st_mode | stat.S_IEXEC)
    return {"path": r, "c1": c1, "c2": c2, "c3": c3, "c4": c4, "bin": shim_dir}


def _run(
    repo: dict[str, object], rev: str | None, *paths: str, fail: bool = False, timeout: str = "1"
) -> subprocess.CompletedProcess[str]:
    env = {
        **os.environ,
        "PATH": f"{repo['bin']}{os.pathsep}{os.environ['PATH']}",
        "CHECK_LATEST_SLEEP_S": "0.2",
        "FAKE_REV": rev or "",
        "FAKE_DOCKER_FAIL": "1" if fail else "",
    }
    return subprocess.run(
        [str(СКРИПТ), "ghcr.io/x/img", timeout, "--", *paths],
        cwd=repo["path"],
        env=env,
        capture_output=True,
        text=True,
    )


def test_fresh_latest_passes(repo: dict[str, object]) -> None:
    """:latest собран из последнего коммита по backend (c3) → 0."""
    r = _run(repo, str(repo["c3"]), "backend")
    assert r.returncode == 0, r.stdout + r.stderr
    assert "содержит последний коммит" in r.stdout


def test_newer_revision_passes(repo: dict[str, object]) -> None:
    """Ревизия новее последнего коммита по компоненту (dispatch собрал голову c4) → 0."""
    r = _run(repo, str(repo["c4"]), "backend")
    assert r.returncode == 0, r.stdout + r.stderr


def test_stale_latest_fails_after_timeout(repo: dict[str, object]) -> None:
    """Головной: :latest из c1, а по backend последний — c3 → ждёт, затем 1 с ::error."""
    # timeout=3: дедлайн считается целыми секундами (`date +%s`), и при 1 с первый
    # опрос может прийти уже за дедлайном — гард честно падает сразу, а тест ждёт
    # строки ожидания. 3 с при паузе 0.2 с дают её гарантированно.
    r = _run(repo, str(repo["c1"]), "backend", timeout="3")
    assert r.returncode == 1, r.stdout + r.stderr
    assert "::error::" in r.stderr and "отстаёт" in r.stderr
    assert "жду билд предшественника" in r.stdout, "гард обязан ЖДАТЬ, а не падать сразу"


def test_revision_from_other_component_does_not_cover(repo: dict[str, object]) -> None:
    """Контроль путей: для frontend последний — c4; :latest из c3 (до c4) → 1."""
    r = _run(repo, str(repo["c3"]), "frontend")
    assert r.returncode == 1, r.stdout + r.stderr


def test_missing_label_fails(repo: dict[str, object]) -> None:
    """Без метки ревизии нельзя утверждать свежесть → 1, а не молчаливый пропуск."""
    r = _run(repo, None, "backend")
    assert r.returncode == 1, r.stdout + r.stderr
    assert "метки нет" in r.stderr


def test_registry_error_fails(repo: dict[str, object]) -> None:
    """Registry не отвечает → 1 (не «проверить нельзя — значит можно»)."""
    r = _run(repo, str(repo["c3"]), "backend", fail=True)
    assert r.returncode == 1, r.stdout + r.stderr


def test_usage_errors_are_distinct(repo: dict[str, object]) -> None:
    """Ошибка вызова — код 2, чтобы её не приняли за «отстаёт»."""
    r = subprocess.run([str(СКРИПТ), "img"], cwd=repo["path"], capture_output=True, text=True)
    assert r.returncode == 2