developer(ET): auto-commit from developer run_id=587

This commit is contained in:
2026-06-10 15:34:46 +03:00
committed by orchestrator-deployer
parent d141280390
commit 13e9618bd2
5 changed files with 1348 additions and 7 deletions

View File

@@ -0,0 +1,200 @@
# ONBOARDING — turnkey-онбординг нового проекта (ORCH-009)
> RUNBOOK. Полный чеклист подключения нового проекта к оркестратору одним проходом.
> Исполнитель — оператор; инструмент — CLI `scripts/onboard_project.py`
> (режимы `plan` — дефолт/dry-run, `apply`, `verify`). Каждый шаг, который CLI выполнить
> не может, помечен **🖐 РУЧНОЙ ШАГ** и снабжён командой проверки результата.
> Архитектура решения — см. «Ссылки» внизу.
Запуск CLI — из корня чекаута репо orchestrator, в venv с `requirements.txt`:
```bash
python3 scripts/onboard_project.py plan \
--name "My Project" --description "зачем проект" \
--repo my-project --prefix MP \
--stack "Python 3.12 + FastAPI" --test-cmd "pytest tests/ -q" \
--prod-port 8600 --staging-port 8601 \
--webhook-url https://openclaw.mva154.duckdns.org/orchestrator/webhook/gitea
```
`plan` печатает полный план **без единой мутации** (ни сети-POST, ни записи на диск);
`apply` — идемпотентный ensure (существующее → `skipped(exists)`, ничего не удаляется);
exit-коды: `0` — чисто, `2` — есть `manual-step`/gap, `1` — ошибка.
---
## 0. Предусловия
Все значения — в `.env` на хосте (секреты в гит не попадают):
| Переменная | Зачем | Проверка |
|-----------|-------|----------|
| `ORCH_PLANE_API_TOKEN` (+`ORCH_PLANE_API_URL`, `ORCH_PLANE_WORKSPACE_SLUG`) | создание/чтение проекта, статусов, лейблов | `curl -s -H "X-API-Key: $TOKEN" $URL/api/v1/workspaces/$SLUG/projects/ \| head -c 200` |
| `ORCH_GITEA_TOKEN` (+`ORCH_GITEA_URL`) | создание репо + webhook | `curl -s -H "Authorization: token $TOKEN" $URL/api/v1/user \| head -c 200` |
| `ORCH_GITEA_WEBHOOK_SECRET` | HMAC webhook (переиспользуется, один на все репо) | есть строка в `.env`; нет → `apply` сгенерирует и выведет |
| `ORCH_PROJECTS_JSON` | текущий реестр — источник merged-вывода | `grep ORCH_PROJECTS_JSON .env` |
Токен Plane должен иметь право создавать проекты в workspace; токен Gitea — создавать репо и
hooks под выбранным owner (`--gitea-owner`, дефолт из конфига).
---
## 1. Слой Plane: проект + статусы + лейблы
Выполняет `apply` (или вручную при недоступности API CE — каждый отказ CLI помечает
`manual-step`, не падает).
1. **Проект**: создаётся с `identifier = --prefix`. Уже существует → передай
`--plane-project-id <uuid>` (ensure распознает и пропустит).
2. **Статусы — точные канонические имена** (22, источник — `plane_sync._PLANE_NAME_TO_KEY`;
опечатка = тихая деградация fail-closed веток):
| Статус | Группа | | Статус | Группа |
|--------|--------|-|--------|--------|
| Backlog | `backlog` | | In Review | `started` |
| Todo | `unstarted` | | Blocked | `started` |
| To Analyse | `unstarted` | | Approved | `started` |
| In Progress | `started` | | Rejected | `started` |
| Analysis | `started` | | **Confirm Deploy** | `started` |
| Architecture | `started` | | Needs Input | `started` |
| Development | `started` | | Done | `completed` |
| Code-Review | `started` | | Cancelled | `cancelled` |
| Review | `started` | | **STOP** | **`cancelled`** |
| Testing | `started` | | Awaiting Deploy | `started` |
| Deploying | `started` | | Monitoring after Deploy | `started` |
⚠️ Код-критично: `STOP` обязан быть в группе `cancelled` (иначе ветка отмены молча не
активируется); в терминальных группах (`completed`/`cancelled`) — ТОЛЬКО
Done/Cancelled/STOP, иначе terminal-detection ложно сочтёт живую задачу терминальной.
3. **Лейблы**: `autoApprove`, `autoDeploy`, `Bug` (имена — из конфига оркестратора; их
отсутствие = fail-safe ручной режим / полный цикл).
4. **🖐 РУЧНОЙ ШАГ — порядок статусов на доске**: drag-and-drop в UI (API не управляет
порядком). Проверка: открой доску проекта — колонки в порядке конвейера.
5. **Workspace-webhook**: уже **существует** (один на весь workspace, создан на уровне
workspace заранее) — CLI его НЕ создаёт, только напоминает проверить:
```bash
docker exec -e PGPASSWORD=plane plane-app-plane-db-1 psql -U plane -d plane -c \
"SELECT id, url, is_active FROM webhooks;"
```
---
## 2. Слой Gitea: репо + per-repo webhook
1. **Репо** `--gitea-owner/--repo`: создаётся пустым (`auto_init=false`; ветку `main` создаст
initial push следующего слоя). Существует → `skipped(exists)`.
2. **Per-repo webhook**: `events: push/pull_request/status`, `content_type: json`,
`branch_filter: *`, URL = `--webhook-url`. **Секрет переиспользуется** из
`ORCH_GITEA_WEBHOOK_SECRET` (приёмник валидирует ОДИН глобальный секрет на все репо;
новый секрет сломал бы HMAC существующих вебхуков). Секрета нет в env → CLI сгенерирует и
выведет строку для `.env`**🖐 РУЧНОЙ ШАГ**: добавить её в `.env` (в гит не коммитить).
Формат и проверка — `docs/operations/SETUP_WEBHOOKS.md`. Проверка:
```bash
curl -s -H "Authorization: token $ORCH_GITEA_TOKEN" \
"$ORCH_GITEA_URL/api/v1/repos/<owner>/<repo>/hooks" | python3 -m json.tool
```
3. **Branch protection `main` НЕ включать** (ADR D10): required-approvals/status-checks ломают
PR-merge API merge-актора конвейера (ложные HOLD). Защита держится конвенцией + скоупом
токенов.
---
## 3. Слой kit: материализация + initial push
1. `apply` рендерит kit (`onboarding/repo-skeleton/`, плейсхолдеры `{{NAME}}` из
`onboarding/placeholders.json`) во временный каталог, докладывает live-copy канона
(`docs/_templates/` 16 скелетов + `docs/_standards/` 3 стандарта — verbatim из текущего
чекаута, BR-2 «канон не форкается») и пушит **ТОЛЬКО в свежесозданный/пустой репо**
(единственный разрешённый push; коммит `feat: onboarding skeleton (ORCH-009 kit)`).
2. Репо непустой → шаг помечается `manual-step`: **🖐 РУЧНОЙ ШАГ** — занеси недостающие
файлы обычным PR с ревью; поверх существующего контента ничего не пушится (BR-9).
3. После рендера не должно остаться ни одного `{{...}}`: CLI падает на этом сам; повторная
проверка — `verify` (скан плейсхолдеров в файлах репо).
---
## 4. Регистрация в реестре оркестратора
> ⚠️ **САМЫЙ ВАЖНЫЙ РУЧНОЙ СЛОЙ.** CLI `.env` прода НЕ правит и контейнер НЕ рестартит
> (инвариант NFR-2) — он только печатает готовую строку.
1. **🖐 РУЧНОЙ ШАГ — env**: возьми из отчёта `apply` строку
`ORCH_PROJECTS_JSON=[...полный merged-массив...]` (существующие записи verbatim + новая в
конец; строка уже провалидирована фактическим парсером реестра) и замени ею строку в `.env`
оркестратора на хосте. Вставляется атомарно одной строкой — ручное слияние JSON не нужно.
2. **🖐 РУЧНОЙ ШАГ — управляемый рестарт оркестратора**: реестр строится при импорте, нужна
перезагрузка процесса. **Self-hosting предупреждение: прод-контейнер один на ВСЕ проекты —
рестарт = групповое окно** (встаёт конвейер всех проектов). Выполняй осознанно: дождись
тихого окна (`GET /queue` — нет бегущих job), затем штатный рестарт по
`docs/operations/INFRA.md`. Проверка после рестарта:
```bash
curl -s http://localhost:8500/health
curl -s http://localhost:8500/queue | python3 -m json.tool | head -30 # реестр жив, конвейер пуст/цел
```
3. TTL-self-heal статусов Plane (300с) рестарта НЕ требует: статусы/лейблы, созданные после
регистрации, подхватятся сами.
---
## 5. Верификация
1. **`verify`-режим CLI** (read-only):
```bash
python3 scripts/onboard_project.py verify --name ... --repo ... --prefix ... \
--plane-project-id <uuid> --stack ... --test-cmd ... --prod-port ... --staging-port ... \
--webhook-url https://openclaw.mva154.duckdns.org/orchestrator/webhook/gitea
```
Проверяет: запись реестра парсится и совпадает по полям; все 22 статуса резолвятся
(включая fail-closed `Confirm Deploy`/`STOP`); лейблы на месте; webhook существует и
активен; kit-файлы в репо (6 промптов, `AGENTS.md`, `INFRA.md`, `_templates`/`_standards`);
нет неразрешённых плейсхолдеров. Любой gap → exit `2` с перечнем.
2. **Smoke на песочнице (ADR D8)** — контур: **staging-оркестратор (порт 8501, изолированная
БД `./data/staging`)** + одноразовый sandbox-проект (рекомендуемые имена: проект
`onboarding-smoke`, префикс `SMK`, репо `onboarding-smoke`):
1. Онборди sandbox самим CLI (слои 13 этого runbook).
2. **🖐 РУЧНОЙ ШАГ**: зарегистрируй sandbox в `ORCH_PROJECTS_JSON` **`.env.staging`**
(не прода!) и перезапусти staging-контейнер (он свободен от прод-инварианта):
`docker compose --profile staging up -d orchestrator-staging`.
3. Создай тестовую задачу в sandbox-проекте → доведи до стадии analysis в песочнице.
4. Критерий PASS: агент по своему промпту **прочитал доку проекта** (следы чтения
`CLAUDE.md`/`AGENTS.md` в выводе) и **записал артефакты** в `docs/work-items/SMK-…/`
по канону `PIPELINE_DOCS.md`.
5. Запротоколируй прогон в «Журнале smoke-прогонов» (ниже). Для приёмки ORCH-009 первый
протокол обязателен.
---
## 6. Откат
CLI ничего не удаляет (BR-9) — откат всегда ручной и осознанный:
| Что создано | Как откатить | Проверка |
|-------------|--------------|----------|
| Plane-проект (+статусы/лейблы) | удалить проект в UI Plane | проект исчез из списка workspace |
| Gitea-репо (+webhook) | удалить репо в UI/API Gitea (webhook умрёт вместе с ним) | `GET /api/v1/repos/<owner>/<repo>` → 404 |
| Строка реестра | убрать запись из `ORCH_PROJECTS_JSON` в `.env` + управляемый рестарт (см. слой 4, то же групповое окно) | `GET /queue` — проекта нет в реестре |
| Sandbox-артефакты smoke | удалить sandbox-проект/репо после прогона (или архивировать) | см. выше |
---
## Журнал smoke-прогонов
| Дата | Оператор | Параметры (проект/префикс/репо) | Контур | Результат (PASS/FAIL) | Протокол |
|------|----------|----------------------------------|--------|------------------------|----------|
| — | — | — (первый прогон фиксируется при приёмке ORCH-009) | staging 8501 | — | — |
---
## Ссылки
- Архитектура решения: `docs/work-items/ORCH-009/06-adr/ADR-001-turnkey-onboarding-kit-and-cli.md`
(D1…D11); сквозной ADR — `docs/architecture/adr/adr-0035-turnkey-project-onboarding.md`.
- Устройство набора шаблонов и словарь плейсхолдеров: `onboarding/README.md`.
- Формат вебхуков: `docs/operations/SETUP_WEBHOOKS.md`; топология и рестарты —
`docs/operations/INFRA.md`.

