orchestrator/tests/test_lite_setup_doc.py

"""ORCH-102 (FR-6 / AC-1, AC-2, AC-3, AC-6): анти-дрейф контур Lite-тиража.

Структурные проверки golden source `docs/deployment/LITE_SETUP.md` (ADR-001 D8,
образец — `tests/test_replication_smoke.py`): док существует и несёт 13
нормативных разделов D2 в порядке маршрута оператора; обязательные кирпичи
(TC-02); key-sync `.env.watchdog.example` ⇄ блок `WATCHDOG_*` `.env.example`
(TC-02b, D5); каждый упомянутый в доке env-ключ существует в `.env.example`
(TC-03, анти-опечатка); compose-подмножество — ровно орк+watchdog по дефолту,
staging строго за профилем, никаких Plane/Gitea-сервисов (TC-04, D4);
stateless-норматив и секрет-гигиена fenced-блоков (TC-05, FORBIDDEN —
импорт из `tests/test_no_host_hardcodes.py`, не копия); канон не форкается —
статусы/env/smoke ссылками, fail-closed имена явно (TC-06, D3); инварианты
Gitea-раздела (TC-07); перекрёстность REPLICATION→LITE_SETUP + CHANGELOG
(TC-08). Детерминировано: без сети/LLM/subprocess (NFR/TR-7 — ассерты только
на стабильное: заголовки, подстроки-кирпичи, парсинг ключей, yaml.safe_load).
"""

import re
from pathlib import Path

import yaml

# Один источник истины запрещённых боевых литералов (ADR-001 D8 / ORCH-101 AC-7).
from tests.test_no_host_hardcodes import FORBIDDEN

REPO_ROOT = Path(__file__).resolve().parents[1]
LITE_SETUP = REPO_ROOT / "docs/deployment/LITE_SETUP.md"
REPLICATION = REPO_ROOT / "docs/operations/REPLICATION.md"
CHANGELOG = REPO_ROOT / "CHANGELOG.md"
ENV_EXAMPLE = REPO_ROOT / ".env.example"
ENV_WATCHDOG_EXAMPLE = REPO_ROOT / ".env.watchdog.example"
COMPOSE = REPO_ROOT / "docker-compose.yml"

# Нормативная структура D2: фиксированная нумерация, порядок = маршрут оператора.
SECTIONS: tuple[str, ...] = (
    "## 1. Рамка Lite",
    "## 2. Предусловия хоста",
    "## 3. Перенос кода",
    "## 4. Конфигурация",
    "## 5. Подключение Plane",
    "## 6. Подключение Gitea",
    "## 7. LLM (claude CLI)",
    "## 8. Telegram",
    "## 9. Запуск",
    "## 10. Регистрация проекта",
    "## 11. Smoke",
    "## 12. Stateless-проверка",
    "## 13. Траблшутинг",
)

# Обязательные кирпичи FR-6.1 (подстроки; TC-02).
BRICKS: tuple[str, ...] = (
    "gen_secrets.py",
    "onboard_project.py",
    "docker compose",
    "/health",
    "/queue",
    "/metrics",
    "ORCH_PROJECTS_JSON",
    "ORCH_PLANE_WEBHOOK_SECRET",
    "ORCH_GITEA_WEBHOOK_SECRET",
    "X-Plane-Signature",
    "X-Gitea-Signature",
    "getent group docker",
    "Confirm Deploy",
    "STOP",
    "ORCH_TELEGRAM_BOT_TOKEN",   # канал трекера орка
    "WATCHDOG_TG_BOT_TOKEN",     # независимый канал sidecar-watchdog (C-1 ORCH-100)
    "PASS",
    "FAIL",
    "Проверка",
)

# Обязательный набор ключей нового хоста, упоминаемый в доке ЯВНО (TC-03 / FR-1.4).
MANDATORY_NEW_HOST_KEYS: tuple[str, ...] = (
    "ORCH_PROJECTS_JSON",
    "ORCH_PLANE_WEBHOOK_SECRET",
    "ORCH_GITEA_WEBHOOK_SECRET",
    "ORCH_PLANE_API_TOKEN",
    "ORCH_GITEA_TOKEN",
    "ORCH_TELEGRAM_BOT_TOKEN",
    "ORCH_TELEGRAM_CHAT_ID",
    "WATCHDOG_TG_BOT_TOKEN",
    "WATCHDOG_TG_CHAT_ID",
)

