#!/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())