174 lines
6.5 KiB
Python
174 lines
6.5 KiB
Python
"""Unit: document summarizer (phase 30, task 03).
|
|
|
|
Covers the ``SUMMARY_MODE`` prompt (marker + instruction, capped user
|
|
content), the code-deterministic ``Source: <source>/<path>`` pointer,
|
|
and the rejection of empty/whitespace model output.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.rag.llm import LLMError
|
|
from app.rag.retriever import TRUNCATION_MARKER
|
|
from app.rag.summarizer import (
|
|
SUMMARY_INSTRUCTION,
|
|
SUMMARY_MODE,
|
|
SYSTEM_PROMPT,
|
|
build_summary_prompt,
|
|
generate_summary,
|
|
)
|
|
|
|
|
|
class _FakeLLM:
|
|
"""Duck-typed stand-in for ``LLMClient`` (``chat`` + ``settings``).
|
|
|
|
Records the messages and the ``model`` kwarg it was called with; can
|
|
return a canned reply or raise (e.g. :class:`LLMError`).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
reply: str | None = "Backups run nightly at 02:00 via the borg schedule.",
|
|
fail: Exception | None = None,
|
|
) -> None:
|
|
self._reply = reply
|
|
self._fail = fail
|
|
self.settings = Settings(_env_file=None) # pyright: ignore[reportCallIssue]
|
|
self.messages: list[dict[str, str]] = []
|
|
self.model: str | None = None
|
|
|
|
async def chat(
|
|
self, messages: list[dict[str, str]], model: str | None = None
|
|
) -> str:
|
|
self.messages = list(messages)
|
|
self.model = model
|
|
if self._fail is not None:
|
|
raise self._fail
|
|
assert self._reply is not None
|
|
return self._reply
|
|
|
|
|
|
# ---------- build_summary_prompt: system ----------
|
|
|
|
|
|
def test_system_prompt_has_marker_and_locked_instruction() -> None:
|
|
assert SYSTEM_PROMPT.startswith(SUMMARY_MODE)
|
|
assert SUMMARY_INSTRUCTION in SYSTEM_PROMPT
|
|
for fragment in (
|
|
"plain-text summary of this document in natural",
|
|
"what it configures/defines",
|
|
"Do not use markdown",
|
|
"Do not invent anything that is not in the document",
|
|
):
|
|
assert fragment in SYSTEM_PROMPT
|
|
system, _ = build_summary_prompt("Homelab", "a.yaml", "content")
|
|
assert system == SYSTEM_PROMPT
|
|
assert SUMMARY_MODE in system # the marker the E2E mock keys on
|
|
|
|
|
|
# ---------- build_summary_prompt: user (capped content) ----------
|
|
|
|
|
|
def test_user_prompt_is_full_content_when_under_cap() -> None:
|
|
content = "services:\n borg:\n port: 9999"
|
|
_, user = build_summary_prompt("Homelab", "a.yaml", content, max_chars=12_000)
|
|
assert user == content
|
|
assert TRUNCATION_MARKER not in user
|
|
|
|
|
|
def test_user_prompt_at_exact_cap_is_not_truncated() -> None:
|
|
content = "z" * 64
|
|
_, user = build_summary_prompt("Homelab", "a.yaml", content, max_chars=64)
|
|
assert user == content
|
|
assert TRUNCATION_MARKER not in user
|
|
|
|
|
|
def test_user_prompt_truncated_with_marker_when_over_custom_cap() -> None:
|
|
content = "x" * 100 + "TAIL"
|
|
_, user = build_summary_prompt("Homelab", "a.yaml", content, max_chars=100)
|
|
assert user == "x" * 100 + "\n" + TRUNCATION_MARKER
|
|
assert "TAIL" not in user # overflow is gone, not squeezed in
|
|
assert user.endswith(TRUNCATION_MARKER)
|
|
|
|
|
|
def test_user_prompt_truncated_at_default_cap() -> None:
|
|
"""No explicit cap → ``BOR_SUMMARY_MAX_CHARS`` (read from the live
|
|
settings, so the test holds for any configured value)."""
|
|
cap = get_settings().summary_max_chars
|
|
content = "y" * (cap + 50)
|
|
_, user = build_summary_prompt("Homelab", "a.yaml", content)
|
|
assert user == "y" * cap + "\n" + TRUNCATION_MARKER
|
|
|
|
|
|
# ---------- generate_summary: pointer + validation ----------
|
|
|
|
|
|
def test_generate_summary_returns_model_text_plus_deterministic_pointer() -> None:
|
|
llm = _FakeLLM(reply="Backups run nightly at 02:00 via the borg schedule.")
|
|
out = asyncio.run(
|
|
generate_summary(llm, source="Homelab", path="backups/borg.yaml", content="c")
|
|
)
|
|
expected = (
|
|
"Backups run nightly at 02:00 via the borg schedule.\n"
|
|
"Source: Homelab/backups/borg.yaml"
|
|
)
|
|
assert out == expected
|
|
assert out.splitlines()[-1] == "Source: Homelab/backups/borg.yaml"
|
|
|
|
|
|
def test_generate_summary_calls_the_configured_summary_model() -> None:
|
|
llm = _FakeLLM(reply="s")
|
|
asyncio.run(generate_summary(llm, source="Homelab", path="a.yaml", content="c"))
|
|
assert llm.model == llm.settings.llm_summary_model # the ``lite`` default
|
|
assert llm.model == "lite"
|
|
assert [m["role"] for m in llm.messages] == ["system", "user"]
|
|
assert SUMMARY_MODE in llm.messages[0]["content"]
|
|
assert llm.messages[1] == {"role": "user", "content": "c"}
|
|
|
|
|
|
def test_generate_summary_strips_model_text_before_appending_pointer() -> None:
|
|
llm = _FakeLLM(reply=" padded summary. \n")
|
|
out = asyncio.run(generate_summary(llm, source="Deployments", path="f.txt", content="c"))
|
|
assert out == "padded summary.\nSource: Deployments/f.txt"
|
|
|
|
|
|
def test_pointer_is_code_deterministic_even_if_model_writes_its_own() -> None:
|
|
"""The pointer must never be model-generated: even a model reply that
|
|
contains a bogus 'Source:' line ends with the code-appended one."""
|
|
llm = _FakeLLM(reply="The document itself says Source: fake/other.yaml inside.")
|
|
out = asyncio.run(generate_summary(llm, source="Homelab", path="real.yaml", content="c"))
|
|
assert out.splitlines()[-1] == "Source: Homelab/real.yaml"
|
|
|
|
|
|
def test_generate_summary_sends_capped_content_to_the_model() -> None:
|
|
"""The cap applies to what the model actually receives (overflow cut
|
|
at the cap + marker) — read from the live settings for any value."""
|
|
llm = _FakeLLM(reply="s")
|
|
content = "w" * (get_settings().summary_max_chars + 50)
|
|
asyncio.run(generate_summary(llm, source="Homelab", path="a.yaml", content=content))
|
|
cap = get_settings().summary_max_chars
|
|
assert llm.messages[1]["content"] == "w" * cap + "\n" + TRUNCATION_MARKER
|
|
|
|
|
|
def test_generate_summary_rejects_whitespace_only_reply() -> None:
|
|
llm = _FakeLLM(reply=" \n\t ")
|
|
with pytest.raises(LLMError, match="empty content"):
|
|
asyncio.run(generate_summary(llm, source="Homelab", path="a.yaml", content="c"))
|
|
|
|
|
|
def test_generate_summary_rejects_empty_reply() -> None:
|
|
llm = _FakeLLM(reply="")
|
|
with pytest.raises(LLMError, match="empty content"):
|
|
asyncio.run(generate_summary(llm, source="Homelab", path="a.yaml", content="c"))
|
|
|
|
|
|
def test_generate_summary_propagates_llm_error_from_client() -> None:
|
|
llm = _FakeLLM(
|
|
fail=LLMError("chat completion from https://aipi.reeseapps.com/v1 failed: boom")
|
|
)
|
|
with pytest.raises(LLMError, match="boom"):
|
|
asyncio.run(generate_summary(llm, source="Homelab", path="a.yaml", content="c"))
|