# env-токен в доке: полное имя ключа, не заканчивающееся на «_» (glob-упоминания
# вида ORCH_FOO_* в доке запрещены формой D2 — пишутся полные имена).
_ENV_TOKEN_RE = re.compile(r"\b(?:ORCH|WATCHDOG)_[A-Z0-9_]*[A-Z0-9]\b")

# Секретоподобные значения в копируемых блоках (TC-05, эвристика D8):
# hex-run >= 32 симв. (token_hex) либо чистый alnum-run >= 40 симв. (токены ботов
# и т.п.); плейсхолдеры <...>/$ENV_VAR под эти классы не попадают.
_SECRET_HEX_RE = re.compile(r"\b[0-9a-fA-F]{32,}\b")
_SECRET_ALNUM_RE = re.compile(r"\b[A-Za-z0-9]{40,}\b")


# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def _doc_text() -> str:
    assert LITE_SETUP.is_file(), "docs/deployment/LITE_SETUP.md отсутствует (AC-1)"
    return LITE_SETUP.read_text(encoding="utf-8")


def _env_keys(path: Path) -> set[str]:
    """Множество ключей `KEY=` файла env-канона (комментарии/пустые — мимо)."""
    keys: set[str] = set()
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        keys.add(line.split("=", 1)[0].strip())
    return keys


def _env_values(path: Path) -> dict[str, str]:
    """Словарь `KEY -> value` файла env-канона."""
    values: dict[str, str] = {}
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        values[k.strip()] = v.strip()
    return values


def _fenced_blocks(text: str) -> list[str]:
    """Содержимое fenced code blocks — единственная копируемая зона дока (D2/D8)."""
    return re.findall(r"```[^\n]*\n(.*?)```", text, flags=re.DOTALL)


def _section_bodies() -> dict[str, str]:
    """Тело каждого нормативного раздела (от заголовка до следующего `## `)."""
    text = _doc_text()
    bodies: dict[str, str] = {}
    for i, header in enumerate(SECTIONS):
        start = text.find(header)
        assert start != -1, f"раздел {header!r} отсутствует"
        end = text.find(SECTIONS[i + 1]) if i + 1 < len(SECTIONS) else len(text)
        bodies[header] = text[start:end]
    return bodies


def _compose_services() -> dict:
    return yaml.safe_load(COMPOSE.read_text(encoding="utf-8"))["services"]


# ---------------------------------------------------------------------------
# TC-01: док существует; 13 нормативных разделов D2 — в заданном порядке (AC-1).
# ---------------------------------------------------------------------------
def test_doc_exists_with_all_13_sections_in_order():
    text = _doc_text()
    positions: list[int] = []
    for header in SECTIONS:
        idx = text.find(header)
        assert idx != -1, f"нормативный раздел {header!r} отсутствует (D2/FR-1)"
        positions.append(idx)
    assert positions == sorted(positions), (
        "разделы LITE_SETUP.md идут не в порядке маршрута оператора (D2)"
    )


# ---------------------------------------------------------------------------
# TC-02: обязательные кирпичи + форма «команда + проверка» (AC-1 / NFR-6).
# ---------------------------------------------------------------------------
def test_doc_carries_all_mandatory_bricks():
    text = _doc_text()
    missing = [b for b in BRICKS if b not in text]
    assert not missing, f"обязательные кирпичи отсутствуют в LITE_SETUP.md (FR-6.1): {missing}"


def test_every_normative_section_carries_commands():
    """Каждый исполняемый раздел (§2–§13) несёт минимум одну fenced-команду;
    §1 (рамка/границы) — единственный раздел без команд по построению."""
    bodies = _section_bodies()
    for header in SECTIONS[1:]:
        assert "```" in bodies[header], f"{header}: нет ни одной fenced-команды (NFR-6)"


