sanguo_llmwiki/tests/integration/test_mcp_tools.py

"""
Integration tests for MCP Tools

测试 MCP Server 工具的端到端功能。
需要设置 RUN_INTEGRATION=1 环境变量才能运行。
"""

import pytest
import os
import tempfile
import asyncio

# 仅在设置 RUN_INTEGRATION=1 时运行
pytestmark = pytest.mark.skipif(
    not os.environ.get("RUN_INTEGRATION"),
    reason="Integration tests require RUN_INTEGRATION=1"
)

from mcp_server.storage import Database
from mcp_server.services import QueryService, ParserService, IndexerService, GraphService, CacheService
from mcp_server.tools import (
    WikiQueryTool,
    WikiStatusTool,
    WikiLintTool,
    CrossLinkerTool,
    TagTaxonomyTool,
    WikiSynthesizeTool,
    DailyUpdateTool,
    MemoryBridgeTool
)


@pytest.fixture
async def setup_mcp():
    """设置 MCP 测试环境"""
    with tempfile.TemporaryDirectory() as tmpdir:
        # 创建测试 wiki vault
        wiki_path = os.path.join(tmpdir, "wiki-vault")
        os.makedirs(wiki_path)

        # 创建测试页面
        os.makedirs(os.path.join(wiki_path, "practices"), exist_ok=True)
        with open(os.path.join(wiki_path, "practices", "test.md"), 'w') as f:
            f.write("""---
name: test-practice
description: Test practice page
metadata:
  type: practice
  tags: [testing, best-practice]
---
# Test Practice

This is a test practice page.

## Key Points

- Point 1
- Point 2

## Related Links

- [[concepts/test-concept]]
- [[skills/test-skill]]
""")

        # 创建概念页面
        os.makedirs(os.path.join(wiki_path, "concepts"), exist_ok=True)
        with open(os.path.join(wiki_path, "concepts", "test-concept.md"), 'w') as f:
            f.write("""---
name: test-concept
description: Test concept page
metadata:
  type: concept
  tags: [concept]
---
# Test Concept

This is a test concept page.

## Related

- [[practices/test-practice]]
""")

        # 创建数据库
        db_path = os.path.join(tmpdir, "test.db")
        db = Database(db_path)
        await db.connect()

        # 初始化索引
        parser = ParserService()
        indexer = IndexerService(db, parser, wiki_path)
        await indexer.rebuild_index()

        # 创建服务
        cache = CacheService(max_size=100)
        query_service = QueryService(db, cache)
        graph_service = GraphService(db)

        yield {
            "db": db,
            "query_service": query_service,
            "parser": parser,
            "indexer": indexer,
            "graph_service": graph_service,
            "wiki_path": wiki_path
        }

        # 清理
        await db.close()


class TestWikiQueryTool:
    """wiki_query 工具测试"""

    @pytest.mark.asyncio
    async def test_search_by_query(self, setup_mcp):
        """测试搜索功能"""
        query_service = setup_mcp["query_service"]
        tool = WikiQueryTool(query_service)

        result = await tool.handle(query="test")
        assert result["success"] is True
        assert "results" in result

    @pytest.mark.asyncio
    async def test_get_page(self, setup_mcp):
        """测试获取单个页面"""
        query_service = setup_mcp["query_service"]
        tool = WikiQueryTool(query_service)

        result = await tool.handle(path="practices/test.md")
        assert result["success"] is True
        assert result["page"]["path"] == "practices/test.md"
        assert result["page"]["title"] == "Test Practice"


class TestWikiStatusTool:
    """wiki_status 工具测试"""

    @pytest.mark.asyncio
    async def test_get_status(self, setup_mcp):
        """测试获取状态"""
        query_service = setup_mcp["query_service"]
        indexer = setup_mcp["indexer"]
        graph_service = setup_mcp["graph_service"]
        tool = WikiStatusTool(query_service, indexer, graph_service)

        result = await tool.handle()
        assert result["success"] is True
        assert "stats" in result
        assert "index_status" in result


