117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
import json
|
|
|
|
import httpx
|
|
|
|
from app.alerts import AlertActions, AlertAnalyzer, AlertPipeline, AlertRepository, SlidingWindowLimiter
|
|
from app.chat import Button, MessageRef, TelegramAdapter
|
|
from app.constants import CHECKMK_DT_ENABLED, FLAG_ON, MATRIX_ALLOWLIST, TELEGRAM_ALLOWLIST
|
|
from app.models import KIND_PROBLEM, KIND_RECOVERY, Event
|
|
from tests.support import Env
|
|
from tests.test_pipeline import Clock, FakeOllama
|
|
|
|
BUNDLE = 60
|
|
|
|
|
|
def problem(service="disk", kind=KIND_PROBLEM):
|
|
return Event(ts=1000, host="h1", service=service, kind=kind, state=2, output="voll")
|
|
|
|
|
|
class ButtonAdapter:
|
|
def __init__(self, name):
|
|
self.name, self.default_target = name, "t1"
|
|
self.sent, self.edits = [], []
|
|
|
|
async def send(self, target, text, buttons=()):
|
|
self.sent.append((text, buttons))
|
|
return MessageRef(self.name, target, "1")
|
|
|
|
async def edit(self, ref, text, buttons=()):
|
|
self.edits.append((text, buttons))
|
|
|
|
async def aclose(self):
|
|
pass
|
|
|
|
|
|
def make_pipeline(env, adapters):
|
|
async def provider():
|
|
return adapters
|
|
|
|
analyzer = AlertAnalyzer(env.store, lambda: FakeOllama(), lambda h, s, n: [], SlidingWindowLimiter(Clock()))
|
|
return AlertPipeline(AlertRepository(env.db), provider, analyzer, lambda: BUNDLE, Clock(),
|
|
actions=AlertActions(env.store))
|
|
|
|
|
|
async def run(pipeline, events):
|
|
await pipeline.handle(events)
|
|
await pipeline.drain()
|
|
await pipeline.wait_idle()
|
|
|
|
|
|
def test_telegram_buttons_for_single_problem_and_downtime_only_when_enabled():
|
|
env = Env(dt_enabled=False)
|
|
actions = AlertActions(env.store)
|
|
extras = actions.build("telegram", [problem()], [7])
|
|
flat = [(b.label, b.data) for row in extras.buttons for b in row]
|
|
assert flat == [("2 h stumm", "mute:7:120"), ("Tiefenanalyse", "deep:7")]
|
|
env.store.set(CHECKMK_DT_ENABLED, FLAG_ON, "t")
|
|
flat = [(b.label, b.data) for row in actions.build("telegram", [problem()], [7]).buttons for b in row]
|
|
assert ("Downtime 2 h", "dt:7:120") in flat
|
|
assert all(len(data.encode()) <= 64 for _, data in flat)
|
|
|
|
|
|
def test_no_extras_for_recovery_bundle_or_empty_allowlist():
|
|
env = Env()
|
|
actions = AlertActions(env.store)
|
|
assert actions.build("telegram", [problem(kind=KIND_RECOVERY)], [1]).buttons == ()
|
|
assert actions.build("telegram", [problem("a"), problem("b")], [1, 2]).buttons == ()
|
|
env.store.set(TELEGRAM_ALLOWLIST, "", "t")
|
|
env.store.set(MATRIX_ALLOWLIST, "", "t")
|
|
assert actions.build("telegram", [problem()], [1]).buttons == ()
|
|
assert actions.build("matrix", [problem()], [1]).hint == ""
|
|
|
|
|
|
def test_matrix_gets_hint_not_buttons():
|
|
env = Env()
|
|
extras = AlertActions(env.store).build("matrix", [problem()], [1])
|
|
assert extras.buttons == () and "!mute 2h" in extras.hint
|
|
|
|
|
|
async def test_pipeline_attaches_buttons_hint_and_keeps_them_on_edit():
|
|
env = Env()
|
|
telegram, matrix = ButtonAdapter("telegram"), ButtonAdapter("matrix")
|
|
pipeline = make_pipeline(env, [telegram, matrix])
|
|
await run(pipeline, [problem()])
|
|
text, buttons = telegram.sent[0]
|
|
assert buttons and "!mute" not in text
|
|
alert_id = env.db.query("SELECT id FROM alerts")[0]["id"]
|
|
assert buttons[0][0].data == f"mute:{alert_id}:120"
|
|
m_text, m_buttons = matrix.sent[0]
|
|
assert m_buttons == () and m_text.endswith("Antwort mit !mute 2h zum Stummschalten")
|
|
assert telegram.edits and telegram.edits[0][1] == buttons
|
|
assert matrix.edits[0][0].endswith("Antwort mit !mute 2h zum Stummschalten")
|
|
|
|
|
|
async def test_pipeline_without_buttons_on_bundles():
|
|
env = Env()
|
|
telegram = ButtonAdapter("telegram")
|
|
await run(make_pipeline(env, [telegram]), [problem("a"), problem("b")])
|
|
assert telegram.sent[0][1] == ()
|
|
|
|
|
|
async def test_telegram_adapter_reply_markup_on_send_and_edit():
|
|
seen = []
|
|
|
|
def handler(request):
|
|
seen.append((request.url.path.rsplit("/", 1)[1], json.loads(request.content)))
|
|
return httpx.Response(200, json={"ok": True, "result": {"message_id": 5}})
|
|
|
|
adapter = TelegramAdapter("1:abc", "42", transport=httpx.MockTransport(handler))
|
|
buttons = ((Button("Ja", "dty:abc"), Button("Nein", "dtn:abc")),)
|
|
await adapter.send("42", "x", buttons=buttons)
|
|
await adapter.edit(MessageRef("telegram", "42", "5"), "y", buttons=buttons)
|
|
await adapter.send("42", "plain")
|
|
for method, body in seen[:2]:
|
|
assert body["reply_markup"] == {"inline_keyboard": [
|
|
[{"text": "Ja", "callback_data": "dty:abc"}, {"text": "Nein", "callback_data": "dtn:abc"}]]}
|
|
assert "reply_markup" not in seen[2][1]
|