orchestrator/tests/test_orch119_business_request.py

"""ORCH-119: source-backed ``00-business-request.md`` (fix the hardcoded ``TBD``).

The Description section of ``00-business-request.md`` must carry the ACTUAL Plane-issue
``description`` instead of the historic hardcoded literal ``TBD``. Because the
self-hosting ``orchestrator`` repo cuts its branch lazily at analyst-job claim (the
deferred path B, ORCH-088), the description must be DURABLE-persisted on the ``tasks``
row at creation time (mirror of ``tasks.title``) so it survives the gap between task
creation and claim.

These tests are network-free: the Gitea ``contents`` POST (httpx) and the deferred
branch-cut coroutines are mocked; ``create_task_atomic`` runs against a real isolated
SQLite DB. Mapping to ``04-test-plan.yaml``:

  * TC-01 — MANDATORY regression: render contains the real description, not ``TBD``.
  * TC-02 — fallback: empty/whitespace/None -> explicit safe marker, never raises.
  * TC-03 — deferred path B: description persisted + rendered at materialisation.
  * TC-04 — direct path A: ``_create_initial_docs`` writes the real description.
  * TC-05 — schema backward-compat: ``_ensure_column`` is additive + idempotent.
  * TC-06 — data integrity: multi-line markdown preserved; Gitea 422 -> no-op.
  * TC-07 — anti-regression: gates / STAGE_TRANSITIONS / QG_CHECKS unchanged.
"""

import asyncio
import base64
import os
import tempfile
from types import SimpleNamespace

import pytest

_test_db = os.path.join(tempfile.gettempdir(), "test_orch119_business_request.db")
os.environ["ORCH_DB_PATH"] = _test_db
os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")

import src.db as _db  # noqa: E402
from src.db import init_db, get_db, create_task_atomic  # noqa: E402
from src.webhooks import plane  # noqa: E402


@pytest.fixture(autouse=True)
def fresh_db(monkeypatch):
    monkeypatch.setattr(_db.settings, "db_path", _test_db)
    if os.path.exists(_test_db):
        os.unlink(_test_db)
    init_db()
    yield


# ---------------------------------------------------------------------------
# TC-01 (MANDATORY regression): the rendered body carries the real request
#        text, NOT the historic hardcoded ``TBD``. Red before the fix
#        (``_render_business_request`` did not exist / body was ``TBD``).
# ---------------------------------------------------------------------------
def test_tc01_render_contains_real_description():
    desc = "Users need source-backed business requests captured from the Plane issue."
    out = plane._render_business_request("ORCH-119", "Source-backed request", desc)

    # The real request text reaches the artifact body...
    assert desc in out
    # ...the header / Work Item ID are still present (unchanged contract)...
    assert "# Business Request: Source-backed request" in out
    assert "Work Item ID: ORCH-119" in out
    # ...and the Description body is NOT the bare ``TBD`` bug.
    body = out.split("## Description", 1)[1]
    assert "TBD" not in body


# ---------------------------------------------------------------------------
# TC-02 (fallback): empty / whitespace / None description -> explicit safe
#        marker (never the bare ``TBD`` bug), and the renderer never raises.
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("desc", ["", "   ", "\n\t  \n", None])
def test_tc02_fallback_marker_no_raise(desc):
    out = plane._render_business_request("ORCH-119", "Name", desc)
    assert "описание отсутствует в источнике" in out
    body = out.split("## Description", 1)[1]
    assert "TBD" not in body


# ---------------------------------------------------------------------------
# TC-03 (deferred path B / self-hosting): description is persisted DURABLE on
#        the tasks row at creation and rendered into the artifact when the
#        deferred branch is materialised at claim (launcher).
# ---------------------------------------------------------------------------
def test_tc03_deferred_path_persists_and_renders(monkeypatch):
    desc = "A durable source-backed description for the deferred path B (claim-time cut)."
    row, created = create_task_atomic(
        "plane-b", "ORCH-201", "orchestrator",
        "feature/ORCH-201-x", "analysis", "Title B", desc,
    )
    assert created

    # Durable: description survives in the tasks row (readable without the network).
    got = get_db().execute(
        "SELECT description FROM tasks WHERE id=?", (row["id"],)
    ).fetchone()
    assert got[0] == desc

    captured = {}

    async def _branch_spy(repo, branch):
        captured["branch_cut"] = (repo, branch)

    async def _docs_spy(repo, branch, work_item_id, name, description=None):
        captured["description"] = description

    # _materialize_deferred_branch imports these names from webhooks.plane at call
    # time, so patching the source module attributes intercepts them.
    monkeypatch.setattr(plane, "_create_gitea_branch", _branch_spy)
    monkeypatch.setattr(plane, "_create_initial_docs", _docs_spy)

    from src.agents.launcher import AgentLauncher
    AgentLauncher()._materialize_deferred_branch(
        "orchestrator", "feature/ORCH-201-x", "ORCH-201", "Title B", desc
    )

    assert captured.get("branch_cut") == ("orchestrator", "feature/ORCH-201-x")
    assert captured.get("description") == desc


