"""Правка пофайлово смонтированного конфига метрик обязана доезжать до контейнера.

ЧТО НЕ ТАК. `git reset --hard` пишет изменившийся файл НОВЫМ инодом, а bind-маунт
одного файла держит прежний. `up -d` этого не видит (сравнивает описание сервиса),
перезагрузка по API тоже: на prom/prometheus:v3.1.0 (17.09) prometheus.yml подменён,
POST /-/reload → rc=0, lastConfigTime новый, а /api/v1/status/config отдаёт прежний
job. Деплой зелёный, конфиг старый — у отказа нет отрицательного признака (#3467).
Так смонтированы prometheus.yml и loki-config.yml на сервере и queries.yml у трёх
postgres-экспортёров (#3486).

ЧТО ПРОВЕРЯЕТСЯ.
1. ops/metrics/recreate-stale-mount.sh ИСПОЛНЯЕТСЯ с подставным `docker`: смотрим
   на совершённые действия (пересоздал / не тронул / упал), а не на текст.
2. Гейт по compose-файлам метрик: у КАЖДОГО пофайлового маунта есть путь доезда в
   той джобе деплоя, что поднимает сервис. Список маунтов берётся ИЗ compose:
   добавят новый — тест покраснеет, пока путь не появится.
"""

from __future__ import annotations

import os
import re
import shutil
import stat
import subprocess
from pathlib import Path

import pytest
import yaml

REPO_ROOT = Path(__file__).resolve().parents[3]
SCRIPT = REPO_ROOT / "ops" / "metrics" / "recreate-stale-mount.sh"
WORKFLOW = REPO_ROOT / ".forgejo" / "workflows" / "deploy-metrics.yml"

SRC = "ops/metrics/loki/loki-config.yml"
DST = "/etc/loki/loki-config.yml"

# Подставной `docker`. Каждый вызов пишется в $FAKE_LOG.
#   compose … ps -q       — id контейнера из $FAKE_CID_FILE (пусто = не запущен);
#                           код из $FAKE_PS_RC;
#   compose … up …        — «пересоздание»: новый id cid-new, и контейнер видит
#                           инод $FAKE_NEW_VIEW (пусто — не видит ничего);
#   exec ID stat -c %i …  — инод, который видит контейнер ID ($FAKE_VIEW/ID).
FAKE_DOCKER = r"""#!/bin/bash
printf '%s\n' "$*" >> "$FAKE_LOG"
cmd="$1"; shift
case "$cmd" in
  exec)
      view="$FAKE_VIEW/$1"
      [ -f "$view" ] || exit 1
      cat "$view"
      ;;
  compose)
      case " $* " in
        *" ps "*)
            [ "${FAKE_PS_RC:-0}" = "0" ] || { echo "compose error" >&2; exit "$FAKE_PS_RC"; }
            cat "$FAKE_CID_FILE"
            ;;
        *" up "*)
            printf 'cid-new\n' > "$FAKE_CID_FILE"
            [ -z "${FAKE_NEW_VIEW:-}" ] || printf '%s\n' "$FAKE_NEW_VIEW" > "$FAKE_VIEW/cid-new"
            ;;
      esac
      ;;
esac
exit 0
"""

# На раннере и на хостах stat — GNU/busybox (`-c %i`); на macOS — BSD (`-f %i`).
# Шим ставится только там, где `-c` не понимают, иначе гейт не запускался бы локально.
STAT_SHIM = (
    '#!/bin/sh\n[ "$1" = "-c" ] && [ "$2" = "%i" ] && exec /usr/bin/stat -f %i "$3"\n'
    'exec /usr/bin/stat "$@"\n'
)
NO_SLEEP = "#!/bin/sh\nexit 0\n"


def _write_exec(path: Path, text: str) -> None:
    path.write_text(text, encoding="utf-8")
    path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)


def _gnu_stat() -> bool:
    return subprocess.run(["stat", "-c", "%i", "/"], capture_output=True).returncode == 0


@pytest.fixture
def host(tmp_path: Path) -> Path:
    """Боевое дерево: скрипт + конфиг; контейнер cid-old видит ТЕКУЩИЙ инод."""
    tree = tmp_path / "opt" / "gendesign"
    (tree / "ops" / "metrics" / "loki").mkdir(parents=True)
    shutil.copy(SCRIPT, tree / "ops" / "metrics" / SCRIPT.name)
    (tree / SRC).write_text("auth_enabled: false\n", encoding="utf-8")

    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    _write_exec(bin_dir / "docker", FAKE_DOCKER)
    _write_exec(bin_dir / "sleep", NO_SLEEP)
    if not _gnu_stat():
        _write_exec(bin_dir / "stat", STAT_SHIM)

    view = tmp_path / "view"
    view.mkdir()
    (view / "cid-old").write_text(f"{_ino(tree)}\n", encoding="utf-8")
    (tmp_path / "cid").write_text("cid-old\n", encoding="utf-8")
    (tmp_path / "log").write_text("", encoding="utf-8")
    return tree


