218 lines
7.4 KiB
Python
218 lines
7.4 KiB
Python
import pytest
|
|
|
|
from app.agent import AgentResult
|
|
from app.checkmk import ServiceState
|
|
from app.commands import AuditEvent, CommandDeps, CommandHandler, IncomingMessage
|
|
from app.ratelimit import FailureLimiter
|
|
|
|
ALLOWED = "111"
|
|
|
|
|
|
def svc(host="web01", service="CPU load", state=2, output="CRIT", downtime=False):
|
|
return ServiceState(host, service, state, True, output, 1, downtime, False)
|
|
|
|
|
|
class FakeReader:
|
|
def __init__(self):
|
|
self.problem_list = [svc(), svc(service="Disk", state=1), svc(service="Mem", state=3, downtime=True)]
|
|
self.services = [svc(), svc(service="Mem", state=0, output="ok")]
|
|
self.fail = False
|
|
|
|
async def problems(self):
|
|
if self.fail:
|
|
raise RuntimeError("http://cmk/secret")
|
|
return self.problem_list
|
|
|
|
async def host_services(self, host):
|
|
if self.fail:
|
|
raise RuntimeError("boom")
|
|
return self.services if host == "web01" else []
|
|
|
|
async def service_status(self, host, service):
|
|
return None
|
|
|
|
async def downtimes(self, host=None):
|
|
return []
|
|
|
|
|
|
class FakeAgent:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def ask(self, question):
|
|
self.calls.append(("ask", question))
|
|
return AgentResult("Antwort <b>fett</b>")
|
|
|
|
async def explain(self, host, service):
|
|
self.calls.append(("explain", host, service))
|
|
return AgentResult("Erklärung")
|
|
|
|
|
|
class Clock:
|
|
now = 0.0
|
|
|
|
def __call__(self):
|
|
return self.now
|
|
|
|
|
|
@pytest.fixture
|
|
def env():
|
|
reader, agent, events, clock = FakeReader(), FakeAgent(), [], Clock()
|
|
handler = CommandHandler(CommandDeps(reader, agent), lambda backend: ALLOWED if backend == "telegram"
|
|
else "@alice:Example.org", events.append,
|
|
limiter=FailureLimiter(5, 60.0, clock))
|
|
return handler, reader, agent, events, clock
|
|
|
|
|
|
def tg(text, sender="111"):
|
|
return IncomingMessage("telegram", sender, "chat1", text)
|
|
|
|
|
|
async def test_plain_chat_ignored(env):
|
|
handler, *_, events, _ = env
|
|
assert await handler.handle(tg("hallo")) is None
|
|
assert events == []
|
|
|
|
|
|
async def test_not_allowed_silent_but_audited(env):
|
|
handler, _, agent, events, _ = env
|
|
assert await handler.handle(tg("/ask x", sender="999")) is None
|
|
assert agent.calls == []
|
|
assert events == [AuditEvent("telegram", "999", "ask", "x", "denied")]
|
|
|
|
|
|
async def test_matrix_allowed_with_domain_case(env):
|
|
handler, *_ = env
|
|
reply = await handler.handle(IncomingMessage("matrix", "@alice:EXAMPLE.org", "!r:x", "!status"))
|
|
assert reply.text.startswith("Probleme: 3")
|
|
|
|
|
|
async def test_rate_limit_ask_only(env):
|
|
handler, _, agent, events, clock = env
|
|
for _ in range(5):
|
|
assert (await handler.handle(tg("/ask hi"))).text.startswith("Antwort")
|
|
blocked = await handler.handle(tg("/ask hi"))
|
|
assert "Zu viele" in blocked.text and len(agent.calls) == 5
|
|
assert events[-1].outcome == "rate_limited"
|
|
assert (await handler.handle(tg("/status"))).text.startswith("Probleme")
|
|
assert "Zu viele" in (await handler.handle(tg("/explain a b"))).text
|
|
clock.now = 61.0
|
|
assert (await handler.handle(tg("/ask hi"))).text.startswith("Antwort")
|
|
|
|
|
|
async def test_rate_limit_per_sender(env):
|
|
handler, *_ = env
|
|
for _ in range(5):
|
|
await handler.handle(tg("/ask hi"))
|
|
other = IncomingMessage("matrix", "@alice:example.org", "!r:x", "!ask hi")
|
|
assert (await handler.handle(other)).text.startswith("Antwort")
|
|
|
|
|
|
async def test_status_counts(env):
|
|
handler, *_ = env
|
|
reply = await handler.handle(tg("/status"))
|
|
assert reply.text == "Probleme: 3 (WARN: 1, CRIT: 1, UNKNOWN: 1), davon in Downtime: 1"
|
|
|
|
|
|
async def test_status_no_problems(env):
|
|
handler, reader, *_ = env
|
|
reader.problem_list = []
|
|
assert "Alles OK" in (await handler.handle(tg("/status"))).text
|
|
|
|
|
|
async def test_problems_limited_to_20(env):
|
|
handler, reader, *_ = env
|
|
reader.problem_list = [svc(service=f"s{i:02}") for i in range(25)]
|
|
reply = await handler.handle(tg("/problems"))
|
|
lines = reply.text.splitlines()
|
|
assert lines[0] == "Probleme (25):" and len(lines) == 1 + 20 + 1
|
|
assert lines[-1] == "... und 5 weitere"
|
|
|
|
|
|
async def test_problems_html_escaped(env):
|
|
handler, reader, *_ = env
|
|
reader.problem_list = [svc(host="<script>", output="a & <b>x</b>")]
|
|
reply = await handler.handle(tg("/problems"))
|
|
assert "<script>" not in reply.html and "<script>" in reply.html
|
|
assert "a & <b>x</b>" in reply.html
|
|
assert "<b>CRIT</b>" in reply.html
|
|
|
|
|
|
async def test_host(env):
|
|
handler, *_ = env
|
|
reply = await handler.handle(tg("/host web01"))
|
|
assert reply.text.splitlines()[0] == "web01: 2 Services, CRIT: 1, OK: 1"
|
|
assert "CRIT CPU load" in reply.text and "Mem" not in reply.text
|
|
assert "nicht gefunden" in (await handler.handle(tg("/host nope"))).text
|
|
assert "Verwendung" in (await handler.handle(tg("/host"))).text
|
|
assert "Ungültig" in (await handler.handle(tg("/host a;b"))).text
|
|
|
|
|
|
async def test_explain(env):
|
|
handler, _, agent, *_ = env
|
|
reply = await handler.handle(tg('/explain web01 "CPU load"'))
|
|
assert reply.text == "Erklärung"
|
|
await handler.handle(tg("/explain web01 CPU load"))
|
|
assert agent.calls == [("explain", "web01", "CPU load")] * 2
|
|
assert "Verwendung" in (await handler.handle(tg("/explain web01"))).text
|
|
assert "Ungültig" in (await handler.handle(tg("/explain bad;host svc"))).text
|
|
|
|
|
|
async def test_ask_escaped_and_limited(env):
|
|
handler, _, agent, *_ = env
|
|
reply = await handler.handle(tg("/ask " + "x" * 800))
|
|
assert len(agent.calls[0][1]) == 500
|
|
assert "<b>fett" in reply.html and "<b>" not in reply.html
|
|
assert "Verwendung" in (await handler.handle(tg("/ask"))).text
|
|
|
|
|
|
async def test_unknown_command(env):
|
|
handler, _, _, events, _ = env
|
|
reply = await handler.handle(tg("/foo"))
|
|
assert "Unbekannter Befehl" in reply.text and events[-1].outcome == "unknown"
|
|
|
|
|
|
async def test_help_uses_prefix_and_registry(env):
|
|
handler, *_ = env
|
|
text = (await handler.handle(tg("/help"))).text
|
|
assert "/status" in text and "/mute" in text
|
|
matrix = await handler.handle(IncomingMessage("matrix", "@alice:example.org", "!r:x", "!help"))
|
|
assert "!status" in matrix.text and "!explain" in matrix.text
|
|
|
|
|
|
async def test_registry_extensible(env):
|
|
handler, *_ = env
|
|
|
|
async def mute(ctx):
|
|
from app.commands import Reply
|
|
return Reply("gemutet")
|
|
|
|
handler.registry["mute"] = mute
|
|
assert (await handler.handle(tg("/mute 2h"))).text == "gemutet"
|
|
assert "mute" in (await handler.handle(tg("/help"))).text
|
|
|
|
|
|
async def test_reader_error_friendly(env):
|
|
handler, reader, _, events, _ = env
|
|
reader.fail = True
|
|
for text in ("/status", "/problems", "/host web01"):
|
|
reply = await handler.handle(tg(text))
|
|
assert "nicht geklappt" in reply.text and "secret" not in reply.text
|
|
assert events[-1].outcome == "error"
|
|
|
|
|
|
async def test_audit_ok_and_args_truncated(env):
|
|
handler, events = env[0], env[3]
|
|
await handler.handle(tg("/ask " + "y" * 600))
|
|
assert events[-1].outcome == "ok" and len(events[-1].args) == 200 and events[-1].sender_id == "111"
|
|
|
|
|
|
async def test_failing_audit_does_not_break(env):
|
|
_, reader, agent, _, _ = env
|
|
|
|
def bad(event):
|
|
raise RuntimeError("x")
|
|
|
|
handler = CommandHandler(CommandDeps(reader, agent), lambda b: ALLOWED, bad)
|
|
assert (await handler.handle(tg("/status"))).text.startswith("Probleme")
|