261 lines
12 KiB
Python
261 lines
12 KiB
Python
import re
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
|
|
from app.agent import TOOL_SCHEMAS
|
|
from app.checkmk import CheckmkClient
|
|
from app.commands import IncomingCallback
|
|
from app.constants import CHECKMK_DT_ENABLED, CHECKMK_DT_MAX_HOURS, FLAG_OFF
|
|
from tests.support import CHAT, TG_OTHER, TG_USER, Env, mx, tg
|
|
|
|
|
|
@pytest.fixture
|
|
def env():
|
|
return Env()
|
|
|
|
|
|
def active(env):
|
|
return env.mutes.active()
|
|
|
|
|
|
async def test_mute_explicit_creates_chat_mute_with_audit(env):
|
|
reply = await env.handler.handle(tg("/mute web* Disk* 2h Wartung am Samstag"))
|
|
mute = active(env)[0]
|
|
assert (mute.host_glob, mute.service_glob, mute.source) == ("web*", "Disk*", "chat")
|
|
assert mute.created_by == f"telegram:{TG_USER}" and mute.reason == "Wartung am Samstag"
|
|
assert "web*" in reply.text and "2 h" in reply.text
|
|
assert "mute.create" in [a for _, a in env.audit_actions()]
|
|
|
|
|
|
async def test_mute_host_only_defaults_to_all_services(env):
|
|
await env.handler.handle(tg("/mute web01 30m"))
|
|
assert active(env)[0].service_glob == "*"
|
|
|
|
|
|
async def test_mute_as_reply_resolves_alert_via_deliveries(env):
|
|
env.alert("web01", "Filesystem [/var]", message_id="55")
|
|
reply = await env.handler.handle(tg("/mute 2h", reply_to="55"))
|
|
mute = active(env)[0]
|
|
assert mute.host_glob == "web01" and "Filesystem" in mute.service_glob
|
|
from app.globs import glob_matches
|
|
assert glob_matches(mute.service_glob, "Filesystem [/var]") and not glob_matches(mute.service_glob, "Filesystem v")
|
|
assert "Stumm" in reply.text
|
|
|
|
|
|
async def test_mute_reply_matrix_with_bang_prefix(env):
|
|
env.alert("db01", "Mem", backend="matrix", target="!room:example.org", message_id="$ev1")
|
|
await env.handler.handle(mx("!mute 1h", reply_to="$ev1"))
|
|
assert active(env)[0].host_glob == "db01"
|
|
|
|
|
|
async def test_mute_reply_unknown_or_missing_reply(env):
|
|
assert "keinem bekannten Alarm" in (await env.handler.handle(tg("/mute 2h", reply_to="999"))).text
|
|
assert "Host fehlt" in (await env.handler.handle(tg("/mute 2h"))).text
|
|
assert active(env) == []
|
|
|
|
|
|
async def test_mute_reply_with_several_services_asks_for_explicit_target(env):
|
|
env.alert("web01", "CPU", message_id="70")
|
|
env.alert("web01", "Mem", message_id="70")
|
|
reply = await env.handler.handle(tg("/mute 2h", reply_to="70"))
|
|
assert "mehrere Services" in reply.text and "web01" in reply.text and active(env) == []
|
|
|
|
|
|
async def test_mute_errors_are_friendly(env):
|
|
for text in ("/mute", "/mute web01", "/mute web01 8d", "/mute web01 0m", "/mute web01 2h " + "x" * 300):
|
|
reply = await env.handler.handle(tg(text))
|
|
assert reply.text and "Traceback" not in reply.text
|
|
assert active(env) == []
|
|
|
|
|
|
async def test_mute_denied_for_non_allowlist(env):
|
|
assert await env.handler.handle(tg("/mute web01 2h", sender=TG_OTHER)) is None
|
|
assert active(env) == [] and env.events[-1].outcome == "denied"
|
|
|
|
|
|
async def test_downtime_requires_confirmation_before_anything_happens(env):
|
|
reply = await env.handler.handle(tg('/downtime web01 "CPU load" 2h Kernel-Update'))
|
|
assert env.client.calls == []
|
|
assert "web01" in reply.text and "CPU load" in reply.text and "2 h" in reply.text and "Kernel-Update" in reply.text
|
|
labels = [b.label for row in reply.buttons for b in row]
|
|
assert labels == ["Bestätigen", "Abbrechen"]
|
|
data = reply.buttons[0][0].data
|
|
assert data.startswith("dty:") and len(data.encode()) <= 64
|
|
assert env.downtime_audit() == ["downtime.request"]
|
|
|
|
|
|
async def test_downtime_confirm_via_callback_by_same_sender(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", reply.buttons[0][0].data))
|
|
assert result.clear_buttons and "Downtime gesetzt" in result.reply.text
|
|
host, service, start, end, comment = env.client.calls[0]
|
|
assert (host, service, end - start) == ("web01", "disk", timedelta(hours=2))
|
|
assert env.client.closed
|
|
assert env.downtime_audit() == ["downtime.request", "downtime.confirm", "downtime.created"]
|
|
|
|
|
|
async def test_downtime_second_confirm_does_nothing(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
data = reply.buttons[0][0].data
|
|
cb = IncomingCallback("telegram", TG_USER, CHAT, "9", data)
|
|
await env.callbacks.handle(cb)
|
|
again = await env.callbacks.handle(cb)
|
|
assert "bereits verwendet" in again.reply.text and len(env.client.calls) == 1
|
|
|
|
|
|
async def test_downtime_cancel_via_callback(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
cancel = reply.buttons[0][1].data
|
|
assert cancel.startswith("dtn:")
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", cancel))
|
|
assert "Abgebrochen" in result.reply.text and env.client.calls == []
|
|
assert env.downtime_audit() == ["downtime.request", "downtime.cancel"]
|
|
|
|
|
|
async def test_downtime_other_allowlisted_user_cannot_confirm(env):
|
|
env.store.set("telegram.allowlist", f"{TG_USER},{TG_OTHER}", "t")
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_OTHER, CHAT, "9", reply.buttons[0][0].data))
|
|
assert "Nur der Absender" in result.reply.text and env.client.calls == [] and not result.clear_buttons
|
|
assert env.downtime_audit()[-1] == "downtime.denied"
|
|
ok = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", reply.buttons[0][0].data))
|
|
assert "Downtime gesetzt" in ok.reply.text
|
|
|
|
|
|
async def test_downtime_non_allowlist_callback_denied_without_effect(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", "999", CHAT, "9", reply.buttons[0][0].data))
|
|
assert result.answer == "Nicht erlaubt." and result.reply is None and env.client.calls == []
|
|
assert env.events[-1].outcome == "denied"
|
|
|
|
|
|
async def test_downtime_ttl_expiry_is_audited(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
env.clock.now += 121
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", reply.buttons[0][0].data))
|
|
assert "abgelaufen" in result.reply.text and env.client.calls == []
|
|
assert env.downtime_audit() == ["downtime.request", "downtime.expired"]
|
|
|
|
|
|
async def test_downtime_matrix_confirm_and_cancel_commands(env):
|
|
reply = await env.handler.handle(mx("!downtime web01 disk 1h"))
|
|
assert reply.buttons == () and "!confirm" in reply.text and "!cancel" in reply.text
|
|
token = re.search(r"!confirm (\S+),", reply.text).group(1)
|
|
assert "Nur der Absender" not in (await env.handler.handle(mx(f"!confirm {token}"))).text
|
|
assert len(env.client.calls) == 1
|
|
reply2 = await env.handler.handle(mx("!downtime web01 disk 1h"))
|
|
token2 = re.search(r"!confirm (\S+),", reply2.text).group(1)
|
|
assert "Abgebrochen" in (await env.handler.handle(mx(f"!cancel {token2}"))).text
|
|
assert len(env.client.calls) == 1
|
|
|
|
|
|
async def test_downtime_confirm_command_wrong_sender_and_garbage(env):
|
|
env.store.set("telegram.allowlist", f"{TG_USER},{TG_OTHER}", "t")
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
token = reply.buttons[0][0].data.split(":")[1]
|
|
assert "Nur der Absender" in (await env.handler.handle(tg(f"/confirm {token}", sender=TG_OTHER))).text
|
|
assert "unbekannt" in (await env.handler.handle(tg("/confirm nope"))).text
|
|
assert "Verwendung" in (await env.handler.handle(tg("/confirm"))).text
|
|
assert env.client.calls == []
|
|
|
|
|
|
async def test_downtime_disabled_feature(env):
|
|
env.store.set(CHECKMK_DT_ENABLED, FLAG_OFF, "t")
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
assert "deaktiviert" in reply.text and reply.buttons == ()
|
|
assert env.downtime_audit() == ["downtime.denied"] and env.client.calls == []
|
|
|
|
|
|
async def test_downtime_disabled_after_request_blocks_confirm(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
env.store.set(CHECKMK_DT_ENABLED, FLAG_OFF, "t")
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", reply.buttons[0][0].data))
|
|
assert "deaktiviert" in result.reply.text and env.client.calls == []
|
|
|
|
|
|
async def test_downtime_max_duration_default_and_setting(env):
|
|
assert "zu lang" in (await env.handler.handle(tg("/downtime web01 disk 25h"))).text
|
|
assert (await env.handler.handle(tg("/downtime web01 disk 24h"))).buttons
|
|
env.store.set(CHECKMK_DT_MAX_HOURS, "4", "t")
|
|
assert "maximal 4 h" in (await env.handler.handle(tg("/downtime web01 disk 5h"))).text
|
|
env.store.set(CHECKMK_DT_MAX_HOURS, "100000", "t")
|
|
assert env.downtime.max_duration() == timedelta(hours=168)
|
|
|
|
|
|
async def test_downtime_non_allowlist_gets_nothing(env):
|
|
assert await env.handler.handle(tg("/downtime web01 disk 2h", sender=TG_OTHER)) is None
|
|
assert env.pending._items == {} and env.client.calls == []
|
|
|
|
|
|
async def test_downtime_service_itself_checks_allowlist(env):
|
|
result = await env.downtime.request("telegram", TG_OTHER, "web01", "disk", timedelta(hours=1), "x")
|
|
assert not result.ok and env.pending._items == {}
|
|
assert "nicht erlaubt" in env.db.query("SELECT new_value FROM audit WHERE action='downtime.denied'")[0][0]
|
|
|
|
|
|
async def test_downtime_checkmk_error_is_reported_and_audited(env):
|
|
env.client.fail = "HTTP 403: nope"
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h"))
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", reply.buttons[0][0].data))
|
|
assert "nicht angenommen" in result.reply.text and "403" in result.reply.text
|
|
assert env.downtime_audit()[-1] == "downtime.error" and env.client.closed
|
|
|
|
|
|
async def test_downtime_reply_mode_uses_exact_names(env):
|
|
env.alert("web01", "Filesystem [/var]", message_id="80")
|
|
reply = await env.handler.handle(tg("/downtime 2h Grund", reply_to="80"))
|
|
token = reply.buttons[0][0].data
|
|
await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", token))
|
|
assert env.client.calls[0][:2] == ("web01", "Filesystem [/var]")
|
|
|
|
|
|
async def test_downtime_comment_control_chars_cleaned_and_html_escaped(env):
|
|
reply = await env.handler.handle(tg("/downtime web01 disk 2h <b>x</b>"))
|
|
assert "<b>" in reply.html and "<b>x</b>" not in reply.html
|
|
|
|
|
|
async def test_downtime_invalid_host_rejected(env):
|
|
reply = await env.handler.handle(tg("/downtime we;b disk 2h"))
|
|
assert "host" in reply.text.lower() and env.pending._items == {}
|
|
|
|
|
|
async def test_button_mute_and_deep_and_downtime(env):
|
|
alert_id = env.alert("web01", "CPU load")
|
|
mute = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "55", f"mute:{alert_id}:120"))
|
|
assert "Stumm" in mute.reply.text and active(env)[0].created_by == f"telegram:{TG_USER}"
|
|
deep = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "55", f"deep:{alert_id}"))
|
|
assert deep.reply.text == "Erklärung" and env.agent.calls == [("explain", "web01", "CPU load")]
|
|
dt = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "55", f"dt:{alert_id}:120"))
|
|
assert dt.reply.buttons and env.client.calls == []
|
|
|
|
|
|
async def test_button_invalid_data(env):
|
|
alert_id = env.alert()
|
|
for data in ("", "mute", "mute:abc:120", "mute:99999:120", "unknown:1", "x" * 80, f"mute:{alert_id}:abc",
|
|
f"mute:{alert_id}:0", f"mute:{alert_id}:99999999", f"dty:{alert_id}:1"):
|
|
result = await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "55", data))
|
|
assert result.reply is None or "Stumm" not in result.reply.text
|
|
assert active(env) == []
|
|
|
|
|
|
async def test_button_deep_rate_limited(env):
|
|
alert_id = env.alert()
|
|
answers = [(await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "55", f"deep:{alert_id}"))).answer
|
|
for _ in range(7)]
|
|
assert answers[-1] == "Zu viele Anfragen. Bitte kurz warten."
|
|
|
|
|
|
def test_agent_has_no_access_to_downtime_or_mutes():
|
|
names = {schema["function"]["name"] for schema in TOOL_SCHEMAS}
|
|
assert not any("downtime_set" in n or "create" in n or "mute" in n for n in names)
|
|
assert all(n.startswith("get_") for n in names)
|
|
assert not any(hasattr(CheckmkClient, attr) for attr in ("create_service_downtime", "create_downtime"))
|
|
|
|
|
|
def test_agent_package_does_not_import_downtime_modules():
|
|
import pathlib
|
|
for path in pathlib.Path("app/agent").glob("*.py"):
|
|
text = path.read_text()
|
|
assert "downtime import" not in text and "from ..downtime" not in text and "mutes" not in text, path
|