import json import httpx import pytest from app.alerts import AlertActions, alert_link, base_url_provider from app.chat import Button, MatrixAdapter, MessageRef, TelegramAdapter from app.constants import CHECKMK_DT_ENABLED, FLAG_ON, MATRIX_ALLOWLIST, TELEGRAM_ALLOWLIST, UI_BASE_URL from app.models import HOST_SERVICE_NAME, KIND_PROBLEM, KIND_RECOVERY, KIND_SYSTEM, Event from tests.support import Env BASE = "https://bot.example.de" LINK = f"{BASE}/alerts#alert-7" def event(kind=KIND_PROBLEM, service="disk"): return Event(ts=1000, host="h1", service=service, kind=kind, state=2, output="voll") def flat(buttons): return [(b.label, b.data, b.url) for row in buttons for b in row] def env_with_link(allowlist=True): env = Env(dt_enabled=False) env.store.set(UI_BASE_URL, BASE, "t") if not allowlist: env.store.set(TELEGRAM_ALLOWLIST, "", "t") env.store.set(MATRIX_ALLOWLIST, "", "t") return env def test_alert_link_formats_anchor_and_ignores_empty_base(): assert alert_link(BASE, 7) == LINK assert alert_link(f"{BASE}/", 7) == LINK assert alert_link(BASE) == f"{BASE}/alerts" assert alert_link(None, 7) is None and alert_link("", 7) is None def test_base_url_provider_reads_setting_and_returns_none_when_empty(): env = Env() provider = base_url_provider(env.store) assert provider() is None env.store.set(UI_BASE_URL, BASE, "t") assert provider() == BASE def test_telegram_details_button_in_the_button_row_with_actions(): actions = AlertActions(env_with_link().store) rows = actions.build("telegram", [event()], [7]).buttons assert flat(rows)[-1] == ("Details", "", LINK) assert [label for label, _, _ in flat(rows)][:2] == ["2 h stumm", "Tiefenanalyse"] assert len(rows) == 1 def test_telegram_details_button_joins_the_downtime_row(): env = env_with_link() env.store.set(CHECKMK_DT_ENABLED, FLAG_ON, "t") rows = AlertActions(env.store).build("telegram", [event()], [7]).buttons assert len(rows) == 2 and flat(rows)[-1][0] == "Details" and flat(rows)[-2][0] == "Downtime 2 h" def test_no_downtime_button_for_host_alerts(): env = env_with_link() env.store.set(CHECKMK_DT_ENABLED, FLAG_ON, "t") labels = [label for label, _, _ in flat(AlertActions(env.store).build( "telegram", [event(service=HOST_SERVICE_NAME)], [7]).buttons)] assert "Downtime 2 h" not in labels and "Details" in labels def test_telegram_link_without_allowlist_is_the_only_button(): actions = AlertActions(env_with_link(allowlist=False).store) assert flat(actions.build("telegram", [event()], [7]).buttons) == [("Details", "", LINK)] def test_link_also_on_recovery_bundle_and_system_messages(): actions = AlertActions(env_with_link().store) assert flat(actions.build("telegram", [event(KIND_RECOVERY)], [7]).buttons) == [("Details", "", LINK)] bundle = actions.build("telegram", [event(), event(service="cpu")], [7, 8]) assert flat(bundle.buttons) == [("Details", "", LINK)] system = actions.build("telegram", [event(KIND_SYSTEM)], [7]) assert flat(system.buttons) == [("Details", "", LINK)] def test_matrix_gets_link_line_after_hint(): extras = AlertActions(env_with_link().store).build("matrix", [event()], [7]) assert extras.buttons == () and extras.hint.splitlines() == ["Antwort mit !mute 2h zum Stummschalten", f"Details: {LINK}"] only_link = AlertActions(env_with_link(allowlist=False).store).build("matrix", [event()], [7]) assert only_link.hint == f"Details: {LINK}" def test_no_link_without_base_url(): env = Env() actions = AlertActions(env.store) assert all(url == "" for _, _, url in flat(actions.build("telegram", [event()], [7]).buttons)) assert "Details" not in actions.build("matrix", [event()], [7]).hint def telegram(handler): return TelegramAdapter("123456:SECRET-token_abc", "42", transport=httpx.MockTransport(handler)) def ok(result): return httpx.Response(200, json={"ok": True, "result": result}) async def test_telegram_sends_url_button_without_callback_data(): seen = {} def handler(request): seen["body"] = json.loads(request.content) return ok({"message_id": 1}) buttons = ((Button("2 h stumm", "mute:7:120"), Button("Details", url=LINK)),) await telegram(handler).send("42", "x", buttons) row = seen["body"]["reply_markup"]["inline_keyboard"][0] assert row == [{"text": "2 h stumm", "callback_data": "mute:7:120"}, {"text": "Details", "url": LINK}] async def test_telegram_retries_without_url_button_when_telegram_rejects_the_url(): bodies = [] def handler(request): bodies.append(json.loads(request.content)) if len(bodies) == 1: return httpx.Response(400, json={"ok": False, "description": "Bad Request: BUTTON_URL_INVALID"}) return ok({"message_id": 9}) buttons = ((Button("2 h stumm", "mute:7:120"), Button("Details", url="http://localhost/alerts")),) ref = await telegram(handler).send("42", "x", buttons) assert ref == MessageRef("telegram", "42", "9") assert bodies[1]["reply_markup"]["inline_keyboard"] == [[{"text": "2 h stumm", "callback_data": "mute:7:120"}]] async def test_telegram_retry_drops_markup_when_only_url_button_existed(): bodies = [] def handler(request): bodies.append(json.loads(request.content)) if len(bodies) == 1: return httpx.Response(400, json={"ok": False, "description": "Bad Request: inline keyboard button URL " "'http://x' is invalid: Wrong HTTP URL"}) return ok({"message_id": 1}) await telegram(handler).send("42", "x", ((Button("Details", url="http://x"),),)) assert "reply_markup" not in bodies[1] async def test_telegram_edit_keeps_url_button_and_falls_back_too(): bodies = [] def handler(request): bodies.append(json.loads(request.content)) if len(bodies) == 1: return httpx.Response(400, json={"ok": False, "description": "Bad Request: BUTTON_URL_INVALID"}) return ok(True) await telegram(handler).edit(MessageRef("telegram", "42", "5"), "neu", ((Button("Details", url=LINK),),)) assert len(bodies) == 2 and "reply_markup" not in bodies[1] async def test_telegram_other_errors_are_not_retried(): calls = [] def handler(request): calls.append(1) return httpx.Response(400, json={"ok": False, "description": "Bad Request: chat not found"}) with pytest.raises(Exception, match="chat not found"): await telegram(handler).send("42", "x", ((Button("Details", url=LINK),),)) assert len(calls) == 1 async def test_matrix_link_text_is_escaped_in_html_body(): seen = {} def handler(request): seen["body"] = json.loads(request.content) return httpx.Response(200, json={"event_id": "$1"}) adapter = MatrixAdapter("https://hs.example.de", "@bot:example.de", "tok", "!r:example.de", transport=httpx.MockTransport(handler)) await adapter.send("!r:example.de", f"h1\nDetails: {BASE}/alerts?a=1&b=<2>#alert-7") body = seen["body"] assert f"Details: {BASE}/alerts?a=1&b=<2>#alert-7" in body["body"] assert "a=1&b=<2>" in body["formatted_body"] and "<2>" not in body["formatted_body"]