From 725791790da1a0732edaa9b073eead9559690b8a Mon Sep 17 00:00:00 2001 From: claude-bot Date: Thu, 11 Jun 2026 21:12:42 +0300 Subject: [PATCH] developer(ET): auto-commit from developer run_id=640 --- tests/test_setup_lite_script.py | 988 ++++++++++++++++++++++++++++++++ 1 file changed, 988 insertions(+) create mode 100644 tests/test_setup_lite_script.py diff --git a/tests/test_setup_lite_script.py b/tests/test_setup_lite_script.py new file mode 100644 index 0000000..6ede2e2 --- /dev/null +++ b/tests/test_setup_lite_script.py @@ -0,0 +1,988 @@ +"""ORCH-104 (TC-01…TC-25, AC-1…AC-12): структурные и unit-проверки +`scripts/setup_lite.py` — интерактивного installer'а Lite-тиража. + +По образцу `tests/test_bootstrap_script.py` (ORCH-103) + ADR-001 ORCH-104 D12: +вся решающая логика скрипта — чистые функции (вердикты предусловий, +классификатор discovery, когерентность портов, рендер env с маркером +managed-файла, builder аргументов onboarding, step-движок), тестируемые без +TTY/сети/docker; интерактив — через инжектируемый I/O (`IO(input_fn=…, +getpass_fn=…, say_fn=…, is_tty=…, env=…)`); файловые сценарии — на tmp_path; +структурная гигиена — ast/эвристики по файлу скрипта (stdlib-only, зеркала +delete/status-needle-наборов, кирпичи gen_secrets.py / onboard_project.py). + +Детерминировано: без сети/docker/LLM (единственный субпроцесс — stdlib-кирпич +gen_secrets.py в TC-13); модуль импортируется по файлу, import не имеет +side effects. +""" + +import ast +import importlib.util +import json + +import pytest + +from pathlib import Path + +# Один источник истины запрещённых боевых литералов (TC-15, ORCH-101 AC-7). +from tests.test_no_host_hardcodes import FORBIDDEN + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPT = REPO_ROOT / "scripts/setup_lite.py" + +# Зеркало FORBIDDEN_DELETE_NEEDLES bootstrap'а (ORCH-103 D9) + API/git-удаления +# (AC-12): delete-операций в installer'е нет ВООБЩЕ — лечение всегда инструкцией. +FORBIDDEN_DELETE_NEEDLES = ( + "volume rm", + "rm -rf", + "down -v", + "compose down", + "rmtree", + "os.remove", + ".unlink", + "push --delete", + 'method="DELETE"', + "method='DELETE'", +) + +# Зеркало FORBIDDEN_STATUS_NEEDLES: собственный канон Plane-статусов в скрипте +# запрещён (статусы — только кирпич onboard_project.py / plane_sync, BR-7/D11). +FORBIDDEN_STATUS_NEEDLES = ( + "Backlog", + "To Analyse", + "Confirm Deploy", + "Code-Review", + "Awaiting Deploy", + "Monitoring after Deploy", +) + +# stdlib-allowlist top-level импортов (D1: 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/setup_lite.py отсутствует (AC-1/FR-1)" + return SCRIPT.read_text(encoding="utf-8") + + +def _load_module(): + spec = importlib.util.spec_from_file_location("setup_lite", SCRIPT) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def _io(mod, answers=(), secrets_=(), say=None, is_tty=True, env=None, yes=False): + """Инжектируемый I/O со скриптованными ответами (D10/NFR-5).""" + answers_it = iter(answers) + secrets_it = iter(secrets_) + return mod.IO( + input_fn=lambda prompt="": next(answers_it), + getpass_fn=lambda prompt="": next(secrets_it), + say_fn=(say.append if say is not None else (lambda s: None)), + is_tty=is_tty, + env=({} if env is None else env), + yes=yes, + ) + + +def _full_facts() -> dict: + """Фикстура «всё установлено» (перечень FR-2; без боевых литералов).""" + return { + "uname": "Linux x86_64", + "docker": True, + "compose_v2": True, + "git": True, + "python3": True, + "node": True, + "node_bin": "/usr/bin/node", + "claude_code_dir": "/usr/lib/node_modules/@anthropic-ai/claude-code", + "claude_creds_readable": True, + "docker_gid": "984", + "uid": 1001, + "gid": 1001, + "home": "/home/operator", + "repos_dir": "/home/operator/repos", + "repos_dir_owner_ok": True, + "ssh_dir": "/home/operator/.orchestrator-ssh", + "ssh_keys": True, + "busy_ports": [], + "pkg_manager": "apt-get", + "repo_root": "/home/operator/orchestrator", + } + + +# --------------------------------------------------------------------------- +# TC-01 / AC-1: CLI — режимы закрыты, дефолт apply (ADR-001 D2), флаги. +# --------------------------------------------------------------------------- +def test_tc01_modes_closed_and_apply_is_default(): + mod = _load_module() + parser = mod.build_arg_parser() + # ОСОЗНАННО зеркально к test_plan_is_default_mode bootstrap'а: у setup_lite + # дефолт — apply-wizard (бизнес-цель «одна команда», ADR-001 D2); безопасность + # дефолта — структурно (фаза 0 ≡ plan, ранний guard .env, per-action consent, + # non-TTY без --yes → exit 2), а не выбором режима. + assert parser.parse_args([]).mode == "apply" + assert parser.parse_args(["plan"]).mode == "plan" # строгий read-only + assert parser.parse_args(["verify"]).mode == "verify" # read-only пост-проверка + with pytest.raises(SystemExit): + parser.parse_args(["wizard"]) # набор режимов закрыт (D2) + args = parser.parse_args([]) + assert args.force is False and args.yes is False + + +# --------------------------------------------------------------------------- +# TC-02 / AC-1: step-движок check→ensure — skip без вызова ensure. +# --------------------------------------------------------------------------- +def test_tc02_engine_skips_done_steps_without_ensure(): + mod = _load_module() + ran: list = [] + steps = ( + ("one", lambda ctx: False, lambda ctx: (ran.append("one"), "ok")[1]), + ("two", lambda ctx: False, lambda ctx: (ran.append("two"), "ok")[1]), + ) + ctx = {"results": {}} + mod.run_steps(steps, ctx) + assert ran == ["one", "two"] + assert ctx["results"] == {"one": "ok", "two": "ok"} + + # повторный прогон по фикстуре «всё выполнено» — каскад skip, ни одной мутации + ran.clear() + done_steps = tuple((n, (lambda ctx: True), e) for n, _, e in steps) + ctx2 = {"results": {}} + mod.run_steps(done_steps, ctx2) + assert ran == [] + assert set(ctx2["results"].values()) == {"skip"} + + +# --------------------------------------------------------------------------- +# TC-03 / AC-1, AC-11: resume — manual-step останавливает (exit 2), повторный +# запуск продолжает с первого незавершённого шага. +# --------------------------------------------------------------------------- +def test_tc03_resume_continues_from_first_unfinished_step(): + mod = _load_module() + ran: list = [] + done = {"a": False, "b": False} + + def ensure_a(ctx): + ran.append("a") + return "ok" + + def ensure_b(ctx): + if not done["b"]: + raise mod.ManualStop("b: выполните ручной шаг") + ran.append("b") + return "ok" + + steps = ( + ("a", lambda ctx: done["a"], ensure_a), + ("b", lambda ctx: done["b"], ensure_b), + ("c", lambda ctx: False, lambda ctx: (ran.append("c"), "ok")[1]), + ) + with pytest.raises(mod.ManualStop): + mod.run_steps(steps, {"results": {}}) + assert ran == ["a"] and "c" not in ran # остановились на manual-step + + # «resume» = повторный запуск: выполненное скипается, продолжаем с первого + # незавершённого (b теперь сделан руками → skip, c выполняется) + done["a"] = done["b"] = True + ran.clear() + ctx = {"results": {}} + mod.run_steps(steps, ctx) + assert ran == ["c"] + assert ctx["results"]["a"] == "skip" and ctx["results"]["b"] == "skip" + + +# --------------------------------------------------------------------------- +# TC-04 / AC-2: вердикты предусловий — полный набор фактов → все OK. +# --------------------------------------------------------------------------- +def test_tc04_full_facts_give_all_ok_and_no_blockers(): + mod = _load_module() + verdicts = mod.prereq_verdicts(_full_facts()) + assert verdicts, "prereq_verdicts вернул пустой перечень (FR-2)" + assert all(v == "OK" for _, v, _ in verdicts), verdicts + assert mod.has_blockers(verdicts) is False + # ни один пункт перечня FR-2 не пропускается молча + items = {item for item, _, _ in verdicts} + for required in ("os", "docker", "compose", "git", "python3", "node", + "claude-code", "claude-auth", "docker-group", "repos-dir", + "ssh", "ports"): + assert required in items, f"пункт {required!r} перечня FR-2 не покрыт" + + +# --------------------------------------------------------------------------- +# TC-05 / AC-2: MISSING с конкретной командой; отказ от согласия → MANUAL, +# команда напечатана, мутация НЕ выполнена. +# --------------------------------------------------------------------------- +def test_tc05_missing_docker_offer_declined_is_manual_without_mutation(): + mod = _load_module() + facts = _full_facts() + facts.update(docker=False, compose_v2=False) + verdicts = dict((i, (v, d)) for i, v, d in mod.prereq_verdicts(facts)) + assert verdicts["docker"][0] == "MISSING" + assert mod.has_blockers(mod.prereq_verdicts(facts)) is True + + command = mod.install_command("apt-get", "docker") + assert command and "apt-get" in command # конкретная команда под менеджер + + transcript: list = [] + executed: list = [] + io = _io(mod, answers=["n"], say=transcript) # инжектированный отказ + status = mod.offer_install("docker", command, io, + runner=lambda cmd: executed.append(cmd)) + assert status == "manual" + assert executed == [], "мутация выполнена несмотря на отказ (AC-2)" + blob = "\n".join(transcript) + assert command in blob, "точная команда не напечатана ДО запроса согласия" + + +def test_tc05b_offer_accepted_runs_and_rechecks(): + mod = _load_module() + executed: list = [] + + class _Proc: + returncode = 0 + + io = _io(mod, answers=["y"]) + status = mod.offer_install( + "git", "sudo apt-get install -y git", io, + runner=lambda cmd: (executed.append(cmd), _Proc())[1], + recheck=lambda: True, + ) + assert status == "ok" and executed == ["sudo apt-get install -y git"] + # re-check фактом не сошёлся → честный MANUAL (не ложный OK) + io2 = _io(mod, answers=["y"]) + status2 = mod.offer_install( + "git", "sudo apt-get install -y git", io2, + runner=lambda cmd: _Proc(), recheck=lambda: False, + ) + assert status2 == "manual" + + +# --------------------------------------------------------------------------- +# TC-06 / AC-2: неопределимый пакетный менеджер → MANUAL со ссылкой на канон; +# uname вне контура → WARN, не падение. +# --------------------------------------------------------------------------- +def test_tc06_unknown_pkg_manager_and_foreign_os(): + mod = _load_module() + assert mod.detect_pkg_manager(which=lambda name: None) is None + assert mod.detect_pkg_manager( + which=lambda name: "/usr/bin/dnf" if name == "dnf" else None) == "dnf" + assert mod.install_command(None, "docker") is None + hint = mod.manual_install_hint("docker") + assert "LITE_SETUP" in hint # ссылка на § канона, не молчаливый пропуск + + facts = _full_facts() + facts["uname"] = "Darwin arm64" + verdicts = dict((i, (v, d)) for i, v, d in mod.prereq_verdicts(facts)) + assert verdicts["os"][0] == "WARN" + assert "вне контура Lite" in verdicts["os"][1] + + +# --------------------------------------------------------------------------- +# TC-07 / AC-3: discovery — две независимые Plane-инсталляции → ровно 2 +# кандидата; выбор пользователя применяется; «ввести вручную» присутствует. +# --------------------------------------------------------------------------- +def _two_plane_containers() -> list: + return [ + {"name": "plane-a-proxy-1", "image": "makeplane/plane-proxy:v0.23.1", + "ports": "0.0.0.0:8080->80/tcp", "project": "plane-a"}, + {"name": "plane-a-api-1", "image": "makeplane/plane-backend:v0.23.1", + "ports": "", "project": "plane-a"}, + {"name": "plane-a-db-1", "image": "postgres:15.7-alpine", + "ports": "", "project": "plane-a"}, + {"name": "plane-b-proxy-1", "image": "makeplane/plane-proxy:v0.23.1", + "ports": "0.0.0.0:9090->80/tcp", "project": "plane-b"}, + {"name": "shop-nginx-1", "image": "nginx:1.25", + "ports": "0.0.0.0:80->80/tcp", "project": "shop"}, + ] + + +def test_tc07_two_plane_installations_listed_and_choice_applied(): + mod = _load_module() + installs = mod.discover_installations(_two_plane_containers()) + planes = [i for i in installs if i["kind"] == "plane"] + assert len(planes) == 2, planes + assert {p["project"] for p in planes} == {"plane-a", "plane-b"} + assert {p["url_port"] for p in planes} == {8080, 9090} + + transcript: list = [] + io = _io(mod, answers=["2"], say=transcript) + chosen = mod.choose_installation("Plane", planes, io) + assert chosen is not None and chosen["url_port"] in (8080, 9090) + blob = "\n".join(transcript) + assert "1." in blob and "2." in blob, "нумерованный список не показан" + assert "вручную" in blob, "пункт «ввести вручную» отсутствует (AC-3)" + + # «ввести вручную» доступен и при ≥2 кандидатах + io0 = _io(mod, answers=["0"]) + assert mod.choose_installation("Plane", planes, io0) is None + + +def test_tc08_single_zero_and_foreign_images(): + mod = _load_module() + containers = _two_plane_containers() + # одна инсталляция → префилл по умолчанию (Enter = подтверждение) + single = [i for i in mod.discover_installations(containers[:3]) + if i["kind"] == "plane"] + assert len(single) == 1 + io = _io(mod, answers=[""]) + chosen = mod.choose_installation("Plane", single, io) + assert chosen is single[0] + + # ноль инсталляций → ручной ввод + честная подсказка про Bundled + transcript: list = [] + io0 = _io(mod, answers=[], say=transcript) + assert mod.choose_installation("Plane", [], io0) is None + assert "BUNDLED_SETUP" in "\n".join(transcript) + + # посторонние образы кандидатами не становятся (строго image-префиксы, D5) + foreign_only = [c for c in containers if c["project"] == "shop"] + assert mod.discover_installations(foreign_only) == [] + + +def test_tc08b_gitea_discovery_by_image_prefix(): + mod = _load_module() + containers = [ + {"name": "git-gitea-1", "image": "docker.gitea.com/gitea:1.24.9", + "ports": "0.0.0.0:3300->3000/tcp", "project": "git"}, + ] + giteas = [i for i in mod.discover_installations(containers) + if i["kind"] == "gitea"] + assert len(giteas) == 1 and giteas[0]["url_port"] == 3300 + + +def test_tc09_docker_failure_is_never_block(monkeypatch): + mod = _load_module() + + def _boom(*args, **kwargs): + raise OSError("docker недоступен") + + monkeypatch.setattr(mod, "_run", _boom) + assert mod.list_containers() is None # ошибка перечисления → None, не исключение + # классификатор на «нет данных» отвечает пустым перечнем (ручной ввод дальше) + assert mod.discover_installations([]) == [] + assert mod.parse_docker_ps("") == [] + + +# --------------------------------------------------------------------------- +# TC-10 / AC-4: цикл запроса — re-prompt с диагнозом, лимит попыток, не +# бесконечный цикл; успешная верификация → значение принято. +# --------------------------------------------------------------------------- +def test_tc10_verify_retry_limit_and_success(): + mod = _load_module() + transcript: list = [] + attempts: list = [] + + def verify_fail(value): + attempts.append(value) + return False, "HTTP 401: токен отклонён" + + io = _io(mod, secrets_=["t1", "t2", "t3", "t4"], say=transcript) + with pytest.raises(mod.ManualStop): + io.ask("ORCH_PLANE_API_TOKEN", "Plane API token", secret=True, + verify=verify_fail, max_tries=3) + assert len(attempts) == 3, "лимит попыток не соблюдён (не бесконечный цикл)" + assert any("401" in line for line in transcript), "re-prompt без диагноза" + + def verify_second(value): + return (value == "good", "HTTP 401") + + io2 = _io(mod, secrets_=["bad", "good"]) + assert io2.ask("ORCH_GITEA_TOKEN", "Gitea token", secret=True, + verify=verify_second) == "good" + + +def test_tc10b_env_prefill_is_used_and_verified(): + mod = _load_module() + seen: list = [] + io = _io(mod, env={"ORCH_GITEA_TOKEN": "from-env"}, is_tty=False, yes=True) + value = io.ask("ORCH_GITEA_TOKEN", "Gitea token", secret=True, + verify=lambda v: (seen.append(v) or True, "")) + assert value == "from-env" and seen == ["from-env"] # верификация выполняется + + +# --------------------------------------------------------------------------- +# TC-11 / AC-4: секрет-гигиена — значение секрета не попадает в транскрипт. +# --------------------------------------------------------------------------- +def test_tc11_secret_values_never_printed(capsys): + mod = _load_module() + secret = "supersecret-token-value-123" + transcript: list = [] + io = _io(mod, secrets_=[secret], say=transcript) + value = io.ask("ORCH_TELEGRAM_BOT_TOKEN", "токен бота", secret=True, + verify=lambda v: (True, "")) + assert value == secret + blob = "\n".join(transcript) + capsys.readouterr().out + assert secret not in blob, "значение секрета напечатано (NFR-3)" + assert "ORCH_TELEGRAM_BOT_TOKEN" in blob or transcript == [], ( + "в транскрипте допустимы только имена ключей" + ) + + +# --------------------------------------------------------------------------- +# TC-12 / AC-4: non-TTY без альтернативы → честный отказ, не зависание. +# --------------------------------------------------------------------------- +def test_tc12_non_tty_is_deterministic_refusal(): + mod = _load_module() + io = _io(mod, is_tty=False) + with pytest.raises(mod.ManualStop): + io.ask("ORCH_PLANE_API_TOKEN", "Plane API token", secret=True) + # consent в non-TTY без --yes — тоже честный отказ + with pytest.raises(mod.ManualStop): + io.consent("выполнить мутацию") + # с --yes согласие дано флагом (headless-consent, D10) + io_yes = _io(mod, is_tty=False, yes=True) + assert io_yes.consent("выполнить мутацию") is True + # non-TTY + непустой дефолт (автодетект) → дефолт принимается, прогон жив + io_def = _io(mod, is_tty=False, yes=True) + assert io_def.ask("ORCH_RUN_UID", "uid", default="1001") == "1001" + + +# --------------------------------------------------------------------------- +# TC-13 / AC-5: сборка .env/.env.watchdog — группы ключей, файл-носитель §4.3, +# свежие 64-hex webhook-секреты кирпичом gen_secrets. +# --------------------------------------------------------------------------- +def test_tc13_split_overrides_watchdog_keys_only_in_watchdog_file(): + mod = _load_module() + answers = { + "ORCH_PLANE_API_URL": "http://127.0.0.1:8080", + "ORCH_TELEGRAM_BOT_TOKEN": "tg-orch", + "WATCHDOG_TG_BOT_TOKEN": "tg-wd", + "WATCHDOG_TG_CHAT_ID": "42", + "WATCHDOG_METRICS_URL": "http://127.0.0.1:8500/metrics", + } + root_ov, wd_ov = mod.split_overrides(answers) + assert "WATCHDOG_TG_BOT_TOKEN" not in root_ov # ловушка файла-носителя §4.3 + assert "WATCHDOG_TG_CHAT_ID" not in root_ov + assert wd_ov["WATCHDOG_TG_BOT_TOKEN"] == "tg-wd" + assert wd_ov["WATCHDOG_TG_CHAT_ID"] == "42" + assert root_ov["ORCH_PLANE_API_URL"] == "http://127.0.0.1:8080" + # когерентный WATCHDOG_METRICS_URL уходит в файл-носитель sidecar'а + assert wd_ov["WATCHDOG_METRICS_URL"] == "http://127.0.0.1:8500/metrics" + + +def test_tc13b_webhook_secrets_are_fresh_64_hex_via_brick(): + mod = _load_module() + first = mod.issue_webhook_secrets() + second = mod.issue_webhook_secrets() + for batch in (first, second): + assert set(batch) == {"ORCH_PLANE_WEBHOOK_SECRET", "ORCH_GITEA_WEBHOOK_SECRET"} + for value in batch.values(): + assert len(value) == 64 and all(c in "0123456789abcdef" for c in value) + assert first != second, "повторный выпуск обязан давать другие значения (AC-5)" + + +def test_tc13c_render_env_keeps_canon_and_adds_marker(): + mod = _load_module() + example = "# шапка канона\nA=1\nB=\n" + rendered = mod.render_env(example, {"B": "v", "NEW": "n"}) + assert rendered.startswith(mod.MANAGED_MARKER), "маркер managed-файла (D6)" + assert "# шапка канона" in rendered and "A=1" in rendered + assert mod.parse_env(rendered) == {"A": "1", "B": "v", "NEW": "n"} + + +def test_tc13d_mandatory_key_groups_covered(): + """Карта обязательных ключей §4.2 LITE_SETUP покрыта константой скрипта.""" + mod = _load_module() + keys = set(mod.MANDATORY_NEW_HOST_KEYS) + for required in ( + "ORCH_PLANE_API_URL", "ORCH_PLANE_WEB_URL", "ORCH_PLANE_WORKSPACE_SLUG", + "ORCH_PLANE_API_TOKEN", "ORCH_GITEA_URL", "ORCH_GITEA_PUBLIC_URL", + "ORCH_GITEA_OWNER", "ORCH_GITEA_TOKEN", "ORCH_PLANE_WEBHOOK_SECRET", + "ORCH_GITEA_WEBHOOK_SECRET", "ORCH_TELEGRAM_BOT_TOKEN", + "ORCH_TELEGRAM_CHAT_ID", "ORCH_RUN_UID", "ORCH_RUN_GID", + "ORCH_DOCKER_GID", "ORCH_HOST_REPOS_DIR", + "ORCH_DEPLOY_PROD_TARGET_PORT", "ORCH_STAGING_PORT", + ): + assert required in keys, f"{required} не покрыт картой §4.2 (AC-5)" + # все имена — канонические (существуют в .env.example) + canon = set() + for line in (REPO_ROOT / ".env.example").read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + canon.add(line.split("=", 1)[0].strip()) + assert keys <= canon, f"вне канона .env.example: {sorted(keys - canon)}" + + +# --------------------------------------------------------------------------- +# TC-14 / AC-5: существующий чужой .env → отказ без force, байт-в-байт; маркер +# managed-файла → resume-ensure; с force — перезапись. +# --------------------------------------------------------------------------- +def test_tc14_existing_foreign_env_is_refused_byte_identical(tmp_path): + mod = _load_module() + env = tmp_path / ".env" + env.write_text("MY_LIVE_KEY=1\n", encoding="utf-8") + before = env.read_bytes() + assert mod.env_file_state(env.read_text(encoding="utf-8")) == "foreign" + assert mod.env_file_state(None) == "absent" + + io = _io(mod, answers=["y"]) + with pytest.raises(mod.ManualStop): + mod.ensure_env_file(str(env), "A=\n", {"A": "x"}, force=False, io=io) + assert env.read_bytes() == before, "чужой .env изменён без force (NFR-7)" + + # с явным force (+согласие) — перезапись выполняется + io2 = _io(mod, answers=["y"]) + mod.ensure_env_file(str(env), "A=\n", {"A": "x"}, force=True, io=io2) + text = env.read_text(encoding="utf-8") + assert text.startswith(mod.MANAGED_MARKER) and "A=x" in text + + +def test_tc14b_managed_file_resume_ensure_keeps_existing_values(tmp_path): + mod = _load_module() + env = tmp_path / ".env" + env.write_text(mod.MANAGED_MARKER + "\nA=keep\nB=\n", encoding="utf-8") + assert mod.env_file_state(env.read_text(encoding="utf-8")) == "managed" + io = _io(mod, answers=[]) + mod.ensure_env_file(str(env), "A=\nB=\nC=\n", {"A": "new", "B": "vb", "C": "vc"}, + force=False, io=io) + got = mod.parse_env(env.read_text(encoding="utf-8")) + assert got["A"] == "keep", "resume-ensure перетёр существующее значение (D6)" + assert got["B"] == "vb" and got["C"] == "vc", "недостающие ключи не дозаполнены" + + +# --------------------------------------------------------------------------- +# TC-15 / AC-5: подсказки-дефолты — только автодетект/канон, без боевых значений. +# --------------------------------------------------------------------------- +def test_tc15_prompt_defaults_come_from_autodetect_without_forbidden(): + mod = _load_module() + defaults = mod.prompt_defaults(_full_facts()) + blob = json.dumps(defaults, ensure_ascii=False) + for literal in FORBIDDEN: + assert literal not in blob, f"боевой литерал {literal!r} в дефолтах промптов" + assert defaults["ORCH_RUN_UID"] == "1001" + assert defaults["ORCH_DOCKER_GID"] == "984" + assert defaults["ORCH_HOST_NODE_BIN"] == "/usr/bin/node" + assert defaults["ORCH_AGENT_HOME_DIR"] == "/home/operator" + assert defaults["ORCH_HOST_CLAUDE_DIR"] == "/home/operator/.claude" + + +def test_tc15b_script_source_carries_no_forbidden_literals(): + src = _source() + offenders = [literal for literal in FORBIDDEN if literal in src] + assert not offenders, f"боевые литералы в setup_lite.py: {offenders}" + + +# --------------------------------------------------------------------------- +# TC-16 / AC-6: когерентная тройка портов одной функцией; занятый порт → +# альтернатива. +# --------------------------------------------------------------------------- +def test_tc16_port_overrides_keep_triple_coherent(): + mod = _load_module() + overrides = mod.port_overrides(8700) + assert overrides["ORCH_DEPLOY_PROD_TARGET_PORT"] == "8700" + assert overrides["WATCHDOG_METRICS_URL"] == "http://127.0.0.1:8700/metrics" + assert overrides["ORCH_POST_DEPLOY_BASE_URL"] == "http://localhost:8700" + # занятый порт → предложена альтернатива (мок busy-check) + assert mod.next_free_port(8500, busy=lambda p: p in (8500, 8501)) == 8502 + + +def test_tc17_staging_port_equal_to_prod_is_fail_closed(): + mod = _load_module() + assert mod.staging_port_ok(8500, 8500) is False # инвариант ORCH-058/101 + assert mod.staging_port_ok(8501, 8500) is True + # в цикле ввода значение не принимается — re-prompt до различного + io = _io(mod, answers=["8500", "8501"]) + value = io.ask("ORCH_STAGING_PORT", "staging-порт", default="8501", + verify=lambda v: (mod.staging_port_ok(int(v), 8500), + "staging-порт обязан отличаться от прод-порта")) + assert value == "8501" + + +# --------------------------------------------------------------------------- +# TC-18 / AC-7: C-1 — одинаковые токены орка и watchdog запрещены машинно. +# --------------------------------------------------------------------------- +def test_tc18_telegram_c1_identical_tokens_refused(): + mod = _load_module() + ok, hint = mod.telegram_c1_verdict("same-token", "same-token") + assert ok is False + assert "C-1" in hint or "ЗАПРЕЩЕНО" in hint.upper(), "отказ без объяснения запрета" + ok2, _ = mod.telegram_c1_verdict("token-a", "token-b") + assert ok2 is True + + +# --------------------------------------------------------------------------- +# TC-19 / AC-8: branch protection — честный FAIL с лечением, без удаления. +# --------------------------------------------------------------------------- +def test_tc19_branch_protection_verdict(): + mod = _load_module() + ok, hint = mod.branch_protection_verdict(200, [{"branch_name": "main"}]) + assert ok is False + assert "6.4" in hint or "13.7" in hint, "лечение без ссылки на норматив §6.4" + assert mod.branch_protection_verdict(200, [])[0] is True + # репо ещё не создан (создаст onboarding) — не FAIL + assert mod.branch_protection_verdict(404, None)[0] is True + + +# --------------------------------------------------------------------------- +# TC-20 / AC-8 (D8): webhook Plane Path Б — только с явного согласия; отказ → +# UI-путь, мутирующий вызов не произведён; после согласия — пост-верификация. +# --------------------------------------------------------------------------- +def _webhook_answers() -> dict: + return { + "plane_db_container": "plane-a-db-1", + "plane_db_user": "plane", + "plane_db_name": "plane", + "plane_db_password": "pw", + "ORCH_PLANE_WORKSPACE_SLUG": "acme", + "ORCH_PLANE_WEBHOOK_SECRET": "a" * 64, + "orchestrator_public_url": "https://orch.example.com", + } + + +def test_tc20_path_b_refused_no_mutation(): + mod = _load_module() + calls: list = [] + + def psql(sql): + calls.append(sql) + return 0, "0" + + transcript: list = [] + io = _io(mod, answers=["n"], say=transcript) # отказ от согласия на SQL + status = mod.plane_webhook_path_b(_webhook_answers(), io, psql) + assert status == "manual" + assert not any("INSERT" in c.upper() for c in calls), ( + "мутирующий SQL выполнен без согласия (D8)" + ) + assert "INSERT" in "\n".join(transcript).upper(), ( + "точный SQL не показан ДО запроса согласия (D8 п.2)" + ) + + +def test_tc20b_path_b_consent_insert_and_postverify(): + mod = _load_module() + state = {"rows": 0} + calls: list = [] + + def psql(sql): + calls.append(sql) + if "INSERT" in sql.upper(): + state["rows"] = 1 + return 0, "" + if "count(*)" in sql: + return 0, str(state["rows"]) + if "SELECT id FROM workspaces" in sql: + return 0, "11111111-2222-3333-4444-555555555555" + return 0, ("https://orch.example.com/webhook/plane|t" if state["rows"] else "") + + io = _io(mod, answers=["y"]) + assert mod.plane_webhook_path_b(_webhook_answers(), io, psql) == "ok" + assert any("INSERT" in c.upper() for c in calls) + # пост-верификация обязательна: INSERT прошёл, но строки нет → НЕ PASS + io2 = _io(mod, answers=["y"]) + + def psql_lost(sql): + if "count(*)" in sql: + return 0, "0" + if "SELECT id FROM workspaces" in sql: + return 0, "11111111-2222-3333-4444-555555555555" + if "INSERT" in sql.upper(): + return 0, "" + return 0, "" + + assert mod.plane_webhook_path_b(_webhook_answers(), io2, psql_lost) == "manual" + + +def test_tc20c_path_b_is_idempotent_skip_when_registered(): + mod = _load_module() + calls: list = [] + + def psql(sql): + calls.append(sql) + return 0, "1" # webhook уже зарегистрирован + + io = _io(mod, answers=["y"]) + assert mod.plane_webhook_path_b(_webhook_answers(), io, psql) == "skipped" + assert not any("INSERT" in c.upper() for c in calls) + + +# --------------------------------------------------------------------------- +# TC-21 / AC-9: запуск — up только после согласия; состав «ровно орк+watchdog»; +# health-контракты; stateless-проверка. +# --------------------------------------------------------------------------- +def test_tc21_composition_verdict_is_exactly_two_services(): + mod = _load_module() + ok, _ = mod.lite_composition_verdict(["orchestrator", "orchestrator-watchdog"]) + assert ok is True + bad, hint = mod.lite_composition_verdict( + ["orchestrator", "orchestrator-watchdog", "orchestrator-staging"]) + assert bad is False and "staging" in hint + assert mod.lite_composition_verdict(["orchestrator"])[0] is False + + +def test_tc21b_health_checks_require_contract_bodies(): + mod = _load_module() + bodies = { + "/health": (200, '{"status":"ok"}'), + "/queue": (200, '{"counts": {}, "recent": []}'), + "/metrics": (200, '{"schema_version": 1}'), + } + + def http(url, headers=None, timeout=10): + for path, resp in bodies.items(): + if url.endswith(path): + return resp + return None, "" + + results = mod.health_checks(http, 8500) + assert all(ok for _, ok, _ in results), results + bodies["/metrics"] = (200, '{"schema_version": 2}') + results2 = mod.health_checks(http, 8500) + assert any(path == "/metrics" and not ok for path, ok, _ in results2) + + +def test_tc21c_step_up_requires_consent_and_checks_composition(monkeypatch): + mod = _load_module() + compose_calls: list = [] + + class _Proc: + returncode = 0 + stderr = "" + stdout = "orchestrator\norchestrator-watchdog\n" + + monkeypatch.setattr(mod, "_compose", + lambda *args, **kw: (compose_calls.append(args), _Proc())[1]) + monkeypatch.setattr(mod, "_http", lambda url, **kw: ( + (200, '{"status":"ok"}') if url.endswith("/health") + else (200, '{"schema_version": 1}') if url.endswith("/metrics") + else (200, '{"counts": {}, "recent": []}'))) + ctx = {"io": _io(mod, answers=["y"]), "answers": {}, "results": {}} + assert mod.step_up(ctx) == "ok" + assert any("up" in c for c in compose_calls), "compose up не вызван" + + # отказ от согласия → MANUAL, ни одного вызова compose-мутации + compose_calls.clear() + ctx2 = {"io": _io(mod, answers=["n"]), "answers": {}, "results": {}} + assert mod.step_up(ctx2) == "manual" + assert not any("up" in c for c in compose_calls), "контур поднят без согласия" + + +def test_tc21d_step_up_fails_on_staging_in_composition(monkeypatch): + mod = _load_module() + + class _Proc: + returncode = 0 + stderr = "" + stdout = "orchestrator\norchestrator-watchdog\norchestrator-staging\n" + + monkeypatch.setattr(mod, "_compose", lambda *args, **kw: _Proc()) + ctx = {"io": _io(mod, answers=["y"]), "answers": {}, "results": {}} + with pytest.raises(mod.SetupError): + mod.step_up(ctx) + + +def test_tc21e_stateless_verdict_flags_foreign_tasks(): + mod = _load_module() + clean = {"counts": {"queued": 0, "running": 0}, "recent": []} + assert mod.stateless_verdict(clean, own_prefixes=("SHP",))[0] is True + dirty = {"counts": {"done": 3}, + "recent": [{"work_item_id": "FOO-61"}]} + ok, hint = mod.stateless_verdict(dirty, own_prefixes=("SHP",)) + assert ok is False and "FOO-61" in hint + own = {"counts": {"done": 1}, "recent": [{"work_item_id": "SHP-1"}]} + assert mod.stateless_verdict(own, own_prefixes=("SHP",))[0] is True + + +# --------------------------------------------------------------------------- +# TC-22 / AC-10: onboarding — builder аргументов чистой функцией; строго +# последовательность plan → согласие → apply → verify; exit 2 кирпича → MANUAL. +# --------------------------------------------------------------------------- +def _project_answers(tmp_path=None) -> dict: + return { + "project_name": "Shop", + "project_description": "магазин", + "project_repo": "shop", + "project_prefix": "SHP", + "project_stack": "python", + "project_test_cmd": "pytest -q", + "project_prod_port": "8600", + "project_staging_port": "8601", + "ORCH_GITEA_OWNER": "acme", + "orchestrator_public_url": "https://orch.example.com", + } + + +def test_tc22_build_onboard_args_is_pure_and_complete(): + mod = _load_module() + args = mod.build_onboard_args(_project_answers(), "plan") + assert args[0].endswith("onboard_project.py") and args[1] == "plan" + pairs = dict(zip(args[2::2], args[3::2])) + assert pairs["--name"] == "Shop" and pairs["--repo"] == "shop" + assert pairs["--prefix"] == "SHP" and pairs["--stack"] == "python" + assert pairs["--test-cmd"] == "pytest -q" + assert pairs["--prod-port"] == "8600" and pairs["--staging-port"] == "8601" + assert pairs["--webhook-url"] == "https://orch.example.com/webhook/gitea" + assert pairs["--gitea-owner"] == "acme" + assert "--json" in args + # детерминизм чистой функции + assert args == mod.build_onboard_args(_project_answers(), "plan") + + +def test_tc22b_onboard_sequence_plan_consent_apply_verify(tmp_path, monkeypatch): + mod = _load_module() + modes: list = [] + report = {"steps": [], "instructions": [ + "Добавь/обнови строку в .env оркестратора: ORCH_PROJECTS_JSON=" + '[{"repo": "shop"}]', + ]} + + class _Proc: + def __init__(self, rc, out=""): + self.returncode = rc + self.stdout = out + self.stderr = "" + + def fake_run(cmd, **kwargs): + mode = next((m for m in ("plan", "apply", "verify") if m in cmd), "?") + modes.append(mode) + if mode == "apply": + return _Proc(0, json.dumps(report)) + if mode == "verify": + return _Proc(2) # остались ручные шаги + return _Proc(0, "план...") + + env = tmp_path / ".env" + env.write_text(mod.MANAGED_MARKER + "\nORCH_PROJECTS_JSON=\n", encoding="utf-8") + monkeypatch.setattr(mod, "_run", fake_run) + monkeypatch.setattr(mod, "_ensure_venv", lambda: "python3") + monkeypatch.setattr(mod, "_http", lambda url, **kw: (200, '{"counts": {}, "recent": []}')) + monkeypatch.setattr(mod, "_compose", lambda *a, **kw: _Proc(0)) + ctx = { + "io": _io(mod, answers=["y", "y", "y"]), + "answers": _project_answers(), + "results": {}, + "paths": {"root_env": str(env), + "root_env_example": str(REPO_ROOT / ".env.example")}, + "args": mod.build_arg_parser().parse_args([]), + } + status = mod.step_onboard(ctx) + assert modes[:2] == ["plan", "apply"] and "verify" in modes, ( + f"последовательность кирпича нарушена: {modes}" + ) + assert status == "manual", "exit 2 кирпича обязан транслироваться как MANUAL" + got = mod.parse_env(env.read_text(encoding="utf-8")) + assert got["ORCH_PROJECTS_JSON"] == '[{"repo": "shop"}]' + + +def test_tc22c_onboard_consent_refused_no_apply(tmp_path, monkeypatch): + mod = _load_module() + modes: list = [] + + class _Proc: + returncode = 0 + stdout = "план..." + stderr = "" + + def fake_run(cmd, **kwargs): + modes.append(next((m for m in ("plan", "apply", "verify") if m in cmd), "?")) + return _Proc() + + monkeypatch.setattr(mod, "_run", fake_run) + monkeypatch.setattr(mod, "_ensure_venv", lambda: "python3") + env = tmp_path / ".env" + env.write_text(mod.MANAGED_MARKER + "\n", encoding="utf-8") + ctx = { + "io": _io(mod, answers=["n"]), # план показан → отказ + "answers": _project_answers(), + "results": {}, + "paths": {"root_env": str(env), + "root_env_example": str(REPO_ROOT / ".env.example")}, + "args": mod.build_arg_parser().parse_args([]), + } + assert mod.step_onboard(ctx) == "manual" + assert modes == ["plan"], f"apply вызван без согласия после плана: {modes}" + + +def test_tc22d_extract_projects_json_from_brick_instructions(): + mod = _load_module() + instructions = [ + "что-то ещё", + 'Добавь/обнови строку: ORCH_PROJECTS_JSON=[{"repo": "shop"}]', + ] + assert mod.extract_projects_json(instructions) == '[{"repo": "shop"}]' + assert mod.extract_projects_json([]) == "" + assert mod.extract_projects_json(["без реестра"]) == "" + + +# --------------------------------------------------------------------------- +# TC-23 / AC-11: контракт exit-кодов — именованные константы 0/2/1. +# --------------------------------------------------------------------------- +def test_tc23_exit_code_contract(): + mod = _load_module() + assert (mod.EXIT_OK, mod.EXIT_MANUAL, mod.EXIT_ERROR) == (0, 2, 1) + assert mod.exit_code_for({"a": "ok", "b": "skip"}) == 0 + assert mod.exit_code_for({"a": "ok", "b": "manual"}) == 2 + assert mod.exit_code_for({}) == 0 + + +# --------------------------------------------------------------------------- +# TC-24 / AC-12: ast-скан — stdlib-only, без модулей платформы; кирпичи и канон +# упомянуты. +# --------------------------------------------------------------------------- +def test_tc24_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 импорты в setup_lite.py (D1): {sorted(set(offenders))}" + + +def test_tc24b_no_platform_imports(): + src = _source() + assert "from src" not in src and "import src" not in src, ( + "setup_lite обязан быть stdlib-only без импортов платформы (D1)" + ) + + +def test_tc24c_canonical_bricks_and_doc_are_referenced(): + src = _source() + assert "gen_secrets.py" in src, "webhook-секреты обязаны идти через gen_secrets.py" + assert "onboard_project.py" in src, "онбординг обязан идти через onboard_project.py" + assert "LITE_SETUP.md" in src, "каждый шаг обязан ссылаться на канон LITE_SETUP.md" + + +# --------------------------------------------------------------------------- +# TC-25 / AC-12: эвристический скан — delete-операций нет; собственного канона +# статусов нет; import модуля без side effects. +# --------------------------------------------------------------------------- +def test_tc25_no_delete_operations(): + src = _source() + offenders = [n for n in FORBIDDEN_DELETE_NEEDLES if n in src] + assert not offenders, ( + f"delete-операции в setup_lite запрещены (no-delete, D1): {offenders}" + ) + + +def test_tc25b_no_own_status_canon(): + src = _source() + offenders = [n for n in FORBIDDEN_STATUS_NEEDLES if n in src] + assert not offenders, ( + f"setup_lite несёт собственный канон статусов (дрейф BR-7): {offenders}; " + "статусы — только onboard_project.py/plane_sync" + ) + + +def test_tc25c_module_import_has_no_side_effects(): + mod1 = _load_module() + mod2 = _load_module() + assert mod1.build_plan() == mod2.build_plan() + + +def test_apply_steps_match_normative_plan(): + """Инвариант D3: имена step-движка = нормативному плану (10 шагов).""" + mod = _load_module() + plan = mod.build_plan() + assert len(plan) == 10, f"нормативный план D3 — 10 шагов, получено {len(plan)}" + assert [name for name, _, _ in mod.APPLY_STEPS] == [name for name, _ in plan] + assert plan[0][0] == "scan" and plan[-1][0] == "report"