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_meta mode 

140 z3 = ZANJ(internal_array_mode="array_list_meta") 

141 path3 = TEST_DATA_PATH / "test_array_mode_array_list_meta.zanj" 

142 z3.save(data, path3) 

143 

144 # Check that all files can be loaded correctly 

145 data1 = z1.read(path1) 

146 data3 = z3.read(path3) 

147 

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

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

150 

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

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

153 

154 

155def test_zanj_meta(): 

156 """Test the meta method of ZANJ""" 

157 # Create a ZANJ instance 

158 z = ZANJ() 

159 

160 # Call the meta method 

161 meta = z.meta() 

162 

163 # Check that it contains the expected fields 

164 assert "zanj_cfg" in meta 

165 assert "sysinfo" in meta 

166 assert "externals_info" in meta 

167 assert "timestamp" in meta 

168 

169 # Check that zanj_cfg contains configuration information 

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

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

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

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

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

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

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

177 

178 

179def test_zanj_externals_info(): 

180 """Test the externals_info method of ZANJ""" 

181 # Create a ZANJ instance with a low threshold 

182 z = ZANJ(external_array_threshold=10) 

183 

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

185 data = { 

186 "name": "externals_test", 

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

188 } 

189 

190 # Save the data 

191 path = TEST_DATA_PATH / "test_externals_info.zanj" 

192 z.save(data, path) 

193 

194 # The externals should be empty after saving 

195 assert len(z._externals) == 0 

196 

197 # Load the data to populate externals 

198 loaded_data = z.read(path) 

199 

200 # Check that the data was loaded correctly 

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

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

203 

204 

205def test_zanj_save_extension(): 

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

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

208 

209 # Save without extension 

210 z = ZANJ() 

211 path_str = str(TEST_DATA_PATH / "test_no_extension") 

212 actual_path = z.save(data, path_str) 

213 

214 # Check that .zanj extension was added 

215 assert actual_path.endswith(".zanj") 

216 assert actual_path == path_str + ".zanj" 

217 

218 # Check that the file exists 

219 assert os.path.exists(actual_path) 

220 

221 # Check that we can load it 

222 loaded_data = z.read(actual_path) 

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

224 

225 # Save with extension already provided 

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

227 actual_path2 = z.save(data, path_with_ext) 

228 

229 # Check that the extension was not added again 

230 assert actual_path2 == path_with_ext 

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

232 

233 # Check that the file exists 

234 assert os.path.exists(actual_path2) 

235 

236 # Check that we can load it 

237 loaded_data2 = z.read(actual_path2) 

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

239 

240 

241def test_zanj_file_not_found(): 

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

243 z = ZANJ() 

244 

245 # Try to read a non-existent file 

246 non_existent_path = TEST_DATA_PATH / "non_existent_file.zanj" 

247 

248 # Should raise FileNotFoundError 

249 with pytest.raises(FileNotFoundError): 

250 z.read(non_existent_path) 

251 

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

253 # First ensure the directory exists 

254 dir_path = TEST_DATA_PATH / "test_dir" 

255 dir_path.mkdir(exist_ok=True) 

256 

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

258 with pytest.raises(FileNotFoundError): 

259 z.read(dir_path) 

260 

261 

262def test_zanj_create_directory(): 

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

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

265 

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

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

268 nested_path = nested_dir / "test_file.zanj" 

269 

270 # Make sure the directory doesn't exist 

271 import shutil 

272 

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

274 shutil.rmtree(TEST_DATA_PATH / "new_dir") 

275 

276 # Save should create all necessary directories 

277 z = ZANJ() 

278 z.save(data, nested_path) 

279 

280 # Check that directories were created 

281 assert nested_dir.exists() 

282 assert nested_dir.is_dir() 

283 

284 # Check that the file exists 

285 assert nested_path.exists() 

286 assert nested_path.is_file() 

287 

288 # Check that we can load it 

289 loaded_data = z.read(nested_path) 

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