"""
測試 JSON 儲存層 - 資料持久化與檔案操作
這些測試驗證：
1. JSON 檔案的讀寫操作
2. 資料完整性與事務性
3. 錯誤處理與復原機制
4. 併發存取控制
"""
import pytest
import json
import tempfile
import os
from pathlib import Path
from decimal import Decimal
from datetime import date, datetime
from typing import List, Dict, Any
from unittest.mock import patch, mock_open
# 這些匯入會失敗，因為我們還沒有實作儲存層
from accounting_mcp.storage import JSONStorage, StorageError
from accounting_mcp.models import Transaction, Account, Category, TransactionType
class TestJSONStorage:
"""測試 JSON 儲存基類"""
def test_storage_initialization_with_existing_file(self, tmp_path: Path) -> None:
"""測試使用已存在檔案初始化儲存"""
data_file = tmp_path / "test_data.json"
initial_data = {"test": "data"}
# 建立初始檔案
with open(data_file, 'w') as f:
json.dump(initial_data, f)
storage = JSONStorage(data_file)
assert storage.file_path == data_file
assert storage.load_data() == initial_data
def test_storage_initialization_with_new_file(self, tmp_path: Path) -> None:
"""測試使用不存在檔案初始化儲存（應建立空檔案）"""
data_file = tmp_path / "new_data.json"
storage = JSONStorage(data_file, default_data={"empty": True})
assert storage.file_path == data_file
assert data_file.exists()
assert storage.load_data() == {"empty": True}
def test_storage_save_and_load(self, tmp_path: Path) -> None:
"""測試資料儲存與載入"""
data_file = tmp_path / "test_data.json"
storage = JSONStorage(data_file)
test_data = {"key": "value", "number": 42}
storage.save_data(test_data)
loaded_data = storage.load_data()
assert loaded_data == test_data
def test_storage_atomic_write(self, tmp_path: Path) -> None:
"""測試原子寫入（避免寫入中斷導致檔案損壞）"""
data_file = tmp_path / "test_data.json"
storage = JSONStorage(data_file)
# 初始資料
storage.save_data({"version": 1})
# 模擬寫入過程中出錯
with patch('builtins.open', side_effect=IOError("磁碟已滿")):
with pytest.raises(StorageError):
storage.save_data({"version": 2})
# 原始資料應該保持不變
loaded_data = storage.load_data()
assert loaded_data == {"version": 1}
def test_storage_backup_on_corruption(self, tmp_path: Path) -> None:
"""測試檔案損壞時的備份復原"""
data_file = tmp_path / "test_data.json"
backup_file = tmp_path / "test_data_backup.json"
# 建立有效的初始資料
storage = JSONStorage(data_file)
storage.save_data({"valid": "data"})
# 手動建立備份
storage._create_backup()
assert backup_file.exists()
# 模擬檔案損壞
with open(data_file, 'w') as f:
f.write("invalid json {")
# 載入時應該從備份復原
loaded_data = storage.load_data()
assert loaded_data == {"valid": "data"}
def test_storage_permission_error_handling(self, tmp_path: Path) -> None:
"""測試檔案權限錯誤處理"""
data_file = tmp_path / "readonly_data.json"
storage = JSONStorage(data_file)
# 建立唯讀檔案
storage.save_data({"test": "data"})
os.chmod(data_file, 0o444) # 唯讀權限
# 嘗試寫入應該拋出權限錯誤
with pytest.raises(StorageError, match="權限"):
storage.save_data({"new": "data"})
# 還原權限以便清理
os.chmod(data_file, 0o644)
class TestTransactionStorage:
"""測試交易記錄儲存"""
def test_save_and_load_transactions(self, tmp_path: Path) -> None:
"""測試儲存與載入交易記錄"""
from accounting_mcp.storage import TransactionStorage
storage = TransactionStorage(tmp_path / "transactions.json")
transactions = [
Transaction(
amount=Decimal("-50.00"),
category="food",
description="午餐"
),
Transaction(
amount=Decimal("1000.00"),
category="income",
description="工資"
)
]
# 儲存交易
storage.save_transactions(transactions)
# 載入交易
loaded_transactions = storage.load_transactions()
assert len(loaded_transactions) == 2
# 按時間倒序，收入交易（後建立）應該在前面
assert loaded_transactions[0].amount == Decimal("1000.00")
assert loaded_transactions[0].category == "income"
assert loaded_transactions[1].amount == Decimal("-50.00")
assert loaded_transactions[1].category == "food"
def test_add_single_transaction(self, tmp_path: Path) -> None:
"""測試新增單筆交易"""
from accounting_mcp.storage import TransactionStorage
storage = TransactionStorage(tmp_path / "transactions.json")
transaction = Transaction(
amount=Decimal("-30.00"),
category="transport",
description="地鐵"
)
# 新增交易
storage.add_transaction(transaction)
# 驗證交易被儲存
transactions = storage.load_transactions()
assert len(transactions) == 1
assert transactions[0].amount == Decimal("-30.00")
assert transactions[0].category == "transport"
def test_transaction_filtering(self, tmp_path: Path) -> None:
"""測試交易篩選功能"""
from accounting_mcp.storage import TransactionStorage
storage = TransactionStorage(tmp_path / "transactions.json")
# 新增多筆交易
transactions = [
Transaction(amount=Decimal("-50.00"), category="food", date=date(2025, 1, 15)),
Transaction(amount=Decimal("-20.00"), category="transport", date=date(2025, 1, 16)),
Transaction(amount=Decimal("-30.00"), category="food", date=date(2025, 1, 17)),
]
for t in transactions:
storage.add_transaction(t)
# 按分類篩選
food_transactions = storage.get_transactions_by_category("food")
assert len(food_transactions) == 2
assert all(t.category == "food" for t in food_transactions)
# 按日期範圍篩選
date_filtered = storage.get_transactions_by_date_range(
start_date=date(2025, 1, 16),
end_date=date(2025, 1, 17)
)
assert len(date_filtered) == 2
def test_transaction_pagination(self, tmp_path: Path) -> None:
"""測試交易分頁功能"""
from accounting_mcp.storage import TransactionStorage
storage = TransactionStorage(tmp_path / "transactions.json")
# 新增多筆交易
for i in range(25):
storage.add_transaction(Transaction(
amount=Decimal(f"-{i+1}.00"),
category="food",
description=f"交易{i+1}"
))
# 測試分頁
page1 = storage.get_transactions(limit=10, offset=0)
page2 = storage.get_transactions(limit=10, offset=10)
page3 = storage.get_transactions(limit=10, offset=20)
assert len(page1) == 10
assert len(page2) == 10
assert len(page3) == 5
# 驗證順序（最新的在前）
assert page1[0].description == "交易25"
assert page1[-1].description == "交易16"
class TestAccountStorage:
"""測試帳戶資訊儲存"""
def test_save_and_load_account(self, tmp_path: Path) -> None:
"""測試儲存與載入帳戶資訊"""
from accounting_mcp.storage import AccountStorage
storage = AccountStorage(tmp_path / "account.json")
account = Account()
account.add_transaction(Transaction(
amount=Decimal("-50.00"),
category="food"
))
# 儲存帳戶
storage.save_account(account)
# 載入帳戶
loaded_account = storage.load_account()
assert loaded_account.balance == Decimal("-50.00")
assert loaded_account.total_transactions == 1
def test_update_account_balance(self, tmp_path: Path) -> None:
"""測試更新帳戶餘額"""
from accounting_mcp.storage import AccountStorage
storage = AccountStorage(tmp_path / "account.json")
# 初始帳戶
account = Account()
storage.save_account(account)
# 更新餘額
storage.update_balance(Decimal("100.00"), 1)
# 驗證更新
updated_account = storage.load_account()
assert updated_account.balance == Decimal("100.00")
assert updated_account.total_transactions == 1
class TestCategoryStorage:
"""測試分類資訊儲存"""
def test_load_default_categories(self, tmp_path: Path) -> None:
"""測試載入預設分類"""
from accounting_mcp.storage import CategoryStorage
storage = CategoryStorage(tmp_path / "categories.json")
categories = storage.load_categories()
# 應該有預設分類
assert len(categories) > 0
# 檢查必要的分類
category_ids = [c.id for c in categories]
assert "food" in category_ids
assert "transport" in category_ids
assert "income" in category_ids
def test_validate_category(self, tmp_path: Path) -> None:
"""測試分類驗證"""
from accounting_mcp.storage import CategoryStorage
storage = CategoryStorage(tmp_path / "categories.json")
# 有效分類
assert storage.is_valid_category("food") == True
# 無效分類
assert storage.is_valid_category("invalid_category") == False
assert storage.is_valid_category("") == False
# 整合測試
class TestStorageIntegration:
"""儲存層整合測試"""
def test_complete_workflow(self, tmp_path: Path) -> None:
"""測試完整的資料流程"""
from accounting_mcp.storage import TransactionStorage, AccountStorage
trans_storage = TransactionStorage(tmp_path / "transactions.json")
acc_storage = AccountStorage(tmp_path / "account.json")
# 1. 建立帳戶
account = Account()
acc_storage.save_account(account)
# 2. 新增交易
transaction = Transaction(
amount=Decimal("-50.00"),
category="food",
description="午餐"
)
trans_storage.add_transaction(transaction)
# 3. 更新帳戶
account.add_transaction(transaction)
acc_storage.save_account(account)
# 4. 驗證資料一致性
loaded_transactions = trans_storage.load_transactions()
loaded_account = acc_storage.load_account()
assert len(loaded_transactions) == 1
assert loaded_account.balance == Decimal("-50.00")
assert loaded_account.total_transactions == 1
def test_concurrent_access_simulation(self, tmp_path: Path) -> None:
"""測試併發存取模擬"""
from accounting_mcp.storage import TransactionStorage
import threading
import time
storage = TransactionStorage(tmp_path / "transactions.json")
results = []
def add_transaction(i):
max_retries = 3
for attempt in range(max_retries):
try:
transaction = Transaction(
amount=Decimal(f"-{i}.00"),
category="test",
description=f"併發測試{i}"
)
storage.add_transaction(transaction)
results.append(f"success-{i}")
return
except Exception as e:
if attempt == max_retries - 1:
results.append(f"error-{i}: {e}")
else:
# 短暫等待後重試
time.sleep(0.01 * (attempt + 1))
# 建立多個執行緒同時寫入
threads = []
for i in range(5):
thread = threading.Thread(target=add_transaction, args=(i,))
threads.append(thread)
for thread in threads:
thread.start()
for thread in threads:
thread.join()
# 驗證大部分操作都成功（允許偶爾的併發衝突）
success_count = len([r for r in results if r.startswith("success")])
assert success_count >= 4 # 至少 4 個成功，允許 1 個失敗
# 驗證成功的交易都被儲存
transactions = storage.load_transactions()
assert len(transactions) == success_count
# 測試工具函式
@pytest.fixture
def sample_transaction() -> Transaction:
"""提供範例交易用於測試"""
return Transaction(
amount=Decimal("-50.00"),
category="food",
description="測試交易"
)
@pytest.fixture
def sample_account() -> Account:
"""提供範例帳戶用於測試"""
account = Account()
account.add_transaction(Transaction(
amount=Decimal("-50.00"),
category="food"
))
return account
