orchestrator/tests/test_notify_issue_links.py

"""ORCH-067 — Group D: clickable issue number in ALL alerts (AC-13, AC-12).

Every orchestrator alert that mentions a work_item_id now renders it as a Plane
hyperlink via the shared ``link_for`` / ``plane_issue_link`` helpers, and degrades
fail-safe to the raw (escaped) number when data is missing. This covers the
dedicated notify_* helpers (notify_approve_requested, notify_error) and asserts
the engine/launcher/security_gate/reconciler alert sites are wired to ``link_for``
— the single DB-resolving helper those sites call. Network is isolated:
send_telegram is replaced with a recorder; the DB is a temp SQLite.

Test ids TC-13, TC-14, TC-15 from 04-test-plan.yaml.
"""

import os
import tempfile

os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")

_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_notify_links.db")
os.environ["ORCH_DB_PATH"] = _test_db

from types import SimpleNamespace  # noqa: E402

import pytest  # noqa: E402

import src.db as db_module  # noqa: E402
import src.projects as projects_mod  # noqa: E402
from src.db import init_db, get_db  # noqa: E402
from src import notifications as N  # noqa: E402

_ORCH_PROJECT_ID = "8da6aa25-a60e-44d6-a1e2-d8ae59aa7d6a"


@pytest.fixture(autouse=True)
def setup_db(monkeypatch):
    monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
    if os.path.exists(_test_db):
        os.unlink(_test_db)
    init_db()
    # Pin repo->project resolution so cross-file registry reloads can't strip
    # 'orchestrator' and break the expected issue URL.
    monkeypatch.setattr(
        projects_mod, "get_project_by_repo",
        lambda repo: (SimpleNamespace(plane_project_id=_ORCH_PROJECT_ID)
                      if repo == "orchestrator" else None),
    )
    yield
    if os.path.exists(_test_db):
        os.unlink(_test_db)


def _set(monkeypatch, **kw):
    s = N._get_settings()
    for k, v in kw.items():
        monkeypatch.setattr(s, k, v, raising=False)


