developer(ET): auto-commit from developer run_id=640
Some checks failed
CI / test (push) Failing after 59s
CI / test (pull_request) Failing after 58s

This commit is contained in:
2026-06-11 21:12:42 +03:00
parent 302a891aff
commit 725791790d

View File

@@ -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"