def _ino(tree: Path) -> int:
    return os.stat(tree / SRC).st_ino


def _stale(tree: Path) -> None:
    """Контейнер держит прежний инод — как после `git reset --hard`."""
    (tree.parent.parent / "view" / "cid-old").write_text(f"{_ino(tree) + 1}\n", encoding="utf-8")


def _run(tree: Path, **env_extra: str) -> tuple[int, str, list[str]]:
    root = tree.parent.parent
    env = {
        "PATH": f"{root / 'bin'}:/usr/bin:/bin:/usr/sbin:/sbin",
        "FAKE_LOG": str(root / "log"),
        "FAKE_CID_FILE": str(root / "cid"),
        "FAKE_VIEW": str(root / "view"),
        **env_extra,
    }
    proc = subprocess.run(
        [
            "sh",
            str(tree / "ops" / "metrics" / SCRIPT.name),
            "gendesign-metrics",
            "docker-compose.metrics.yml",
            "loki",
            SRC,
            DST,
        ],
        cwd="/",  # скрипт обязан сам встать в корень дерева
        env=env,
        capture_output=True,
        text=True,
    )
    calls = [c for c in (root / "log").read_text(encoding="utf-8").splitlines() if c]
    return proc.returncode, proc.stdout + proc.stderr, calls


def _recreates(calls: list[str]) -> list[str]:
    return [c for c in calls if "--force-recreate" in c]


# ── Что скрипт делает на самом деле ──────────────────────────────────────────


def test_fresh_inode_touches_nothing(host: Path) -> None:
    """Контейнер видит текущий файл — пересоздавать нельзя: это рвёт работу зря."""
    rc, out, calls = _run(host)
    assert rc == 0, out
    assert not _recreates(calls), f"пересоздан без расхождения инода: {calls}"


def test_stale_inode_recreates_that_service_and_confirms(host: Path) -> None:
    """Старый инод → пересоздать ИМЕННО этот сервис и убедиться, что доехало."""
    _stale(host)
    rc, out, calls = _run(host, FAKE_NEW_VIEW=str(_ino(host)))
    assert rc == 0, out
    recreates = _recreates(calls)
    assert len(recreates) == 1, f"ожидалось одно пересоздание, было: {calls}"
    assert recreates[0].split()[-1] == "loki", f"пересоздан не тот сервис: {recreates[0]}"
    assert "--no-deps" in recreates[0], f"пересоздание тянет зависимости: {recreates[0]}"
    assert "-p gendesign-metrics -f docker-compose.metrics.yml" in recreates[0], recreates[0]
    assert any(c.startswith("exec cid-new ") for c in calls), (
        f"после пересоздания инод НОВОГО контейнера не сверен: {calls}"
    )


def test_still_stale_after_recreate_fails_the_deploy(host: Path) -> None:
    """Пересоздание не помогло — красный деплой, а не зелёный со старым конфигом."""
    _stale(host)
    rc, out, _calls = _run(host, FAKE_NEW_VIEW=str(_ino(host) + 1))
    assert rc != 0, f"контейнер всё ещё на старом иноде, а скрипт вышел с 0:\n{out}"
    assert SRC in out, f"в логе не назван файл, который не доехал:\n{out}"


def test_unreadable_container_view_counts_as_stale(host: Path) -> None:
    """Инод в контейнере не прочитался — это расхождение, а не «доехало»."""
    (host.parent.parent / "view" / "cid-old").unlink()
    rc, out, calls = _run(host, FAKE_NEW_VIEW=str(_ino(host)))
    assert rc == 0, out
    assert _recreates(calls), f"непрочитанный инод сочтён доехавшим: {calls}"


def test_not_running_container_is_left_alone(host: Path) -> None:
    """Контейнера нет — старый инод держать некому; поднимать сервис тут нельзя."""
    (host.parent.parent / "cid").write_text("", encoding="utf-8")
    rc, out, calls = _run(host)
    assert rc == 0, out
    assert not _recreates(calls), f"поднят незапущенный сервис: {calls}"


def test_compose_failure_is_not_read_as_not_running(host: Path) -> None:
    """Упавший `compose ps` — не «контейнер не запущен»: иначе зелёный выход вслепую."""
    _stale(host)
    rc, out, calls = _run(host, FAKE_PS_RC="1")
    assert rc != 0, f"ошибка compose проглочена, скрипт вышел с 0:\n{out}"
    assert not _recreates(calls), calls


def test_missing_host_file_is_not_success(host: Path) -> None:
    """Файла на хосте нет — сверять не с чем; молча выйти с 0 нельзя."""
    (host / SRC).unlink()
    rc, out, _calls = _run(host)
    assert rc != 0, f"отсутствующий файл сочтён доехавшим:\n{out}"


# ── Гейт: у каждого пофайлового маунта есть путь доезда ──────────────────────

