Accounting MCP Server

""" 集成测试 - 测试完整的 MCP 服务器功能 这些测试验证： 1. MCP 服务器启动和响应 2. 工具调用的端到端流程 3. 资源访问的完整性 4. 错误处理机制 """ import pytest import pytest_asyncio import asyncio import json import subprocess import sys from pathlib import Path import tempfile import shutil from decimal import Decimal from accounting_mcp.server import AccountingMCPServer from accounting_mcp.models import Transaction from accounting_mcp.storage import StorageManager class TestMCPServerIntegration: """MCP 服务器集成测试""" @pytest_asyncio.fixture async def temp_server(self): """创建临时 MCP 服务器用于测试""" with tempfile.TemporaryDirectory() as temp_dir: server = AccountingMCPServer(data_dir=temp_dir) yield server @pytest.mark.asyncio async def test_server_initialization(self, temp_server): """测试服务器初始化""" assert temp_server.server is not None assert temp_server.storage is not None assert temp_server.tools is not None assert temp_server.resources is not None @pytest.mark.asyncio async def test_tool_add_transaction(self, temp_server): """测试添加交易工具""" result = await temp_server.tools.add_transaction( amount=-50.00, category="food", description="测试午餐" ) assert result["success"] == True assert result["transaction"]["amount"] == -50.00 assert result["transaction"]["category"] == "food" assert result["new_balance"] == -50.00 @pytest.mark.asyncio async def test_tool_get_balance(self, temp_server): """测试获取余额工具""" # 先添加一些交易 await temp_server.tools.add_transaction(-30, "food", "早餐") await temp_server.tools.add_transaction(1000, "income", "工资") result = await temp_server.tools.get_balance(detailed=True) assert result["success"] == True assert result["balance"] == 970.00 assert result["total_transactions"] == 2 assert "monthly_stats" in result @pytest.mark.asyncio async def test_tool_list_transactions(self, temp_server): """测试查询交易列表工具""" # 添加多笔交易 await temp_server.tools.add_transaction(-50, "food", "午餐") await temp_server.tools.add_transaction(-20, "transport", "地铁") await temp_server.tools.add_transaction(500, "income", "奖金") result = await temp_server.tools.list_transactions(limit=10) assert result["success"] == True assert len(result["transactions"]) == 3 assert result["pagination"]["total_count"] == 3 @pytest.mark.asyncio async def test_tool_monthly_summary(self, temp_server): """测试月度汇总工具""" # 添加交易 await temp_server.tools.add_transaction(-100, "food", "餐饮") await temp_server.tools.add_transaction(-50, "transport", "交通") await temp_server.tools.add_transaction(1000, "income", "收入") from datetime import datetime now = datetime.now() result = await temp_server.tools.get_monthly_summary( year=now.year, month=now.month ) assert result["success"] == True assert result["summary"]["totals"]["income"] == 1000.0 assert result["summary"]["totals"]["expense"] == 150.0 assert result["summary"]["totals"]["net_flow"] == 850.0 assert result["summary"]["transaction_count"] == 3 @pytest.mark.asyncio async def test_resource_transactions(self, temp_server): """测试交易记录资源""" # 添加一些交易 await temp_server.tools.add_transaction(-30, "food", "早餐") await temp_server.tools.add_transaction(-15, "transport", "公交") result = await temp_server.resources.get_resource("transactions://all") assert result["success"] == True assert result["mimeType"] == "application/json" # 解析内容 content = json.loads(result["content"]) assert "transactions" in content assert len(content["transactions"]) == 2 assert "metadata" in content @pytest.mark.asyncio async def test_resource_categories(self, temp_server): """测试分类资源""" result = await temp_server.resources.get_resource("categories://list") assert result["success"] == True assert result["mimeType"] == "application/json" # 解析内容 content = json.loads(result["content"]) assert "categories" in content assert "by_type" in content["categories"] assert len(content["categories"]["all"]) > 0 @pytest.mark.asyncio async def test_resource_summary(self, temp_server): """测试汇总资源""" # 添加交易 await temp_server.tools.add_transaction(100, "income", "收入") result = await temp_server.resources.get_resource("summary://current") assert result["success"] == True assert result["mimeType"] == "application/json" # 解析内容 content = json.loads(result["content"]) assert "account_summary" in content assert "monthly_summary" in content assert content["account_summary"]["balance"] == 100.0 @pytest.mark.asyncio async def test_error_handling(self, temp_server): """测试错误处理""" # 无效金额 result = await temp_server.tools.add_transaction( amount=0, # 无效金额 category="food" ) assert result["success"] == False assert "error" in result # 无效分类 result = await temp_server.tools.add_transaction( amount=-50, category="invalid_category" ) assert result["success"] == False # 无效资源 result = await temp_server.resources.get_resource("invalid://resource") assert result["success"] == False class TestCommandLineInterface: """命令行接口测试""" @pytest.fixture def temp_project_dir(self): """创建临时项目目录""" # 复制项目文件到临时目录进行测试 with tempfile.TemporaryDirectory() as temp_dir: project_dir = Path(temp_dir) / "accounting-mcp-demo" # 复制必要文件 current_dir = Path(__file__).parent.parent shutil.copytree(current_dir / "accounting_mcp", project_dir / "accounting_mcp") shutil.copy(current_dir / "test_client.py", project_dir / "test_client.py") yield project_dir def test_server_help(self, temp_project_dir): """测试服务器帮助信息""" result = subprocess.run( [sys.executable, "-m", "accounting_mcp.server", "--help"], cwd=temp_project_dir, capture_output=True, text=True ) assert result.returncode == 0 assert "记账 MCP 服务器" in result.stdout assert "--data-dir" in result.stdout assert "--log-level" in result.stdout # 性能测试 class TestPerformance: """性能测试""" @pytest.mark.asyncio async def test_bulk_transactions_performance(self): """测试批量交易性能""" import time with tempfile.TemporaryDirectory() as temp_dir: server = AccountingMCPServer(data_dir=temp_dir) # 添加100笔交易，测试性能 start_time = time.time() for i in range(100): await server.tools.add_transaction( amount=-(i+1), category="food", description=f"测试交易 {i+1}" ) end_time = time.time() duration = end_time - start_time # 100笔交易应该在合理时间内完成 assert duration < 10.0 # 不超过10秒 # 验证数据完整性 balance_result = await server.tools.get_balance() expected_balance = -sum(range(1, 101)) # -5050 assert balance_result["balance"] == expected_balance # 验证查询性能 start_time = time.time() list_result = await server.tools.list_transactions(limit=50) end_time = time.time() assert end_time - start_time < 1.0 # 查询应该在1秒内完成 assert len(list_result["transactions"]) == 50 # 数据一致性测试 class TestDataConsistency: """数据一致性测试""" @pytest.mark.asyncio async def test_transaction_account_consistency(self): """测试交易和账户数据一致性""" with tempfile.TemporaryDirectory() as temp_dir: server = AccountingMCPServer(data_dir=temp_dir) # 添加一系列交易 transactions = [ (1000, "income", "工资"), (-200, "food", "餐饮"), (-50, "transport", "交通"), (-30, "entertainment", "电影"), (100, "income", "奖金") ] expected_balance = Decimal("0") for amount, category, desc in transactions: await server.tools.add_transaction(amount, category, desc) expected_balance += Decimal(str(amount)) # 验证余额一致性 balance_result = await server.tools.get_balance() assert Decimal(str(balance_result["balance"])) == expected_balance assert balance_result["total_transactions"] == len(transactions) # 验证交易记录一致性 list_result = await server.tools.list_transactions(limit=10) assert len(list_result["transactions"]) == len(transactions) # 验证月度汇总一致性 summary_result = await server.tools.get_monthly_summary() assert summary_result["summary"]["transaction_count"] == len(transactions) assert summary_result["summary"]["totals"]["net_flow"] == float(expected_balance)