awg-dash/tests/deploy/validate_compose.py

#!/usr/bin/env python3
"""validate_compose.py — статическая проверка compose-файлов стенда (задача #21).

Проверяет инварианты, которые нужны acceptance criteria #21 и не проверяются
самим `docker compose config`:

  deploy/docker-compose.dev.yml
    * приложение слушает контейнерный порт 8080 и публикуется на хост;
    * /config и /data смонтированы (состояние переживает restart);
    * healthcheck ходит в /api/v1/health;
    * выданы NET_ADMIN-возможности, проброшен /dev/net/tun, включён ip_forward;
    * включён ограниченный json-file-лог (иначе логи съедят диск на dev-хосте).

  deploy/docker-compose.test.yml
    * сервис netns привилегированный (нужен CAP_SYS_ADMIN/NET_ADMIN для `ip netns add`);
    * entrypoint — tests/netns/run.sh, отчёт пишется в /artifacts;
    * существуют все файлы, на которые ссылаются тома.

  deploy/docker-compose.stub.yml
    * подменяет entrypoint на deploy/api_stub.py и явно помечен как DEV-ONLY.

Запуск: make test-manifests / python3 tests/deploy/validate_compose.py
Коды возврата: 0 — ок (только warnings), 1 — есть ошибки.
"""

from __future__ import annotations

import argparse
import pathlib
import re
import sys

try:
    import yaml
except ImportError:  # pragma: no cover - зависит от окружения
    yaml = None  # type: ignore[assignment]

ROOT = pathlib.Path(__file__).resolve().parents[2]
DEV = ROOT / "deploy" / "docker-compose.dev.yml"
TEST = ROOT / "deploy" / "docker-compose.test.yml"
STUB = ROOT / "deploy" / "docker-compose.stub.yml"

HEALTH_PATH = "/api/v1/health"


def load(path: pathlib.Path) -> dict:
    if yaml is None:
        raise SystemExit("нужен PyYAML: pip install -r requirements-dev.txt (или make venv)")
    return yaml.safe_load(path.read_text(encoding="utf-8")) or {}


def _ports(service: dict) -> list[str]:
    return [str(p) for p in service.get("ports", []) or []]


def _volumes(service: dict) -> list[str]:
    return [str(v) for v in service.get("volumes", []) or []]


def _mount_targets(service: dict) -> list[str]:
    """Точки монтирования внутри контейнера.

    Строки вида `${AWG_DASH_HOST_DIR:-/opt/awg-dash-dev}/config:/config` содержат
    ':' внутри подстановки переменной, поэтому перед разбором подстановки заменяем
    на плейсхолдер без двоеточий.
    """
    targets = []
    for spec in _volumes(service):
        clean = re.sub(r"\$\{[^}]*\}", "SRC", spec)
        parts = clean.rsplit(":", 2)
        if len(parts) >= 2:
            # 2 части: src:target; 3 части: src:target:mode
            targets.append(parts[-2] if len(parts) == 3 else parts[-1])
    return targets


def validate_dev(errors: list[str], warnings: list[str], require_dockerfile: bool) -> None:
    if not DEV.exists():
        errors.append(f"{DEV.name}: файл не найден")
        return
    doc = load(DEV)
    services = doc.get("services") or {}
    svc = services.get("awg-dash")
    if not svc:
        errors.append("dev: нет сервиса awg-dash")
        return

    if not doc.get("name"):
        errors.append("dev: не задан `name` — compose project не изолирован от других стендов")

    ports = _ports(svc)
    if "8080" not in " ".join(ports):
        errors.append("dev: порт 8080 не публикуется наружу")
    targets = _mount_targets(svc)
    for needle in ("/config", "/data"):
        if needle not in targets:
            errors.append(f"dev: не смонтирован {needle} (состояние не переживёт restart — AC-2)")

    health = (svc.get("healthcheck") or {}).get("test")
    if not health or HEALTH_PATH not in jsonish(health):
        errors.append(f"dev: healthcheck не ходит в {HEALTH_PATH}")

    caps = [str(c) for c in svc.get("cap_add", []) or []]
    if "NET_ADMIN" not in caps:
        errors.append("dev: нет cap_add NET_ADMIN (iptables/ipset/ip rule не заработают)")
    devices = " ".join(str(d) for d in svc.get("devices", []) or [])
    if "/dev/net/tun" not in devices:
        errors.append("dev: не проброшен /dev/net/tun")
    if str((svc.get("sysctls") or {}).get("net.ipv4.ip_forward", "")) not in ("1", "true", "True"):
        errors.append("dev: не включён net.ipv4.ip_forward")

    logging = svc.get("logging") or {}
    if str(logging.get("driver", "")) != "json-file":
        warnings.append("dev: не задан json-file logging — на dev-хосте логи могут расти без ограничений")

    dockerfile = ((svc.get("build") or {}).get("dockerfile")) or "deploy/Dockerfile"
    if not (ROOT / dockerfile).exists():
        msg = f"dev: build.dockerfile {dockerfile} отсутствует в дереве"
        (errors if require_dockerfile else warnings).append(
            msg + ("" if require_dockerfile else " (ожидаемо до мержа #18)")
        )


