homelab-codex-ws/services/kb-query/tests/test_search.py

"""Unit tests for /search's core logic (app.search.run_search) -- no real DB, no real Ollama.
Same mocking style as packages/kb-retrieval/tests/test_retrieval.py, extended with an
`envelope` table fixture for the join app/db.py adds on top of kb_retrieval."""
from __future__ import annotations

import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))

from fastapi.testclient import TestClient  # noqa: E402

from app.main import app  # noqa: E402
from app.search import run_search  # noqa: E402


class _FakeConn:
    """summaries: [(envelope_id, dist), ...] -- cascade stage-1 pre-filter. chunks_by_envelope:
    envelope_id -> [(chunk_index, text, dist), ...]. envelopes: envelope_id -> {"source": ...,
    "entities": [...]}. summary_texts: envelope_id -> {"summary": ..., "tags": [...]} -- the
    document_summary row fetched for the result header (app/db.py fetch_summaries)."""

    def __init__(
        self, summaries=None, chunks_by_envelope=None, envelopes=None, summary_texts=None,
        mail_chunks_by_source=None,
    ):
        self._summaries = list(summaries or [])
        self._chunks_by_envelope = chunks_by_envelope or {}
        self._envelopes = envelopes or {}
        self._summary_texts = summary_texts or {}
        self._mail_chunks_by_source = mail_chunks_by_source or {}

    async def fetch(self, query, *params):
        if "FROM document_summary" in query and "= ANY" in query:
            (envelope_ids, _model) = params
            return [
                {"envelope_id": eid, "summary": s["summary"], "tags": s["tags"]}
                for eid, s in self._summary_texts.items()
                if eid in envelope_ids
            ]
        if "FROM document_summary" in query:
            _, _model, limit = params
            return [{"envelope_id": eid, "dist": dist} for eid, dist in self._summaries[:limit]]
        if "JOIN envelope" in query:  # hybrid's direct summaryless-source chunk scan
            _, sources, limit = params
            rows = [
                {"envelope_id": eid, "chunk_index": idx, "text": text, "dist": dist}
                for source in sources
                for eid, idx, text, dist in self._mail_chunks_by_source.get(source, [])
            ]
            rows.sort(key=lambda r: r["dist"])
            return rows[:limit]
        if "FROM document_chunk" in query and "= ANY" in query:
            _, envelope_ids, limit = params
            rows = [
                {"envelope_id": eid, "chunk_index": idx, "text": text, "dist": dist}
                for eid in envelope_ids
                for idx, text, dist in self._chunks_by_envelope.get(eid, [])
            ]
            rows.sort(key=lambda r: r["dist"])
            return rows[:limit]
        if "FROM document_chunk" in query:  # flat path
            _, limit = params
            rows = [
                {"envelope_id": eid, "chunk_index": idx, "text": text, "dist": dist}
                for eid, chunk_list in self._chunks_by_envelope.items()
                for idx, text, dist in chunk_list
            ]
            rows.sort(key=lambda r: r["dist"])
            return rows[:limit]
        if "FROM envelope" in query:
            (envelope_ids,) = params
            return [
                {"id": eid, "source": self._envelopes[eid]["source"], "entities": self._envelopes[eid]["entities"]}
                for eid in envelope_ids
                if eid in self._envelopes
            ]
        raise AssertionError(f"unexpected query: {query}")


class _FakeSession:
    """run_search no longer talks HTTP itself -- embedding goes through the router
    (below), so the session is just passed through untouched."""


class _FakeRouter:
    """Stands in for app.embed_router.EmbedRouter: returns a fixed embedding and the name
    of the backend that 'served' it, mirroring EmbedRouter.embed's contract."""

    def __init__(self, backend="solaria", primary_name="solaria"):
        self._backend = backend
        self.primary = type("B", (), {"name": primary_name})()
        self.embed_calls: list[str] = []

    async def embed(self, session, text):
        self.embed_calls.append(text)
        return [0.01] * 1024, self._backend


