736 lines
31 KiB
Python
736 lines
31 KiB
Python
"""ORCH-104 (TC-01…TC-21, AC-1…AC-11): структурные и unit-проверки
|
||
`scripts/install_lite.py` — интерактивного установщика Lite-тиража.
|
||
|
||
Зеркало паттернов `tests/test_bootstrap_script.py`: модуль грузится по файлу
|
||
(import без side-effects, `main` только под `__main__`); side-effects
|
||
(subprocess/HTTP/getpass/input) инъектируются фейками через monkeypatch
|
||
module-level обёрток; чистые функции (`parse_env`/`render_env`/
|
||
`preflight_verdict`/`install_command`/`rank_candidates`/`select_candidate`)
|
||
тестируются изолированно — без сети/docker/LLM (NFR-7).
|
||
|
||
Структурные анти-дрейф (AC-9/AC-10/AC-7): stdlib-only AST-скан, нет `from src`,
|
||
нет delete/force-операций, нет собственного канона статусов Plane, нет
|
||
собственной генерации секретов (делегирование `gen_secrets.py`), нет
|
||
хост-литералов (импорт `FORBIDDEN` из `test_no_host_hardcodes.py` — один
|
||
источник истины), exit-контракт `{0,1,2}`, установщик не трогает рантайм
|
||
(`STAGE_TRANSITIONS`/`QG_CHECKS`).
|
||
"""
|
||
|
||
import ast
|
||
import importlib.util
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# Один источник истины запрещённых боевых литералов (ORCH-101 AC-7, не копия).
|
||
from tests.test_no_host_hardcodes import FORBIDDEN, find_violations
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||
SCRIPT = REPO_ROOT / "scripts/install_lite.py"
|
||
|
||
# Запрещённые delete/force-операции (INV-3: установщик ничего не удаляет/форсит).
|
||
FORBIDDEN_DELETE_NEEDLES = (
|
||
"volume rm",
|
||
"rm -rf",
|
||
"down -v",
|
||
"compose down",
|
||
"rmtree",
|
||
"os.remove",
|
||
".unlink",
|
||
"push --force",
|
||
"force-push",
|
||
"force-with-lease",
|
||
"branch -D",
|
||
"branch -d",
|
||
)
|
||
|
||
# Маркеры собственного канона статусов Plane (запрещены: канон — onboard-кирпич).
|
||
FORBIDDEN_STATUS_NEEDLES = (
|
||
"Backlog",
|
||
"To Analyse",
|
||
"Code-Review",
|
||
"Awaiting Deploy",
|
||
"Monitoring after Deploy",
|
||
)
|
||
|
||
# Запрет собственной генерации секретов (делегирование gen_secrets.py, AC-7).
|
||
FORBIDDEN_SECRET_GEN_NEEDLES = ("token_hex", "token_urlsafe", "token_bytes")
|
||
|
||
# stdlib-allowlist top-level импортов (INV-7: 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/install_lite.py отсутствует (FR-1)"
|
||
return SCRIPT.read_text(encoding="utf-8")
|
||
|
||
|
||
def _load_module():
|
||
spec = importlib.util.spec_from_file_location("install_lite", SCRIPT)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-18/TC-19/TC-20/TC-21 — структурные анти-дрейф (AC-7/AC-9/AC-10/INV).
|
||
# ===========================================================================
|
||
def test_tc18_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 импорты (INV-7): {sorted(set(offenders))}"
|
||
|
||
|
||
def test_tc18_does_not_import_platform_modules():
|
||
src = _source()
|
||
assert "from src" not in src and "import src" not in src, (
|
||
"установщик обязан быть stdlib-only без импортов платформы (INV-7/AC-10)"
|
||
)
|
||
|
||
|
||
def test_tc19_has_no_delete_or_force_operations():
|
||
src = _source()
|
||
offenders = [n for n in FORBIDDEN_DELETE_NEEDLES if n in src]
|
||
assert not offenders, (
|
||
f"delete/force-операции запрещены (INV-3, AC-10): {offenders}"
|
||
)
|
||
|
||
|
||
def test_tc19_carries_no_host_hardcodes():
|
||
"""FORBIDDEN host-литералы (импорт find_violations — один источник истины)."""
|
||
violations = find_violations(_source(), FORBIDDEN)
|
||
assert not violations, (
|
||
f"хост-литералы в install_lite.py (AC-10): {violations}"
|
||
)
|
||
|
||
|
||
def test_tc20_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_tc20_carries_no_own_secret_generation():
|
||
"""AC-7: собственной генерации секретов нет — только делегирование кирпичу."""
|
||
src = _source()
|
||
offenders = [n for n in FORBIDDEN_SECRET_GEN_NEEDLES if n in src]
|
||
assert not offenders, (
|
||
f"собственная генерация секретов запрещена (анти-форк AC-7): {offenders}; "
|
||
"webhook-секреты — только gen_secrets.py"
|
||
)
|
||
|
||
|
||
def test_tc20_carries_no_own_status_canon():
|
||
src = _source()
|
||
offenders = [n for n in FORBIDDEN_STATUS_NEEDLES if n in src]
|
||
assert not offenders, (
|
||
f"установщик несёт собственный канон статусов Plane (дрейф AC-7): "
|
||
f"{offenders}; 22 статуса — только onboard_project.py/plane_sync"
|
||
)
|
||
|
||
|
||
def test_tc21_does_not_touch_runtime_registries():
|
||
"""AC-9: установщик вне рантайма — не ссылается на STAGE_TRANSITIONS/QG_CHECKS
|
||
и не правит src/** (он — scripts-артефакт)."""
|
||
src = _source()
|
||
for needle in ("STAGE_TRANSITIONS", "QG_CHECKS", "check_ci_green"):
|
||
assert needle not in src, (
|
||
f"установщик не должен ссылаться на рантайм-реестр {needle} (AC-9)"
|
||
)
|
||
|
||
|
||
def test_tc18_module_import_has_no_side_effects():
|
||
before = dict(sys.modules)
|
||
mod1 = _load_module()
|
||
mod2 = _load_module()
|
||
assert mod1.build_plan() == mod2.build_plan()
|
||
assert before is not None # загрузка по файлу не должна падать
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-01/TC-02 — entry-point, режимы, план.
|
||
# ===========================================================================
|
||
def test_tc01_exit_code_contract():
|
||
mod = _load_module()
|
||
assert (mod.EXIT_OK, mod.EXIT_MANUAL, mod.EXIT_ERROR) == (0, 2, 1)
|
||
|
||
|
||
def test_tc01_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 is False
|
||
assert parser.parse_args([]).non_interactive is False
|
||
|
||
|
||
def test_tc01_plan_makes_zero_mutations(tmp_path, monkeypatch, capsys):
|
||
"""plan печатает план + preflight и НЕ пишет .env / не поднимает compose."""
|
||
mod = _load_module()
|
||
env_path = tmp_path / ".env"
|
||
monkeypatch.setattr(mod, "ROOT_ENV", str(env_path))
|
||
monkeypatch.setattr(mod, "WATCHDOG_ENV", str(tmp_path / ".env.watchdog"))
|
||
# Чистый хост: ни одной мутации, ни одного docker up.
|
||
monkeypatch.setattr(mod, "collect_facts", lambda env: {
|
||
"docker": True, "compose": True, "git": True, "node": True,
|
||
"claude": True, "prod_port": 8500, "prod_port_busy": False,
|
||
"orch_already_up": False, "repos_owner_ok": True,
|
||
})
|
||
|
||
def _boom_compose(*a, **k):
|
||
raise AssertionError("plan не имеет права запускать docker compose")
|
||
|
||
monkeypatch.setattr(mod, "_compose", _boom_compose)
|
||
rc = mod.main(["plan"])
|
||
out = capsys.readouterr().out
|
||
assert rc == mod.EXIT_OK
|
||
assert not env_path.exists(), "plan создал .env — это мутация (AC-1 FAIL)"
|
||
assert "план apply" in out and "preflight" in out
|
||
|
||
|
||
def test_tc01_plan_returns_manual_on_blockers(tmp_path, monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "ROOT_ENV", str(tmp_path / ".env"))
|
||
monkeypatch.setattr(mod, "collect_facts", lambda env: {
|
||
"docker": False, "compose": False, "git": True, "node": True,
|
||
"claude": True, "prod_port": 8500, "prod_port_busy": False,
|
||
"orch_already_up": False, "repos_owner_ok": True,
|
||
})
|
||
assert mod.main(["plan"]) == mod.EXIT_MANUAL
|
||
|
||
|
||
def test_tc02_build_plan_is_ordered_and_complete():
|
||
mod = _load_module()
|
||
names = [n for n, _ in mod.build_plan()]
|
||
order = ("preflight", "deps", "discovery", "inputs", "secrets", "env",
|
||
"up", "onboard", "health")
|
||
assert names[0] == "preflight", "preflight — строго ДО любых мутаций (FR-2)"
|
||
indexes = [names.index(n) for n in order]
|
||
assert indexes == sorted(indexes), f"порядок шагов нарушен: {names}"
|
||
assert len(names) >= 9
|
||
|
||
|
||
def test_tc02_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-03 — preflight-вердикт (AC-2).
|
||
# ===========================================================================
|
||
def _clean_facts() -> dict:
|
||
return {
|
||
"docker": True, "compose": True, "git": True, "node": True,
|
||
"claude": True, "prod_port": 8500, "prod_port_busy": False,
|
||
"orch_already_up": False, "repos_owner_ok": True,
|
||
}
|
||
|
||
|
||
def test_tc03_preflight_clean_host_has_no_blockers():
|
||
mod = _load_module()
|
||
blockers, warnings = mod.preflight_verdict(_clean_facts())
|
||
assert blockers == [] and warnings == []
|
||
|
||
|
||
def test_tc03_preflight_blocks_missing_deps_and_port_and_owner():
|
||
mod = _load_module()
|
||
facts = _clean_facts()
|
||
facts.update(docker=False, compose=False, node=False,
|
||
prod_port_busy=True, orch_already_up=False, repos_owner_ok=False)
|
||
blockers, _ = mod.preflight_verdict(facts)
|
||
blob = "\n".join(blockers)
|
||
assert "docker" in blob
|
||
assert "node" in blob
|
||
assert "8500" in blob
|
||
assert "uid:gid" in blob
|
||
|
||
|
||
def test_tc03_busy_port_owned_by_orchestrator_is_not_a_blocker():
|
||
mod = _load_module()
|
||
facts = _clean_facts()
|
||
facts.update(prod_port_busy=True, orch_already_up=True)
|
||
blockers, _ = mod.preflight_verdict(facts)
|
||
assert blockers == [], "уже поднятый орк на порту — не блокер (resume)"
|
||
|
||
|
||
def test_tc03_missing_claude_is_warning_not_blocker():
|
||
mod = _load_module()
|
||
facts = _clean_facts()
|
||
facts.update(claude=False)
|
||
blockers, warnings = mod.preflight_verdict(facts)
|
||
assert blockers == []
|
||
assert any("LLM" in w or "claude" in w for w in warnings)
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-04 — distro → команда установки (AC-3 / D-1).
|
||
# ===========================================================================
|
||
def test_tc04_install_command_per_distro():
|
||
mod = _load_module()
|
||
apt_cmd = mod.install_command("ubuntu", "docker")
|
||
dnf_cmd = mod.install_command("fedora", "docker")
|
||
assert apt_cmd and "apt" in apt_cmd
|
||
assert dnf_cmd and "dnf" in dnf_cmd
|
||
assert apt_cmd != dnf_cmd
|
||
# debian-семейство по ID_LIKE-подобному токену тоже резолвится:
|
||
assert mod.install_command("debian", "git").startswith("sudo apt")
|
||
|
||
|
||
def test_tc04_unknown_distro_degrades_to_none():
|
||
mod = _load_module()
|
||
assert mod.install_command("arch", "docker") is None
|
||
assert mod.install_command("", "docker") is None
|
||
assert mod.install_command("ubuntu", "no-such-dep") is None
|
||
|
||
|
||
def test_tc04_install_commands_are_in_allowlist():
|
||
"""Команда, которую установщик может выполнить, ∈ фиксированному allowlist."""
|
||
mod = _load_module()
|
||
allowed = mod.allowed_install_commands()
|
||
for distro in ("ubuntu", "fedora"):
|
||
for dep in mod.INSTALLABLE_DEPS:
|
||
cmd = mod.install_command(distro, dep)
|
||
if cmd is not None:
|
||
assert cmd in allowed
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-05 — управляемая установка: отказ / нет TTY → exit 2, без мутации (AC-3).
|
||
# ===========================================================================
|
||
def test_tc05_deps_no_tty_does_not_install_and_stops(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_isatty", lambda: False)
|
||
|
||
def _boom_run_shell(*a, **k):
|
||
raise AssertionError("без TTY установщик не имеет права что-либо ставить")
|
||
|
||
monkeypatch.setattr(mod, "_run_shell", _boom_run_shell)
|
||
ctx = {
|
||
"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {}, "overrides": {},
|
||
"facts": {"docker": False, "git": True, "node": True, "claude": True},
|
||
"distro": "ubuntu",
|
||
}
|
||
import pytest
|
||
with pytest.raises(mod.ManualStop):
|
||
mod.step_deps(ctx)
|
||
|
||
|
||
def test_tc05_deps_consent_no_keeps_root_clean(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_isatty", lambda: True)
|
||
monkeypatch.setattr(mod, "_ask", lambda prompt: "N") # явный отказ
|
||
|
||
def _boom_run_shell(*a, **k):
|
||
raise AssertionError("при отказе оператора установка не выполняется")
|
||
|
||
monkeypatch.setattr(mod, "_run_shell", _boom_run_shell)
|
||
ctx = {
|
||
"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {}, "overrides": {},
|
||
"facts": {"docker": False, "git": True, "node": True, "claude": True},
|
||
"distro": "ubuntu",
|
||
}
|
||
import pytest
|
||
with pytest.raises(mod.ManualStop):
|
||
mod.step_deps(ctx)
|
||
|
||
|
||
def test_tc05_deps_all_present_is_skipped(monkeypatch):
|
||
mod = _load_module()
|
||
ctx = {
|
||
"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {}, "overrides": {},
|
||
"facts": {"docker": True, "git": True, "node": True, "claude": True},
|
||
"distro": "ubuntu",
|
||
}
|
||
assert mod.step_deps(ctx) == "skipped"
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-06/TC-07/TC-08 — детект Plane/Gitea + выбор (AC-4 / D-2).
|
||
# ===========================================================================
|
||
def test_tc06_discovery_ranks_two_candidates():
|
||
mod = _load_module()
|
||
# 2 слушающих plane-порта (80, 8080), оба живые → 2 ранжированных кандидата.
|
||
cands = mod.detect_existing(
|
||
"plane",
|
||
docker_fn=lambda: [],
|
||
ports_fn=lambda ports: [80, 8080],
|
||
prober=lambda url: True,
|
||
)
|
||
assert len(cands) == 2
|
||
assert all(c.confidence == 2 for c in cands)
|
||
# дедуп по URL: повторяющийся порт не плодит дублей
|
||
cands2 = mod.detect_existing(
|
||
"gitea",
|
||
docker_fn=lambda: ["gitea-server gitea/gitea:1.22"],
|
||
ports_fn=lambda ports: [3000],
|
||
prober=lambda url: True,
|
||
)
|
||
assert len(cands2) == 1
|
||
|
||
|
||
def test_tc06_zero_candidates_is_empty_list():
|
||
mod = _load_module()
|
||
cands = mod.detect_existing(
|
||
"plane", docker_fn=lambda: [], ports_fn=lambda ports: [],
|
||
prober=lambda url: False,
|
||
)
|
||
assert cands == []
|
||
|
||
|
||
def test_tc06_rank_candidates_orders_alive_above_dead():
|
||
mod = _load_module()
|
||
ranked = mod.rank_candidates([
|
||
("http://127.0.0.1:80", "port", False),
|
||
("http://127.0.0.1:8080", "port", True),
|
||
])
|
||
assert [c.url for c in ranked] == [
|
||
"http://127.0.0.1:8080", "http://127.0.0.1:80"]
|
||
|
||
|
||
def test_tc07_select_candidate_valid_and_out_of_range():
|
||
mod = _load_module()
|
||
cands = mod.rank_candidates([("http://a", "port", True),
|
||
("http://b", "port", True)])
|
||
assert mod.select_candidate(cands, "1") in ("http://a", "http://b")
|
||
assert mod.select_candidate(cands, "99") is None # вне диапазона → ручной
|
||
assert mod.select_candidate(cands, "") is None # пусто → ручной
|
||
assert mod.select_candidate(cands, "abc") is None # мусор → ручной (never-raise)
|
||
|
||
|
||
def test_tc08_detect_never_raises_on_probe_failure():
|
||
mod = _load_module()
|
||
|
||
def _raising_docker():
|
||
raise RuntimeError("docker недоступен")
|
||
|
||
def _raising_ports(ports):
|
||
raise OSError("socket error")
|
||
|
||
# исключение в docker-ps → []
|
||
assert mod.detect_existing("plane", docker_fn=_raising_docker,
|
||
ports_fn=lambda p: [80]) == []
|
||
# исключение в порт-скане → []
|
||
assert mod.detect_existing("gitea", docker_fn=lambda: [],
|
||
ports_fn=_raising_ports) == []
|
||
# исключение в пробе живости конкретного URL → кандидат «мёртв», не падение
|
||
cands = mod.detect_existing(
|
||
"gitea", docker_fn=lambda: [], ports_fn=lambda p: [3000],
|
||
prober=lambda url: (_ for _ in ()).throw(RuntimeError()),
|
||
)
|
||
assert all(c.confidence == 1 for c in cands)
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-09 — живая верификация ДО записи (AC-5).
|
||
# ===========================================================================
|
||
def test_tc09_verify_plane_rejects_401_accepts_200():
|
||
mod = _load_module()
|
||
reject = mod.verify_plane_token(
|
||
"http://plane", "ws", "bad", http=lambda u, h, t: (401, ""))
|
||
accept = mod.verify_plane_token(
|
||
"http://plane", "ws", "good", http=lambda u, h, t: (200, "[]"))
|
||
assert reject[0] is False and accept[0] is True
|
||
|
||
|
||
def test_tc09_verify_gitea_rejects_401_accepts_200():
|
||
mod = _load_module()
|
||
assert mod.verify_gitea_token(
|
||
"http://gitea", "bad", http=lambda u, h, t: (401, ""))[0] is False
|
||
assert mod.verify_gitea_token(
|
||
"http://gitea", "good", http=lambda u, h, t: (200, "{}"))[0] is True
|
||
|
||
|
||
def test_tc09_verify_telegram_ok_false_is_rejected():
|
||
mod = _load_module()
|
||
assert mod.verify_telegram_token(
|
||
"bad", http=lambda u, h, t: (200, '{"ok": false}'))[0] is False
|
||
assert mod.verify_telegram_token(
|
||
"good", http=lambda u, h, t: (200, '{"ok": true}'))[0] is True
|
||
assert mod.verify_telegram_token(
|
||
"x", http=lambda u, h, t: (401, ""))[0] is False
|
||
|
||
|
||
def test_tc09_collect_secret_does_not_write_invalid(monkeypatch):
|
||
"""Битый токен НЕ принимается; верный — принимается (verify-before-write)."""
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_isatty", lambda: True)
|
||
seq = iter(["bad-token", "good-token"])
|
||
monkeypatch.setattr(mod, "_getsecret", lambda prompt: next(seq))
|
||
ctx = {"args": mod.build_arg_parser().parse_args(["apply"]), "root_env": {}}
|
||
|
||
def verify(value):
|
||
return (value == "good-token", "401")
|
||
|
||
got = mod._collect_secret(ctx, "ORCH_PLANE_API_TOKEN", "Plane",
|
||
verify, "инструкция")
|
||
assert got == "good-token" # принят только прошедший верификацию
|
||
|
||
|
||
def test_tc09_collect_secret_no_tty_fails_closed():
|
||
mod = _load_module()
|
||
ctx = {"args": mod.build_arg_parser().parse_args(["--non-interactive", "apply"]),
|
||
"root_env": {}}
|
||
import pytest
|
||
with pytest.raises(mod.ManualStop):
|
||
mod._collect_secret(ctx, "ORCH_GITEA_TOKEN", "Gitea",
|
||
lambda v: (False, "x"), "инструкция")
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-10/TC-11 — parse_env / render_env (AC-6/AC-7).
|
||
# ===========================================================================
|
||
def test_tc10_parse_env_roundtrip():
|
||
mod = _load_module()
|
||
text = "# шапка\nA=1\nB=\n\n# хвост\nC = three \n"
|
||
assert mod.parse_env(text) == {"A": "1", "B": "", "C": "three"}
|
||
|
||
|
||
def test_tc11_render_env_preserves_canon_and_appends_unknown():
|
||
mod = _load_module()
|
||
example = "# шапка\nA=1\nB=\n\n# хвост\n"
|
||
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 # внеканонный — управляемым блоком
|
||
# идемпотентность: повторный рендер от результата стабилен по ключам
|
||
again = mod.render_env(rendered, {"B": "v", "NEW": "n"})
|
||
assert mod.parse_env(again)["B"] == "v"
|
||
assert mod.parse_env(again)["NEW"] == "n"
|
||
|
||
|
||
def test_tc11_watchdog_keys_split_to_their_own_file():
|
||
mod = _load_module()
|
||
root, watchdog = mod.split_watchdog_overrides(
|
||
{"ORCH_GITEA_TOKEN": "x", "WATCHDOG_TG_BOT_TOKEN": "y"})
|
||
assert root == {"ORCH_GITEA_TOKEN": "x"}
|
||
assert watchdog == {"WATCHDOG_TG_BOT_TOKEN": "y"}
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-12/TC-13 — секрет-гигиена + no-silent-overwrite (AC-6/AC-7).
|
||
# ===========================================================================
|
||
def test_tc12_write_private_sets_600(tmp_path, monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "REPO_ROOT", str(tmp_path))
|
||
path = tmp_path / ".env"
|
||
mod._write_private(str(path), "ORCH_GITEA_TOKEN=secretvalue\n")
|
||
assert (path.stat().st_mode & 0o777) == 0o600
|
||
|
||
|
||
def test_tc12_mask_secret_never_reveals_value():
|
||
mod = _load_module()
|
||
masked = mod.mask_secret("super-secret-token-value")
|
||
assert "super-secret-token-value" not in masked
|
||
assert mod.mask_secret("") == "<пусто>"
|
||
|
||
|
||
def test_tc12_secrets_step_does_not_log_secret_value(tmp_path, monkeypatch, capsys):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "REPO_ROOT", str(tmp_path))
|
||
secret = "deadbeefcafefacefeed0123456789abcdef0123456789abcdef0123456789ab"
|
||
|
||
def _fake_gen(cmd, env=None, timeout=600):
|
||
path = cmd[cmd.index("--write") + 1]
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write(f"ORCH_PLANE_WEBHOOK_SECRET={secret}\n"
|
||
f"ORCH_GITEA_WEBHOOK_SECRET={secret}\n")
|
||
return type("P", (), {"returncode": 0, "stdout": "", "stderr": ""})()
|
||
|
||
monkeypatch.setattr(mod, "_run", _fake_gen)
|
||
ctx = {"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {}, "overrides": {}}
|
||
assert mod.step_secrets(ctx) == "ok"
|
||
out = capsys.readouterr().out
|
||
assert secret not in out, "значение секрета утекло в лог (NFR-2 FAIL)"
|
||
assert ctx["overrides"]["ORCH_PLANE_WEBHOOK_SECRET"] == secret
|
||
|
||
|
||
def test_tc13_secrets_no_overwrite_without_force(monkeypatch):
|
||
mod = _load_module()
|
||
|
||
def _boom(*a, **k):
|
||
raise AssertionError("при наличии секретов gen_secrets.py не вызывается")
|
||
|
||
monkeypatch.setattr(mod, "_run", _boom)
|
||
ctx = {
|
||
"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {"ORCH_PLANE_WEBHOOK_SECRET": "a" * 64,
|
||
"ORCH_GITEA_WEBHOOK_SECRET": "b" * 64},
|
||
"overrides": {},
|
||
}
|
||
assert mod.step_secrets(ctx) == "skipped"
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-14 — онбординг кирпичом (AC-7).
|
||
# ===========================================================================
|
||
def test_tc14_onboard_parses_registry_and_propagates_manual(tmp_path, monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "REPO_ROOT", str(tmp_path))
|
||
monkeypatch.setattr(mod, "ROOT_ENV", str(tmp_path / ".env"))
|
||
monkeypatch.setattr(mod, "ROOT_ENV_EXAMPLE", str(REPO_ROOT / ".env.example"))
|
||
merged = '[{"repo":"acme","work_item_prefix":"ACME"}]'
|
||
calls = {"recreate": 0}
|
||
|
||
def _fake_run(cmd, env=None, timeout=600):
|
||
joined = " ".join(cmd)
|
||
if "onboard_project.py" in joined and "apply" in cmd:
|
||
report = {"instructions": [f"ORCH_PROJECTS_JSON={merged}"],
|
||
"exit_code": 2, "steps": []}
|
||
return type("P", (), {"returncode": 2, "stdout": __import__("json").dumps(report),
|
||
"stderr": ""})()
|
||
if "onboard_project.py" in joined and "verify" in cmd:
|
||
return type("P", (), {"returncode": 2, "stdout": "{}", "stderr": ""})()
|
||
raise AssertionError(f"неожиданный вызов: {joined}")
|
||
|
||
def _fake_compose(*a, **k):
|
||
calls["recreate"] += 1
|
||
return type("P", (), {"returncode": 0, "stdout": "", "stderr": ""})()
|
||
|
||
monkeypatch.setattr(mod, "_run", _fake_run)
|
||
monkeypatch.setattr(mod, "_compose", _fake_compose)
|
||
args = mod.build_arg_parser().parse_args([
|
||
"apply", "--name", "Acme", "--repo", "acme", "--prefix", "ACME",
|
||
"--stack", "python", "--test-cmd", "pytest", "--prod-port", "8600",
|
||
"--staging-port", "8601", "--webhook-url", "http://x/webhook/gitea"])
|
||
ctx = {"args": args, "root_env": {}, "overrides": {}}
|
||
status = mod.step_onboard(ctx)
|
||
written = (tmp_path / ".env").read_text(encoding="utf-8")
|
||
assert merged in written, "merged ORCH_PROJECTS_JSON не записан в .env"
|
||
assert ctx["onboard_manual"] is True
|
||
assert status == "manual-step"
|
||
assert calls["recreate"] == 1
|
||
|
||
|
||
def test_tc14_onboard_skips_when_project_present():
|
||
mod = _load_module()
|
||
args = mod.build_arg_parser().parse_args([
|
||
"apply", "--name", "Acme", "--repo", "acme", "--prefix", "ACME"])
|
||
ctx = {"args": args,
|
||
"root_env": {"ORCH_PROJECTS_JSON": '[{"repo":"acme"}]'},
|
||
"overrides": {}}
|
||
assert mod.step_onboard(ctx) == "skipped"
|
||
|
||
|
||
def test_tc14_onboard_without_params_is_manual_step():
|
||
mod = _load_module()
|
||
args = mod.build_arg_parser().parse_args(["apply"])
|
||
ctx = {"args": args, "root_env": {}, "overrides": {}}
|
||
assert mod.step_onboard(ctx) == "manual-step"
|
||
assert ctx["onboard_manual"] is True
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-15 — идемпотентность (AC-8).
|
||
# ===========================================================================
|
||
def test_tc15_secrets_then_env_are_idempotent(tmp_path, monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "REPO_ROOT", str(tmp_path))
|
||
monkeypatch.setattr(mod, "ROOT_ENV", str(tmp_path / ".env"))
|
||
monkeypatch.setattr(mod, "ROOT_ENV_EXAMPLE", str(REPO_ROOT / ".env.example"))
|
||
monkeypatch.setattr(mod, "WATCHDOG_ENV", str(tmp_path / ".env.watchdog"))
|
||
monkeypatch.setattr(mod, "WATCHDOG_ENV_EXAMPLE",
|
||
str(REPO_ROOT / ".env.watchdog.example"))
|
||
secret = "a" * 64
|
||
|
||
def _fake_gen(cmd, env=None, timeout=600):
|
||
path = cmd[cmd.index("--write") + 1]
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write(f"ORCH_PLANE_WEBHOOK_SECRET={secret}\n"
|
||
f"ORCH_GITEA_WEBHOOK_SECRET={secret}\n")
|
||
return type("P", (), {"returncode": 0, "stdout": "", "stderr": ""})()
|
||
|
||
monkeypatch.setattr(mod, "_run", _fake_gen)
|
||
args = mod.build_arg_parser().parse_args(["apply"])
|
||
ctx = {"args": args, "root_env": {}, "overrides": {}}
|
||
# первый прогон: секреты выпущены, env собран
|
||
assert mod.step_secrets(ctx) == "ok"
|
||
assert mod.step_env(ctx) == "ok"
|
||
first = (tmp_path / ".env").read_text(encoding="utf-8")
|
||
# второй прогон тех же шагов поверх собранного .env
|
||
ctx2 = {"args": args,
|
||
"root_env": mod.parse_env(first), "overrides": {}}
|
||
assert mod.step_secrets(ctx2) == "skipped" # уже есть → не перевыпуск
|
||
assert mod.step_env(ctx2) == "ok"
|
||
second = (tmp_path / ".env").read_text(encoding="utf-8")
|
||
assert mod.parse_env(second)["ORCH_PLANE_WEBHOOK_SECRET"] == secret
|
||
assert mod.parse_env(first) == mod.parse_env(second) # значения стабильны
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-16 — no-TTY manual_checkpoint fail-closed (AC-8/AC-11).
|
||
# ===========================================================================
|
||
def test_tc16_manual_checkpoint_no_tty_raises(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_isatty", lambda: False)
|
||
|
||
def _boom_ask(prompt):
|
||
raise AssertionError("без TTY input() вызываться не должен")
|
||
|
||
monkeypatch.setattr(mod, "_ask", _boom_ask)
|
||
import pytest
|
||
with pytest.raises(mod.ManualStop):
|
||
mod.manual_checkpoint("step", ["do x"], lambda: (True, ""))
|
||
|
||
|
||
def test_tc16_manual_checkpoint_verifies_after_confirm(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_isatty", lambda: True)
|
||
monkeypatch.setattr(mod, "_ask", lambda prompt: "")
|
||
# верификация проходит со второй попытки
|
||
seq = iter([(False, "ещё нет"), (True, "")])
|
||
mod.manual_checkpoint("step", ["do x"], lambda: next(seq), max_tries=3)
|
||
|
||
|
||
# ===========================================================================
|
||
# TC-17 — health-гейт (AC-11).
|
||
# ===========================================================================
|
||
def test_tc17_health_pass_on_200(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_http",
|
||
lambda url, headers=None, timeout=10: (200, "{}"))
|
||
ctx = {"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {}, "overrides": {}}
|
||
assert mod.step_health(ctx) == "ok"
|
||
|
||
|
||
def test_tc17_health_fail_raises_install_error(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "_http",
|
||
lambda url, headers=None, timeout=10: (503, ""))
|
||
monkeypatch.setattr(mod, "_compose",
|
||
lambda *a, **k: type("P", (), {"stdout": "", "stderr": ""})())
|
||
monkeypatch.setattr(mod, "time",
|
||
type("T", (), {"monotonic": staticmethod(lambda: 1e9),
|
||
"sleep": staticmethod(lambda s: None)}))
|
||
ctx = {"args": mod.build_arg_parser().parse_args(["apply"]),
|
||
"root_env": {}, "overrides": {}}
|
||
import pytest
|
||
with pytest.raises(mod.InstallError):
|
||
mod.step_health(ctx)
|
||
|
||
|
||
def test_tc17_main_verify_reports_fail_as_exit_error(monkeypatch):
|
||
mod = _load_module()
|
||
monkeypatch.setattr(mod, "ROOT_ENV", os.path.join(str(REPO_ROOT), "nonexistent.env"))
|
||
monkeypatch.setattr(mod, "_http",
|
||
lambda url, headers=None, timeout=10: (503, ""))
|
||
assert mod.main(["verify"]) == mod.EXIT_ERROR
|