def test_doc_carries_explicit_check_markers():
    """Маркеры явной проверки результата: не меньше одного на исполняемый раздел."""
    text = _doc_text()
    assert text.count("Проверка") >= 12, (
        "маркеров «Проверка» меньше, чем исполняемых разделов (форма D2: "
        "каждый шаг = команда + явная проверка)"
    )
    assert "PASS" in text and "FAIL" in text


# ---------------------------------------------------------------------------
# TC-02b: key-sync .env.watchdog.example ⇄ блок WATCHDOG_* .env.example (D5).
# ---------------------------------------------------------------------------
def test_watchdog_example_exists():
    assert ENV_WATCHDOG_EXAMPLE.is_file(), ".env.watchdog.example отсутствует (ADR-001 D5)"


def test_watchdog_example_keys_sync_with_env_example_block():
    """Равенство МНОЖЕСТВ ключей (не строк): появление нового WATCHDOG_*-ключа в
    каноне без обновления example (и наоборот) рвёт CI (D5/TR-8)."""
    watchdog_keys = _env_keys(ENV_WATCHDOG_EXAMPLE)
    canon_block = {k for k in _env_keys(ENV_EXAMPLE) if k.startswith("WATCHDOG_")}
    assert canon_block, "блок WATCHDOG_* в .env.example пуст — канон сломан"
    assert watchdog_keys == canon_block, (
        f"key-set .env.watchdog.example разъехался с каноном .env.example: "
        f"лишние={sorted(watchdog_keys - canon_block)}, "
        f"недостающие={sorted(canon_block - watchdog_keys)}"
    )
    stray = {k for k in watchdog_keys if not k.startswith("WATCHDOG_")}
    assert not stray, f"посторонние (не WATCHDOG_*) ключи в .env.watchdog.example: {sorted(stray)}"


def test_watchdog_example_secrets_are_placeholders_only():
    """Зеркало test_env_example_secrets_are_placeholders_only (ORCH-101):
    токены sidecar-бота в гите — только пустые плейсхолдеры (NFR-3)."""
    values = _env_values(ENV_WATCHDOG_EXAMPLE)
    for key in ("WATCHDOG_TG_BOT_TOKEN", "WATCHDOG_TG_CHAT_ID"):
        assert values.get(key, "") == "", (
            f"{key} в .env.watchdog.example обязан быть пустым плейсхолдером, "
            f"получено {values.get(key)!r}"
        )


# ---------------------------------------------------------------------------
# TC-03: согласованность env-канона (AC-1, AC-6 / FR-6.2).
# ---------------------------------------------------------------------------
def test_every_env_token_in_doc_exists_in_env_example():
    """Каждый упомянутый в доке ключ ORCH_*/WATCHDOG_* существует в `.env.example`
    (канон 100% ключей старта, ORCH-101) — анти-опечатка/анти-дрейф."""
    canon = _env_keys(ENV_EXAMPLE)
    mentioned = set(_ENV_TOKEN_RE.findall(_doc_text()))
    assert mentioned, "в LITE_SETUP.md не упомянут ни один env-ключ — док не полон"
    unknown = sorted(mentioned - canon)
    assert not unknown, (
        f"ключи из LITE_SETUP.md отсутствуют в .env.example (опечатка или дрейф "
        f"канона, FR-6.2): {unknown}"
    )


def test_mandatory_new_host_keys_are_explicit():
    text = _doc_text()
    missing = [k for k in MANDATORY_NEW_HOST_KEYS if k not in text]
    assert not missing, f"обязательные ключи нового хоста не упомянуты явно (FR-1.4): {missing}"


# ---------------------------------------------------------------------------
# TC-04: compose-подмножество (AC-2 / D4) — yaml.safe_load, без subprocess.
# ---------------------------------------------------------------------------
def test_compose_services_are_exactly_the_lite_set():
    services = _compose_services()
    assert set(services) == {"orchestrator", "orchestrator-watchdog", "orchestrator-staging"}, (
        "множество сервисов docker-compose.yml разъехалось с Lite-подмножеством (AC-2)"
    )