def _mk_task(wid="ORCH-067", repo="orchestrator", title="notify links",
             plane_issue_id="iss-1", stage="development"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, title, "
        "plane_issue_id) VALUES (?, ?, ?, ?, ?, ?, ?)",
        ("p1", wid, repo, "feature/ORCH-067-x", stage, title, plane_issue_id),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


def _record_send(monkeypatch):
    calls = []

    def _fake(text, disable_notification=False):
        calls.append({"text": text, "silent": disable_notification})
        return 1

    monkeypatch.setattr(N, "send_telegram", _fake)
    monkeypatch.setattr(N, "update_task_tracker", lambda task_id: None)
    return calls


# --------------------------------------------------------------------------- #
# TC-13 / AC-13 — notify_approve_requested: number clickable, CTA + single ping
# --------------------------------------------------------------------------- #
def test_tc13_approve_requested_number_clickable(monkeypatch):
    _set(monkeypatch, plane_web_url="https://plane.example.org",
         plane_workspace_slug="acme", gitea_public_url="https://git.example.org",
         gitea_owner="orchteam")
    tid = _mk_task(plane_issue_id="iss-1")
    calls = _record_send(monkeypatch)

    N.notify_approve_requested(tid)

    assert len(calls) == 1                          # exactly one notifying ping
    assert calls[0]["silent"] is not True
    text = calls[0]["text"]
    expected = (
        f"https://plane.example.org/acme/projects/{_ORCH_PROJECT_ID}"
        f"/issues/iss-1/"
    )
    assert f'<a href="{expected}">ORCH-067</a>' in text   # clickable number
    assert "Approved" in text                       # call-to-action preserved


# --------------------------------------------------------------------------- #
# TC-14 / AC-13, AC-12 — notify_error: clickable when data present, else raw
# --------------------------------------------------------------------------- #
def test_tc14_notify_error_clickable(monkeypatch):
    _set(monkeypatch, plane_web_url="https://plane.example.org",
         plane_workspace_slug="acme")
    tid = _mk_task(plane_issue_id="iss-1")
    calls = _record_send(monkeypatch)

    N.notify_error(tid, "boom happened")

    assert len(calls) == 1
    text = calls[0]["text"]
    assert ">ORCH-067</a>" in text                  # number is a link
    assert "ERROR" in text and "boom happened" in text


def test_tc14_notify_error_degrades_raw_number(monkeypatch):
    # No usable Plane base -> raw (unlinked) number, alert still sent, no crash.
    _set(monkeypatch, plane_web_url="", plane_api_url="")
    tid = _mk_task(plane_issue_id="iss-1")
    calls = _record_send(monkeypatch)

    N.notify_error(tid, "boom")

    text = calls[0]["text"]
    assert "ORCH-067" in text
    assert "<a href=" not in text


def test_tc14_notify_error_escapes_error_text(monkeypatch):
    # The error string is html-escaped so it can't break the <a>/HTML markup.
    _set(monkeypatch, plane_web_url="https://plane.example.org",
         plane_workspace_slug="acme")
    tid = _mk_task(plane_issue_id="iss-1")
    calls = _record_send(monkeypatch)

    N.notify_error(tid, "<script> & </script>")

    text = calls[0]["text"]
    assert "<script>" not in text
    assert "&lt;script&gt;" in text and "&amp;" in text
    # The clickable number's anchor is still well-formed.
    assert text.count("<a href=") == text.count("</a>")


# --------------------------------------------------------------------------- #
# TC-15 / AC-13 — link_for is the DB-resolving helper the alert sites call
# --------------------------------------------------------------------------- #
def test_tc15_link_for_by_work_item_id(monkeypatch):
    # Sites holding only a work_item_id (launcher deploy-fail, security_gate,
    # reconciler, engine QG-fail) call link_for(wid) -> resolves repo + issue id
    # from the DB and returns a clickable number.
    _set(monkeypatch, plane_web_url="https://plane.example.org",
         plane_workspace_slug="acme")
    _mk_task(wid="ORCH-067", plane_issue_id="iss-1")

    out = N.link_for("ORCH-067")
    expected = (
        f"https://plane.example.org/acme/projects/{_ORCH_PROJECT_ID}"
        f"/issues/iss-1/"
    )
    assert out == f'<a href="{expected}">ORCH-067</a>'


def test_tc15_link_for_by_task_id(monkeypatch):
    # Sites holding a task_id (launcher agent-fail, engine) call link_for(wid, tid).
    _set(monkeypatch, plane_web_url="https://plane.example.org",
         plane_workspace_slug="acme")
    tid = _mk_task(wid="ORCH-067", plane_issue_id="iss-7")

    out = N.link_for("ORCH-067", tid)
    assert ">ORCH-067</a>" in out and "/issues/iss-7/" in out


def test_tc15_link_for_unknown_task_degrades(monkeypatch):
    # No matching DB row -> raw number, never raises.
    _set(monkeypatch, plane_web_url="https://plane.example.org",
         plane_workspace_slug="acme")
    out = N.link_for("ORCH-999")
    assert out == "ORCH-999"
    assert "<a href=" not in out


@pytest.mark.parametrize("module_name", [
    "src.stage_engine",
    "src.agents.launcher",
    "src.security_gate",
    "src.reconciler",
])
def test_tc15_alert_modules_wire_link_for(module_name):
    """The representative alert modules call the shared link_for helper, so their
    work_item_id alerts render a clickable number (not a bare string). Checked at
    source level since some sites import link_for function-locally."""
    import importlib
    import inspect
    mod = importlib.import_module(module_name)
    src = inspect.getsource(mod)
    assert "link_for(" in src, f"{module_name} must use link_for in its alerts"