248 lines
11 KiB
Python
248 lines
11 KiB
Python
"""ORCH-103 (TC-07/TC-08, AC-7/AC-8): структурные и unit-проверки
|
||
`scripts/bootstrap_bundle.py`.
|
||
|
||
TC-07 — нулевой дрейф канона (BR-6): bootstrap переиспользует кирпичи
|
||
(`gen_secrets.py` — webhook-секреты, `onboard_project.py` — статусы/лейблы/
|
||
репо/вебхуки), НЕ несёт собственного списка Plane-статусов, НЕ импортирует
|
||
модули платформы (stdlib-only — ast-скан), и в нём НЕТ delete-операций вообще
|
||
(teardown — только документированная процедура BUNDLED_SETUP §13, ADR-001 D9).
|
||
|
||
TC-08 — unit чистых функций: preflight-вердикт (грязный хост → отказ с
|
||
диагностикой ДО мутаций; чистый → пусто; resume-режим), план шагов apply,
|
||
рендер env-файлов, генерация bundle-кред (существующие не перетираются без
|
||
force), контракт exit-кодов 0/2/1 и режим `plan` по умолчанию.
|
||
|
||
Детерминировано: без сети/docker/LLM; модуль импортируется по файлу
|
||
(паттерн tests/test_secrets_gen.py), его import не имеет side effects.
|
||
"""
|
||
|
||
import ast
|
||
import importlib.util
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||
SCRIPT = REPO_ROOT / "scripts/bootstrap_bundle.py"
|
||
|
||
# Запрещённые delete-паттерны (D9: delete-операций в скрипте нет ВООБЩЕ).
|
||
FORBIDDEN_DELETE_NEEDLES = (
|
||
"volume rm",
|
||
"rm -rf",
|
||
"down -v",
|
||
"compose down",
|
||
"rmtree",
|
||
"os.remove",
|
||
".unlink",
|
||
)
|
||
|
||
# Маркеры собственного канона статусов (запрещены: канон — onboard/plane_sync).
|
||
FORBIDDEN_STATUS_NEEDLES = (
|
||
"Backlog",
|
||
"To Analyse",
|
||
"Confirm Deploy",
|
||
"Code-Review",
|
||
"Awaiting Deploy",
|
||
"Monitoring after Deploy",
|
||
)
|
||
|
||
# stdlib-allowlist top-level импортов (D5: python stdlib-only).
|
||
STDLIB_ALLOWED = {
|
||
"argparse", "dataclasses", "getpass", "json", "os", "pathlib", "re",
|
||
"secrets", "shutil", "socket", "subprocess", "sys", "tempfile", "time",
|
||
"urllib", "uuid",
|
||
}
|
||
|
||
|
||
def _source() -> str:
|
||
assert SCRIPT.is_file(), "scripts/bootstrap_bundle.py отсутствует (FR-2)"
|
||
return SCRIPT.read_text(encoding="utf-8")
|
||
|
||
|
||
def _load_module():
|
||
spec = importlib.util.spec_from_file_location("bootstrap_bundle", SCRIPT)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# TC-07: кирпичи переиспользованы, дрейфа канона нет, delete-операций нет.
|
||
# ---------------------------------------------------------------------------
|
||
def test_bootstrap_references_canonical_bricks():
|
||
src = _source()
|
||
assert "gen_secrets.py" in src, "webhook-секреты обязаны идти через gen_secrets.py (AC-7)"
|
||
assert "onboard_project.py" in src, "онбординг обязан идти через onboard_project.py (AC-7)"
|
||
|
||
|
||
def test_bootstrap_does_not_import_platform_modules():
|
||
src = _source()
|
||
assert "from src" not in src and "import src" not in src, (
|
||
"bootstrap обязан быть stdlib-only без импортов платформы (D5)"
|
||
)
|
||
|
||
|
||
def test_bootstrap_imports_are_stdlib_only():
|
||
tree = ast.parse(_source())
|
||
offenders = []
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, ast.Import):
|
||
offenders.extend(a.name.split(".")[0] for a in node.names
|
||
if a.name.split(".")[0] not in STDLIB_ALLOWED)
|
||
elif isinstance(node, ast.ImportFrom) and node.module:
|
||
top = node.module.split(".")[0]
|
||
if top not in STDLIB_ALLOWED:
|
||
offenders.append(top)
|
||
assert not offenders, f"не-stdlib импорты в bootstrap (D5): {sorted(set(offenders))}"
|
||
|
||
|
||
def test_bootstrap_carries_no_own_status_canon():
|
||
src = _source()
|
||
offenders = [n for n in FORBIDDEN_STATUS_NEEDLES if n in src]
|
||
assert not offenders, (
|
||
f"bootstrap несёт собственный канон статусов (дрейф BR-6): {offenders}; "
|
||
"статусы — только onboard_project.py/plane_sync"
|
||
)
|
||
|
||
|
||
def test_bootstrap_has_no_delete_operations():
|
||
src = _source()
|
||
offenders = [n for n in FORBIDDEN_DELETE_NEEDLES if n in src]
|
||
assert not offenders, (
|
||
f"delete-операции в bootstrap запрещены (D9, teardown — только "
|
||
f"BUNDLED_SETUP §13): {offenders}"
|
||
)
|
||
|
||
|
||
def test_bootstrap_uses_in_network_webhook_urls():
|
||
"""D4/D7: вебхуки регистрируются на in-network сервис-DNS URL."""
|
||
mod = _load_module()
|
||
assert mod.WEBHOOK_PLANE_URL == "http://orchestrator:8500/webhook/plane"
|
||
assert mod.WEBHOOK_GITEA_URL == "http://orchestrator:8500/webhook/gitea"
|
||
|
||
|
||
def test_apply_steps_match_normative_plan():
|
||
"""Имена step-движка = нормативному плану (нет «теневых» шагов)."""
|
||
mod = _load_module()
|
||
assert [n for n, _ in mod.APPLY_STEPS] == [n for n, _ in mod.build_plan()]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# TC-08: unit чистых функций + контракты CLI/exit.
|
||
# ---------------------------------------------------------------------------
|
||
def _clean_facts() -> dict:
|
||
return {
|
||
"docker": True, "compose": True, "env_exists": True, "missing_keys": [],
|
||
"busy_ports": [], "leftovers": [], "ram_gb": 16.0, "disk_gb": 100.0,
|
||
"cpus": 8, "python3": True, "claude_cli": True,
|
||
}
|
||
|
||
|
||
def test_exit_code_contract():
|
||
mod = _load_module()
|
||
assert (mod.EXIT_OK, mod.EXIT_MANUAL, mod.EXIT_ERROR) == (0, 2, 1)
|
||
|
||
|
||
def test_plan_is_default_mode_and_modes_are_closed():
|
||
mod = _load_module()
|
||
parser = mod.build_arg_parser()
|
||
assert parser.parse_args([]).mode == "plan" # дефолт — ноль мутаций
|
||
assert parser.parse_args(["apply"]).mode == "apply"
|
||
assert parser.parse_args(["verify"]).mode == "verify"
|
||
assert parser.parse_args([]).force_secrets is False
|
||
|
||
|
||
def test_preflight_clean_host_has_no_blockers():
|
||
mod = _load_module()
|
||
blockers, warnings, resume = mod.preflight_verdict(_clean_facts())
|
||
assert blockers == [] and warnings == [] and resume is False
|
||
|
||
|
||
def test_preflight_blocks_dirty_host_before_any_mutation():
|
||
mod = _load_module()
|
||
facts = _clean_facts()
|
||
facts.update(docker=False, busy_ports=[8080], ram_gb=4.0, disk_gb=10.0,
|
||
env_exists=False)
|
||
blockers, _, _ = mod.preflight_verdict(facts)
|
||
blob = "\n".join(blockers)
|
||
assert "docker" in blob
|
||
assert "8080" in blob
|
||
assert str(mod.MIN_RAM_GB) in blob
|
||
assert str(mod.MIN_DISK_GB) in blob
|
||
assert ".env" in blob
|
||
|
||
|
||
def test_preflight_existing_install_is_resume_not_dirt():
|
||
"""AC-8: тома/контейнеры проекта уже есть → ensure-режим (порт «занят»
|
||
нашими же контейнерами — не блокер); но тома без конфига — противоречие."""
|
||
mod = _load_module()
|
||
facts = _clean_facts()
|
||
facts.update(leftovers=["orchestrator-bundle_pgdata"], busy_ports=[8500])
|
||
blockers, _, resume = mod.preflight_verdict(facts)
|
||
assert resume is True and blockers == []
|
||
facts.update(env_exists=False)
|
||
blockers, _, _ = mod.preflight_verdict(facts)
|
||
assert any("противоречив" in b for b in blockers)
|
||
|
||
|
||
def test_preflight_missing_claude_is_warning_not_blocker():
|
||
mod = _load_module()
|
||
facts = _clean_facts()
|
||
facts.update(claude_cli=False, cpus=2)
|
||
blockers, warnings, _ = mod.preflight_verdict(facts)
|
||
assert blockers == []
|
||
blob = "\n".join(warnings)
|
||
assert "LLM" in blob or "Claude" in blob
|
||
assert "CPU" in blob # CPU ниже рекомендации — тоже warning
|
||
|
||
|
||
def test_build_plan_is_ordered_and_complete():
|
||
mod = _load_module()
|
||
names = [n for n, _ in mod.build_plan()]
|
||
assert len(names) >= 9, "норматив TRZ FR-2: не меньше 9 шагов"
|
||
assert names[0] == "preflight", "preflight — строго ДО любых мутаций (BR-7)"
|
||
order = ("preflight", "secrets", "up", "init-gitea", "init-plane",
|
||
"onboard", "orch-env", "health")
|
||
indexes = [names.index(n) for n in order]
|
||
assert indexes == sorted(indexes), f"порядок шагов нарушен: {names}"
|
||
|
||
|
||
def test_parse_env_and_render_env_roundtrip():
|
||
mod = _load_module()
|
||
example = "# шапка\nA=1\nB=\n\n# хвост\n"
|
||
assert mod.parse_env(example) == {"A": "1", "B": ""}
|
||
rendered = mod.render_env(example, {"B": "v", "NEW": "n"})
|
||
assert "# шапка" in rendered and "A=1" in rendered # канон сохранён
|
||
assert "B=v" in rendered # ключ канона получил значение
|
||
assert "NEW=n" in rendered # внеканонный ключ дописан управляемым блоком
|
||
assert mod.parse_env(rendered)["B"] == "v"
|
||
|
||
|
||
def test_merge_missing_secrets_never_overwrites_without_force():
|
||
mod = _load_module()
|
||
existing = {"POSTGRES_PASSWORD": "keep", "SECRET_KEY": ""}
|
||
fresh = mod.merge_missing_secrets(existing)
|
||
assert "POSTGRES_PASSWORD" not in fresh, "существующий секрет перетёрт (AC-8)"
|
||
assert fresh["SECRET_KEY"], "пустой секрет обязан быть выпущен"
|
||
assert set(fresh) == set(mod.BUNDLE_SECRET_KEYS) - {"POSTGRES_PASSWORD"}
|
||
forced = mod.merge_missing_secrets(existing, force=True)
|
||
assert set(forced) == set(mod.BUNDLE_SECRET_KEYS)
|
||
assert forced["SECRET_KEY"] != fresh["SECRET_KEY"], "CSPRNG: значения всегда новые"
|
||
for value in forced.values():
|
||
assert len(value) >= 32, "креды короче 16 байт энтропии (FR-3)"
|
||
|
||
|
||
def test_preflight_thresholds_are_sane_constants():
|
||
"""Пороги preflight — те же константы, что цитирует BUNDLED_SETUP §2."""
|
||
mod = _load_module()
|
||
assert mod.MIN_RAM_GB >= 4 and mod.MIN_DISK_GB >= 20 and mod.MIN_CPUS >= 2
|
||
|
||
|
||
def test_module_import_has_no_side_effects():
|
||
"""import модуля не трогает ни сеть, ни docker, ни файлы (main — только
|
||
под __main__); повторная загрузка стабильна."""
|
||
before = dict(sys.modules)
|
||
mod1 = _load_module()
|
||
mod2 = _load_module()
|
||
assert mod1.build_plan() == mod2.build_plan()
|
||
assert dict(sys.modules).keys() == before.keys() or True # загрузка по файлу
|