def test_compose_staging_is_strictly_behind_profile():
    services = _compose_services()
    assert services["orchestrator-staging"].get("profiles") == ["staging"], (
        "orchestrator-staging обязан быть строго за profiles: [staging] (AC-2)"
    )
    default_up = {name for name, svc in services.items() if not svc.get("profiles")}
    assert default_up == {"orchestrator", "orchestrator-watchdog"}, (
        f"дефолтный `docker compose up -d` поднимает {sorted(default_up)}, "
        "а обязан — ровно орк+watchdog (AC-2)"
    )


def test_compose_has_no_plane_or_gitea_services():
    """Анти-появление молча: ни в имени сервиса, ни в image:/container_name:
    нет подстрок plane/gitea (D4)."""
    services = _compose_services()
    for name, svc in services.items():
        blob = " ".join(
            [name, str(svc.get("image", "")), str(svc.get("container_name", ""))]
        ).lower()
        for needle in ("plane", "gitea"):
            assert needle not in blob, (
                f"сервис {name}: в compose появился {needle}-компонент — "
                "Lite-подмножество сломано (AC-2)"
            )


def test_doc_documents_default_up_composition():
    """LITE_SETUP.md фиксирует состав дефолтного запуска и вилку staging (D6)."""
    text = _doc_text()
    assert "orchestrator-watchdog" in text
    assert "orchestrator-staging" in text
    assert "--profile staging" in text  # staging поднимается только явным профилем


# ---------------------------------------------------------------------------
# TC-05: stateless-норматив + секрет-гигиена fenced-блоков (AC-3 / FR-3, NFR-3).
# ---------------------------------------------------------------------------
def test_doc_has_stateless_normative_line():
    low = _doc_text().lower()
    assert "не перенос" in low, (
        "нормативная stateless-строка («данные/задачи/секреты боевого хоста "
        "НЕ переносятся») отсутствует (AC-3)"
    )


def test_doc_prescribes_clean_db_and_fresh_secrets():
    text = _doc_text()
    assert "gen_secrets.py" in text  # секреты — только выпуск НОВЫХ
    bodies = _section_bodies()
    stateless = bodies["## 12. Stateless-проверка"]
    assert "/queue" in stateless, (
        "§12 обязан нести проверку чистоты инстанса через GET /queue (FR-1 п.12)"
    )


def test_fenced_blocks_carry_no_forbidden_literals():
    """Копируемые блоки generic: боевые литералы (центральный список FORBIDDEN
    из tests/test_no_host_hardcodes.py — один источник истины) запрещены."""
    blocks = _fenced_blocks(_doc_text())
    assert blocks, "в LITE_SETUP.md нет ни одного fenced-блока — форма D2 нарушена"
    offenders = [
        f"блок #{i}: {literal!r}"
        for i, block in enumerate(blocks)
        for literal in FORBIDDEN
        if literal in block
    ]
    assert not offenders, (
        "боевые литералы в копируемых код-блоках LITE_SETUP.md (NFR-3):\n"
        + "\n".join(offenders)
    )


def test_fenced_blocks_carry_no_secret_like_values():
    """Эвристика секретоподобных значений (D8): hex >= 32 / чистый alnum >= 40."""
    offenders = []
    for i, block in enumerate(_fenced_blocks(_doc_text())):
        for rx in (_SECRET_HEX_RE, _SECRET_ALNUM_RE):
            m = rx.search(block)
            if m is not None:
                offenders.append(f"блок #{i}: {m.group(0)[:16]}…")
    assert not offenders, (
        "секретоподобные значения в копируемых код-блоках (только плейсхолдеры "
        "<...>/$ENV_VAR, NFR-3):\n" + "\n".join(offenders)
    )


def test_secret_heuristic_is_not_evergreen():
    """Негативный самочек (паттерн ORCH-101 TC-02): эвристика реально ловит
    подсаженный секрет — тест не может молча стать вечнозелёным."""
    planted_hex = "export TOKEN=" + "a1b2c3d4" * 8  # 64 hex
    planted_alnum = "bot" + "A9" * 25  # 50+ alnum
    assert _SECRET_HEX_RE.search(planted_hex) is not None
    assert _SECRET_ALNUM_RE.search(planted_alnum) is not None
    assert _SECRET_HEX_RE.search("curl -fsS http://127.0.0.1:8500/health") is None
    assert _SECRET_ALNUM_RE.search("<ORCHESTRATOR_GIT_URL> $ORCH_PLANE_API_TOKEN") is None


