checkbot/tests/test_pipeline_phase6c.py

from app.alerts import AlertActions
from app.chat import Target
from app.constants import UI_BASE_URL
from app.models import (HOST_SERVICE_NAME, KIND_RECOVERY, KIND_REMINDER, SUPPRESS_NOT_REPORTED,
                        SUPPRESS_SEND_LIMIT, Event)
from tests.test_pipeline import BUNDLE, Env, FakeAdapter, FakeOllama, problem


def hosts(count):
    return [problem(host=f"h{i}") for i in range(count)]


def rows(env):
    return sorted(env.repo.recent(), key=lambda r: r["id"])


async def test_messages_over_the_limit_are_not_sent_and_marked_in_history():
    env = Env(FakeOllama(), send_limit=lambda: 2)
    await env.handle(hosts(5))
    assert len(env.adapter.sent) == 2
    limited = [r for r in rows(env) if r["suppress_reason"] == SUPPRESS_SEND_LIMIT]
    assert [r["host"] for r in limited] == ["h2", "h3", "h4"]
    assert all(r["suppressed"] == 1 and r["ai_status"] == "skipped" for r in limited)
    assert len(env.ollama.calls) == 2


async def test_collective_notice_once_per_minute_and_target():
    env = Env(FakeOllama(), send_limit=lambda: 2)
    await env.handle(hosts(5))
    await env.pipeline._flush_overflow()
    assert len(env.adapter.sent) == 3
    assert "3 weitere Alarme in der letzten Minute, siehe Verlauf" in env.adapter.sent[2][1]
    env.clock.now += 30
    await env.handle([problem(host="late1"), problem(host="late2"), problem(host="late3")])
    await env.pipeline._flush_overflow()
    assert len(env.adapter.sent) == 3
    env.clock.now += 40
    await env.pipeline._flush_overflow()
    assert len(env.adapter.sent) == 4 and "weitere" in env.adapter.sent[3][1]


async def test_collective_notice_for_single_alert_uses_singular():
    env = Env(FakeOllama(), send_limit=lambda: 1)
    await env.handle(hosts(2))
    await env.pipeline._flush_overflow()
    assert "1 weiterer Alarm in der letzten Minute" in env.adapter.sent[-1][1]


async def test_collective_notice_has_ui_link_only_when_base_url_is_set():
    plain = Env(FakeOllama(), send_limit=lambda: 1)
    await plain.handle(hosts(2))
    await plain.pipeline._flush_overflow()
    assert "http" not in plain.adapter.sent[-1][1]
    linked = Env(FakeOllama(), send_limit=lambda: 1, base_url=lambda: "https://bot.example.de")
    await linked.handle(hosts(2))
    await linked.pipeline._flush_overflow()
    assert "https://bot.example.de/alerts" in linked.adapter.sent[-1][1]


async def test_limit_counts_per_target():
    tg, mx = FakeAdapter("telegram", "tg1"), FakeAdapter("matrix", "!r:s")
    env = Env(FakeOllama(), adapter_list=[tg, mx], send_limit=lambda: 1)
    await env.handle(hosts(2))
    assert len(tg.sent) == 1 and len(mx.sent) == 1
    keys = {key for key, _count in env.pipeline._limiter.due_notices()}
    assert keys == {Target("telegram", "tg1"), Target("matrix", "!r:s")}


async def test_ai_edits_do_not_count_against_the_limit():
    env = Env(FakeOllama(), send_limit=lambda: 1)
    await env.handle([problem()])
    assert len(env.adapter.sent) == 1 and len(env.adapter.edits) == 1
    assert env.pipeline._limiter.acquire(Target("fake", "chat1"), 1) is False
    assert [r["suppress_reason"] for r in rows(env)] == [None]


async def test_reminders_count_against_the_limit():
    env = Env(FakeOllama(), send_limit=lambda: 2)
    events = [Event(ts=1000, host=f"h{i}", service="disk", kind=KIND_REMINDER, state=2, output="voll")
              for i in range(3)]
    await env.handle(events)
    assert len(env.adapter.sent) == 2
    assert [r["suppress_reason"] for r in rows(env)].count(SUPPRESS_SEND_LIMIT) == 1


