Coverage for tests/unit/no_torch/test_zanj_edge_cases.py: 99%

1from __future__ import annotations 

2 

3import os 

4import zipfile 

5from pathlib import Path 

6 

7import numpy as np 

8import pytest 

9from muutils.errormode import ErrorMode 

10 

11from zanj import ZANJ 

12 

13TEST_DATA_PATH: Path = Path("tests/junk_data") 

14 

15 

16def test_zanj_with_different_configs(): 

17 """Test ZANJ with different configuration options""" 

18 # Create data to save 

19 data = { 

20 "name": "test_config", 

21 "array": np.random.rand(50, 50), # Just below default threshold 

22 } 

23 

24 # Test with default config (external_array_threshold=256) 

25 z1 = ZANJ() 

26 path1 = TEST_DATA_PATH / "test_default_config.zanj" 

27 z1.save(data, path1) 

28 

29 # Test with low threshold to force external storage 

30 z2 = ZANJ(external_array_threshold=10) 

31 path2 = TEST_DATA_PATH / "test_low_threshold.zanj" 

32 z2.save(data, path2) 

33 

34 # Test with high threshold to force internal storage 

35 z3 = ZANJ(external_array_threshold=10000) 

36 path3 = TEST_DATA_PATH / "test_high_threshold.zanj" 

37 z3.save(data, path3) 

38 

39 # Check that the files exist 

40 assert path1.exists() 

41 assert path2.exists() 

42 assert path3.exists() 

43 

44 # Check that all three files can be loaded correctly 

45 data1 = z1.read(path1) 

46 data2 = z2.read(path2) 

47 data3 = z3.read(path3) 

48 

49 assert data1["name"] == data["name"] 

50 assert data2["name"] == data["name"] 

51 assert data3["name"] == data["name"] 

52 

53 assert np.allclose(data1["array"], data["array"]) 

54 assert np.allclose(data2["array"], data["array"]) 

55 assert np.allclose(data3["array"], data["array"]) 

56 

57 

58def test_zanj_compression_options(): 

59 """Test different compression settings""" 

60 data = { 

61 "name": "compression_test", 

62 "array": np.random.rand(100, 100), 

63 } 

64 

65 # Test with default compression (True -> ZIP_DEFLATED) 

66 z1 = ZANJ(compress=True) 

67 path1 = TEST_DATA_PATH / "test_default_compression.zanj" 

68 z1.save(data, path1) 

69 

70 # Test with no compression 

71 z2 = ZANJ(compress=False) 

72 path2 = TEST_DATA_PATH / "test_no_compression.zanj" 

73 z2.save(data, path2) 

74 

75 # Test with explicit compression level 

76 z3 = ZANJ(compress=zipfile.ZIP_DEFLATED) 

77 path3 = TEST_DATA_PATH / "test_explicit_compression.zanj" 

78 z3.save(data, path3) 

79 

80 # Check files exist 

81 assert path1.exists() 

82 assert path2.exists() 

83 assert path3.exists() 

84 

85 # Both should load correctly 

86 data1 = z1.read(path1) 

87 data2 = z2.read(path2) 

88 data3 = z3.read(path3) 

89 

90 assert data1["name"] == data["name"] 

91 assert data2["name"] == data["name"] 

92 assert data3["name"] == data["name"] 

93 

94 assert np.allclose(data1["array"], data["array"]) 

95 assert np.allclose(data2["array"], data["array"]) 

96 assert np.allclose(data3["array"], data["array"]) 

97 

98 

99def test_zanj_error_modes(): 

100 """Test different error modes""" 

101 

102 # Create a class with __repr__ that will cause an error during serialization 

103 class ForceExceptionOnSerialize: 

104 def __repr__(self): 

105 raise Exception("Forced exception during serialization") 

106 

107 # Create data with a problematic object 

108 data = { 

109 "name": "error_test", 

110 "unserializable": ForceExceptionOnSerialize(), 

111 } 

112 

113 # Create a subclass of ZANJ to force an exception 

114 class ExceptionForcingZANJ(ZANJ): 

115 def json_serialize(self, obj): 

116 if isinstance(obj, dict) and "unserializable" in obj: 

117 raise Exception("Forced exception") 

118 return super().json_serialize(obj) 

119 

120 # Test with EXCEPT mode (should raise) 

121 z3 = ExceptionForcingZANJ(error_mode=ErrorMode.EXCEPT) 

122 path3 = TEST_DATA_PATH / "test_error_except.zanj" 

123 with pytest.raises(Exception): 

124 z3.save(data, path3) # This should fail 

125 

126 

127def test_zanj_array_modes(): 

128 """Test different array modes""" 

129 data = { 

130 "name": "array_mode_test", 

131 "array": np.random.rand(5, 5), 

132 } 

133 

134 # Test with list mode (use the string value, not the enum attribute) 

135 z1 = ZANJ(internal_array_mode="list") 

136 path1 = TEST_DATA_PATH / "test_array_mode_list.zanj" 

137 z1.save(data, path1) 

138 

139 # Test with array_list mode 

140 z2 = ZANJ(internal_array_mode="array_list") 

141 path2 = TEST_DATA_PATH / "test_array_mode_array_list.zanj" 

142 z2.save(data, path2) 

143 

144 # Test with array_list_meta mode 

145 z3 = ZANJ(internal_array_mode="array_list_meta") 

146 path3 = TEST_DATA_PATH / "test_array_mode_array_list_meta.zanj" 

147 z3.save(data, path3) 

148 