class TestRunSearchHappyPath:
    async def test_cascade_hit_joins_envelope_and_shapes_paperless_link(self):
        conn = _FakeConn(
            summaries=[("paperless:119", 0.1)],
            chunks_by_envelope={"paperless:119": [(2, "hit text", 0.34)]},
            envelopes={"paperless:119": {"source": "paperless", "entities": []}},
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "polisa PZU", "cascade", "claude-haiku-4-5"
        )
        assert result["query"] == "polisa PZU"
        assert result["mode"] == "cascade"
        assert result["sol_status"] == "up"
        assert result["embed_backend"] == "solaria"
        assert len(result["results"]) == 1
        hit = result["results"][0]
        assert hit["envelope_id"] == "paperless:119"
        assert hit["source"] == "paperless"
        assert hit["dist"] == 0.34
        assert hit["chunk_index"] == 2
        assert hit["text"] == "hit text"
        assert hit["link"] == "https://paper.kapala.org/documents/119/details"

    async def test_flat_mode_skips_cascade_stage1(self):
        conn = _FakeConn(
            chunks_by_envelope={"paperless:1": [(0, "a", 0.2)]},
            envelopes={"paperless:1": {"source": "paperless", "entities": []}},
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "q", "flat", "claude-haiku-4-5"
        )
        assert result["mode"] == "flat"
        assert len(result["results"]) == 1

    async def test_summary_attached_when_document_summary_row_exists(self):
        conn = _FakeConn(
            summaries=[("paperless:119", 0.1)],
            chunks_by_envelope={"paperless:119": [(2, "hit text", 0.34)]},
            envelopes={"paperless:119": {"source": "paperless", "entities": []}},
            summary_texts={"paperless:119": {"summary": "Polisa OC 2024", "tags": ["ubezpieczenia"]}},
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "polisa PZU", "cascade", "claude-haiku-4-5"
        )
        hit = result["results"][0]
        assert hit["summary"] == "Polisa OC 2024"
        assert hit["summary_tags"] == ["ubezpieczenia"]

    async def test_summary_defaults_to_none_when_no_document_summary_row(self):
        conn = _FakeConn(
            summaries=[("paperless:119", 0.1)],
            chunks_by_envelope={"paperless:119": [(2, "hit text", 0.34)]},
            envelopes={"paperless:119": {"source": "paperless", "entities": []}},
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "polisa PZU", "cascade", "claude-haiku-4-5"
        )
        hit = result["results"][0]
        assert hit["summary"] is None
        assert hit["summary_tags"] == []

    async def test_hybrid_mode_merges_cascade_and_mail_branches(self):
        conn = _FakeConn(
            summaries=[("paperless:1", 0.3)],
            chunks_by_envelope={"paperless:1": [(0, "doc text", 0.3)]},
            envelopes={
                "paperless:1": {"source": "paperless", "entities": []},
                "<msgid@example.com>": {
                    "source": "gmail",
                    "entities": [{"type": "headers", "from": None, "subject": "s", "date_raw": "d"}],
                },
            },
            mail_chunks_by_source={"gmail": [("<msgid@example.com>", 0, "mail text", 0.2)]},
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "q", "hybrid", "claude-haiku-4-5"
        )
        assert result["mode"] == "hybrid"
        envelope_ids = [r["envelope_id"] for r in result["results"]]
        assert envelope_ids == ["<msgid@example.com>", "paperless:1"]

    async def test_gmail_hit_carries_header_metadata_not_a_link(self):
        conn = _FakeConn(
            summaries=[("<msgid@example.com>", 0.1)],
            chunks_by_envelope={"<msgid@example.com>": [(0, "body text", 0.4)]},
            envelopes={
                "<msgid@example.com>": {
                    "source": "gmail",
                    "entities": [
                        {"type": "headers", "from": {"name": "A", "address": "a@b.com"}, "subject": "s", "date_raw": "d"}
                    ],
                }
            },
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "q", "cascade", "claude-haiku-4-5"
        )
        hit = result["results"][0]
        assert hit["source"] == "gmail"
        assert hit["subject"] == "s"
        assert hit["link"] is None
        assert hit["mail_ui_url"] is None


class TestRunSearchEmbedBackendMarking:
    async def test_fallback_embed_marks_backend_piha_and_sol_status_down(self):
        # Task spec: the response must say WHICH backend embedded the query (quality
        # debugging), and sol_status must reflect the router's world view -- the frontend
        # renders "down" as "offline (fallback embed)".
        conn = _FakeConn(
            summaries=[("paperless:119", 0.1)],
            chunks_by_envelope={"paperless:119": [(2, "hit text", 0.34)]},
            envelopes={"paperless:119": {"source": "paperless", "entities": []}},
        )
        result = await run_search(
            conn, _FakeSession(), _FakeRouter(backend="piha"), "q", "cascade", "claude-haiku-4-5"
        )
        assert result["embed_backend"] == "piha"
        assert result["sol_status"] == "down"
        assert len(result["results"]) == 1


