71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
"""Tests for app/services/llm.py — graceful failure and context building."""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.services.llm import summarize, _build_context
|
|
from app.services.search import SearchResult
|
|
|
|
|
|
def _entry(text: str, severity: str = "INFO", ts: str = "2026-05-06T21:00:00+00:00") -> SearchResult:
|
|
return SearchResult(
|
|
entry_id="x",
|
|
source_id="svc",
|
|
sequence=0,
|
|
timestamp_iso=ts,
|
|
severity=severity,
|
|
text=text,
|
|
matched_patterns=[],
|
|
repeat_count=1,
|
|
out_of_order=False,
|
|
rank=0.0,
|
|
)
|
|
|
|
|
|
def test_summarize_returns_none_on_connection_error():
|
|
with patch("app.services.llm.httpx.post", side_effect=ConnectionError("refused")):
|
|
result = summarize("ollama crashed", [_entry("failed")], "http://bad", "llama3")
|
|
assert result is None
|
|
|
|
|
|
def test_summarize_returns_none_on_http_error():
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.side_effect = Exception("404")
|
|
with patch("app.services.llm.httpx.post", return_value=mock_resp):
|
|
result = summarize("ollama crashed", [_entry("failed")], "http://host", "llama3")
|
|
assert result is None
|
|
|
|
|
|
def test_summarize_returns_none_on_empty_response():
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.return_value = None
|
|
mock_resp.json.return_value = {"response": ""}
|
|
with patch("app.services.llm.httpx.post", return_value=mock_resp):
|
|
result = summarize("query", [_entry("x")], "http://host", "llama3")
|
|
assert result is None
|
|
|
|
|
|
def test_summarize_returns_text_on_success():
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.return_value = None
|
|
mock_resp.json.return_value = {"response": "Ollama exited with code 1."}
|
|
with patch("app.services.llm.httpx.post", return_value=mock_resp):
|
|
result = summarize("ollama crashed", [_entry("Failed")], "http://host", "llama3")
|
|
assert result == "Ollama exited with code 1."
|
|
|
|
|
|
def test_build_context_sorts_errors_first():
|
|
entries = [
|
|
_entry("info message", severity="INFO"),
|
|
_entry("critical crash", severity="CRITICAL"),
|
|
_entry("warn spike", severity="WARN"),
|
|
]
|
|
ctx = _build_context(entries)
|
|
lines = ctx.splitlines()
|
|
assert "CRITICAL" in lines[0]
|
|
assert "WARN" in lines[1]
|
|
|
|
|
|
def test_summarize_empty_entries_returns_none():
|
|
result = summarize("query", [], "http://host", "model")
|
|
assert result is None
|