claude-bot
69a4aaab99
notify_approve_requested now embeds two HTML <a> links into the single notifying approve-gate message: a Gitea branch-view link to 01-brd.md and a Plane issue browser link. Adds ORCH_PLANE_WEB_URL (external Plane web URL, fallback to plane_api_url) with a loopback-guard that omits the Plane link when the resolved base is localhost/empty (no broken localhost URLs in prod). Each link is built independently and omitted on missing data; the message and the "flip to Approved" call to action are always sent as exactly one ping. The shared send_telegram helper is left untouched (min blast radius for the self-hosting prod container). Dynamic labels are html.escaped; parse_mode=HTML preserved. QG registry / stages / approve handler unchanged. Docs updated in-PR: CHANGELOG, .env.example, INFRA env map. Tests: test_notify_approve_links.py, test_analysis_approve_flow_links.py. Refs: ORCH-017 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""ORCH-017: tests for the direct BRD + Plane links in the approve-gate ping.
|
|
|
|
`notify_approve_requested` builds ONE notifying Telegram message that embeds:
|
|
* a Gitea branch-view link to docs/work-items/<WI>/01-brd.md (AC-1)
|
|
* a Plane issue browser link (AC-2)
|
|
|
|
Both links use external base URLs with documented fallbacks (AC-3), degrade
|
|
gracefully when data is missing / the Plane base is loopback (AC-6), keep the
|
|
'flip to Approved' call to action (AC-4), send exactly one notifying message
|
|
(AC-5) and stay HTML-safe (AC-7).
|
|
|
|
Network is isolated: send_telegram is replaced with an in-test recorder, the DB
|
|
is a temp SQLite seeded by a fixture. Mapping to acceptance criteria is in each
|
|
test's docstring (test ids TC-01..TC-08 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_approve_links.db")
|
|
os.environ["ORCH_DB_PATH"] = _test_db
|
|
|
|
from unittest.mock import MagicMock, patch # noqa: E402
|
|
|
|
import pytest # noqa: E402
|
|
|
|
import src.db as db_module # noqa: E402
|
|
from src.db import init_db, get_db # noqa: E402
|
|
from src import notifications as N # noqa: E402
|
|
|
|
# Captured at import time, BEFORE the conftest autouse fixture stubs it to a
|
|
# no-op, so TC-08 can exercise the REAL send_telegram (parse_mode=HTML) end-to-end.
|
|
_ORIG_SEND_TELEGRAM = N.send_telegram
|
|
|
|
# orchestrator repo -> default project registry uuid (src/projects.py).
|
|
_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()
|
|
yield
|
|
if os.path.exists(_test_db):
|
|
os.unlink(_test_db)
|
|
|
|
|
|
def _mk_task(wid="ORCH-017", repo="orchestrator",
|
|
branch="feature/ORCH-017-brd-plane-telegram",
|
|
plane_issue_id="11112222-3333-4444-5555-666677778888",
|
|
title="Links in approve ping", stage="analysis"):
|
|
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, branch, stage, title, plane_issue_id),
|
|
)
|
|
tid = cur.lastrowid
|
|
conn.commit()
|
|
conn.close()
|
|
return tid
|
|
|
|
|
|
def _set(monkeypatch, **kw):
|
|
"""Set settings attrs on the singleton notifications actually reads."""
|
|
s = N._get_settings()
|
|
for k, v in kw.items():
|
|
monkeypatch.setattr(s, k, v, raising=False)
|
|
|
|
|
|
def _record_send(monkeypatch):
|
|
"""Replace send_telegram with a recorder; returns the calls list."""
|
|
calls = []
|
|
|
|
def _fake(text, disable_notification=False):
|
|
calls.append({"text": text, "silent": disable_notification})
|
|
return 1
|
|
|
|
monkeypatch.setattr(N, "send_telegram", _fake)
|
|
# Tracker refresh is irrelevant here and would hit send_telegram too -> stub.
|
|
monkeypatch.setattr(N, "update_task_tracker", lambda task_id: None)
|
|
return calls
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-01 — BRD link (Gitea branch-view), AC-1 / AC-3
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc01_brd_link_present(monkeypatch):
|
|
tid = _mk_task()
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_url="http://localhost:3000", gitea_owner="orchteam")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
|
|
assert len(calls) == 1
|
|
text = calls[0]["text"]
|
|
expected = (
|
|
'https://git.example.org/orchteam/orchestrator/src/branch/'
|
|
'feature/ORCH-017-brd-plane-telegram/docs/work-items/ORCH-017/01-brd.md'
|
|
)
|
|
assert expected in text
|
|
assert f'<a href="{expected}">' in text # clickable, points at 01-brd.md
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-02 — Plane issue link (external web URL + workspace + project + issue id)
|
|
# AC-2 / AC-3
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc02_plane_link_present(monkeypatch):
|
|
tid = _mk_task(plane_issue_id="abcd-issue-uuid")
|
|
_set(monkeypatch, plane_web_url="https://plane.example.org",
|
|
plane_api_url="http://localhost:8091", plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
|
|
text = calls[0]["text"]
|
|
expected = (
|
|
f"https://plane.example.org/acme/projects/{_ORCH_PROJECT_ID}"
|
|
f"/issues/abcd-issue-uuid/"
|
|
)
|
|
assert expected in text
|
|
assert f'<a href="{expected}">' in text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-03 — fallback chain: gitea_public_url -> gitea_url, plane_web_url -> plane_api_url
|
|
# AC-3
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc03_url_fallbacks(monkeypatch):
|
|
tid = _mk_task(plane_issue_id="iss-1")
|
|
_set(monkeypatch,
|
|
gitea_public_url="", gitea_url="https://git-fallback.example.org",
|
|
gitea_owner="orchteam",
|
|
plane_web_url="", plane_api_url="https://plane-fallback.example.org",
|
|
plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
|
|
text = calls[0]["text"]
|
|
# BRD link uses gitea_url fallback.
|
|
assert "https://git-fallback.example.org/orchteam/orchestrator/" in text
|
|
# Plane link uses plane_api_url fallback (non-loopback -> allowed).
|
|
assert (
|
|
f"https://plane-fallback.example.org/acme/projects/{_ORCH_PROJECT_ID}"
|
|
f"/issues/iss-1/"
|
|
) in text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-04 — the 'flip to Approved' call to action is preserved. AC-4
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc04_keeps_approved_call_to_action(monkeypatch):
|
|
tid = _mk_task()
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_owner="orchteam", plane_web_url="https://plane.example.org",
|
|
plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
assert "Approved" in calls[0]["text"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-05 — exactly one notifying (non-silent) message. AC-5
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc05_single_notifying_message(monkeypatch):
|
|
tid = _mk_task()
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_owner="orchteam", plane_web_url="https://plane.example.org",
|
|
plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0]["silent"] is not True # notifying ping, not silent
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-06 — graceful: no branch / no plane_issue_id -> still one message, missing
|
|
# links omitted, no exception. AC-6
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc06_graceful_missing_branch_and_issue(monkeypatch):
|
|
tid = _mk_task(branch=None, plane_issue_id=None)
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_owner="orchteam", plane_web_url="https://plane.example.org",
|
|
plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid) # must not raise
|
|
|
|
assert len(calls) == 1
|
|
text = calls[0]["text"]
|
|
assert "Approved" in text # message still sent
|
|
assert "01-brd.md" not in text # BRD link omitted (no branch)
|
|
assert "/issues/" not in text # Plane link omitted (no issue id)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-07 — Plane base unusable (web url empty + api url empty) -> Plane link
|
|
# dropped, BRD link stays, orchestrator survives. AC-6
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc07_plane_base_empty_drops_plane_link_keeps_brd(monkeypatch):
|
|
tid = _mk_task()
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_owner="orchteam",
|
|
plane_web_url="", plane_api_url="", plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
|
|
text = calls[0]["text"]
|
|
assert "01-brd.md" in text # BRD link survives
|
|
assert "/issues/" not in text # Plane link dropped
|
|
|
|
|
|
def test_tc07b_loopback_plane_base_dropped(monkeypatch):
|
|
"""Loopback fallback (plane_api_url=localhost) must NOT emit a broken link."""
|
|
tid = _mk_task()
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_owner="orchteam",
|
|
plane_web_url="", plane_api_url="http://localhost:8091",
|
|
plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
|
|
text = calls[0]["text"]
|
|
assert "localhost" not in text # no loopback URL leaks into the ping
|
|
assert "/issues/" not in text # Plane link dropped by loopback-guard
|
|
assert "01-brd.md" in text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# TC-08 — HTML safety: parse_mode=HTML preserved + dynamic parts escaped + valid
|
|
# <a> markup. AC-7
|
|
# --------------------------------------------------------------------------- #
|
|
def test_tc08_html_escaped_and_valid_markup(monkeypatch):
|
|
# work_item_id with an ampersand exercises html.escape on the dynamic label.
|
|
tid = _mk_task(wid="ORCH&17")
|
|
_set(monkeypatch, gitea_public_url="https://git.example.org",
|
|
gitea_owner="orchteam", plane_web_url="https://plane.example.org",
|
|
plane_workspace_slug="acme")
|
|
calls = _record_send(monkeypatch)
|
|
|
|
N.notify_approve_requested(tid)
|
|
text = calls[0]["text"]
|
|
|
|
# Dynamic work_item_id escaped in the header (no raw '&' before a word).
|
|
assert "ORCH&17" in text
|
|
# Well-formed anchor markup: equal number of opening/closing tags.
|
|
assert text.count("<a href=") == text.count("</a>")
|
|
assert text.count("<a href=") >= 1
|
|
|
|
|
|
def test_tc08b_send_telegram_keeps_parse_mode_html(monkeypatch):
|
|
"""End-to-end through the REAL send_telegram: payload still parse_mode=HTML."""
|
|
# Restore the genuine send_telegram (conftest stubbed it to a no-op).
|
|
monkeypatch.setattr(N, "send_telegram", _ORIG_SEND_TELEGRAM)
|
|
monkeypatch.setattr(N, "update_task_tracker", lambda task_id: None)
|
|
_set(monkeypatch, telegram_bot_token="T", telegram_chat_id="C",
|
|
gitea_public_url="https://git.example.org", gitea_owner="orchteam",
|
|
plane_web_url="https://plane.example.org", plane_workspace_slug="acme")
|
|
tid = _mk_task()
|
|
|
|
with patch("src.notifications.httpx") as hx:
|
|
resp = MagicMock()
|
|
resp.json.return_value = {"ok": True, "result": {"message_id": 9}}
|
|
hx.post.return_value = resp
|
|
N.notify_approve_requested(tid)
|
|
|
|
assert hx.post.call_count == 1
|
|
payload = hx.post.call_args.kwargs["json"]
|
|
assert payload["parse_mode"] == "HTML"
|
|
assert payload["disable_notification"] is False # notifying
|
|
assert "<a href=" in payload["text"]
|