151 lines
6.7 KiB
Python
151 lines
6.7 KiB
Python
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from summary import Summarizer, SummaryError, load_summarizer
|
|
|
|
|
|
@contextmanager
|
|
def llm_server(body=None, status=200, delay=0, redirect=None):
|
|
requests = []
|
|
response = body if body is not None else {"choices": [{"message": {"content": "Rückruf morgen um zehn Uhr."}, "finish_reason": "stop"}]}
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
requests.append({"path": self.path, "authorization": self.headers.get("Authorization"),
|
|
"body": json.loads(self.rfile.read(int(self.headers["Content-Length"])))})
|
|
time.sleep(delay)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
if redirect:
|
|
self.send_header("Location", redirect)
|
|
self.end_headers()
|
|
try:
|
|
self.wfile.write(json.dumps(response).encode())
|
|
except BrokenPipeError:
|
|
pass
|
|
|
|
def log_message(self, *_):
|
|
pass
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}/v1", requests
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|
|
|
|
|
|
def test_openai_compatible_request_contains_text_model_and_optional_bearer():
|
|
with llm_server() as (base_url, requests):
|
|
summarize = Summarizer(base_url + "/", "local-model", "test-api-key")
|
|
assert summarize("Bitte morgen um zehn Uhr zurückrufen.", "meeting") == "Rückruf morgen um zehn Uhr."
|
|
assert requests[0]["path"] == "/v1/chat/completions"
|
|
assert requests[0]["authorization"] == "Bearer test-api-key"
|
|
payload = requests[0]["body"]
|
|
assert payload["model"] == "local-model"
|
|
assert payload["stream"] is False
|
|
assert payload["messages"][0]["role"] == "system"
|
|
assert "Aufgaben" in payload["messages"][0]["content"]
|
|
assert payload["messages"][1] == {"role": "user", "content": "Bitte morgen um zehn Uhr zurückrufen."}
|
|
assert "test-api-key" not in json.dumps(payload)
|
|
Summarizer(base_url, "model-without-key")("Text", "short")
|
|
assert requests[1]["authorization"] is None
|
|
assert "fünf" in requests[1]["body"]["messages"][0]["content"]
|
|
|
|
|
|
@pytest.mark.parametrize("body", [
|
|
{}, {"choices": []}, {"choices": [{"message": {"content": ""}}]},
|
|
{"choices": [{"message": {"content": None}}]},
|
|
{"choices": [{"message": {"content": "Halbe Antwort"}, "finish_reason": "length"}]},
|
|
{"choices": [{"message": {"content": "<think>Unfertiger Gedankengang"}}]},
|
|
])
|
|
def test_rejects_missing_empty_truncated_and_unfinished_responses(body):
|
|
with llm_server(body) as (base_url, _):
|
|
with pytest.raises(SummaryError, match="keine vollständige Zusammenfassung"):
|
|
Summarizer(base_url, "local-model")("Text", "standard")
|
|
|
|
|
|
def test_reasoning_is_not_exposed_as_summary():
|
|
body = {"choices": [{"message": {"content": "<think>Interner Gedankengang</think>\nFertige Zusammenfassung."}}]}
|
|
with llm_server(body) as (base_url, _):
|
|
assert Summarizer(base_url, "local-model")("Text", "standard") == "Fertige Zusammenfassung."
|
|
|
|
|
|
def test_http_errors_do_not_expose_provider_body_or_api_key():
|
|
with llm_server({"error": "secret-api-key private transcript"}, status=401) as (base_url, _):
|
|
with pytest.raises(SummaryError, match="HTTP 401") as error:
|
|
Summarizer(base_url, "local-model", "secret-api-key")("private transcript", "standard")
|
|
assert "secret-api-key" not in str(error.value)
|
|
assert "private transcript" not in str(error.value)
|
|
|
|
|
|
def test_no_redirect_can_forward_transcript_or_credentials():
|
|
with llm_server() as (target, target_requests):
|
|
with llm_server(status=307, redirect=target + "/chat/completions") as (base_url, _):
|
|
with pytest.raises(SummaryError, match="HTTP 307"):
|
|
Summarizer(base_url, "local-model", "test-key")("Text", "standard")
|
|
assert not target_requests
|
|
|
|
|
|
def test_timeout_is_retryable_summary_error():
|
|
with llm_server(delay=.1) as (base_url, _):
|
|
with pytest.raises(SummaryError, match="nicht rechtzeitig"):
|
|
Summarizer(base_url, "local-model", timeout=.01)("Text", "standard")
|
|
|
|
|
|
def test_long_transcript_is_fully_processed_before_merging(monkeypatch):
|
|
summarizer = Summarizer("http://localhost/v1", "model", chunk_chars=1000)
|
|
transcript = " ".join(f"Wort{number}" for number in range(600))
|
|
seen = []
|
|
|
|
def complete(text, style, partial=False):
|
|
seen.append((text, style, partial))
|
|
return f"Teil {len(seen)}" if partial else "Gesamtergebnis"
|
|
|
|
monkeypatch.setattr(summarizer, "complete", complete)
|
|
assert summarizer(transcript, "meeting") == "Gesamtergebnis"
|
|
chunks = [text for text, _, partial in seen if partial]
|
|
assert " ".join(chunks) == transcript
|
|
assert all(len(part) <= 1000 for part in chunks)
|
|
assert seen[-1] == ("\n\n".join(f"Teil {n}" for n in range(1, len(chunks) + 1)), "meeting", False)
|
|
|
|
|
|
def test_non_reducing_model_does_not_loop_forever(monkeypatch):
|
|
summarizer = Summarizer("http://localhost/v1", "model", chunk_chars=1000)
|
|
monkeypatch.setattr(summarizer, "complete", lambda text, *args, **kwargs: text)
|
|
with pytest.raises(SummaryError, match="nicht ausreichend gekürzt"):
|
|
summarizer("wort " * 1000, "meeting")
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["file:///tmp/model", "http://user:secret@localhost/v1", "http://localhost/v1?key=secret", "http://localhost/v1#section"])
|
|
def test_invalid_base_urls_are_rejected(url):
|
|
with pytest.raises(ValueError, match="LLM_BASE_URL"):
|
|
Summarizer(url, "model")
|
|
|
|
|
|
def test_env_config_is_optional_and_passed_to_client(monkeypatch):
|
|
for name in ["LLM_BASE_URL", "LLM_MODEL", "LLM_API_KEY", "LLM_TIMEOUT_SECONDS", "LLM_CHUNK_CHARS"]:
|
|
monkeypatch.delenv(name, raising=False)
|
|
assert load_summarizer() is None
|
|
monkeypatch.setenv("LLM_BASE_URL", "http://localhost:1234/v1")
|
|
with pytest.raises(SummaryError, match="LLM_MODEL"):
|
|
load_summarizer()("Transkript", "meeting")
|
|
monkeypatch.setenv("LLM_MODEL", "configured-model")
|
|
monkeypatch.setenv("LLM_API_KEY", "test-key")
|
|
monkeypatch.setenv("LLM_TIMEOUT_SECONDS", "45")
|
|
monkeypatch.setenv("LLM_CHUNK_CHARS", "8000")
|
|
client = load_summarizer()
|
|
assert (client.base_url, client.model, client.api_key, client.timeout, client.chunk_chars) == (
|
|
"http://localhost:1234/v1", "configured-model", "test-key", 45, 8000)
|