#!/usr/bin/env python3 """Детерминированный гейт ревью. Прогоняет всё, у чего есть объективный оракул, и печатает сводку. В отличие от `task test`/`task lint` не останавливается на первом отказе — ревьюверу нужна полная картина, а не первая упавшая команда. Использование: scripts/gate.py [] base-rev — база для диффа. По умолчанию merge-base с master; на самом master — HEAD~1. Шаги выбираются по изменённым файлам: правка документации не гоняет тесты и линтеры. Пропущенный шаг всегда виден в сводке с причиной — молча пропущенная проверка даёт ложное ощущение проверенности, а это ровно то, ради чего гейт и заводился. Коды возврата: 0 — красных шагов нет, 1 — есть. Статусы: OK, FAIL (краснит гейт), WARN (виден, но не блокирует), SKIP. """ import os import shutil import subprocess import sys from pathlib import Path OUT_DIR = Path("tmp/gate") OK, FAIL, WARN, SKIP = "OK", "FAIL", "WARN", "SKIP" summary: list[tuple[str, str, str]] = [] def record(status: str, name: str, hint: str = "") -> None: summary.append((status, name, hint)) def git(*args: str) -> str: return subprocess.run( ["git", *args], capture_output=True, text=True, check=True ).stdout.strip() def base_rev(argv: list[str]) -> str: if len(argv) > 1: return argv[1] on_master = git("rev-parse", "--abbrev-ref", "HEAD") == "master" has_master = subprocess.run( ["git", "rev-parse", "--verify", "-q", "master"], capture_output=True ).returncode == 0 if has_master and not on_master: return git("merge-base", "HEAD", "master") return "HEAD~1" def changed_files(base: str) -> list[str]: """Изменённые файлы: закоммиченное относительно базы + рабочее дерево + новые. Берём объединение намеренно: гейт гоняют и до коммита, и после, и лишний прогон шага дешевле пропущенного. """ files = set(git("diff", "--name-only", base).splitlines()) files |= set(git("ls-files", "--others", "--exclude-standard").splitlines()) return sorted(f for f in files if f) def run(name: str, cmd: list[str], env: dict[str, str] | None = None) -> bool: """Выполняет шаг, складывает вывод в tmp/gate/.log.""" log = OUT_DIR / f"{name}.log" full_env = {**os.environ, **(env or {})} proc = subprocess.run(cmd, capture_output=True, text=True, env=full_env) log.write_text(proc.stdout + proc.stderr, encoding="utf-8") return proc.returncode == 0 def step(name: str, cmd: list[str], hint: str = "", env: dict[str, str] | None = None) -> bool: ok = run(name, cmd, env) record(OK, name) if ok else record( FAIL, name, f"{hint + ' → ' if hint else ''}{OUT_DIR}/{name}.log" ) return ok def main() -> int: OUT_DIR.mkdir(parents=True, exist_ok=True) base = base_rev(sys.argv) changed = changed_files(base) go_changed = any(f.endswith(".go") for f in changed) deps_changed = any(f in ("go.mod", "go.sum") for f in changed) migrations_changed = any(f.startswith("internal/store/migrations/") for f in changed) config_changed = any(f.startswith("internal/config/") for f in changed) code_changed = go_changed or deps_changed no_code = "нет изменений в .go/go.mod — код не трогали" print(f"== gate: база диффа {base}, изменённых файлов {len(changed)} ==") if not code_changed: print(" код не менялся — go-шаги пропускаются, см. сводку") # --- Компиляция и статика --- if code_changed: step("build", ["go", "build", "./..."], "не собирается") step("vet", ["go", "vet", "./..."]) if shutil.which("golangci-lint"): # Кеш линтера — СВОЙ на дерево, а не общий на машину. # # Общий `~/.cache/golangci-lint` отдаёт замечания, привязанные к # путям ЧУЖОГО worktree того же модуля: конвейер задач работает в # `tmp/wt-*`, и прогон в одном дереве красил соседнее сообщением про # файл, которого в нём нет (воспроизведено дважды за одну задачу, # оба раза `../…/internal/store/store.go: time.Now forbidden` — а # исключение `^internal/(ident|store)/` на такой путь не # распространяется). Краснота по причине, не связанной с # изменением, приучает не читать красноту — та же цена, что у # записи 2026-08-04 в docs/review.md. step( "lint", ["golangci-lint", "run"], env={"GOLANGCI_LINT_CACHE": str((OUT_DIR / "golangci").resolve())}, ) else: record(SKIP, "lint", "golangci-lint не установлен (task setup)") unformatted = [ f for f in subprocess.run( ["gofmt", "-l", "."], capture_output=True, text=True ).stdout.split() if not f.startswith("tmp/") ] if unformatted: record(FAIL, "gofmt", "не отформатировано: " + " ".join(unformatted)) else: record(OK, "gofmt") else: for name in ("build", "vet", "lint", "gofmt"): record(SKIP, name, no_code) # --- Тесты --- if code_changed: tests_ok = step("test", ["go", "test", "-count=1", "./..."]) # Флаки: повторный прогон. Тест, который иногда зелёный, не является # оракулом ни для чего, поэтому расхождение — находка не ниже major. # Стабильно красный набор флаки не проверяем: его разбирает шаг test. if tests_ok: if run("test-repeat", ["go", "test", "-count=1", "./..."]): record(OK, "flaky") else: record(FAIL, "flaky", "прогон 1 зелёный, прогон 2 красный — флаки-тест") else: record(SKIP, "flaky", "набор красный — сперва чиним test") else: record(SKIP, "test", no_code) record(SKIP, "flaky", no_code) # --- Гонки --- if not code_changed: record(SKIP, "race", no_code) elif shutil.which("gcc"): step( "race", ["go", "test", "-race", "-count=1", "./..."], "гонка либо сборка тестов — смотри лог", env={"CGO_ENABLED": "1"}, ) else: record(SKIP, "race", "нет gcc: -race требует cgo. Гонки НЕ проверены") # --- Покрытие изменённых строк --- if not go_changed: record(SKIP, "diff-coverage", "нет изменений в .go") elif run("cover", ["go", "test", "-count=1", f"-coverprofile={OUT_DIR}/cover.out", "./..."]): if run("diff-coverage", ["python3", "scripts/diff-coverage.py", f"{OUT_DIR}/cover.out", base]): record(OK, "diff-coverage", (OUT_DIR / "diff-coverage.log").read_text().splitlines()[0]) else: record(SKIP, "diff-coverage", f"не удалось посчитать → {OUT_DIR}/diff-coverage.log") else: record(SKIP, "diff-coverage", f"прогон с профилем не собрался → {OUT_DIR}/cover.log") # --- Миграции на чистой схеме --- if migrations_changed or go_changed: step( "migrations", ["go", "test", "-count=1", "-run", "Migration", "./internal/store/..."], "миграции не накатываются с нуля", ) else: record(SKIP, "migrations", "миграции и код не менялись") # --- ER-схема синхронна с миграциями --- # Меняем структуру — обновляем ER-схему в том же change. Проверка по # диффу, потому и живёт здесь, а не в правиле линтера. if migrations_changed: if "docs/database.md" in changed: record(OK, "er-schema") else: record(FAIL, "er-schema", "миграция изменена, а docs/database.md — нет") # --- Образцы конфигурации синхронны с его структурой --- # Конвенция: config.example.toml самодокументируемый и полный. Забытое поле # обнаруживается не тестом, а тем, что через полгода никто не знает о его # существовании, — поэтому проверяем механически. if config_changed: samples = [s for s in ("config.example.toml", "config.docker.toml") if s not in changed] if samples: record(FAIL, "config-samples", "internal/config изменён, а образцы — нет: " + ", ".join(samples)) else: record(OK, "config-samples") # --- Секреты --- # Гоняем всегда: секрет утекает из любого файла, не только из кода. Для # healthlog это ещё и данные о здоровье — они чувствительнее токенов. if shutil.which("gitleaks"): step("gitleaks", ["gitleaks", "git", "--no-banner"], "возможен секрет в истории/индексе") else: record(SKIP, "gitleaks", "gitleaks не установлен") # --- Данные не утекли в репозиторий --- # Каталог ./data (база + сырой архив) под .gitignore. Попадание любого его # файла в индекс означает утечку выгрузок Apple Health в историю git, # откуда их уже не убрать простым коммитом. tracked_data = [f for f in git("ls-files").splitlines() if f.startswith("data/") or f.endswith((".db", ".db-wal", ".db-shm"))] if tracked_data: record(FAIL, "no-health-data", "данные о здоровье под контролем версий: " + ", ".join(tracked_data[:5])) else: record(OK, "no-health-data") # --- Уязвимости зависимостей --- # Не блокирует: находка тут — состояние зависимостей и тулчейна, а не диффа. if not code_changed: record(SKIP, "govulncheck", no_code) elif shutil.which("govulncheck"): if run("govulncheck", ["govulncheck", "./..."]): record(OK, "govulncheck") else: # Ненулевой код возврата означает и найденные уязвимости, и отказ # самого инструмента (чаще всего код не собирается). Различаем: без # этого «уязвимостей: 0» выглядит как проверка, которой не было. found = (OUT_DIR / "govulncheck.log").read_text().count("\nVulnerability #") if found: record(WARN, "govulncheck", f"достижимо из кода уязвимостей: {found} → {OUT_DIR}/govulncheck.log") else: record(SKIP, "govulncheck", f"не отработал (обычно код не собирается) → {OUT_DIR}/govulncheck.log") else: record(SKIP, "govulncheck", "govulncheck не установлен (task setup)") # --- Сводка --- print("\n== сводка ==") for status, name, hint in summary: print(f"{status:<5} {name:<15} {hint}") if any(s == FAIL for s, _, _ in summary): print("\nГЕЙТ КРАСНЫЙ — опиниативные проходы не запускаются") return 1 print("\nгейт зелёный (шаги WARN и SKIP см. в сводке — они идут в находки" " и в границы покрытия)") return 0 if __name__ == "__main__": sys.exit(main())