class TestRunSearchNoGoodResults:
    async def test_results_above_no_answer_threshold_are_still_returned_unfiltered(self):
        # Plan §7: the 0.55 "no answer" colour threshold is a frontend concern -- the API
        # must not silently drop/hide a poor match, only report its true dist so the caller
        # (UI or eval harness) can apply that policy itself.
        conn = _FakeConn(
            summaries=[("paperless:1", 0.6)],
            chunks_by_envelope={"paperless:1": [(0, "unrelated text", 0.62)]},
            envelopes={"paperless:1": {"source": "paperless", "entities": []}},
        )
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "unrelated query", "cascade", "claude-haiku-4-5"
        )
        assert len(result["results"]) == 1
        assert result["results"][0]["dist"] == 0.62

    async def test_no_summaries_yields_empty_results_not_an_error(self):
        conn = _FakeConn(summaries=[], chunks_by_envelope={}, envelopes={})
        session = _FakeSession()
        result = await run_search(
            conn, session, _FakeRouter(), "nothing matches", "cascade", "claude-haiku-4-5"
        )
        assert result["results"] == []


class _FakePool:
    """asyncpg pool stand-in: `async with pool.acquire() as conn` yields the fake conn."""

    def __init__(self, conn):
        self._conn = conn

    def acquire(self):
        conn = self._conn

        class _Acquire:
            async def __aenter__(self):
                return conn

            async def __aexit__(self, *exc_info):
                return False

        return _Acquire()


def _client_with_fake_state(conn) -> TestClient:
    # Same trick as tests/test_frontend.py: TestClient is NOT entered as a context manager,
    # so `lifespan` (live DSN + Ollama) never runs and we inject app.state ourselves.
    app.state.pool = _FakePool(conn)
    app.state.http = _FakeSession()
    app.state.embed_router = _FakeRouter()
    return TestClient(app)


def _mixed_corpus_conn() -> _FakeConn:
    """One paperless envelope reachable via the summary pre-filter + one gmail envelope
    reachable ONLY through hybrid's summaryless-source branch -- so the mode a request
    actually took is visible in the results, not just in the echoed `mode` field."""
    return _FakeConn(
        summaries=[("paperless:1", 0.3)],
        chunks_by_envelope={"paperless:1": [(0, "doc text", 0.3)]},
        envelopes={
            "paperless:1": {"source": "paperless", "entities": []},
            "<msgid@example.com>": {
                "source": "gmail",
                "entities": [{"type": "headers", "from": None, "subject": "s", "date_raw": "d"}],
            },
        },
        mail_chunks_by_source={"gmail": [("<msgid@example.com>", 0, "mail text", 0.2)]},
    )


class TestSearchEndpointModeDefault:
    """HTTP-level contract of `mode` (kb/phases/kb-m5-faza-mailowa.md DoD (d), bramka §8
    PASS on the full corpus 2026-08-06): no `mode` parameter => hybrid."""

    def test_default_without_mode_param_takes_the_hybrid_path(self):
        client = _client_with_fake_state(_mixed_corpus_conn())
        response = client.get("/search", params={"q": "q"})
        assert response.status_code == 200
        body = response.json()
        assert body["mode"] == "hybrid"
        # The mail envelope is unreachable from the cascade/flat paths in this fixture,
        # so its presence proves the hybrid branch really ran (not just a relabelled cascade).
        assert [r["envelope_id"] for r in body["results"]] == [
            "<msgid@example.com>",
            "paperless:1",
        ]

    def test_explicit_mode_flat_still_takes_the_flat_path(self):
        client = _client_with_fake_state(_mixed_corpus_conn())
        body = client.get("/search", params={"q": "q", "mode": "flat"}).json()
        assert body["mode"] == "flat"
        assert [r["envelope_id"] for r in body["results"]] == ["paperless:1"]

    def test_explicit_mode_cascade_still_takes_the_cascade_path(self):
        client = _client_with_fake_state(_mixed_corpus_conn())
        body = client.get("/search", params={"q": "q", "mode": "cascade"}).json()
        assert body["mode"] == "cascade"
        assert [r["envelope_id"] for r in body["results"]] == ["paperless:1"]

    def test_unknown_mode_is_rejected(self):
        client = _client_with_fake_state(_mixed_corpus_conn())
        assert client.get("/search", params={"q": "q", "mode": "turbo"}).status_code == 422