130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
"""M8 MCP 服务测试。
|
|
|
|
验证工具存在 + 格式化逻辑 + 降级行为,不依赖真实嵌入/检索。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
from mcp_server.tools import (
|
|
_fmt_results,
|
|
mcp,
|
|
search_company_news,
|
|
search_news,
|
|
search_sentiment_trend,
|
|
search_stock_events,
|
|
)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 工具存在性
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_mcp_server_has_name() -> None:
|
|
assert mcp.name == "A股DeepResearch"
|
|
|
|
|
|
def test_all_five_tools_registered() -> None:
|
|
tool_names = [getattr(t, "name", "") for t in mcp._tool_manager._tools.values()] # type: ignore[union-attr]
|
|
expected = {
|
|
"search_news", "search_company_news", "search_industry_news",
|
|
"search_stock_events", "search_sentiment_trend",
|
|
}
|
|
assert set(tool_names) == expected
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _fmt_results
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _hit(title: str = "测试标题", source: str = "cls", score: float = 0.9,
|
|
sentiment: str = "positive", stock_codes: list[str] | None = None,
|
|
company_names: list[str] | None = None, summary: str = "摘要",
|
|
publish_time: str = "2026-06-16T10:00:00",
|
|
url: str = "https://example.com/1") -> dict:
|
|
return {
|
|
"title": title, "url": url, "source": source, "score": score,
|
|
"publish_time": publish_time,
|
|
"event": {
|
|
"sentiment": sentiment, "importance": 4, "event_type": "重大合同",
|
|
"stock_codes": stock_codes or [], "company_names": company_names or [],
|
|
"industries": ["动力电池"], "summary": summary,
|
|
},
|
|
}
|
|
|
|
|
|
def test_fmt_results_contains_title_and_source() -> None:
|
|
hits = [_hit("宁德时代签百亿合同", "cls")]
|
|
out = _fmt_results(hits, "宁德时代")
|
|
assert "百亿合同" in out
|
|
assert "cls" in out
|
|
assert "0.9" in out
|
|
|
|
|
|
def test_fmt_results_includes_event_fields() -> None:
|
|
hits = [_hit(
|
|
company_names=["宁德时代"], stock_codes=["300750"],
|
|
)]
|
|
out = _fmt_results(hits, "查询")
|
|
assert "300750" in out
|
|
assert "宁德时代" in out
|
|
assert "动力电池" in out
|
|
assert "摘要" in out
|
|
|
|
|
|
def test_fmt_results_empty_returns_hint() -> None:
|
|
out = _fmt_results([], "无结果")
|
|
assert "未找到" in out and "无结果" in out
|
|
|
|
|
|
def test_fmt_results_multiple_hits() -> None:
|
|
hits = [_hit(f"测试{i}") for i in range(3)]
|
|
out = _fmt_results(hits, "查询")
|
|
assert "共 3 条" in out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 工具:降级行为(无精确命中时走纯语义)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@patch("mcp_server.tools._search")
|
|
def test_search_company_news_falls_back_on_empty(mock_search) -> None:
|
|
"""company 精确命中 0 条时,降级为无 filter 纯语义搜索。"""
|
|
mock_search.side_effect = [
|
|
[], # 第一次:精确匹配 0 条
|
|
[_hit("fallback")], # 降级: 纯语义
|
|
]
|
|
out = search_company_news("查询", company="不存在的公司")
|
|
assert "fallback" in out
|
|
|
|
|
|
@patch("mcp_server.tools._search")
|
|
def test_search_stock_events_normalizes_code(mock_search) -> None:
|
|
"""stock_code 应去掉后缀,统一大写。"""
|
|
mock_search.return_value = [_hit("结果")]
|
|
out = search_stock_events("查询", stock_code="300750.SZ")
|
|
mock_search.assert_called() # code 应为 '300750'
|
|
assert "结果" in out
|
|
|
|
|
|
@patch("mcp_server.tools._search")
|
|
def test_search_sentiment_trend_includes_stats(mock_search) -> None:
|
|
mock_search.return_value = [
|
|
_hit("a", sentiment="positive"),
|
|
_hit("b", sentiment="positive"),
|
|
_hit("c", sentiment="negative"),
|
|
_hit("d", sentiment="neutral"),
|
|
]
|
|
out = search_sentiment_trend("查询", sentiment="all", top_k=10)
|
|
assert "利好 2" in out
|
|
assert "利空 1" in out
|
|
assert "中性 1" in out
|
|
assert "共 4 条" in out
|
|
|
|
|
|
@patch("mcp_server.tools._search")
|
|
def test_search_news_passthrough(mock_search) -> None:
|
|
mock_search.return_value = [_hit("结果")]
|
|
out = search_news("查询")
|
|
assert "结果" in out
|