# ---------------------------------------------------------------------------
# TC-04 (direct path A / serial gate not applicable): _create_initial_docs
#        writes the real description into the Gitea contents body.
# ---------------------------------------------------------------------------
def test_tc04_direct_path_renders_description(monkeypatch):
    desc = "Direct path A description that must reach the artifact body verbatim."
    captured = {}

    class _Resp:
        status_code = 201

        def raise_for_status(self):
            pass

    class _Client:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def post(self, url, json=None, headers=None, timeout=None):
            captured["content"] = base64.b64decode(json["content"]).decode()
            return _Resp()

    monkeypatch.setattr(
        plane, "httpx", SimpleNamespace(AsyncClient=lambda *a, **k: _Client())
    )

    asyncio.run(
        plane._create_initial_docs("orchestrator", "feature/x", "ORCH-119", "Name", desc)
    )

    assert desc in captured["content"]
    assert "## Description\n\nTBD\n" not in captured["content"]


# ---------------------------------------------------------------------------
# TC-05 (schema backward-compat): _ensure_column adds tasks.description on a
#        pre-existing table without it; idempotent on re-run; creation works.
# ---------------------------------------------------------------------------
def test_tc05_schema_backward_compat(monkeypatch, tmp_path):
    db_path = str(tmp_path / "bc.db")
    monkeypatch.setattr(_db.settings, "db_path", db_path)

    # Simulate an OLD tasks table WITHOUT the description column.
    conn = _db.get_db()
    conn.executescript(
        """
        CREATE TABLE tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            plane_id TEXT, work_item_id TEXT, repo TEXT NOT NULL,
            branch TEXT, stage TEXT DEFAULT 'created', plane_issue_id TEXT, title TEXT
        );
        """
    )
    conn.commit()
    conn.close()

    # init_db must add the column idempotently and never fail.
    init_db()
    cols = [r[1] for r in _db.get_db().execute("PRAGMA table_info(tasks)").fetchall()]
    assert "description" in cols

    init_db()  # re-run: _ensure_column is a no-op (idempotent)

    # Task creation works and persists the description.
    row, created = create_task_atomic(
        "p1", "ORCH-1", "orchestrator", "b", "analysis", "T", "a real description"
    )
    assert created
    got = get_db().execute(
        "SELECT description FROM tasks WHERE id=?", (row["id"],)
    ).fetchone()
    assert got[0] == "a real description"


# ---------------------------------------------------------------------------
# TC-06 (data integrity): multi-line markdown with special chars is preserved
#        verbatim (no truncation/escaping); Gitea 422 (file exists) -> no-op
#        (single create attempt, body NOT overwritten).
# ---------------------------------------------------------------------------
def test_tc06_multiline_preserved_and_idempotent_422(monkeypatch):
    desc = (
        "# Heading\n\n- bullet with `inline code`\n"
        "- second *italic* and __bold__\n\n"
        "Paragraph with special chars: <>&\"' and a trailing word."
    )
    out = plane._render_business_request("ORCH-119", "N", desc)
    assert desc in out  # preserved verbatim, no truncation/escaping

    calls = []

    class _Resp:
        status_code = 422  # Gitea: file already exists

        def raise_for_status(self):
            raise AssertionError("422 must be a no-op, never raised")

    class _Client:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def post(self, url, json=None, headers=None, timeout=None):
            calls.append(json)
            return _Resp()

    monkeypatch.setattr(
        plane, "httpx", SimpleNamespace(AsyncClient=lambda *a, **k: _Client())
    )

    asyncio.run(
        plane._create_initial_docs("orchestrator", "b", "ORCH-119", "N", desc)
    )
    # Exactly one create attempt; no follow-up PUT/overwrite of the existing body.
    assert len(calls) == 1


# ---------------------------------------------------------------------------
# TC-07 (anti-regression): the pipeline machinery is untouched —
#        00-business-request.md stays informational (not gate-parsed).
# ---------------------------------------------------------------------------
def test_tc07_gates_unchanged():
    from src.stages import STAGE_TRANSITIONS
    from src.qg.checks import QG_CHECKS

    # Stage graph intact.
    for st in ("created", "analysis", "architecture", "development",
               "review", "testing", "deploy-staging", "deploy", "done"):
        assert st in STAGE_TRANSITIONS

    # The named checks still exist with their canonical names.
    for chk in ("check_analysis_complete", "check_architecture_done",
                "check_ci_green", "check_tests_passed"):
        assert chk in QG_CHECKS

    # 00-business-request.md is informational: no check is keyed on it.
    assert not any("business" in k.lower() for k in QG_CHECKS)