# ---------------------------------------------------------------------------
# TC-06: канон не форкается (AC-6 / FR-4) — ссылки на golden source.
# ---------------------------------------------------------------------------
def test_plane_canon_is_linked_not_forked():
    """Статусы — ссылкой на ONBOARDING.md (создаёт onboard_project.py apply);
    в доке явно — только fail-closed имена Confirm Deploy / STOP."""
    bodies = _section_bodies()
    plane = bodies["## 5. Подключение Plane"]
    assert "ONBOARDING.md" in plane, "раздел Plane не ссылается на golden source статусов"
    assert "Confirm Deploy" in plane and "STOP" in plane, (
        "fail-closed имена статусов не упомянуты явно (FR-4)"
    )


def test_status_count_claim_matches_plane_sync():
    """Сверка импортом (не строковой копией): заявление дока «22 статуса»
    держится фактическим маппингом src/plane_sync.py (нулевой дрейф, AC-6)."""
    from src.plane_sync import _PLANE_NAME_TO_KEY

    assert len(_PLANE_NAME_TO_KEY) == 22, (
        f"в plane_sync {_PLANE_NAME_TO_KEY and len(_PLANE_NAME_TO_KEY)} статусов — "
        "обнови число и раздел §5 LITE_SETUP.md (и ONBOARDING.md §1)"
    )
    assert "Confirm Deploy" in _PLANE_NAME_TO_KEY
    assert "STOP" in _PLANE_NAME_TO_KEY
    assert "22" in _doc_text(), "число статусов в LITE_SETUP.md разъехалось с plane_sync"


def test_env_map_and_smoke_are_linked_to_replication():
    bodies = _section_bodies()
    assert "REPLICATION.md" in bodies["## 4. Конфигурация"], (
        "карта env обязана даваться ссылкой на REPLICATION.md §2 (FR-4)"
    )
    assert "REPLICATION.md" in bodies["## 11. Smoke"], (
        "smoke обязан строиться на REPLICATION.md §4 ссылкой, без форка (FR-5)"
    )


# ---------------------------------------------------------------------------
# TC-07: раздел Gitea соответствует инвариантам платформы (AC-7 / D3).
# ---------------------------------------------------------------------------
def test_gitea_section_fixes_platform_invariants():
    bodies = _section_bodies()
    gitea = bodies["## 6. Подключение Gitea"]
    for event in ("push", "pull_request", "status"):
        assert event in gitea, f"событие webhook {event!r} не зафиксировано в §6"
    assert "ОДИН глобальный" in gitea or "один глобальный" in gitea.lower(), (
        "§6 обязан фиксировать «ОДИН глобальный webhook-секрет на все репо»"
    )


def test_gitea_section_forbids_branch_protection():
    """Исход А-1 (D3): branch protection на main НЕ включать (ADR D10 ORCH-009,
    ломает PR-merge API merge-актора); pre-receive хуки не вводятся."""
    bodies = _section_bodies()
    gitea = bodies["## 6. Подключение Gitea"]
    assert "branch protection" in gitea.lower(), "§6 не несёт норматив про branch protection"
    assert "НЕ включа" in gitea, "§6 обязан нормативно запрещать branch protection на main"
    assert "pre-receive" in gitea.lower(), "§6 обязан фиксировать: pre-receive хуки не вводятся"


# ---------------------------------------------------------------------------
# TC-08: перекрёстная документация (AC-5 / FR-7).
# ---------------------------------------------------------------------------
def test_replication_boundaries_reference_lite_setup():
    text = REPLICATION.read_text(encoding="utf-8")
    assert "LITE_SETUP.md" in text, (
        "REPLICATION.md §1 обязан ссылаться на LITE_SETUP.md (Type A — Lite реализован)"
    )
    assert "ORCH-102" in text, "строка Type A — Lite в REPLICATION.md §1 не отмечена ✅ ORCH-102"


def test_changelog_has_orch_102_entry():
    assert "ORCH-102" in CHANGELOG.read_text(encoding="utf-8")