View File

@@ -12,30 +12,36 @@ Internal URL: `http://127.0.0.1:8500/`
---
## Gitea Webhook
## Gitea Webhook (per-repo)
**Создан автоматически через API.**
Gitea-webhook — **per-repo**: создаётся для КАЖДОГО подключаемого к оркестратору репозитория
(`<repo>` ниже). Для новых проектов его создаёт onboarding-CLI
(`scripts/onboard_project.py apply`) — полный процесс см. `docs/operations/ONBOARDING.md`;
команды ниже — для ручной проверки/пересоздания на любом репо.
- URL: `https://openclaw.mva154.duckdns.org/orchestrator/webhook/gitea`
- Events: `push`, `pull_request`, `status`
- Secret: значение `ORCH_GITEA_WEBHOOK_SECRET` в `.env`
- Secret: значение `ORCH_GITEA_WEBHOOK_SECRET` в `.env` — **ОДИН глобальный секрет на все
репо** (приёмник валидирует только его; новый секрет на одном репо сломал бы HMAC остальных —
при ротации меняется на всех репо разом)
- Signature header: `X-Gitea-Signature` (HMAC-SHA256 hex digest)
### Проверка
```bash
GITEA_TOKEN=$(grep ORCH_GITEA_TOKEN /home/slin/repos/orchestrator/.env | cut -d= -f2)
curl -s "http://localhost:3000/api/v1/repos/admin/enduro-trails/hooks" \
curl -s "http://localhost:3000/api/v1/repos/<owner>/<repo>/hooks" \
-H "Authorization: token ${GITEA_TOKEN}" | python3 -m json.tool
```
### Пересоздание (если нужно)
```bash
GITEA_WEBHOOK_SECRET=$(openssl rand -hex 20)
# Обновить в .env: ORCH_GITEA_WEBHOOK_SECRET=<new_secret>
# Секрет переиспользуй из .env (ORCH_GITEA_WEBHOOK_SECRET); генерируй новый ТОЛЬКО при
# первичной настройке/осознанной ротации (и обнови вебхуки ВСЕХ репо):
GITEA_WEBHOOK_SECRET=$(grep ORCH_GITEA_WEBHOOK_SECRET /home/slin/repos/orchestrator/.env | cut -d= -f2)
curl -X POST "http://localhost:3000/api/v1/repos/admin/enduro-trails/hooks" \
curl -X POST "http://localhost:3000/api/v1/repos/<owner>/<repo>/hooks" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H "Content-Type: application/json" \
-d '{

View File

@@ -0,0 +1,116 @@
"""ORCH-009 TC-21: pipeline invariants are untouched by the onboarding capability.
The onboarding kit/CLI lives entirely OUTSIDE the runtime (NFR-1): `src/**` is
byte-for-byte untouched. These tests pin that contract:
* a literal snapshot of ``STAGE_TRANSITIONS`` (the stage machine) and of the
``QG_CHECKS`` registry — any drift fails loudly;
* ``src/**`` never references the onboarding tree (no runtime coupling);
* the CLI's read-only imports from ``src`` stay within the CLOSED list of
ADR-001 D4 (ORCH-009) — extending the list requires an ADR update;
* kit prompt templates name only real quality gates (no phantom ``check_*``).
"""
import ast
import os
import re
from src.qg.checks import QG_CHECKS
from src.stages import STAGE_TRANSITIONS
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_SCRIPT_PATH = os.path.join(_REPO_ROOT, "scripts", "onboard_project.py")
_KIT_AGENTS = os.path.join(_REPO_ROOT, "onboarding", "repo-skeleton", ".openclaw", "agents")
# Literal snapshot of the stage machine (src/stages.py). Byte-exact NFR-1 pin:
# the onboarding work item must not move a single edge/agent/gate.
_EXPECTED_TRANSITIONS = {
"created": {"next": "analysis", "agent": "analyst", "qg": None},
"analysis": {"next": "architecture", "agent": "architect", "qg": "check_analysis_approved"},
"architecture": {"next": "development", "agent": "developer", "qg": "check_architecture_done"},
"development": {"next": "review", "agent": "reviewer", "qg": "check_ci_green"},
"review": {"next": "testing", "agent": "tester", "qg": "check_reviewer_verdict"},
"testing": {"next": "deploy-staging", "agent": "deployer", "qg": "check_tests_passed"},
"deploy-staging": {"next": "deploy", "agent": "deployer", "qg": "check_staging_status"},
"deploy": {"next": "done", "agent": None, "qg": "check_deploy_status"},
"done": {"next": None, "agent": None, "qg": None},
"cancelled": {"next": None, "agent": None, "qg": None},
}
# Snapshot of the QG registry KEYS (src/qg/checks.py::QG_CHECKS).
_EXPECTED_QG_KEYS = {
"check_analysis_approved",
"check_analysis_complete",
"check_architecture_done",
"check_ci_green",
"check_review_approved",
"check_tests_passed",
"check_reviewer_verdict",
"check_tests_local",
"check_deploy_status",
"check_staging_status",
"check_branch_mergeable",
"check_staging_image_fresh",
"check_security_gate",
"check_coverage_gate",
}
# Closed read-only import list of the onboarding CLI (ADR-001 D4 ORCH-009).
_ALLOWED_SRC_IMPORTS = {"src.config", "src.plane_sync", "src.projects"}
def test_tc21_stage_transitions_snapshot():
assert STAGE_TRANSITIONS == _EXPECTED_TRANSITIONS, (
"STAGE_TRANSITIONS drifted — ORCH-009 must not touch the stage machine (NFR-1)"
)
def test_tc21_qg_checks_registry_snapshot():
assert set(QG_CHECKS) == _EXPECTED_QG_KEYS, (
"QG_CHECKS registry drifted — ORCH-009 must not touch the gates (NFR-1)"
)
def test_tc21_src_never_references_onboarding():
"""No runtime coupling: src/** must not import/reference the onboarding tree."""
offenders = []
for root, _dirs, files in os.walk(os.path.join(_REPO_ROOT, "src")):
for name in files:
if not name.endswith(".py"):
continue
path = os.path.join(root, name)
with open(path, encoding="utf-8") as f:
if "onboard" in f.read().lower():
offenders.append(os.path.relpath(path, _REPO_ROOT))
assert not offenders, f"src/** references onboarding: {offenders}"
def test_tc21_cli_src_imports_stay_in_closed_list():
"""ADR-001 D4: the CLI may import ONLY src.config / src.plane_sync / src.projects."""
with open(_SCRIPT_PATH, encoding="utf-8") as f:
tree = ast.parse(f.read())
found = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith("src"):
found.add(node.module)
elif isinstance(node, ast.Import):
for alias in node.names:
if alias.name.startswith("src"):
found.add(alias.name)
assert found, "the CLI must use the closed src imports (round-trip via real parser)"
assert found <= _ALLOWED_SRC_IMPORTS, (
f"onboard_project.py imports outside the closed ADR D4 list: "
f"{sorted(found - _ALLOWED_SRC_IMPORTS)} — extend ONLY via an ADR update"
)
def test_tc21_kit_prompts_name_only_real_gates():
"""A kit prompt naming a phantom gate would mislead every onboarded project."""
pattern = re.compile(r"check_[a-z_]+")
for name in sorted(os.listdir(_KIT_AGENTS)):
path = os.path.join(_KIT_AGENTS, name)
with open(path, encoding="utf-8") as f:
text = f.read()
for gate in sorted(set(pattern.findall(text))):
assert gate in QG_CHECKS, (
f"kit prompt {name} references gate {gate!r} absent from QG_CHECKS"
)

View File

@@ -0,0 +1,414 @@
"""ORCH-009: structural tests of the onboarding kit (`onboarding/repo-skeleton/`).
Covers test-plan TC-01 (kit completeness), TC-03..TC-08 (prompt-template canon
52d/92), TC-19 (INFRA.md template sections) and TC-20 (ONBOARDING.md runbook).
Pure-text structural checks: NO network, NO agent runs (NFR-5). The kit prompt
templates are checked separately from the live orchestrator prompts
(`tests/test_agent_prompts_canon.py`) — the two trees must not be confused
(ADR-001 D1 ORCH-009).
"""
import json
import os
import re
import pytest
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_ONBOARDING = os.path.join(_REPO_ROOT, "onboarding")
_KIT = os.path.join(_ONBOARDING, "repo-skeleton")
_RUNBOOK = os.path.join(_REPO_ROOT, "docs", "operations", "ONBOARDING.md")
_AGENTS = ("analyst", "architect", "developer", "reviewer", "tester", "deployer")
# The 5 mandatory XML sections, in normative order (canon 52d, AC-2).
_SECTIONS = ("context", "task", "deliverables", "constraints", "output_format")
# The 6 mandatory 52c schema fields (mirrors src/frontmatter.py::REQUIRED_FIELDS,
# kept literal here on purpose: kit tests must not import src/ — NFR-1 hygiene).
_SCHEMA_FIELDS = ("work_item", "stage", "author_agent", "status", "created_at", "model_used")
# Role -> stage value(s) the template's example schema must pin (FR-2).
_STAGE_BY_ROLE = {
"analyst": ("analysis",),
"architect": ("architecture",),
"developer": ("development",),
"reviewer": ("review",),
"tester": ("testing",),
"deployer": ("deploy-staging", "deploy"),
}
def _read(*parts: str) -> str:
with open(os.path.join(_REPO_ROOT, *parts), encoding="utf-8") as f:
return f.read()
def _kit(*parts: str) -> str:
with open(os.path.join(_KIT, *parts), encoding="utf-8") as f:
return f.read()
def _prompt(agent: str) -> str:
return _kit(".openclaw", "agents", f"{agent}.md")
def _fenced_blocks(text: str) -> list[str]:
"""Return the body of every ``` fenced code block (the *copyable* examples)."""
blocks: list[str] = []
inside = False
buf: list[str] = []
for line in text.splitlines():
if line.lstrip().startswith("```"):
if inside:
blocks.append("\n".join(buf))
buf = []
inside = not inside
continue
if inside:
buf.append(line)
return blocks
# --------------------------------------------------------------------------- #
# TC-01 — kit completeness (AC-1 / FR-1)
# --------------------------------------------------------------------------- #
_REQUIRED_FILES = [
".openclaw/agents/analyst.md",
".openclaw/agents/architect.md",
".openclaw/agents/developer.md",
".openclaw/agents/reviewer.md",
".openclaw/agents/tester.md",
".openclaw/agents/deployer.md",
"CLAUDE.md",
"AGENTS.md",
"CONTRIBUTING.md",
"README.md",
"CHANGELOG.md",
".env.example",
"docs/ARCHITECTURE.md",
"docs/PIPELINE.md",
"docs/PRODUCT_VISION.md",
"docs/operations/INFRA.md",
"docs/architecture/adr/README.md",
"docs/work-items/.gitkeep",
"docs/history/.gitkeep",
]
def test_tc01_kit_contains_all_required_elements():
"""TC-01: every FR-1 element of the skeleton is present (6 prompts + carcass)."""
missing = [
rel for rel in _REQUIRED_FILES
if not os.path.isfile(os.path.join(_KIT, *rel.split("/")))
]
assert not missing, f"onboarding/repo-skeleton is missing: {missing}"
def test_tc01_kit_readme_and_placeholder_dictionary_exist():
"""TC-01/D1: onboarding/README.md + placeholders.json (single source of truth)."""
assert os.path.isfile(os.path.join(_ONBOARDING, "README.md"))
payload = json.loads(_read("onboarding", "placeholders.json"))
assert isinstance(payload, dict) and payload, "placeholders.json must be a non-empty dict"
for name, meta in payload.items():
assert re.fullmatch(r"[A-Z][A-Z0-9_]*", name), f"bad placeholder name {name!r}"
for key in ("description", "required", "default", "example"):
assert key in meta, f"placeholders.json[{name}] lacks {key!r}"
def test_kit_does_not_fork_the_canon():
"""BR-2/D3: no second editable copy of the canon inside the kit.
`docs/_templates/` and `docs/_standards/` are live-copied by the script at
materialisation time and must NOT be stored in the skeleton.
"""
for forbidden in ("docs/_templates", "docs/_standards"):
assert not os.path.exists(os.path.join(_KIT, *forbidden.split("/"))), (
f"kit must not store an editable canon copy: {forbidden}"
)
# --------------------------------------------------------------------------- #
# D2 — placeholder dictionary bijection (declared <-> used)
# --------------------------------------------------------------------------- #
_PLACEHOLDER_RE = re.compile(r"\{\{([A-Z][A-Z0-9_]*)\}\}")
def _kit_files() -> list[str]:
out = []
for root, _dirs, files in os.walk(_KIT):
for name in files:
out.append(os.path.join(root, name))
return out
def test_placeholder_dictionary_bijection():
"""D2: every placeholder used in the kit is declared, every declared is used."""
declared = set(json.loads(_read("onboarding", "placeholders.json")))
used: set[str] = set()
for path in _kit_files():
with open(path, encoding="utf-8") as f:
used.update(_PLACEHOLDER_RE.findall(f.read()))
assert used == declared, (
f"placeholder drift: used-not-declared={sorted(used - declared)}, "
f"declared-not-used={sorted(declared - used)}"
)
# --------------------------------------------------------------------------- #
# TC-03 — 5 XML sections in normative order (AC-2)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("agent", _AGENTS)
def test_tc03_five_xml_sections_in_normative_order(agent):
"""Real section tags sit on their own line; inline backticked mentions
(e.g. «см. `<output_format>`» inside <task>) must not be mistaken for them
(same disambiguation as the ORCH-092 <escalation> check)."""
text = _prompt(agent)
positions = []
for section in _SECTIONS:
open_m = re.search(rf"(?m)^<{section}>\s*$", text)
close_m = re.search(rf"(?m)^</{section}>\s*$", text)
assert open_m, f"kit {agent}.md missing <{section}> on its own line"
assert close_m, f"kit {agent}.md missing </{section}> on its own line"
positions.append(open_m.start())
assert positions == sorted(positions), (
f"kit {agent}.md sections out of normative order "
f"context→task→deliverables→constraints→output_format"
)
# --------------------------------------------------------------------------- #
# TC-04 — <escalation> at dev/reviewer/tester; bans in «❌ → ✅» form (AC-2)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("agent", ("developer", "reviewer", "tester"))
def test_tc04_escalation_section_after_success_criteria(agent):
text = _prompt(agent)
open_m = re.search(r"(?m)^<escalation>\s*$", text)
close_m = re.search(r"(?m)^</escalation>\s*$", text)
assert open_m and close_m, f"kit {agent}.md is missing the <escalation> section"
success_m = re.search(r"(?m)^</success_criteria>\s*$", text)
assert success_m and open_m.start() > success_m.start(), (
f"kit {agent}.md must place <escalation> after </success_criteria>"
)
@pytest.mark.parametrize("agent", _AGENTS)
def test_tc04_bans_use_cross_check_format(agent):
text = _prompt(agent)
assert "" in text and "" in text, (
f"kit {agent}.md must format bans as «❌ X → ✅ Y»"
)
# --------------------------------------------------------------------------- #
# TC-05 — each template directs the agent to the project docs (AC-2 / BR-3)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("agent", _AGENTS)
def test_tc05_prompt_directs_agent_to_docs(agent):
text = _prompt(agent)
for marker in (
"CLAUDE.md", # passport, read BEFORE work
"AGENTS.md", # docs map / entry point
"docs/ARCHITECTURE.md", # architecture doc
"docs/work-items/", # artefact home
"PIPELINE_DOCS.md", # docs standard
"docs/_templates/", # skeletons
):
assert marker in text, f"kit {agent}.md does not reference {marker!r}"
@pytest.mark.parametrize("agent", ("developer", "reviewer"))
def test_tc05_changelog_duty_present(agent):
assert "CHANGELOG.md" in _prompt(agent), (
f"kit {agent}.md must carry the CHANGELOG update duty"
)
def test_tc05_architect_carries_adr_rules():
text = _prompt("architect")
assert "06-adr/" in text, "kit architect.md must route decisions to 06-adr/"
assert "docs/architecture/adr/" in text, (
"kit architect.md must carry the cross-cutting ADR rule"
)
# --------------------------------------------------------------------------- #
# TC-06 — 52c schema emission + byte-exact machine-verdict keys (AC-2)
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("agent", _AGENTS)
def test_tc06_six_schema_fields_named(agent):
text = _prompt(agent)
for field in _SCHEMA_FIELDS:
assert field in text, f"kit {agent}.md does not mention schema field {field!r}"
@pytest.mark.parametrize("agent", _AGENTS)
def test_tc06_schema_pins_role_author_and_stage(agent):
text = _prompt(agent)
assert f"author_agent: {agent}" in text, (
f"kit {agent}.md does not pin 'author_agent: {agent}'"
)
for stage in _STAGE_BY_ROLE[agent]:
assert f"stage: {stage}" in text, f"kit {agent}.md does not pin 'stage: {stage}'"
def test_tc06_machine_verdict_keys_byte_exact():
reviewer = _prompt("reviewer")
assert "verdict:" in reviewer
assert "APPROVED" in reviewer and "REQUEST_CHANGES" in reviewer
tester = _prompt("tester")
assert "result:" in tester
assert "PASS" in tester and "FAIL" in tester
deployer = _prompt("deployer")
assert "staging_status:" in deployer
assert "deploy_status:" in deployer
assert "security_status:" in deployer
assert "SUCCESS" in deployer and "FAILED" in deployer
@pytest.mark.parametrize("agent", _AGENTS)
def test_tc06_dates_and_models_are_placeholders(agent):
"""Anti-pattern ORCH-092: no literal date/model inside copyable examples."""
text = _prompt(agent)
assert "created_at: <YYYY-MM-DD>" in text, (
f"kit {agent}.md must use the created_at: <YYYY-MM-DD> placeholder"
)
assert "date +%F" in text, (
f"kit {agent}.md must instruct to substitute the actual date (date +%F)"
)
for block in _fenced_blocks(text):
assert re.search(r"created_at:\s*\d", block) is None, (
f"kit {agent}.md hardcodes a literal created_at date in a copyable block"
)
assert re.search(r"model_used:\s*claude", block) is None, (
f"kit {agent}.md hardcodes a literal model in a copyable block"
)
# --------------------------------------------------------------------------- #
# TC-07 — reviewer-gate on documentation (AC-3 / BR-4)
# --------------------------------------------------------------------------- #
def test_tc07_reviewer_gate_docs_not_updated_means_request_changes():
text = _prompt("reviewer")
assert "REQUEST_CHANGES" in text
assert "НЕ обновлена" in text, (
"kit reviewer.md must carry the mandatory gate: docs NOT updated -> "
"verdict: REQUEST_CHANGES"
)
# --------------------------------------------------------------------------- #
# TC-08 — language policy: 5 ru + deployer en (AC-4 / D9)
# --------------------------------------------------------------------------- #
_CYRILLIC = re.compile(r"[а-яА-ЯёЁ]")
@pytest.mark.parametrize("agent", ("analyst", "architect", "developer", "reviewer", "tester"))
def test_tc08_ru_canon_for_five_roles(agent):
assert _CYRILLIC.search(_prompt(agent)), (
f"kit {agent}.md must follow the ru canon (ADR-001 D9 ORCH-009)"
)
def test_tc08_deployer_is_english():
text = _prompt("deployer")
assert not _CYRILLIC.search(text), (
"kit deployer.md must stay 100% English (safety-critical canon, D9)"
)
assert "Do NOT translate" in text, (
"kit deployer.md must carry the language-note guard"
)
# --------------------------------------------------------------------------- #
# TC-19 — INFRA.md template: mandatory sections (AC-10 / FR-3)
# --------------------------------------------------------------------------- #
def test_tc19_infra_template_mandatory_sections():
text = _kit("docs", "operations", "INFRA.md")
assert "Топология" in text, "INFRA template lacks the topology section"
assert "{{PROD_PORT}}" in text and "{{STAGING_PORT}}" in text, (
"INFRA template must parametrise prod/staging ports"
)
assert "env" in text.lower(), "INFRA template lacks the env map section"
assert ".env.example" in text, "INFRA template lacks the .env.example canon rule"
assert "Границы доступа" in text, "INFRA template lacks the access-boundaries section"
assert "общего хоста" in text or "общий хост" in text, (
"INFRA template lacks the shared-host risk warnings"
)
assert "секрет" in text.lower(), "INFRA template lacks the secrets rule"
def test_tc19_orchestrator_own_infra_untouched_sections():
"""AC-10: the orchestrator's own INFRA.md keeps its self-hosting warnings."""
own = _read("docs", "operations", "INFRA.md")
assert "orchestrator" in own and "8500" in own, (
"docs/operations/INFRA.md of the orchestrator must stay the self-hosting runbook"
)
# --------------------------------------------------------------------------- #
# TC-20 — runbook ONBOARDING.md covers all layers in order (AC-11 / FR-6)
# --------------------------------------------------------------------------- #
def test_tc20_runbook_exists_and_layer_order():
assert os.path.isfile(_RUNBOOK), "docs/operations/ONBOARDING.md is missing"
text = _read("docs", "operations", "ONBOARDING.md")
# All BR-1 layers, in sequence.
anchors = ["Предусловия", "Plane", "Gitea", "kit", "Регистрация", "Верификация", "Откат"]
positions = []
for anchor in anchors:
idx = text.find(anchor)
assert idx != -1, f"ONBOARDING.md lacks the {anchor!r} layer"
positions.append(idx)
assert positions == sorted(positions), (
f"ONBOARDING.md layers out of order: {anchors}"
)
def test_tc20_runbook_manual_steps_and_selfhosting_warning():
text = _read("docs", "operations", "ONBOARDING.md")
assert "ручной шаг" in text.lower() or "РУЧНОЙ ШАГ" in text, (
"ONBOARDING.md must explicitly mark manual steps"
)
assert "рестарт" in text.lower(), (
"ONBOARDING.md must describe the operator-managed restart step"
)
assert "self-hosting" in text or "групповое окно" in text, (
"ONBOARDING.md must warn that a prod restart is a group-wide window"
)
# Plane workspace-webhook already exists: verify, never create (Ф-6).
assert "workspace" in text.lower(), "ONBOARDING.md must cover the workspace webhook"
assert "существует" in text, (
"ONBOARDING.md must state the Plane workspace-webhook already exists"
)
def test_tc20_runbook_verification_and_smoke_journal():
text = _read("docs", "operations", "ONBOARDING.md")
assert "verify" in text, "ONBOARDING.md must document the verify mode"
assert "8501" in text, "ONBOARDING.md smoke contour must be staging (8501) — D8"
assert "Журнал smoke-прогонов" in text, (
"ONBOARDING.md must carry the smoke-run journal section (D8)"
)
assert "onboard_project.py" in text, "ONBOARDING.md must reference the CLI"
def test_setup_webhooks_generalised():
"""TRZ §2: SETUP_WEBHOOKS.md is generalised per-repo + references the runbook."""
text = _read("docs", "operations", "SETUP_WEBHOOKS.md")
assert "ONBOARDING.md" in text, (
"SETUP_WEBHOOKS.md must reference docs/operations/ONBOARDING.md"
)
assert "<repo>" in text or "{repo}" in text, (
"SETUP_WEBHOOKS.md per-repo section must be generalised, not enduro-hardcoded"
)

View File

@@ -0,0 +1,605 @@
"""ORCH-009: tests of the operator onboarding CLI (`scripts/onboard_project.py`).
Covers test-plan TC-02 (live-copy of the canon), TC-09..TC-11 (render /
anti-leak / referential integrity), TC-12 (registry round-trip through the
actual parser), TC-13..TC-16 (plan: Plane/Gitea completeness + dry-run with
zero mutations), TC-17..TC-18 (idempotent & safe apply).
All tests are deterministic and offline (NFR-5): the Plane/Gitea clients are
replaced with in-memory fakes; git is replaced with a recording runner. The
script module is loaded via importlib (pattern: tests/test_staging_check_b6.py).
"""
import importlib.util
import json
import os
import re
import pytest
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_SCRIPT_PATH = os.path.join(_REPO_ROOT, "scripts", "onboard_project.py")
def _load_module():
spec = importlib.util.spec_from_file_location("onboard_project", _SCRIPT_PATH)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.fixture(scope="module")
def mod():
return _load_module()
_WEBHOOK_URL = "https://orchestrator.example.org/webhook/gitea"
def _params(**over):
p = {
"PROJECT_NAME": "Demo Project",
"PROJECT_DESCRIPTION": "Демо-проект для проверки онбординга",
"REPO": "demo-project",
"GITEA_OWNER": "admin",
"WORK_ITEM_PREFIX": "DEMO",
"PLANE_PROJECT_ID": "11111111-2222-3333-4444-555555555555",
"STACK": "Python 3.12 + FastAPI + SQLite",
"TEST_CMD": "pytest tests/ -q",
"PROD_PORT": "8600",
"STAGING_PORT": "8601",
}
p.update(over)
return p
def _step(report, step_id):
matches = [s for s in report.steps if s.id == step_id]
assert matches, f"report has no step {step_id!r}: {[s.id for s in report.steps]}"
return matches[0]
# --------------------------------------------------------------------------- #
# Fakes — the only network touchpoints of the script, replaced in-memory.
# --------------------------------------------------------------------------- #
class FakePlane:
def __init__(self, mod, project=None, states=None, labels=None,
refuse_create_states=False, refuse_create_labels=False,
refuse_create_project=False):
self._mod = mod
self.project = project
self.states = list(states or [])
self.labels = list(labels or [])
self.mutations = []
self.refuse_create_states = refuse_create_states
self.refuse_create_labels = refuse_create_labels
self.refuse_create_project = refuse_create_project
# GET probes
def get_project(self, project_id):
if self.project and self.project.get("id") == project_id:
return self.project
return None
def find_project_by_identifier(self, identifier):
if self.project and self.project.get("identifier") == identifier:
return self.project
return None
def list_states(self, project_id):
return list(self.states)
def list_labels(self, project_id):
return list(self.labels)
# mutations
def create_project(self, name, identifier):
if self.refuse_create_project:
raise self._mod.ManualStep("Plane CE: projects API not available")
self.mutations.append(("create_project", name, identifier))
self.project = {"id": "plane-uuid-created", "name": name, "identifier": identifier}
return self.project
def create_state(self, project_id, name, group):
if self.refuse_create_states:
raise self._mod.ManualStep("Plane CE: states API not available")
self.mutations.append(("create_state", name, group))
state = {"id": f"uuid-{name}", "name": name, "group": group}
self.states.append(state)
return state
def create_label(self, project_id, name):
if self.refuse_create_labels:
raise self._mod.ManualStep("Plane CE: labels API not available")
self.mutations.append(("create_label", name))
label = {"id": f"uuid-{name}", "name": name}
self.labels.append(label)
return label
class FakeGitea:
def __init__(self, repo=None, hooks=None, files=None):
self.repo = repo
self.hooks = list(hooks or [])
self.files = dict(files or {}) # repo path -> text (for verify)
self.mutations = []
def get_repo(self, owner, repo):
return self.repo
def list_hooks(self, owner, repo):
return list(self.hooks)
def create_repo(self, owner, name, description=""):
self.mutations.append(("create_repo", owner, name))
self.repo = {"name": name, "owner": {"login": owner}, "empty": True}
return self.repo
def create_hook(self, owner, repo, url, secret, events):
self.mutations.append(("create_hook", url, tuple(events)))
hook = {"id": 1, "active": True, "config": {"url": url}, "events": list(events)}
self.hooks.append(hook)
return hook
# verify helpers
def get_file_text(self, owner, repo, path):
return self.files.get(path)
def list_dir(self, owner, repo, path):
prefix = path.rstrip("/") + "/"
names = {
rel[len(prefix):].split("/", 1)[0]
for rel in self.files
if rel.startswith(prefix)
}
return sorted(names) or None
def _full_states(mod):
return [
{"id": f"uuid-{name}", "name": name, "group": group}
for name, group in mod.STATE_GROUPS.items()
]
def _full_labels(mod):
return [{"id": f"uuid-{name}", "name": name} for name in mod.label_names()]
# --------------------------------------------------------------------------- #
# TC-02 — materialisation live-copies the canon (BR-2 / D3)
# --------------------------------------------------------------------------- #
def test_tc02_materialise_live_copies_canon(mod, tmp_path):
dest = tmp_path / "repo"
written = mod.materialize_kit(_params(), str(dest))
assert written, "materialize_kit wrote nothing"
templates = os.listdir(dest / "docs" / "_templates")
standards = os.listdir(dest / "docs" / "_standards")
assert len(templates) >= 16, f"expected >=16 canonical skeletons, got {len(templates)}"
assert len(standards) >= 3, f"expected >=3 standards, got {len(standards)}"
# Verbatim copy — byte-equal to the live canon of the orchestrator checkout.
for rel in ("PIPELINE_DOCS.md", "HANDOFF_PROTOCOL.md", "TRACEABILITY.md"):
src_path = os.path.join(_REPO_ROOT, "docs", "_standards", rel)
with open(src_path, encoding="utf-8") as f:
canon = f.read()
with open(dest / "docs" / "_standards" / rel, encoding="utf-8") as f:
copied = f.read()
assert copied == canon, f"{rel} must be live-copied verbatim (BR-2)"
# --------------------------------------------------------------------------- #
# TC-09 / TC-10 — render: no unresolved placeholders, no orc leaks (AC-5)
# --------------------------------------------------------------------------- #
def test_tc09_render_resolves_all_placeholders(mod):
rendered = mod.render_kit_in_memory(_params())
assert rendered, "render_kit_in_memory returned nothing"
for rel, content in rendered.items():
unresolved = mod.find_unresolved(content)
assert not unresolved, f"{rel} keeps unresolved placeholders: {unresolved}"
def test_tc10_no_orchestrator_specific_leaks(mod):
rendered = mod.render_kit_in_memory(_params())
joined = "\n".join(rendered.values())
assert re.search(r"ORCH-\d", joined) is None, (
"rendered kit leaks an ORCH-NNN work-item literal where the project "
"prefix belongs (TC-10)"
)
assert "8500" not in joined and "8501" not in joined, (
"rendered kit leaks the orchestrator prod/staging ports"
)
assert "self-hosting" not in joined.lower(), (
"rendered kit leaks the orchestrator self-hosting rules"
)
# The project's own parameters actually got substituted.
assert "DEMO-" in joined, "the project's work-item prefix was not substituted"
assert "demo-project" in joined, "the repo name was not substituted"
assert "8600" in joined and "8601" in joined, "ports were not substituted"
def test_render_is_a_pure_replace(mod):
text = "prefix {{WORK_ITEM_PREFIX}}-12 on port {{PROD_PORT}}"
out = mod.render(text, {"WORK_ITEM_PREFIX": "AB", "PROD_PORT": "9000"})
assert out == "prefix AB-12 on port 9000"
assert mod.find_unresolved("a {{LEFT_OVER}} b") == ["{{LEFT_OVER}}"]
# --------------------------------------------------------------------------- #
# TC-11 — referential integrity of rendered prompts/AGENTS.md (AC-5)
# --------------------------------------------------------------------------- #
_PATH_TOKEN = re.compile(
r"(?:docs/[\w./\-]+|\.openclaw/agents/[\w.\-]+|CLAUDE\.md|AGENTS\.md|"
r"CONTRIBUTING\.md|README\.md|CHANGELOG\.md|\.env\.example)"
)
def _static_paths(text: str) -> set[str]:
out = set()
for token in _PATH_TOKEN.findall(text):
token = token.rstrip(".,;:)`'\"")
# dynamic/illustrative tokens are not checkable paths
if any(ch in token for ch in "<>*{}") or "NNN" in token:
continue
out.add(token)
return out
def test_tc11_referenced_paths_exist_in_materialised_tree(mod, tmp_path):
dest = tmp_path / "repo"
mod.materialize_kit(_params(), str(dest))
sources = [
os.path.join(dest, ".openclaw", "agents", f"{a}.md")
for a in ("analyst", "architect", "developer", "reviewer", "tester", "deployer")
]
sources.append(os.path.join(dest, "AGENTS.md"))
broken = []
for src_file in sources:
with open(src_file, encoding="utf-8") as f:
for ref in _static_paths(f.read()):
target = os.path.join(dest, *ref.split("/"))
if not (os.path.isfile(target) or os.path.isdir(target.rstrip("/"))):
broken.append((os.path.relpath(src_file, dest), ref))
assert not broken, f"kit references non-existent paths: {broken}"
# --------------------------------------------------------------------------- #
# TC-12 — registry round-trip through the ACTUAL parser (AC-6)
# --------------------------------------------------------------------------- #
def test_tc12_registry_round_trip_through_actual_parser(mod):
from src.projects import _parse_projects_json
existing = [
{
"plane_project_id": "7a79f0a9-5278-49cd-9007-9a338f238f9c",
"repo": "enduro-trails",
"work_item_prefix": "ET",
"name": "enduro-trails",
},
{
"plane_project_id": "8da6aa25-a60e-44d6-a1e2-d8ae59aa7d6a",
"repo": "orchestrator",
"work_item_prefix": "ORCH",
"name": "orchestrator",
},
]
params = _params()
entry = mod.build_registry_entry(params)
standalone, merged = mod.merged_projects_json(entry, json.dumps(existing))
# standalone entry parses on its own
solo = _parse_projects_json(f"[{standalone}]")
assert solo is not None and len(solo) == 1
parsed = _parse_projects_json(merged)
assert parsed is not None and len(parsed) == 3, "merged registry must carry all 3"
# existing entries survive verbatim (no loss, no distortion)
for i, exp in enumerate(existing):
assert parsed[i].plane_project_id == exp["plane_project_id"]
assert parsed[i].repo == exp["repo"]
assert parsed[i].work_item_prefix == exp["work_item_prefix"]
assert parsed[i].name == exp["name"]
# the new entry carries the source params
new = parsed[2]
assert new.plane_project_id == params["PLANE_PROJECT_ID"]
assert new.repo == params["REPO"]
assert new.work_item_prefix == params["WORK_ITEM_PREFIX"]
assert new.name == params["PROJECT_NAME"]
def test_tc12_merge_is_idempotent_no_duplicates(mod):
from src.projects import _parse_projects_json
params = _params()
entry = mod.build_registry_entry(params)
once = json.dumps([entry])
_standalone, merged = mod.merged_projects_json(entry, once)
parsed = _parse_projects_json(merged)
assert parsed is not None and len(parsed) == 1, (
"re-merging an already-registered project must not duplicate it"
)
# --------------------------------------------------------------------------- #
# TC-13 — plan: exact Plane statuses (22) + groups + labels (AC-7)
# --------------------------------------------------------------------------- #
def test_state_groups_match_plane_name_to_key(mod):
from src.plane_sync import _PLANE_NAME_TO_KEY
assert set(mod.STATE_GROUPS) == set(_PLANE_NAME_TO_KEY), (
"STATE_GROUPS must cover exactly the canonical Plane status names"
)
# Code-critical constraints (ADR-001 D5): STOP joins the cancelled group
# (ORCH-090 fail-closed cancel); only Done/Cancelled/STOP are terminal —
# otherwise terminal-detection (ORCH-068) falsely terminates live tasks.
assert mod.STATE_GROUPS["STOP"] == "cancelled"
assert mod.STATE_GROUPS["Done"] == "completed"
assert mod.STATE_GROUPS["Cancelled"] == "cancelled"
terminal = {n for n, g in mod.STATE_GROUPS.items() if g in ("completed", "cancelled")}
assert terminal == {"Done", "Cancelled", "STOP"}
def test_tc13_plan_covers_all_statuses_and_labels(mod):
from src.plane_sync import _PLANE_NAME_TO_KEY
plane = FakePlane(mod)
gitea = FakeGitea()
report = mod.run_plan(_params(), plane, gitea, webhook_url=_WEBHOOK_URL)
for name in _PLANE_NAME_TO_KEY:
step = _step(report, f"plane.state:{name}")
assert step.status == mod.PLANNED, f"status {name!r} not planned: {step.status}"
stop = _step(report, "plane.state:STOP")
assert "cancelled" in stop.detail, "STOP step must pin the cancelled group"
for label in mod.label_names():
assert _step(report, f"plane.label:{label}").status == mod.PLANNED
assert set(mod.label_names()) == {"autoApprove", "autoDeploy", "Bug"}
# known UI-only steps are flagged manual, never silently dropped (D5)
assert _step(report, "plane.board-order").status == mod.MANUAL
assert _step(report, "plane.workspace-webhook").status == mod.MANUAL
# --------------------------------------------------------------------------- #
# TC-14 — Plane API refusal degrades to manual-step, never a crash (AC-7)
# --------------------------------------------------------------------------- #
def test_tc14_plane_refusal_becomes_manual_step(mod, tmp_path):
plane = FakePlane(
mod,
project={"id": _params()["PLANE_PROJECT_ID"], "identifier": "DEMO"},
refuse_create_states=True,
refuse_create_labels=True,
)
gitea = FakeGitea(
repo={"name": "demo-project", "empty": False},
hooks=[{"id": 1, "active": True, "config": {"url": _WEBHOOK_URL}}],
)
report = mod.run_apply(
_params(), plane, gitea,
webhook_url=_WEBHOOK_URL, git_runner=lambda cmd, cwd: 0,
workdir=str(tmp_path),
)
state_steps = [s for s in report.steps if s.id.startswith("plane.state:")]
assert state_steps and all(s.status == mod.MANUAL for s in state_steps), (
"refused Plane state creation must degrade to manual-step"
)
for s in state_steps:
assert "ONBOARDING.md" in s.detail, "manual-step must link the runbook"
label_steps = [s for s in report.steps if s.id.startswith("plane.label:")]
assert label_steps and all(s.status == mod.MANUAL for s in label_steps)
assert report.exit_code == 2, "manual steps -> exit code 2"
# --------------------------------------------------------------------------- #
# TC-15 / TC-16 — plan: Gitea layer complete; dry-run mutates NOTHING (AC-8)
# --------------------------------------------------------------------------- #
def test_tc15_plan_contains_gitea_repo_webhook_and_push(mod):
plane = FakePlane(mod)
gitea = FakeGitea()
report = mod.run_plan(_params(), plane, gitea, webhook_url=_WEBHOOK_URL)
assert _step(report, "gitea.repo").status == mod.PLANNED
hook = _step(report, "gitea.webhook")
assert hook.status == mod.PLANNED
for event in ("push", "pull_request", "status"):
assert event in hook.detail, f"webhook plan must name event {event!r}"
assert "HMAC" in hook.detail or "secret" in hook.detail.lower(), (
"webhook plan must mention the HMAC secret (kept out of git)"
)
push = _step(report, "kit.push")
assert push.status == mod.PLANNED
assert "push" in push.detail.lower()
def test_tc16_plan_is_a_pure_dry_run(mod, monkeypatch):
plane = FakePlane(mod)
gitea = FakeGitea()
def _boom(*_a, **_kw): # plan must never materialise or push
raise AssertionError("plan mode touched the disk / git")
monkeypatch.setattr(mod, "materialize_kit", _boom)
monkeypatch.setattr(mod, "initial_push", _boom)
report = mod.run_plan(_params(), plane, gitea, webhook_url=_WEBHOOK_URL)
assert plane.mutations == [], "plan made a Plane mutation"
assert gitea.mutations == [], "plan made a Gitea mutation"
assert report.steps, "plan produced an empty report"
def test_secret_never_leaks_into_report(mod):
plane = FakePlane(mod)
gitea = FakeGitea()
report = mod.run_plan(
_params(), plane, gitea, webhook_url=_WEBHOOK_URL,
webhook_secret="super-secret-hmac-value",
)
dumped = json.dumps(report.to_dict(), ensure_ascii=False)
assert "super-secret-hmac-value" not in dumped, (
"the webhook HMAC secret leaked into the report (NFR-3)"
)
# --------------------------------------------------------------------------- #
# TC-17 — apply is idempotent: existing entities -> skipped(exists) (AC-9)
# --------------------------------------------------------------------------- #
def test_tc17_second_apply_skips_everything_existing(mod, tmp_path):
params = _params()
plane = FakePlane(
mod,
project={"id": params["PLANE_PROJECT_ID"], "identifier": "DEMO"},
states=_full_states(mod),
labels=_full_labels(mod),
)
gitea = FakeGitea(
repo={"name": params["REPO"], "empty": False},
hooks=[{"id": 7, "active": True, "config": {"url": _WEBHOOK_URL}}],
)
git_calls = []
report = mod.run_apply(
params, plane, gitea, webhook_url=_WEBHOOK_URL,
git_runner=lambda cmd, cwd: git_calls.append((cmd, cwd)) or 0,
workdir=str(tmp_path),
)
assert plane.mutations == [], "idempotent apply must not re-create Plane entities"
assert gitea.mutations == [], "idempotent apply must not re-create Gitea entities"
assert git_calls == [], "apply must NEVER push into a non-empty existing repo"
assert _step(report, "plane.project").status == mod.SKIPPED
for name in mod.STATE_GROUPS:
assert _step(report, f"plane.state:{name}").status == mod.SKIPPED
for label in mod.label_names():
assert _step(report, f"plane.label:{label}").status == mod.SKIPPED
assert _step(report, "gitea.repo").status == mod.SKIPPED
assert _step(report, "gitea.webhook").status == mod.SKIPPED
assert _step(report, "kit.push").status == mod.MANUAL, (
"non-empty repo -> kit push degrades to a manual step, never an overwrite"
)
summary = report.to_dict()
for key in ("created", "skipped", "manual"):
assert key in summary["totals"], f"report totals lack {key!r}"
# --------------------------------------------------------------------------- #
# TC-18 — apply runs no restarts / no prod-.env edits / git only (NFR-2)
# --------------------------------------------------------------------------- #
def test_tc18_fresh_apply_runs_git_only_inside_workdir(mod, tmp_path):
params = _params()
plane = FakePlane(mod)
gitea = FakeGitea()
calls = []
def recorder(cmd, cwd):
calls.append((list(cmd), cwd))
return 0
report = mod.run_apply(
params, plane, gitea, webhook_url=_WEBHOOK_URL,
git_runner=recorder, workdir=str(tmp_path),
)
assert calls, "fresh empty repo: the initial push pipeline must run"
for cmd, cwd in calls:
assert cmd[0] == "git", f"only git may be executed, got: {cmd}"
assert cwd and str(tmp_path) in cwd, (
f"git must run only inside the materialisation workdir, got cwd={cwd}"
)
joined = " ".join(" ".join(c) for c, _ in calls)
assert "docker" not in joined and "restart" not in joined
assert _step(report, "kit.push").status == mod.CREATED
assert ("create_repo", "admin", "demo-project") in gitea.mutations
hook_calls = [m for m in gitea.mutations if m[0] == "create_hook"]
assert hook_calls and hook_calls[0][1] == _WEBHOOK_URL
assert set(hook_calls[0][2]) == {"push", "pull_request", "status"}
def test_tc18_source_has_no_container_or_env_mutation_ops(mod):
with open(_SCRIPT_PATH, encoding="utf-8") as f:
source = f.read()
lowered = source.lower()
assert "docker" not in lowered, "the script must not touch containers (NFR-2)"
assert "systemctl" not in lowered
assert "compose" not in lowered
assert re.search(r"open\([^)]*\.env[^)]*['\"][wa]", source) is None, (
"the script must never WRITE any .env (read-only access allowed)"
)
# --------------------------------------------------------------------------- #
# verify — registry / states / labels / webhook / kit completeness (FR-5)
# --------------------------------------------------------------------------- #
def _verify_files(mod):
params = _params()
rendered = mod.render_kit_in_memory(params)
files = dict(rendered)
for i in range(16):
files[f"docs/_templates/{i:02d}-skeleton.md"] = "x"
for name in ("PIPELINE_DOCS.md", "HANDOFF_PROTOCOL.md", "TRACEABILITY.md"):
files[f"docs/_standards/{name}"] = "x"
return files
def test_verify_all_green(mod):
params = _params()
plane = FakePlane(
mod,
project={"id": params["PLANE_PROJECT_ID"], "identifier": "DEMO"},
states=_full_states(mod),
labels=_full_labels(mod),
)
gitea = FakeGitea(
repo={"name": params["REPO"], "empty": False},
hooks=[{"id": 1, "active": True, "config": {"url": _WEBHOOK_URL}}],
files=_verify_files(mod),
)
entry = mod.build_registry_entry(params)
_, merged = mod.merged_projects_json(entry, "[]")
report = mod.run_verify(
params, plane, gitea, webhook_url=_WEBHOOK_URL, projects_raw=merged,
)
gaps = [s for s in report.steps if s.status == mod.GAP]
assert not gaps, f"verify reported gaps on a fully onboarded project: {gaps}"
def test_verify_flags_missing_failclosed_statuses(mod):
params = _params()
states = [s for s in _full_states(mod) if s["name"] not in ("STOP", "Confirm Deploy")]
plane = FakePlane(
mod,
project={"id": params["PLANE_PROJECT_ID"], "identifier": "DEMO"},
states=states,
labels=_full_labels(mod),
)
gitea = FakeGitea(
repo={"name": params["REPO"], "empty": False},
hooks=[{"id": 1, "active": True, "config": {"url": _WEBHOOK_URL}}],
files=_verify_files(mod),
)
entry = mod.build_registry_entry(params)
_, merged = mod.merged_projects_json(entry, "[]")
report = mod.run_verify(
params, plane, gitea, webhook_url=_WEBHOOK_URL, projects_raw=merged,
)
states_step = _step(report, "verify.plane.states")
assert states_step.status == mod.GAP
assert "STOP" in states_step.detail and "Confirm Deploy" in states_step.detail, (
"verify must name the missing fail-closed statuses explicitly"
)
assert report.exit_code == 2