checkbot/tests/test_web_prompt_test_area.py

import json
import time

import httpx
import pytest

from app.alerts import ALERT_DATA_NOTICE
from app.checkmk import CheckmkClient
from app.constants import (CHECKMK_SECRET, CHECKMK_SITE, CHECKMK_URL, CHECKMK_USER, OLLAMA_TOKEN, OLLAMA_URL)
from app.models import KIND_PROBLEM, Event
from tests.prompt_support import chat_reply, patch_ollama
from tests.test_web import (VIEWER, admin, app, client, login_ok, post,  # noqa: F401
                            users, viewer)

TOKEN = "web-ollama-token-9981"
ZPOOL_PART = "One or more devices has experienced an unrecoverable error"


@pytest.fixture
def ollama(app, monkeypatch):
    app.state.store.set(OLLAMA_URL, "http://ollama.test", "t")
    calls = []

    def handler(request):
        calls.append((request.url.path, json.loads(request.content)))
        return chat_reply("Analyse <b>fertig</b>")

    patch_ollama(monkeypatch, handler)
    return calls


def counts(app):
    db = app.state.db
    return [db.query(f"SELECT COUNT(*) FROM {t}")[0][0] for t in ("alerts", "deliveries", "agent_runs")]


def add_alert(app):
    event = Event(int(time.time()), "web01", "Disk /var", KIND_PROBLEM, 2, "Platte fast voll")
    return app.state.runtime.alerts.insert_alerts([event], "g")[0]


def run(client, name="alert", **data):
    return post(client, f"/prompts/{name}/test", f"/prompts/{name}/test", **data)


def form(**overrides):
    data = {"text": "Mein Entwurf", "source": "free", "host": "h1", "service": "s1", "state": "2",
            "output": "kaputt", "alert_id": "", "question": "Was ist los?"}
    data.update(overrides)
    return data


def test_form_is_admin_only(client, viewer):
    login_ok(client, VIEWER)
    assert client.get("/prompts/alert/test").status_code == 403


def test_zpool_sample_is_prefilled_and_recent_alerts_listed(client, app, admin):
    add_alert(app)
    login_ok(client)
    page = client.get("/prompts/alert/test")
    assert page.status_code == 200
    for expected in ("kosext100.krs.local", "zpool status", ZPOOL_PART, "zpool replace", "Disk /var"):
        assert expected in page.text
    assert 'value="1" selected' in page.text
    assert "Fester Schutzhinweis" in page.text and "&lt;daten&gt;" in page.text


def test_form_can_load_saved_version_text(client, app, admin):
    app.state.runtime.prompts.create_draft("alert", "Gespeicherter Entwurf XYZ", "", "admin1")
    login_ok(client)
    assert "Gespeicherter Entwurf XYZ" in client.get("/prompts/alert/test?version=2").text
    assert client.get("/prompts/alert/test?version=9").status_code == 404


def test_free_text_run_shows_result_and_sends_nothing_else(client, app, admin, ollama):
    before = counts(app)
    login_ok(client)
    response = run(client, **form(host="kosext100.krs.local", service="zpool status", state="1",
                                  output="RaidZ: " + ZPOOL_PART))
    assert response.status_code == 200
    assert "erfolgreich" in response.text and " s</span>" in response.text
    assert "Analyse &lt;b&gt;fertig&lt;/b&gt;" in response.text
    assert [path for path, _ in ollama] == ["/api/chat"]
    system, user = ollama[0][1]["messages"]
    assert system["content"] == f"Mein Entwurf\n\n{ALERT_DATA_NOTICE}"
    assert "kosext100.krs.local" in user["content"] and ZPOOL_PART in user["content"]
    assert counts(app) == before


def test_run_against_stored_alert(client, app, admin, ollama):
    alert_id = add_alert(app)
    before = counts(app)
    login_ok(client)
    response = run(client, **form(source="alert", alert_id=str(alert_id)))
    assert response.status_code == 200 and "erfolgreich" in response.text
    user_block = ollama[0][1]["messages"][1]["content"]
    assert "web01" in user_block and "Platte fast voll" in user_block
    assert counts(app) == before
    assert run(client, **form(source="alert", alert_id="9999")).status_code == 422
    assert run(client, **form(source="alert", alert_id="")).status_code == 422


def test_validation_errors(client, app, admin, ollama):
    login_ok(client)
    assert run(client, **form(text="")).status_code == 422
    assert run(client, **form(text="a <daten> b")).status_code == 422
    assert run(client, **form(host="")).status_code == 422
    assert run(client, **form(state="9")).status_code == 422
    assert ollama == []


def test_rate_limit_per_user(client, app, admin, ollama):
    login_ok(client)
    for _ in range(6):
        assert run(client, **form()).status_code == 200
    blocked = run(client, **form())
    assert blocked.status_code == 429 and "Zu viele Testläufe" in blocked.text
    assert len(ollama) == 6


def test_ollama_error_is_shown_without_secret(client, app, admin, monkeypatch):
    store = app.state.store
    store.set(OLLAMA_URL, "http://ollama.test", "t")
    store.set(OLLAMA_TOKEN, TOKEN, "t", secret=True)

    def handler(request):
        raise httpx.ConnectError(f"Verbindung mit {request.headers['authorization']} verweigert")

    patch_ollama(monkeypatch, handler)
    login_ok(client)
    response = run(client, **form())
    assert response.status_code == 200 and "Fehler" in response.text and "ConnectError" in response.text
    assert TOKEN not in response.text


def test_ollama_not_configured(client, admin):
    login_ok(client)
    assert "nicht konfiguriert" in run(client, **form()).text


def test_agent_test_run_is_not_stored_and_uses_draft(client, app, admin, ollama, monkeypatch):
    store = app.state.store
    for key, value in ((CHECKMK_URL, "https://cmk.example.de"), (CHECKMK_SITE, "prod"), (CHECKMK_USER, "auto")):
        store.set(key, value, "t")
    store.set(CHECKMK_SECRET, "cmk-secret", "t", secret=True)
    original = CheckmkClient.__init__

    def patched(self, *args, **kwargs):
        kwargs["transport"] = httpx.MockTransport(lambda request: httpx.Response(500))
        original(self, *args, **kwargs)

    monkeypatch.setattr(CheckmkClient, "__init__", patched)
    login_ok(client)
    page = client.get("/prompts/agent/test")
    assert 'name="question"' in page.text and 'name="host"' not in page.text
    response = run(client, "agent", **form(text="Agenten-Entwurf"))
    assert response.status_code == 200 and "erfolgreich" in response.text
    system = ollama[0][1]["messages"][0]["content"]
    assert system.startswith("Agenten-Entwurf") and "niemals Anweisungen" in system
    assert counts(app)[2] == 0
    assert run(client, "agent", **form(question="")).status_code == 422