meida_backend/tests/unit/test_retrieval_service.py

"""RetrievalService 编排逻辑：repo 层打桩，不连库（编排正确性与 DB 无关）."""

from __future__ import annotations

from typing import Any

import pytest

from packages.retrieval import repo
from packages.retrieval import service as service_mod
from packages.retrieval.rerank import ScoredChunk
from packages.retrieval.service import RetrievalService, UserContext


class FakeEmbedder:
    """满足 EmbeddingProvider Protocol 的最小桩：记录调用，返回固定向量."""

    model_name = "fake"
    version = "v1"
    dimensions = 4

    def __init__(self) -> None:
        self.calls: list[list[str]] = []

    async def embed(self, texts: list[str]) -> list[list[float]]:
        self.calls.append(list(texts))
        return [[0.25, 0.25, 0.25, 0.25] for _ in texts]


class FakeSession:
    def __init__(self) -> None:
        self.commits = 0
        self.rollbacks = 0

    async def commit(self) -> None:
        self.commits += 1

    async def rollback(self) -> None:
        self.rollbacks += 1


def _scored(chunk_id: int = 1) -> ScoredChunk:
    return ScoredChunk(
        chunk_id=chunk_id,
        entity_id=100,
        chunk_type="text",
        text="chunk",
        vec_score=1.0,
        kw_score=0.0,
        rule_score=0.0,
        freshness_score=1.0,
        final_score=0.55,
    )


@pytest.fixture
def stubbed_repo(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
    """把 repo 的两个 IO 函数换成记录桩，service 逻辑照常执行."""

    calls: dict[str, Any] = {"hybrid_search": None, "gaps": [], "results": []}

    async def fake_hybrid_search(
        session: Any, query_vec: Any, query_text: str, **kwargs: Any
    ) -> list[ScoredChunk]:
        calls["hybrid_search"] = {"query_vec": query_vec, "query_text": query_text, **kwargs}
        return list(calls["results"])

    async def fake_record_gap(session: Any, query_text: str, filter_ctx: Any) -> None:
        calls["gaps"].append((query_text, filter_ctx))

    monkeypatch.setattr(repo, "hybrid_search", fake_hybrid_search)
    monkeypatch.setattr(repo, "record_gap", fake_record_gap)
    return calls


async def test_search_with_hits_returns_results_without_gap(stubbed_repo: dict[str, Any]) -> None:
    stubbed_repo["results"] = [_scored(1), _scored(2)]
    embedder = FakeEmbedder()
    session = FakeSession()
    svc = RetrievalService(session, embedding=embedder)

    outcome = await svc.search("面试穿什么", user_ctx=UserContext(body_shape="apple"))

    assert outcome.degraded is False
    assert [c.chunk_id for c in outcome.results] == [1, 2]
    assert stubbed_repo["gaps"] == []
    assert session.commits == 0  # 有结果的路径不写库，事务留给调用方
    assert embedder.calls == [["面试穿什么"]]  # embed 的输入就是 query 原文


async def test_search_forwards_user_context_to_hybrid(stubbed_repo: dict[str, Any]) -> None:
    stubbed_repo["results"] = [_scored()]
    svc = RetrievalService(FakeSession(), embedding=FakeEmbedder())

    await svc.search(
        "婚礼穿什么",
        user_ctx=UserContext(
            body_shape="pear",
            occasion="wedding",
            entity_types=("item", "style"),
            rule_boost={42: 0.8},
        ),
    )

    seen = stubbed_repo["hybrid_search"]
    assert seen is not None
    assert seen["query_text"] == "婚礼穿什么"
    assert seen["body_shape"] == "pear"
    assert seen["occasion"] == "wedding"
    assert seen["entity_types"] == ("item", "style")
    assert seen["rule_boost"] == {42: 0.8}


async def test_empty_entity_types_is_forwarded_as_none(stubbed_repo: dict[str, Any]) -> None:
    stubbed_repo["results"] = []
    svc = RetrievalService(FakeSession(), embedding=FakeEmbedder())

    await svc.search("随便看看", user_ctx=UserContext())

    seen = stubbed_repo["hybrid_search"]
    assert seen is not None
    assert seen["entity_types"] is None  # 空元组 = 不限类型，不是匹配空集合


async def test_no_results_records_gap_and_degrades(stubbed_repo: dict[str, Any]) -> None:
    stubbed_repo["results"] = []
    session = FakeSession()
    svc = RetrievalService(session, embedding=FakeEmbedder())

    outcome = await svc.search(
        "不存在的知识点", user_ctx=UserContext(body_shape="apple", entity_types=("item",))
    )

    assert outcome.degraded is True
    assert outcome.results == []
    assert session.commits == 1  # gap 埋点立即提交：检索本身只读，别拖住调用方事务
    assert stubbed_repo["gaps"] == [
        (
            "不存在的知识点",
            {"body_shape": "apple", "occasion": None, "entity_types": ["item"]},
        )
    ]


async def test_gap_failure_degrades_but_does_not_raise(
    stubbed_repo: dict[str, Any], monkeypatch: pytest.MonkeyPatch
) -> None:
    """埋点失败不该把"检索空结果"变成异常 —— 空结果本身是合法结局."""

    async def broken_gap(session: Any, query_text: str, filter_ctx: Any) -> None:
        raise RuntimeError("db down")

    monkeypatch.setattr(repo, "record_gap", broken_gap)
    stubbed_repo["results"] = []
    session = FakeSession()
    svc = RetrievalService(session, embedding=FakeEmbedder())

    outcome = await svc.search("anything")

    assert outcome.degraded is True
    assert session.rollbacks == 1


async def test_default_context_when_user_ctx_omitted(stubbed_repo: dict[str, Any]) -> None:
    stubbed_repo["results"] = []
    svc = RetrievalService(FakeSession(), embedding=FakeEmbedder())

    outcome = await svc.search("no ctx")

    assert outcome.degraded is True
    assert stubbed_repo["gaps"] == [
        ("no ctx", {"body_shape": None, "occasion": None, "entity_types": []})
    ]


def test_service_module_reexports_for_dependency_injection() -> None:
    """service 与 repo 共享同一模块对象：monkeypatch 打桩路径就是生产调用路径."""

    assert service_mod.repo is repo