Coverage for tests/test_storage.py: 100%

1"""測試存儲後端""" 

2 

3import pytest 

4import os 

5from dataclasses import dataclass 

6from autocrud import AutoCRUD, MemoryStorage, DiskStorage, SerializerFactory 

7 

8 

9@dataclass 

10class Product: 

11 name: str 

12 description: str 

13 price: float 

14 category: str 

15 

16 

17class TestMemoryStorage: 

18 """測試內存存儲""" 

19 

20 def test_memory_storage_basic_operations(self, sample_product_data): 

21 """測試內存存儲基本操作""" 

22 storage = MemoryStorage() 

23 crud = AutoCRUD(model=Product, storage=storage, resource_name="products") 

24 

25 # 創建 

26 product = crud.create(sample_product_data) 

27 assert product['name'] == sample_product_data['name'] 

28 

29 # 獲取 

30 retrieved = crud.get(product['id']) 

31 assert retrieved is not None 

32 assert retrieved['name'] == sample_product_data['name'] 

33 

34 # 更新 

35 updated_data = {"name": "更新產品", "description": "更新描述", "price": 999.0, "category": "測試"} 

36 updated = crud.update(product['id'], updated_data) 

37 assert updated['name'] == "更新產品" 

38 

39 # 刪除 

40 deleted = crud.delete(product['id']) 

41 assert deleted is True 

42 assert crud.get(product['id']) is None 

43 

44 def test_memory_storage_persistence(self, sample_product_data): 

45 """測試內存存儲數據不會在實例間保持""" 

46 # 第一個存儲實例 

47 storage1 = MemoryStorage() 

48 crud1 = AutoCRUD(model=Product, storage=storage1, resource_name="products") 

49 product = crud1.create(sample_product_data) 

50 

51 # 第二個存儲實例（應該是空的） 

52 storage2 = MemoryStorage() 

53 crud2 = AutoCRUD(model=Product, storage=storage2, resource_name="products") 

54 

55 # 在新實例中查找數據 

56 result = crud2.get(product['id']) 

57 assert result is None # 數據不應該存在 

58 

59 all_products = crud2.list_all() 

60 assert len(all_products) == 0 # 應該是空的 

61 

62 

63class TestDiskStorage: 

64 """測試磁碟存儲""" 

65 

66 def test_disk_storage_basic_operations(self, temp_dir, sample_product_data): 

67 """測試磁碟存儲基本操作""" 

68 storage = DiskStorage(temp_dir) 

69 crud = AutoCRUD(model=Product, storage=storage, resource_name="products") 

70 

71 # 創建 

72 product = crud.create(sample_product_data) 

73 assert product['name'] == sample_product_data['name'] 

74 

75 # 檢查文件是否創建 

76 files = os.listdir(temp_dir) 

77 assert len(files) == 1 

78 assert files[0].endswith('.data') 

79 

80 # 獲取 

81 retrieved = crud.get(product['id']) 

82 assert retrieved is not None 

83 assert retrieved['name'] == sample_product_data['name'] 

84 

85 # 更新 

86 updated_data = {"name": "更新產品", "description": "更新描述", "price": 999.0, "category": "測試"} 

87 updated = crud.update(product['id'], updated_data) 

88 assert updated['name'] == "更新產品" 

89 

90 # 刪除 

91 deleted = crud.delete(product['id']) 

92 assert deleted is True 

93 assert crud.get(product['id']) is None 

94 

95 # 檢查文件是否刪除 

96 files = os.listdir(temp_dir) 

97 assert len(files) == 0 

98 

99 def test_disk_storage_persistence(self, temp_dir, sample_product_data): 

100 """測試磁碟存儲數據持久化""" 

101 # 第一個存儲實例 

102 storage1 = DiskStorage(temp_dir) 

103 crud1 = AutoCRUD(model=Product, storage=storage1, resource_name="products") 

104 product = crud1.create(sample_product_data) 

105 product_id = product['id'] 

106 

107 # 第二個存儲實例（應該能讀取相同數據） 

108 storage2 = DiskStorage(temp_dir) 

109 crud2 = AutoCRUD(model=Product, storage=storage2, resource_name="products") 

110 

111 # 在新實例中查找數據 

112 retrieved = crud2.get(product_id) 

113 assert retrieved is not None 

114 assert retrieved['name'] == sample_product_data['name'] 

115 

116 all_products = crud2.list_all() 

117 assert len(all_products) == 1 

118 assert product_id in all_products 

119 

120 def test_disk_storage_with_different_serializers(self, temp_dir, sample_product_data): 

121 """測試磁碟存儲使用不同序列化器""" 

122 serializer_types = ['json', 'pickle'] 

123 

124 for serializer_type in serializer_types: 

125 # 使用不同的子目錄避免衝突 

126 storage_dir = os.path.join(temp_dir, serializer_type) 

127 os.makedirs(storage_dir, exist_ok=True) 

128 

129 serializer = SerializerFactory.create(serializer_type) 

130 storage = DiskStorage(storage_dir, serializer=serializer) 

131 crud = AutoCRUD(model=Product, storage=storage, resource_name="products") 

132 

133 # 基本操作測試 

134 product = crud.create(sample_product_data) 

135 retrieved = crud.get(product['id']) 

136 

137 assert retrieved is not None 

138 assert retrieved['name'] == sample_product_data['name'] 

139 assert retrieved['price'] == sample_product_data['price'] 

140 

141 

142class TestStorageComparison: 

143 """測試存儲比較""" 

144 

145 def test_memory_vs_disk_performance(self, temp_dir, sample_product_data): 

146 """簡單的性能比較測試""" 

147 import time 

148 

149 # 內存存儲性能 

150 memory_storage = MemoryStorage() 

151 memory_crud = AutoCRUD(model=Product, storage=memory_storage, resource_name="products") 

152 

153 start_time = time.time() 

154 for i in range(10): # 減少次數以加快測試 

155 test_data = {**sample_product_data, "name": f"產品{i}"} 

156 memory_crud.create(test_data) 

157 memory_time = time.time() - start_time 

158 

159 # 磁碟存儲性能 

160 disk_storage = DiskStorage(temp_dir) 

161 disk_crud = AutoCRUD(model=Product, storage=disk_storage, resource_name="products") 

162 

163 start_time = time.time() 

164 for i in range(10): 

165 test_data = {**sample_product_data, "name": f"產品{i}"} 

166 disk_crud.create(test_data) 

167 disk_time = time.time() - start_time 

168 

169 # 磁碟應該比內存慢（但這只是一般期望，不是嚴格要求） 

170 assert memory_time >= 0 # 至少要大於0 

171 assert disk_time >= 0 

172 

173 # 檢查數據正確性 

174 memory_products = memory_crud.list_all() 

175 disk_products = disk_crud.list_all() 

176 

177 assert len(memory_products) == 10 

178 assert len(disk_products) == 10