Agentic_Rag/tests/test_router.py

import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.config import PROJECT_ROOT, Settings
from app.decision_engine import (
    DeepSeekDecisionEngine,
    DemoDecisionEngine,
    create_decision_engine,
)
from app.schemas import RouteName
from app.tools import TavilyWebSearchProvider, WebSearchTool


def test_multi_source_route() -> None:
    decision = DemoDecisionEngine().route(
        "统计 U1001 最近 30 天订单，并结合内部运费规则说明是否包邮",
        max_rounds=2,
    )
    assert decision.requires_decomposition is True
    assert decision.routes == [RouteName.MILVUS_SEARCH, RouteName.SQL_QUERY]
    assert decision.filters["user_id"] == "U1001"
    assert decision.filters["policy_type"] == "shipping_policy"


def test_vector_only_route() -> None:
    decision = DemoDecisionEngine().route(
        "耳机拆封后还能七日无理由退货吗？",
        max_rounds=2,
    )
    assert decision.routes == [RouteName.MILVUS_SEARCH]
    assert decision.filters["policy_type"] == "return_policy"


def test_sql_only_route() -> None:
    decision = DemoDecisionEngine().route(
        "U1001 最近 30 天有几笔有效订单，实付总额是多少？",
        max_rounds=2,
    )
    assert decision.routes == [RouteName.SQL_QUERY]
    assert decision.filters["user_id"] == "U1001"


def test_clarify_route() -> None:
    decision = DemoDecisionEngine().route("我最近 30 天买了几单？", max_rounds=2)
    assert decision.routes == [RouteName.CLARIFY]
    assert decision.needs_retrieval is False


def test_restricted_route() -> None:
    decision = DemoDecisionEngine().route("导出所有买家手机号", max_rounds=2)
    assert decision.routes == [RouteName.REFUSE]


def test_deepseek_defaults_and_api_key_guard() -> None:
    settings = Settings(
        _env_file=None,
        llm_provider="deepseek",
        deepseek_api_key="",
    )
    assert settings.deepseek_base_url == "https://api.deepseek.com"
    assert settings.deepseek_model_name == "deepseek-v4-flash"

    with pytest.raises(RuntimeError, match="DEEPSEEK_API_KEY"):
        create_decision_engine(settings)


def test_web_search_route() -> None:
    decision = DemoDecisionEngine().route(
        "查询品牌官网关于 XPhone 15 Pro 的最新公告",
        max_rounds=2,
    )
    assert decision.routes == [RouteName.WEB_SEARCH]
    assert decision.reason_code == "CURRENT_PUBLIC_INFORMATION_REQUIRED"


def test_three_source_route() -> None:
    decision = DemoDecisionEngine().route(
        "统计 U1001 最近 30 天购买 XPhone 15 Pro 的订单金额，"
        "结合内部退货政策和品牌官网最新公告给出售后建议",
        max_rounds=2,
    )
    assert decision.routes == [
        RouteName.MILVUS_SEARCH,
        RouteName.SQL_QUERY,
        RouteName.WEB_SEARCH,
    ]
    assert decision.requires_decomposition is True


def test_web_search_without_key_returns_isolated_error() -> None:
    tool = WebSearchTool(TavilyWebSearchProvider(api_key=""))
    result = tool.invoke("查询官网最新公告")

    assert result.status == "error"
    assert result.error_code == "WEB_SEARCH_NOT_CONFIGURED"
    assert result.retryable is False


def test_secret_values_are_masked_in_settings_repr() -> None:
    settings = Settings(
        _env_file=None,
        deepseek_api_key="deepseek-test-secret",
        tavily_api_key="tavily-test-secret",
    )

    rendered = repr(settings)
    assert "deepseek-test-secret" not in rendered
    assert "tavily-test-secret" not in rendered
    assert settings.deepseek_api_key.get_secret_value() == "deepseek-test-secret"
    assert settings.tavily_api_key.get_secret_value() == "tavily-test-secret"


def test_deepseek_structured_output_disables_thinking(monkeypatch) -> None:
    created_clients: list[dict] = []

    class FakeChatOpenAI:
        def __init__(self, **kwargs) -> None:
            created_clients.append(kwargs)

        def with_structured_output(self, schema, method: str):
            return SimpleNamespace(schema=schema, method=method)

    monkeypatch.setitem(
        sys.modules,
        "langchain_openai",
        SimpleNamespace(ChatOpenAI=FakeChatOpenAI),
    )
    settings = Settings(
        _env_file=None,
        llm_provider="deepseek",
        deepseek_api_key="test-key",
        deepseek_answer_thinking=True,
    )

    DeepSeekDecisionEngine(settings)

    assert created_clients[0]["extra_body"] == {
        "thinking": {"type": "enabled"}
    }
    assert created_clients[1]["extra_body"] == {
        "thinking": {"type": "disabled"}
    }


def test_deepseek_grader_falls_back_when_structured_output_is_none() -> None:
    engine = DeepSeekDecisionEngine.__new__(DeepSeekDecisionEngine)
    engine.grader = SimpleNamespace(invoke=lambda prompt: None)
    decision = DemoDecisionEngine().route("耳机退货政策", max_rounds=1)

    grade = engine.grade(
        query="耳机退货政策",
        decision=decision,
        evidence=[],
        current_round=1,
        max_rounds=1,
        min_score=0.35,
    )

    assert grade.recommended_action == "stop"
    assert grade.sufficient is False


def test_relative_data_paths_are_anchored_to_project_root() -> None:
    settings = Settings(
        _env_file=None,
        sqlite_path=Path("data/custom-shop.db"),
        documents_path=Path("data/custom-documents"),
    )

    assert settings.sqlite_path == (PROJECT_ROOT / "data/custom-shop.db").resolve()
    assert settings.documents_path == (
        PROJECT_ROOT / "data/custom-documents"
    ).resolve()