Coverage for tests/test_persistence.py: 96%
101 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-23 23:00 +0800
1"""測試持久化功能"""
3import os
4import tempfile
5import pytest
6from dataclasses import dataclass
7from autocrud import AutoCRUD, DiskStorage, SerializerFactory
10@dataclass
11class Product:
12 name: str
13 description: str
14 price: float
15 category: str
18class TestDiskPersistence:
19 """測試磁碟存儲持久化功能"""
21 def test_disk_persistence_basic(self, temp_dir):
22 """測試基本持久化功能"""
23 # 第一階段：創建數據並保存
24 storage1 = DiskStorage(temp_dir)
25 crud1 = AutoCRUD(model=Product, storage=storage1, resource_name="products")
27 # 創建一些產品
28 products_data = [
29 {"name": "筆記本電腦", "description": "高性能筆記本", "price": 25000.0, "category": "電子產品"},
30 {"name": "無線滑鼠", "description": "高精度無線滑鼠", "price": 800.0, "category": "電子產品"},
31 {"name": "咖啡豆", "description": "精品咖啡豆", "price": 450.0, "category": "食品"},
32 ]
34 created_products = []
35 for product_data in products_data:
36 product = crud1.create(product_data)
37 created_products.append(product)
39 assert len(created_products) == 3
41 # 第二階段：重新載入數據
42 storage2 = DiskStorage(temp_dir)
43 crud2 = AutoCRUD(model=Product, storage=storage2, resource_name="products")
45 all_products = crud2.list_all()
46 assert len(all_products) == 3
48 # 驗證產品內容
49 for product_id, product in all_products.items():
50 assert 'name' in product
51 assert 'description' in product
52 assert 'price' in product
53 assert 'category' in product
55 def test_disk_persistence_updates(self, temp_dir):
56 """測試持久化更新操作"""
57 # 創建初始數據
58 storage1 = DiskStorage(temp_dir)
59 crud1 = AutoCRUD(model=Product, storage=storage1, resource_name="products")
61 product = crud1.create({
62 "name": "筆記本電腦",
63 "description": "普通筆記本",
64 "price": 25000.0,
65 "category": "電子產品"
66 })
67 product_id = product["id"]
69 # 更新數據
70 updated_product = crud1.update(product_id, {
71 "name": "高端筆記本電腦",
72 "description": "高端遊戲筆記本",
73 "price": 30000.0,
74 "category": "電子產品"
75 })
77 assert updated_product["name"] == "高端筆記本電腦"
78 assert updated_product["price"] == 30000.0
80 # 重新載入驗證更新
81 storage2 = DiskStorage(temp_dir)
82 crud2 = AutoCRUD(model=Product, storage=storage2, resource_name="products")
84 loaded_product = crud2.get(product_id)
85 assert loaded_product["name"] == "高端筆記本電腦"
86 assert loaded_product["price"] == 30000.0
88 def test_disk_persistence_deletions(self, temp_dir):
89 """測試持久化刪除操作"""
90 # 創建初始數據
91 storage1 = DiskStorage(temp_dir)
92 crud1 = AutoCRUD(model=Product, storage=storage1, resource_name="products")
94 product1 = crud1.create({"name": "產品1", "description": "測試產品1", "price": 100.0, "category": "測試"})
95 product2 = crud1.create({"name": "產品2", "description": "測試產品2", "price": 200.0, "category": "測試"})
97 # 刪除一個產品
98 deleted = crud1.delete(product1["id"])
99 assert deleted is True
101 # 驗證只剩一個產品
102 all_products = crud1.list_all()
103 assert len(all_products) == 1
105 # 重新載入驗證刪除
106 storage2 = DiskStorage(temp_dir)
107 crud2 = AutoCRUD(model=Product, storage=storage2, resource_name="products")
109 loaded_products = crud2.list_all()
110 assert len(loaded_products) == 1
112 remaining_product = list(loaded_products.values())[0]
113 assert remaining_product["name"] == "產品2"
116class TestSerializerPersistence:
117 """測試不同序列化器的持久化"""
119 @pytest.mark.parametrize("serializer_type", ["json", "pickle", "msgpack"])
120 def test_different_serializers_persistence(self, temp_dir, serializer_type):
121 """測試不同序列化器的持久化"""
122 test_data = {"name": "測試產品", "description": "測試描述", "price": 100.0, "category": "測試"}
124 try:
125 # 為每個序列化器使用不同的目錄
126 storage_dir = os.path.join(temp_dir, serializer_type)
127 os.makedirs(storage_dir, exist_ok=True)
129 # 創建特定序列化器的存儲
130 serializer = SerializerFactory.create(serializer_type)
131 storage = DiskStorage(storage_dir, serializer=serializer)
132 crud = AutoCRUD(
133 model=Product, storage=storage, resource_name="products"
134 )
136 # 創建數據
137 product = crud.create(test_data)
138 assert product["name"] == test_data["name"]
139 assert product["price"] == test_data["price"]
141 # 重新載入
142 storage2 = DiskStorage(storage_dir, serializer=serializer)
143 crud2 = AutoCRUD(
144 model=Product, storage=storage2, resource_name="products"
145 )
147 loaded_products = crud2.list_all()
148 assert len(loaded_products) == 1
150 loaded_product = list(loaded_products.values())[0]
151 assert loaded_product["name"] == test_data["name"]
152 assert loaded_product["price"] == test_data["price"]
154 except ImportError as e:
155 if serializer_type == "msgpack":
156 pytest.skip(f"msgpack not available: {e}")
157 else:
158 raise
160 def test_persistence_file_corruption_handling(self, temp_dir):
161 """測試持久化文件損壞處理"""
162 # 創建損壞的文件
163 corrupt_file = os.path.join(temp_dir, "products:test.data")
164 with open(corrupt_file, 'w') as f:
165 f.write("{ invalid json content")
167 # 應該能處理損壞的文件並創建新的存儲
168 storage = DiskStorage(temp_dir)
169 crud = AutoCRUD(model=Product, storage=storage, resource_name="products")
171 # 應該能正常創建數據
172 product = crud.create({
173 "name": "測試產品",
174 "description": "測試描述",
175 "price": 100.0,
176 "category": "測試"
177 })
179 assert product["name"] == "測試產品"
181 def test_persistence_directory_creation(self, temp_dir):
182 """測試持久化時自動創建目錄"""
183 nested_dir = os.path.join(temp_dir, "nested", "path")
185 # 目錄不應該存在
186 assert not os.path.exists(nested_dir)
188 # 創建存儲應該自動創建目錄
189 storage = DiskStorage(nested_dir)
190 crud = AutoCRUD(model=Product, storage=storage, resource_name="products")
192 product = crud.create({
193 "name": "測試產品",
194 "description": "測試描述",
195 "price": 100.0,
196 "category": "測試"
197 })
199 # 目錄應該被創建
200 assert os.path.exists(nested_dir)
202 # 重新載入應該正常工作
203 storage2 = DiskStorage(nested_dir)
204 crud2 = AutoCRUD(model=Product, storage=storage2, resource_name="products")
206 loaded_products = crud2.list_all()
207 assert len(loaded_products) == 1