# Производные файлы: их рендерит сам деплой, в git их нет.
#   alertmanager.yml — rm + рендер, пересоздание по ALERTMANAGER_RERENDERED;
#   alertmanager_targets.gen.yml — усечение на месте (`: >`), инод сохраняется.
RENDERED = {
    "ops/metrics/alertmanager/alertmanager.yml",
    "ops/metrics/prometheus/alertmanager_targets.gen.yml",
}
# Источник маунта: ./путь[:${VAR:?…}]:/куда[:ro]. `${…}` внутри источника — файл,
# выбираемый ролью (alloy-apps.alloy / alloy-infra.alloy).
MOUNT = re.compile(r"^\./(?P<src>(?:\$\{[^}]*\}|[^:])+):(?P<dst>/[^:]+)")
COMPOSE_FILES = {
    "docker-compose.metrics.yml": "gendesign-metrics",
    "docker-compose.metrics-agent.yml": "gendesign-metrics-agent",
}


def _jobs_for(compose_name: str, profiles: list[str]) -> list[str]:
    """Джобы деплоя, поднимающие сервис: сервер — одна; агенты — по профилю."""
    if compose_name == "docker-compose.metrics.yml":
        return ["server"]
    return [f"agent-{p}" for p in profiles] or ["agent-apps", "agent-infra"]


def _commands(job: str) -> str:
    """ssh-скрипт джобы: комментарии выкинуты, продолжения строк склеены.

    Комментарии — потому что разбор дефекта содержит его формулировку дословно.
    """
    spec = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
    steps = [s for s in spec["jobs"][job]["steps"] if "ssh-action" in str(s.get("uses"))]
    assert len(steps) == 1, f"в джобе {job} не ровно один ssh-шаг — гейт ослеп"
    kept = [ln for ln in steps[0]["with"]["script"].splitlines() if not ln.lstrip().startswith("#")]
    return re.sub(r"\\\n\s*", " ", "\n".join(kept))


def _file_mounts(compose_name: str) -> list[tuple[str, str, str, list[str]]]:
    spec = yaml.safe_load((REPO_ROOT / compose_name).read_text(encoding="utf-8"))
    found = []
    for service, body in spec["services"].items():
        for vol in body.get("volumes") or []:
            m = MOUNT.match(vol) if isinstance(vol, str) else None
            if not m or (REPO_ROOT / m["src"]).is_dir() or m["src"] in RENDERED:
                continue
            found.append((service, m["src"], m["dst"], body.get("profiles") or []))
    return found


def _arrival(commands: str, project: str, compose_name: str, service: str, src: str, dst: str):
    script_call = re.search(
        rf"recreate-stale-mount\.sh\s+{re.escape(project)}\s+{re.escape(compose_name)}"
        rf"\s+{re.escape(service)}\s+{re.escape(src)}\s+{re.escape(dst)}[^\n]*",
        commands,
    )
    forced = re.search(
        rf"--force-recreate(?:\s+--no-deps)?\s+{re.escape(service)}(?![\w-])[^\n]*", commands
    )
    return script_call or forced


@pytest.mark.parametrize("compose_name", sorted(COMPOSE_FILES))
def test_every_single_file_mount_has_a_way_to_arrive(compose_name: str) -> None:
    project = COMPOSE_FILES[compose_name]
    mounts = _file_mounts(compose_name)
    assert mounts, f"в {compose_name} не найдено ни одного пофайлового маунта — маска ослепла"
    missing = []
    for service, src, dst, profiles in mounts:
        for job in _jobs_for(compose_name, profiles):
            hit = _arrival(_commands(job), project, compose_name, service, src, dst)
            if not hit:
                missing.append(f"{job}: {service} {src} → {dst}")
            elif re.search(r"\|\|\s*(true|:)\s*$", hit.group(0)):
                missing.append(f"{job}: {service} — отказ проглочен: {hit.group(0).strip()}")
    assert not missing, (
        "пофайловые маунты без пути доезда: правка файла ляжет на диск, контейнер "
        "останется на старом иноде, деплой зелёный:\n  " + "\n  ".join(missing)
    )


def test_gate_sees_the_known_mounts() -> None:
    """Проверка на себя: маска обязана находить маунты, ради которых гейт заведён."""
    server = {(s, src) for s, src, _d, _p in _file_mounts("docker-compose.metrics.yml")}
    assert ("prometheus", "ops/metrics/prometheus/prometheus.yml") in server, server
    assert ("loki", "ops/metrics/loki/loki-config.yml") in server, server
    agent = {(s, src) for s, src, _d, _p in _file_mounts("docker-compose.metrics-agent.yml")}
    for exporter in ("gendesign", "tradein", "infra"):
        assert (f"postgres-exporter-{exporter}", "ops/metrics/postgres/queries.yml") in agent, agent


def test_gate_would_notice_a_mount_without_arrival() -> None:
    """Сервис без пути доезда гейт обязан НЕ найти — иначе он зелёный по построению."""
    commands = _commands("server")
    assert not _arrival(
        commands, "gendesign-metrics", "docker-compose.metrics.yml", "nosuch", SRC, DST
    ), "гейт нашёл путь доезда у несуществующего сервиса — маска слишком широкая"