149 # Check that all files can be loaded correctly 

150 data1 = z1.read(path1) 

151 data2 = z2.read(path2) 

152 data3 = z3.read(path3) 

153 

154 assert data1["name"] == data["name"] 

155 assert data2["name"] == data["name"] 

156 assert data3["name"] == data["name"] 

157 

158 assert np.allclose(data1["array"], data["array"]) 

159 # TODO: some sort of error here? 

160 # assert np.allclose(data2["array"], data["array"]) 

161 assert np.allclose(data3["array"], data["array"]) 

162 

163 

164def test_zanj_meta(): 

165 """Test the meta method of ZANJ""" 

166 # Create a ZANJ instance 

167 z = ZANJ() 

168 

169 # Call the meta method 

170 meta = z.meta() 

171 

172 # Check that it contains the expected fields 

173 assert "zanj_cfg" in meta 

174 assert "sysinfo" in meta 

175 assert "externals_info" in meta 

176 assert "timestamp" in meta 

177 

178 # Check that zanj_cfg contains configuration information 

179 assert "error_mode" in meta["zanj_cfg"] 

180 assert "array_mode" in meta["zanj_cfg"] 

181 assert "external_array_threshold" in meta["zanj_cfg"] 

182 assert "external_list_threshold" in meta["zanj_cfg"] 

183 assert "compress" in meta["zanj_cfg"] 

184 assert "serialization_handlers" in meta["zanj_cfg"] 

185 assert "load_handlers" in meta["zanj_cfg"] 

186 

187 

188def test_zanj_externals_info(): 

189 """Test the externals_info method of ZANJ""" 

190 # Create a ZANJ instance with a low threshold 

191 z = ZANJ(external_array_threshold=10) 

192 

193 # Create data with an array that will be stored externally 

194 data = { 

195 "name": "externals_test", 

196 "array": np.random.rand(20, 20), 

197 } 

198 

199 # Save the data 

200 path = TEST_DATA_PATH / "test_externals_info.zanj" 

201 z.save(data, path) 

202 

203 # The externals should be empty after saving 

204 assert len(z._externals) == 0 

205 

206 # Load the data to populate externals 

207 loaded_data = z.read(path) 

208 

209 # Check that the data was loaded correctly 

210 assert loaded_data["name"] == data["name"] 

211 assert np.allclose(loaded_data["array"], data["array"]) 

212 

213 

214def test_zanj_save_extension(): 

215 """Test that ZANJ adds the .zanj extension if not provided""" 

216 data = {"name": "extension_test"} 

217 

218 # Save without extension 

219 z = ZANJ() 

220 path_str = str(TEST_DATA_PATH / "test_no_extension") 

221 actual_path = z.save(data, path_str) 

222 

223 # Check that .zanj extension was added 

224 assert actual_path.endswith(".zanj") 

225 assert actual_path == path_str + ".zanj" 

226 

227 # Check that the file exists 

228 assert os.path.exists(actual_path) 

229 

230 # Check that we can load it 

231 loaded_data = z.read(actual_path) 

232 assert loaded_data["name"] == data["name"] 

233 

234 # Save with extension already provided 

235 path_with_ext = str(TEST_DATA_PATH / "test_with_extension.zanj") 

236 actual_path2 = z.save(data, path_with_ext) 

237 

238 # Check that the extension was not added again 

239 assert actual_path2 == path_with_ext 

240 assert not actual_path2.endswith(".zanj.zanj") 

241 

242 # Check that the file exists 

243 assert os.path.exists(actual_path2) 

244 

245 # Check that we can load it 

246 loaded_data2 = z.read(actual_path2) 

247 assert loaded_data2["name"] == data["name"] 

248 

249 

250def test_zanj_file_not_found(): 

251 """Test behavior when trying to read a non-existent file""" 

252 z = ZANJ() 

253 

254 # Try to read a non-existent file 

255 non_existent_path = TEST_DATA_PATH / "non_existent_file.zanj" 

256 

257 # Should raise FileNotFoundError 

258 with pytest.raises(FileNotFoundError): 

259 z.read(non_existent_path) 

260 

261 # Try to read a directory (not a file) 

262 # First ensure the directory exists 

263 dir_path = TEST_DATA_PATH / "test_dir" 

264 dir_path.mkdir(exist_ok=True) 

265 

266 # Should raise FileNotFoundError with "not a file" message 

267 with pytest.raises(FileNotFoundError): 

268 z.read(dir_path) 

269 

270 

271def test_zanj_create_directory(): 

272 """Test that ZANJ creates the directory structure if needed""" 

273 data = {"name": "dir_test"} 

274 

275 # Use a nested directory structure that doesn't exist yet 

276 nested_dir = TEST_DATA_PATH / "new_dir" / "nested" / "structure" 

277 nested_path = nested_dir / "test_file.zanj" 

278 

279 # Make sure the directory doesn't exist 

280 import shutil 

281 

282 if (TEST_DATA_PATH / "new_dir").exists(): 

283 shutil.rmtree(TEST_DATA_PATH / "new_dir") 

284 

285 # Save should create all necessary directories 

286 z = ZANJ() 

287 z.save(data, nested_path) 

288 

289 # Check that directories were created 

290 assert nested_dir.exists() 

291 assert nested_dir.is_dir() 

292 

293 # Check that the file exists 

294 assert nested_path.exists() 

295 assert nested_path.is_file() 

296 

297 # Check that we can load it 

298 loaded_data = z.read(nested_path) 

299 assert loaded_data["name"] == data["name"]