checkbot/tests/test_ui_link.py

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"<b>h1</b>\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&amp;b=&lt;2&gt;" in body["formatted_body"] and "<2>" not in body["formatted_body"]