Richie
41592491dc
treefmt / nix fmt (pull_request) Failing after 7s
pytest / pytest (pull_request) Failing after 12s
build_systems / build-brain (pull_request) Successful in 44s
build_systems / build-bob (pull_request) Successful in 46s
build_systems / build-leviathan (pull_request) Successful in 54s
build_systems / build-rhapsody-in-green (pull_request) Successful in 59s
build_systems / build-jeeves (pull_request) Successful in 2m29s
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Tests for EPUB search HTTP model adapters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from python.ebook_search.answer import answer_query
|
|
from python.ebook_search.config import EbookSearchConfig, RerankConfig
|
|
from python.ebook_search.embeddings import embed_texts
|
|
from python.ebook_search.search import SearchResult
|
|
|
|
|
|
def test_answer_query_uses_httpx_chat_completions(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_post(url: str, **kwargs: object) -> httpx.Response:
|
|
captured["url"] = url
|
|
captured["kwargs"] = kwargs
|
|
return httpx.Response(
|
|
200,
|
|
json={"choices": [{"message": {"content": "grounded answer"}}]},
|
|
request=httpx.Request("POST", url),
|
|
)
|
|
|
|
monkeypatch.setattr(httpx, "post", fake_post)
|
|
config = EbookSearchConfig(
|
|
rerank=RerankConfig(enabled=False),
|
|
vllm_base_url="https://ollama.com/v1",
|
|
vllm_api_key="secret",
|
|
chat_model="deepseek-v4-flash",
|
|
)
|
|
|
|
answer = answer_query("question", [SearchResult(chunk_id=1, text="source", source_title="Book")], config)
|
|
|
|
assert answer == "grounded answer"
|
|
assert captured["url"] == "https://ollama.com/v1/chat/completions"
|
|
kwargs = captured["kwargs"]
|
|
assert isinstance(kwargs, dict)
|
|
assert kwargs["headers"] == {"Authorization": "Bearer secret"}
|
|
payload = kwargs["json"]
|
|
assert isinstance(payload, dict)
|
|
assert payload["model"] == "deepseek-v4-flash"
|
|
|
|
|
|
def test_embed_texts_uses_httpx_embeddings(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
vector = [0.0] * 1024
|
|
|
|
def fake_post(url: str, **kwargs: object) -> httpx.Response:
|
|
captured["url"] = url
|
|
captured["kwargs"] = kwargs
|
|
return httpx.Response(
|
|
200,
|
|
json={"data": [{"embedding": vector}]},
|
|
request=httpx.Request("POST", url),
|
|
)
|
|
|
|
monkeypatch.setattr(httpx, "post", fake_post)
|
|
config = EbookSearchConfig(
|
|
rerank=RerankConfig(enabled=False),
|
|
embedding_base_url="http://bob:8000/v1",
|
|
embedding_model="qwen3-embedding-0.6b",
|
|
)
|
|
|
|
embeddings = embed_texts(["hello"], config)
|
|
|
|
assert embeddings == [vector]
|
|
assert captured["url"] == "http://bob:8000/v1/embeddings"
|
|
kwargs = captured["kwargs"]
|
|
assert isinstance(kwargs, dict)
|
|
assert kwargs["headers"] == {}
|
|
assert kwargs["json"] == {"model": "qwen3-embedding-0.6b", "input": ["hello"]}
|
|
|
|
|
|
def test_embed_texts_rejects_bad_response_shape(monkeypatch) -> None:
|
|
def fake_post(url: str, **_kwargs: object) -> httpx.Response:
|
|
return httpx.Response(200, json={"data": [{}]}, request=httpx.Request("POST", url))
|
|
|
|
monkeypatch.setattr(httpx, "post", fake_post)
|
|
config = EbookSearchConfig(rerank=RerankConfig(enabled=False))
|
|
|
|
with pytest.raises(RuntimeError, match="Embedding request failed"):
|
|
embed_texts(["hello"], config)
|