async def test_window_slides_so_later_alerts_pass_again():
    env = Env(FakeOllama(), send_limit=lambda: 1)
    await env.handle(hosts(2))
    env.clock.now += 61
    await env.handle([problem(host="later")])
    assert len(env.adapter.sent) == 2


async def test_short_mode_sends_no_output_and_no_ai():
    env = Env(FakeOllama(), content_mode=lambda backend: "kurz")
    await env.handle([problem()])
    text = env.adapter.sent[0][1]
    assert "<b>h1</b>" in text and "disk" in text and "CRIT" in text and "Platte voll" not in text
    assert env.adapter.edits == [] and env.ollama.calls == []
    assert rows(env)[0]["ai_status"] == "skipped"


async def test_mixed_modes_split_text_and_only_full_target_gets_ai_edit():
    tg, mx = FakeAdapter("telegram", "tg1"), FakeAdapter("matrix", "!r:s")
    modes = {"telegram": "kurz", "matrix": "voll"}
    env = Env(FakeOllama(), adapter_list=[tg, mx], content_mode=modes.get)
    await env.handle([problem()])
    assert "Platte voll" not in tg.sent[0][1] and "Platte voll" in mx.sent[0][1]
    assert tg.edits == [] and len(mx.edits) == 1 and "Analyse" in mx.edits[0][1]
    assert len(env.ollama.calls) == 1
    assert len(rows(env)) == 1


async def test_group_key_contains_modes_so_changed_mode_is_not_bundled_with_older_window():
    modes = {"fake": "voll"}
    env = Env(FakeOllama(), content_mode=lambda backend: modes[backend])
    await env.handle([problem(service="a")])
    env.clock.now += 5
    await env.handle([problem(service="b")])
    assert len(env.adapter.sent) == 1
    modes["fake"] = "kurz"
    await env.handle([problem(service="c")])
    assert len(env.adapter.sent) == 2 and "Platte voll" not in env.adapter.sent[1][1]


async def test_host_problem_and_recovery_use_host_texts():
    env = Env(FakeOllama())
    await env.handle([problem(service=HOST_SERVICE_NAME, output="no route", since_ts=900)])
    assert "<b>Host DOWN</b>" in env.adapter.sent[0][1] and "no route" in env.adapter.sent[0][1]
    env.clock.now += BUNDLE + 40
    recovery = Event(ts=4600, host="h1", service=HOST_SERVICE_NAME, kind=KIND_RECOVERY, state=0,
                     previous_state=2, problem_since_ts=1000)
    await env.handle([recovery])
    assert "<b>Host wieder UP</b> nach 1 h (zuletzt DOWN)" in env.adapter.sent[1][1]
    assert len(env.ollama.calls) == 1


async def test_baseline_reset_lets_recovery_pass_despite_only_suppressed_history():
    env = Env(FakeOllama())
    env.repo.insert_alerts([problem(suppressed=True, suppress_reason="downtime")], "g")
    recovery = Event(ts=2000, host="h1", service="disk", kind=KIND_RECOVERY, state=0, previous_state=2)
    await env.handle([recovery])
    assert env.adapter.sent == [] and rows(env)[-1]["suppress_reason"] == SUPPRESS_NOT_REPORTED
    env.pipeline.reset_reported([("h1", "disk")])
    env.clock.now += BUNDLE + 1
    await env.handle([recovery])
    assert len(env.adapter.sent) == 1


async def test_recovery_after_baseline_problem_without_history_is_sent():
    env = Env(FakeOllama())
    recovery = Event(ts=2000, host="h1", service="disk", kind=KIND_RECOVERY, state=0, previous_state=2)
    await env.handle([recovery])
    assert len(env.adapter.sent) == 1 and "wieder OK" in env.adapter.sent[0][1]


async def test_details_link_added_to_pipeline_messages_when_base_url_set():
    mx = FakeAdapter("matrix", "!r:s")
    env = Env(FakeOllama(), adapter_list=[mx])
    env.store.set(UI_BASE_URL, "https://bot.example.de", "t")
    env.pipeline._actions = AlertActions(env.store)
    await env.handle([problem()])
    alert_id = rows(env)[0]["id"]
    assert f"Details: https://bot.example.de/alerts#alert-{alert_id}" in mx.sent[0][1]
    assert f"Details: https://bot.example.de/alerts#alert-{alert_id}" in mx.edits[0][1]