class TestWikiLintTool:
    """wiki_lint 工具测试"""

    @pytest.mark.asyncio
    async def test_lint_all_pages(self, setup_mcp):
        """测试审计所有页面"""
        parser = setup_mcp["parser"]
        query_service = setup_mcp["query_service"]
        graph_service = setup_mcp["graph_service"]
        wiki_path = setup_mcp["wiki_path"]
        tool = WikiLintTool(parser, query_service, graph_service, wiki_path)

        result = await tool.handle(level="basic")
        assert result["success"] is True
        assert "summary" in result

    @pytest.mark.asyncio
    async def test_lint_single_page(self, setup_mcp):
        """测试审计单个页面"""
        parser = setup_mcp["parser"]
        query_service = setup_mcp["query_service"]
        graph_service = setup_mcp["graph_service"]
        wiki_path = setup_mcp["wiki_path"]
        tool = WikiLintTool(parser, query_service, graph_service, wiki_path)

        result = await tool.handle(path="practices/test.md")
        assert result["success"] is True
        assert "issues" in result


class TestCrossLinkerTool:
    """cross_linker 工具测试"""

    @pytest.mark.asyncio
    async def test_find_missing_links(self, setup_mcp):
        """测试查找缺失链接"""
        graph_service = setup_mcp["graph_service"]
        tool = CrossLinkerTool(graph_service)

        result = await tool.handle(action="find_missing")
        assert result["success"] is True
        assert "missing_links" in result


class TestTagTaxonomyTool:
    """tag_taxonomy 工具测试"""

    @pytest.mark.asyncio
    async def test_get_taxonomy(self, setup_mcp):
        """测试获取标签分类"""
        query_service = setup_mcp["query_service"]
        tool = TagTaxonomyTool(query_service)

        result = await tool.handle()
        assert result["success"] is True
        assert "taxonomy" in result


class TestWikiSynthesizeTool:
    """wiki_synthesize 工具测试"""

    @pytest.mark.asyncio
    async def test_synthesize_topic(self, setup_mcp):
        """测试主题综合"""
        query_service = setup_mcp["query_service"]
        graph_service = setup_mcp["graph_service"]
        tool = WikiSynthesizeTool(query_service, graph_service)

        result = await tool.handle(topic="test")
        assert result["success"] is True
        assert "synthesis" in result


class TestDailyUpdateTool:
    """daily_update 工具测试"""

    @pytest.mark.asyncio
    async def test_daily_update(self, setup_mcp):
        """测试日常维护"""
        query_service = setup_mcp["query_service"]
        indexer = setup_mcp["indexer"]
        wiki_path = setup_mcp["wiki_path"]
        tool = DailyUpdateTool(query_service, indexer, wiki_path)

        result = await tool.handle()
        assert result["success"] is True
        assert "updated" in result


class TestMemoryBridgeTool:
    """memory_bridge 工具测试"""

    @pytest.mark.asyncio
    async def test_bridge_retrieval(self, setup_mcp):
        """测试记忆桥接"""
        query_service = setup_mcp["query_service"]
        tool = MemoryBridgeTool(query_service)

        result = await tool.handle(query="test", source_tool="claude")
        assert result["success"] is True
        assert "results" in result


class TestEndToEnd:
    """端到端工作流测试"""

    @pytest.mark.asyncio
    async def test_full_workflow(self, setup_mcp):
        """测试完整工作流：索引 -> 查询 -> 链接"""
        db = setup_mcp["db"]
        parser = setup_mcp["parser"]
        indexer = setup_mcp["indexer"]
        query_service = setup_mcp["query_service"]
        wiki_path = setup_mcp["wiki_path"]

        # 1. 创建新页面
        new_page_path = os.path.join(wiki_path, "practices", "new.md")
        with open(new_page_path, 'w') as f:
            f.write("""---
name: new-practice
description: New practice page
metadata:
  type: practice
  tags: [new, practice]
---
# New Practice

This is a new practice page.

## Links

- [[concepts/test-concept]]
""")

        # 2. 重新索引
        await indexer.incremental_update()

        # 3. 查询新页面
        result = await query_service.get_page("practices/new.md")
        assert result is not None
        assert result.title == "New Practice"

        # 4. 验证链接
        links = await db.get_links("practices/new.md")
        assert "concepts/test-concept" in links