989 lines
44 KiB
Python
989 lines
44 KiB
Python
"""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"
|