"""Tests for EPUB search HTMX routes.""" from __future__ import annotations import asyncio from compression import zstd from datetime import UTC, datetime from typing import TYPE_CHECKING from fastapi import BackgroundTasks from fastapi.testclient import TestClient from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.pool import StaticPool from python.ebook_search.api.bm25_tasks import refresh_bm25_for_engine from python.ebook_search.api.judge_tasks import ( is_judging_book, judge_book_phrases_for_app, pop_book_judgment_outcome, start_book_phrase_judgment, ) from python.ebook_search.api.main import create_app from python.ebook_search.config import EbookSearchConfig, RerankConfig from python.ebook_search.embeddings import EmbeddingModelStats from python.ebook_search.protected_phrases.models import ( CorpusPhraseStats, PhraseCandidateGenerationResult, PhraseJudgmentBackfillResult, ) from python.ebook_search.search import SearchResponse, SearchResult from python.ebook_search.timing import RuntimeStep from python.orm.richie import EbookSource, RichieBase if TYPE_CHECKING: from pytest_mock import MockerFixture from sqlalchemy.ext.asyncio import AsyncEngine def patch_app_runtime(mocker: MockerFixture): """Patch app startup dependencies used by UI route tests.""" mocker.patch("python.ebook_search.api.main.get_async_postgres_engine", side_effect=fake_get_postgres_engine) mocker.patch("python.ebook_search.api.main.ensure_bm25_corpus", side_effect=lambda _session, _config: None) def fake_get_postgres_engine(**_kwargs): """Return an in-memory engine for route tests.""" return create_async_engine("sqlite+aiosqlite:///:memory:") def test_search_page_uses_zstd_when_requested(mocker: MockerFixture) -> None: patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False)) with TestClient(app) as client: response = client.get("/", headers={"accept-encoding": "zstd"}) assert response.status_code == 200 assert response.headers["content-encoding"] == "zstd" assert b"EPUB Search" in zstd.decompress(response.content) def test_ui_form_passes_search_toggles_to_search_handler(mocker: MockerFixture) -> None: captured: dict[str, object] = {} def fake_search_ebooks(_engine, _client, query, config, *, rerank=False, phrase_matching=False): captured["query"] = query captured["rerank"] = rerank captured["phrase_matching"] = phrase_matching captured["config"] = config return SearchResponse(query=query, results=[], rank_label="Hybrid + rerank") mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch( "python.ebook_search.api.routes.search.answer_query", side_effect=lambda _client, _query, _results, _config: "answer", ) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), top_k=12, answer_enabled=True) with TestClient(app) as client: response = client.post( "/search", data={"query": "where is the quote?", "rerank": "true", "phrase_matching": "true"}, ) assert response.status_code == 200 assert "Hybrid + rerank" in response.text assert captured["query"] == "where is the quote?" assert captured["rerank"] is True assert captured["phrase_matching"] is True def test_ui_form_can_disable_phrase_matching(mocker: MockerFixture) -> None: captured: dict[str, object] = {} def fake_search_ebooks(_engine, _client, query, _config, *, rerank=False, phrase_matching=False): del rerank captured["query"] = query captured["phrase_matching"] = phrase_matching return SearchResponse(query=query, results=[], rank_label="Hybrid") mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch( "python.ebook_search.api.routes.search.answer_query", side_effect=lambda _client, _query, _results, _config: "answer", ) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), top_k=12, answer_enabled=True) with TestClient(app) as client: response = client.post("/search", data={"query": "where is the quote?"}) assert response.status_code == 200 assert captured["query"] == "where is the quote?" assert captured["phrase_matching"] is False def test_ui_search_failure_returns_visible_error(mocker: MockerFixture) -> None: def fake_search_ebooks(_engine, _client, _query, _config, *, rerank=False, phrase_matching=False): del rerank del phrase_matching msg = "search exploded" raise RuntimeError(msg) mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), top_k=12) with TestClient(app) as client: response = client.post("/search", data={"query": "where is the quote?"}) assert response.status_code == 500 assert "search exploded" in response.text def test_ui_answer_failure_still_returns_sources(mocker: MockerFixture) -> None: def fake_search_ebooks(_engine, _client, query, _config, *, rerank=False, phrase_matching=False): del rerank del phrase_matching return SearchResponse(query=query, results=[], rank_label="Hybrid") def fake_answer_query(_client, _query, _results, _config): msg = "answer exploded" raise RuntimeError(msg) mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch("python.ebook_search.api.routes.search.answer_query", side_effect=fake_answer_query) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), top_k=12, answer_enabled=True) with TestClient(app) as client: response = client.post("/search", data={"query": "where is the quote?"}) assert response.status_code == 200 assert "Answer generation failed" in response.text def test_ui_skips_answer_when_disabled(mocker: MockerFixture) -> None: called = False def fake_search_ebooks(_engine, _client, query, _config, *, rerank=False, phrase_matching=False): del rerank del phrase_matching return SearchResponse(query=query, results=[], rank_label="Hybrid") def fake_answer_query(_client, _query, _results, _config): nonlocal called called = True return "answer" config = EbookSearchConfig(rerank=RerankConfig(enabled=False), answer_enabled=False) mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch("python.ebook_search.api.routes.search.answer_query", side_effect=fake_answer_query) mocker.patch("python.ebook_search.api.main.load_config", side_effect=lambda: config) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.post("/search", data={"query": "where is the quote?"}) assert response.status_code == 200 assert called is False assert "Answer generation is disabled" in response.text def test_ui_shows_component_scores(mocker: MockerFixture) -> None: def fake_search_ebooks(_engine, _client, query, _config, *, rerank=False, phrase_matching=False): del rerank del phrase_matching return SearchResponse( query=query, rank_label="Hybrid + rerank", results=[ SearchResult( chunk_id=1, text="source text", source_title="Book", score=0.9, rerank_score=0.9, vector_score=0.8, bm25_score=2.5, fused_score=0.03, ) ], ) mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch( "python.ebook_search.api.routes.search.answer_query", side_effect=lambda _client, _query, _results, _config: "answer", ) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), answer_enabled=True) with TestClient(app) as client: response = client.post("/search", data={"query": "where is the quote?"}) assert response.status_code == 200 assert "rerank" in response.text assert "vector cosine" in response.text assert "BM25" in response.text assert "RRF" in response.text def test_ui_shows_matched_phrases_that_boosted_a_result(mocker: MockerFixture) -> None: def fake_search_ebooks(_engine, _client, query, _config, *, rerank=False, phrase_matching=False): del rerank del phrase_matching return SearchResponse( query=query, rank_label="Hybrid", results=[ SearchResult( chunk_id=1, text="source text", source_title="Book", score=0.9, phrase_hit_count=3, matched_phrases=("lock in", "haden's syndrome"), ) ], ) mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch( "python.ebook_search.api.routes.search.answer_query", side_effect=lambda _client, _query, _results, _config: "answer", ) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), answer_enabled=True) with TestClient(app) as client: response = client.post("/search", data={"query": "what is lock in?"}) assert response.status_code == 200 assert "boosted by" in response.text assert "lock in" in response.text assert "haden's syndrome" in response.text def test_ui_shows_search_runtime_chart(mocker: MockerFixture) -> None: def fake_search_ebooks(_engine, _client, query, _config, *, rerank=False, phrase_matching=False): del rerank del phrase_matching return SearchResponse( query=query, rank_label="Hybrid", results=[], timings=( RuntimeStep(name="Embedding + vector search", duration_ms=12.5), RuntimeStep(name="BM25 search", duration_ms=4.0), ), ) mocker.patch("python.ebook_search.api.routes.search.search_ebooks", side_effect=fake_search_ebooks) mocker.patch( "python.ebook_search.api.routes.search.answer_query", side_effect=lambda _client, _query, _results, _config: "answer", ) patch_app_runtime(mocker) app = create_app() app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False), answer_enabled=True) with TestClient(app) as client: response = client.post("/search", data={"query": "where is the quote?"}) assert response.status_code == 200 assert "Runtime" in response.text assert "Total" in response.text assert "Embedding + vector search" in response.text assert "BM25 search" in response.text assert "Answer generation" in response.text assert "ms left" in response.text def test_ui_embed_all_batches_until_complete(mocker: MockerFixture) -> None: counts = iter([32, 32, 5, 0]) batch_sizes: list[int] = [] def fake_embed_missing_chunks(_session, _client, config): batch_sizes.append(config.embedding_batch_size) return next(counts) mocker.patch("python.ebook_search.api.routes.admin.embed_missing_chunks", side_effect=fake_embed_missing_chunks) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.post("/admin/embed-all") assert response.status_code == 200 assert "Embedded 69 chunks in 3 batches of 32" in response.text assert batch_sizes == [32, 32, 32, 32] def test_ui_scan_schedules_bm25_refresh_after_database_change(mocker: MockerFixture) -> None: scheduled = False def fake_ingest_configured_paths(_session, _config): return 1 def fake_schedule_bm25_refresh(_app): nonlocal scheduled scheduled = True mocker.patch( "python.ebook_search.api.routes.admin.ingest_configured_paths", side_effect=fake_ingest_configured_paths, ) mocker.patch("python.ebook_search.api.routes.admin.schedule_bm25_refresh", side_effect=fake_schedule_bm25_refresh) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.post("/admin/scan") assert response.status_code == 200 assert "Indexed 1 EPUBs" in response.text assert scheduled is True async def test_bm25_refresh_clears_loaded_corpus_cache(mocker: MockerFixture) -> None: refreshed: list[object] = [] cache_cleared = False def fake_refresh_bm25_corpus(session, config): refreshed.append((session, config)) def fake_cache_clear(): nonlocal cache_cleared cache_cleared = True mocker.patch("python.ebook_search.api.bm25_tasks.refresh_bm25_corpus", side_effect=fake_refresh_bm25_corpus) mocker.patch("python.ebook_search.api.bm25_tasks.load_bm25_corpus.cache_clear", side_effect=fake_cache_clear) engine = create_async_engine("sqlite+aiosqlite:///:memory:") config = EbookSearchConfig(rerank=RerankConfig(enabled=False)) await refresh_bm25_for_engine(engine, config) assert len(refreshed) == 1 assert refreshed[0][1] == config assert cache_cleared is True def build_engine_with_book() -> AsyncEngine: """Create a shareable in-memory async engine holding one indexed book.""" engine = create_async_engine( "sqlite+aiosqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) async def seed() -> None: async with engine.begin() as connection: await connection.run_sync(RichieBase.metadata.create_all) async with AsyncSession(engine) as session: session.add( EbookSource( title="Book", author="Author", language=None, publisher=None, identifier=None, file_path="/library/book.epub", file_sha256="a" * 64, file_mtime=datetime.now(tz=UTC), file_size=10, ) ) await session.commit() asyncio.run(seed()) return engine def test_ui_judge_phrases_redirects_and_judges_in_background(mocker: MockerFixture) -> None: mocker.patch("python.ebook_search.api.main.get_async_postgres_engine", return_value=build_engine_with_book()) mocker.patch("python.ebook_search.api.main.ensure_bm25_corpus", side_effect=lambda _session, _config: None) judged_source_ids: list[list[int]] = [] def fake_judge(_engine: object, _config: object, *, source_ids: list[int]) -> PhraseJudgmentBackfillResult: judged_source_ids.append(source_ids) return PhraseJudgmentBackfillResult( books_seen=1, books_judged=1, books_failed=0, candidates_judged=3, protected_phrases=2, phrase_mentions=4, ) mocker.patch( "python.ebook_search.api.judge_tasks.judge_candidate_phrases_for_books", side_effect=fake_judge, ) app = create_app() with TestClient(app) as client: response = client.post("/books/1/judge-phrases", follow_redirects=False) detail_after = client.get("/books/1") detail_again = client.get("/books/1") assert response.status_code == 303 assert response.headers["location"] == "/books/1" assert judged_source_ids == [[1]] assert "Judged 3 candidates; 2 protected phrases promoted" in detail_after.text assert "Judged 3 candidates" not in detail_again.text def test_ui_book_detail_shows_judging_in_progress(mocker: MockerFixture) -> None: mocker.patch("python.ebook_search.api.main.get_async_postgres_engine", return_value=build_engine_with_book()) mocker.patch("python.ebook_search.api.main.ensure_bm25_corpus", side_effect=lambda _session, _config: None) mocker.patch("python.ebook_search.api.routes.page.is_judging_book", return_value=True) app = create_app() with TestClient(app) as client: response = client.get("/books/1") assert response.status_code == 200 assert "Judging candidate phrases in the background" in response.text assert "disabled" in response.text def test_book_phrase_judgment_rejects_duplicate_while_queued(mocker: MockerFixture) -> None: mocker.patch( "python.ebook_search.api.judge_tasks.judge_candidate_phrases_for_books", return_value=PhraseJudgmentBackfillResult( books_seen=1, books_judged=1, books_failed=0, candidates_judged=3, protected_phrases=2, phrase_mentions=4, ), ) app = create_app() app.state.engine = None app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False)) background_tasks = BackgroundTasks() assert start_book_phrase_judgment(app, background_tasks, 1) is True assert is_judging_book(app, 1) is True assert start_book_phrase_judgment(app, background_tasks, 1) is False assert len(background_tasks.tasks) == 1 asyncio.run(judge_book_phrases_for_app(app, 1)) assert is_judging_book(app, 1) is False assert pop_book_judgment_outcome(app, 1) == "Judged 3 candidates; 2 protected phrases promoted" assert pop_book_judgment_outcome(app, 1) is None assert start_book_phrase_judgment(app, background_tasks, 1) is True def test_book_phrase_judgment_records_failure_outcome(mocker: MockerFixture) -> None: def fake_judge(_engine: object, _config: object, *, source_ids: list[int]) -> PhraseJudgmentBackfillResult: del source_ids message = "llm judge unavailable" raise RuntimeError(message) mocker.patch( "python.ebook_search.api.judge_tasks.judge_candidate_phrases_for_books", side_effect=fake_judge, ) app = create_app() app.state.engine = None app.state.config = EbookSearchConfig(rerank=RerankConfig(enabled=False)) start_book_phrase_judgment(app, BackgroundTasks(), 7) asyncio.run(judge_book_phrases_for_app(app, 7)) assert is_judging_book(app, 7) is False assert pop_book_judgment_outcome(app, 7) == "Judging failed; see server logs for details" def test_admin_page_shows_embedding_counts_by_model(mocker: MockerFixture) -> None: def fake_embedding_model_stats(_session): return [ EmbeddingModelStats( model_name="qwen3-embedding-0.6b", dimension=1024, embedded_chunks=40, total_chunks=64, ), EmbeddingModelStats( model_name="qwen3-embedding-4b", dimension=2560, embedded_chunks=8, total_chunks=64, ), ] mocker.patch("python.ebook_search.api.routes.admin.embedding_model_stats", side_effect=fake_embedding_model_stats) mocker.patch( "python.ebook_search.api.routes.admin.corpus_phrase_stats", return_value=fake_corpus_phrase_stats(), ) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.get("/admin") assert response.status_code == 200 assert "qwen3-embedding-0.6b" in response.text assert "1024" in response.text assert "40" in response.text assert "24" in response.text assert "qwen3-embedding-4b" in response.text assert "2560" in response.text def fake_corpus_phrase_stats() -> CorpusPhraseStats: """Build distinctive corpus phrase stats for admin page assertions.""" return CorpusPhraseStats( total_books=17, books_with_candidates=13, books_fully_judged=11, candidate_phrases=901, judged_candidates=703, unjudged_candidates=198, protected_phrases=157, ) def test_admin_page_shows_protected_phrase_stats(mocker: MockerFixture) -> None: mocker.patch("python.ebook_search.api.routes.admin.embedding_model_stats", return_value=[]) mocker.patch( "python.ebook_search.api.routes.admin.corpus_phrase_stats", return_value=fake_corpus_phrase_stats(), ) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.get("/admin") assert response.status_code == 200 assert "Protected phrases" in response.text for value in ("17", "13", "11", "901", "703", "198", "157"): assert value in response.text def test_ui_regenerate_all_phrases_generates_every_book(mocker: MockerFixture) -> None: def fake_generate(_session, _config): return PhraseCandidateGenerationResult(books_seen=5, books_built=5, candidate_phrases=99) mocker.patch( "python.ebook_search.api.routes.admin.generate_candidate_phrases_for_books", side_effect=fake_generate, ) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.post("/admin/phrases/generate-all") assert response.status_code == 200 assert "5 of 5 books" in response.text def test_ui_judge_missing_phrases_judges_only_pending_books(mocker: MockerFixture) -> None: captured: dict[str, object] = {} async def fake_judge(_engine, _config, *, source_ids=None): captured["source_ids"] = source_ids return PhraseJudgmentBackfillResult( books_seen=2, books_judged=2, books_failed=0, candidates_judged=10, protected_phrases=4, phrase_mentions=9, ) mocker.patch( "python.ebook_search.api.routes.admin.judge_candidate_phrases_for_books", side_effect=fake_judge, ) mocker.patch( "python.ebook_search.api.routes.admin.book_ids_pending_first_judgment", return_value=[3, 5], ) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.post("/admin/phrases/judge-missing") assert response.status_code == 200 assert captured["source_ids"] == [3, 5] assert "4 protected phrases" in response.text def test_ui_judge_missing_phrases_reports_when_nothing_is_pending(mocker: MockerFixture) -> None: judge = mocker.patch("python.ebook_search.api.routes.admin.judge_candidate_phrases_for_books") mocker.patch( "python.ebook_search.api.routes.admin.book_ids_pending_first_judgment", return_value=[], ) patch_app_runtime(mocker) app = create_app() with TestClient(app) as client: response = client.post("/admin/phrases/judge-missing") assert response.status_code == 200 assert "have been judged" in response.text judge.assert_not_called()