patrickandClaude Sonnet 5.5
cdcf7f673a
Telegram: Plain-Text-Fallback bei Parse-Fehler, tag-sichere Kürzung auf 4096. Matrix: bytebewusste Kürzung. Downtime: lesender Abgleich nach dem POST, generische Fehlermeldung im Chat. Agent: Platzhalter für nicht ausgeführte Tool-Calls, Text im letzten Schritt erzwingen, Zeitreserve, Kompaktierung, Limit pro Antwort. CLAUDE.md: Fallbacks für Phase 6d. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
from datetime import timedelta
|
|
|
|
from app.checkmk import Downtime
|
|
from app.commands import IncomingCallback
|
|
from tests.support import CHAT, TG_USER, Env, tg
|
|
|
|
|
|
class ReaderStub:
|
|
def __init__(self, downtimes=None, fail=False):
|
|
self._downtimes = downtimes or []
|
|
self.fail = fail
|
|
self.closed = False
|
|
|
|
async def downtimes(self, host=None):
|
|
if self.fail:
|
|
raise RuntimeError("boom")
|
|
return self._downtimes
|
|
|
|
async def aclose(self):
|
|
self.closed = True
|
|
|
|
|
|
async def confirm(env):
|
|
reply = await env.handler.handle(tg('/downtime web01 "CPU load" 2h Wartung'))
|
|
return await env.callbacks.handle(IncomingCallback("telegram", TG_USER, CHAT, "9", reply.buttons[0][0].data))
|
|
|
|
|
|
async def test_downtime_confirmed_when_reader_finds_matching_entry():
|
|
env = Env()
|
|
result = await confirm(env)
|
|
assert "Downtime gesetzt" in result.reply.text
|
|
assert env.downtime_audit()[-1] == "downtime.created"
|
|
|
|
|
|
async def test_downtime_unconfirmed_when_reader_finds_nothing():
|
|
env = Env()
|
|
env.downtime._reader_factory = lambda: ReaderStub([])
|
|
result = await confirm(env)
|
|
assert "gesendet, aber nicht bestätigt" in result.reply.text
|
|
assert env.downtime_audit()[-1] == "downtime.unconfirmed"
|
|
|
|
|
|
async def test_downtime_unconfirmed_when_reader_fails():
|
|
env = Env()
|
|
env.downtime._reader_factory = lambda: ReaderStub(fail=True)
|
|
result = await confirm(env)
|
|
assert "gesendet, aber nicht bestätigt" in result.reply.text
|
|
|
|
|
|
async def test_downtime_unconfirmed_when_no_reader_configured():
|
|
env = Env()
|
|
env.downtime._reader_factory = lambda: None
|
|
result = await confirm(env)
|
|
assert "gesendet, aber nicht bestätigt" in result.reply.text
|
|
|
|
|
|
async def test_downtime_not_confirmed_for_mismatching_entry():
|
|
env = Env()
|
|
wrong = Downtime("otherhost", "CPU load", "2026-11-06T08:00:00Z", "2026-11-06T10:00:00Z", "x", "a")
|
|
env.downtime._reader_factory = lambda: ReaderStub([wrong])
|
|
result = await confirm(env)
|
|
assert "gesendet, aber nicht bestätigt" in result.reply.text
|
|
|
|
|
|
async def test_rejection_message_hides_detail_but_keeps_http_status():
|
|
env = Env()
|
|
env.client.fail = "HTTP 403: nope, denied-secret"
|
|
result = await confirm(env)
|
|
assert "nicht angenommen" in result.reply.text and "403" in result.reply.text
|
|
assert "denied-secret" not in result.reply.text
|
|
|
|
|
|
async def test_rejection_message_hides_exception_class_name():
|
|
env = Env()
|
|
|
|
class BrokenClient:
|
|
async def create_service_downtime(self, *a, **k):
|
|
raise ValueError("weirdly broken")
|
|
|
|
async def aclose(self):
|
|
pass
|
|
|
|
env.downtime._client_factory = lambda: BrokenClient()
|
|
result = await confirm(env)
|
|
assert "ValueError" not in result.reply.text
|
|
assert "nicht angenommen" in result.reply.text
|