def validate_test(errors: list[str], warnings: list[str]) -> None:
    if not TEST.exists():
        errors.append(f"{TEST.name}: файл не найден")
        return
    doc = load(TEST)
    services = doc.get("services") or {}
    svc = services.get("netns")
    if not svc:
        errors.append("test: нет сервиса netns (netns-харнесс)")
        return
    if svc.get("privileged") is not True:
        errors.append("test: сервис netns не privileged — `ip netns add` внутри контейнера не сработает")
    entry = " ".join(str(x) for x in svc.get("entrypoint", []) or [])
    if "tests/netns/run.sh" not in entry:
        errors.append("test: entrypoint сервиса netns не указывает на tests/netns/run.sh")
    if not (ROOT / "tests" / "netns" / "run.sh").exists():
        errors.append("test: нет файла tests/netns/run.sh")
    if "/artifacts" not in _mount_targets(svc):
        warnings.append("test: отчёт стенда не выносится в /artifacts")
    tmpfs = " ".join(str(t) for t in svc.get("tmpfs", []) or [])
    if "/tmp" not in tmpfs:
        warnings.append("test: /tmp не tmpfs — логи стенда останутся в слое контейнера")


def validate_stub(errors: list[str], warnings: list[str]) -> None:
    if not STUB.exists():
        warnings.append("stub: оверрайд с health-заглушкой отсутствует")
        return
    doc = load(STUB)
    svc = (doc.get("services") or {}).get("awg-dash")
    if not svc:
        errors.append("stub: нет сервиса awg-dash")
        return
    entry = " ".join(str(x) for x in svc.get("entrypoint", []) or [])
    if "api_stub.py" not in entry:
        errors.append("stub: entrypoint не указывает на deploy/api_stub.py")
    if not (ROOT / "deploy" / "api_stub.py").exists():
        errors.append("stub: нет файла deploy/api_stub.py")
    if "DEV" not in STUB.read_text(encoding="utf-8").upper():
        warnings.append("stub: в файле нет явной пометки DEV-ONLY")


def jsonish(value: object) -> str:
    if isinstance(value, list):
        return " ".join(str(v) for v in value)
    return str(value)


def validate(require_dockerfile: bool = False) -> tuple[list[str], list[str]]:
    errors: list[str] = []
    warnings: list[str] = []
    validate_dev(errors, warnings, require_dockerfile)
    validate_test(errors, warnings)
    validate_stub(errors, warnings)
    return errors, warnings


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Проверка compose-файлов dev/тест-стенда awg-dash (#21)")
    ap.add_argument("--require-dockerfile", action="store_true", help="требовать наличие deploy/Dockerfile (#18 смёржен)")
    ap.add_argument("--quiet", action="store_true")
    args = ap.parse_args(argv)

    errors, warnings = validate(args.require_dockerfile)
    if not args.quiet:
        for warning in warnings:
            print(f"[warn] {warning}")
        for error in errors:
            print(f"[error] {error}")
    if errors:
        print(f"validate-compose: {len(errors)} ошибок, {len(warnings)} предупреждений")
        return 1
    print(f"validate-compose: ok ({len(warnings)} предупреждений)")
    return 0


if __name__ == "__main__":
    sys.exit(main())