"""
整合測試 - 測試完整的 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)
