Coverage for tests / unit / test_utils.py: 100%

1""" 

2Unit tests for utility functions. 

3""" 

4 

5import io 

6import zipfile 

7from importlib import metadata 

8from pathlib import Path 

9 

10import pytest 

11 

12from superset_io.utils import ( 

13 get_version, 

14 validate_assets_bundle_structure, 

15 zipfile_buffer_from_folder, 

16 zipfile_buffer_from_zipfile, 

17) 

18 

19 

20class TestZipfileBufferFromFolder: 

21 """Tests for zipfile_buffer_from_folder function.""" 

22 

23 def test_create_zip_from_folder(self, tmp_path): 

24 """Test creating a ZIP from a folder structure.""" 

25 # Create a test folder structure 

26 test_dir = tmp_path / "test_folder" 

27 test_dir.mkdir() 

28 

29 # Create some files 

30 (test_dir / "file1.txt").write_text("Content 1") 

31 (test_dir / "file2.txt").write_text("Content 2") 

32 subdir = test_dir / "subdir" 

33 subdir.mkdir() 

34 (subdir / "file3.txt").write_text("Content 3") 

35 

36 # Create ZIP 

37 zip_buffer = zipfile_buffer_from_folder(test_dir) 

38 

39 # Verify the ZIP 

40 assert isinstance(zip_buffer, io.BytesIO) 

41 

42 with zipfile.ZipFile(zip_buffer, "r") as zf: 

43 assert "test_folder/file1.txt" in zf.namelist() 

44 assert "test_folder/file2.txt" in zf.namelist() 

45 assert "test_folder/subdir/file3.txt" in zf.namelist() 

46 

47 # Verify file contents 

48 assert zf.read("test_folder/file1.txt").decode() == "Content 1" 

49 assert zf.read("test_folder/file2.txt").decode() == "Content 2" 

50 assert zf.read("test_folder/subdir/file3.txt").decode() == "Content 3" 

51 

52 def test_nonexistent_folder(self): 

53 """Test that non-existent folder raises ValueError.""" 

54 with pytest.raises(ValueError, match="Not a folder"): 

55 zipfile_buffer_from_folder("/nonexistent/path") 

56 

57 def test_file_instead_of_folder(self, tmp_path): 

58 """Test that passing a file instead of folder raises ValueError.""" 

59 test_file = tmp_path / "test.txt" 

60 test_file.write_text("test") 

61 

62 with pytest.raises(ValueError, match="Not a folder"): 

63 zipfile_buffer_from_folder(test_file) 

64 

65 

66class TestZipfileBufferFromZipfile: 

67 """Tests for zipfile_buffer_from_zipfile function.""" 

68 

69 def test_copy_existing_zip(self, tmp_path): 

70 """Test copying an existing ZIP file to a buffer.""" 

71 # Create a test ZIP file 

72 zip_path = tmp_path / "test.zip" 

73 with zipfile.ZipFile(zip_path, "w") as zf: 

74 zf.writestr("file1.txt", "Content 1") 

75 zf.writestr("subdir/file2.txt", "Content 2") 

76 

77 # Copy to buffer 

78 zip_buffer = zipfile_buffer_from_zipfile(zip_path) 

79 

80 # Verify the buffer 

81 assert isinstance(zip_buffer, io.BytesIO) 

82 

83 with zipfile.ZipFile(zip_buffer, "r") as zf: 

84 assert "file1.txt" in zf.namelist() 

85 assert "subdir/file2.txt" in zf.namelist() 

86 assert zf.read("file1.txt").decode() == "Content 1" 

87 

88 def test_nonexistent_zip(self): 

89 """Test that non-existent ZIP file raises error on read.""" 

90 # The function will try to open the file, which will raise FileNotFoundError 

91 # when the file is read. The function doesn't check existence beforehand. 

92 non_existent = Path("/nonexistent/path.zip") 

93 

94 # This will raise FileNotFoundError when trying to open the file 

95 with pytest.raises(FileNotFoundError): 

96 zipfile_buffer_from_zipfile(non_existent) 

97 

98 

99class TestValidateAssetsBundleStructure: 

100 """Tests for validate_assets_bundle_structure function.""" 

101 

102 def test_valid_structure(self): 

103 """Test validation of a valid Superset assets bundle.""" 

104 # Create a valid ZIP structure 

105 zip_buffer = io.BytesIO() 

106 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: 

107 zf.writestr("assets_export_20240101/metadata.yaml", "version: 1.0") 

108 zf.writestr("assets_export_20240101/dashboards/test.yaml", "title: Test") 

109 zf.writestr("assets_export_20240101/charts/", "") # Empty directory 

110 zf.writestr("assets_export_20240101/datasets/", "") 

111 

112 zip_buffer.seek(0) 

113 

114 # Should not raise any exception 

115 validate_assets_bundle_structure(zip_buffer) 

116 

117 def test_valid_structure_with_bytes(self): 

118 """Test validation with bytes input.""" 

119 zip_buffer = io.BytesIO() 

120 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: 

121 zf.writestr("assets_export_20240101/metadata.yaml", "version: 1.0") 

122 zf.writestr("assets_export_20240101/dashboards/test.yaml", "title: Test") 

123 

124 zip_bytes = zip_buffer.getvalue() 

125 

126 # Should not raise any exception 

127 validate_assets_bundle_structure(zip_bytes) 

128 

129 def test_valid_structure_with_path(self, tmp_path): 

130 """Test validation with Path input.""" 

131 zip_path = tmp_path / "test.zip" 

132 with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: 

133 zf.writestr("assets_export_20240101/metadata.yaml", "version: 1.0") 

134 zf.writestr("assets_export_20240101/dashboards/test.yaml", "title: Test") 

135 

136 # Should not raise any exception 

137 validate_assets_bundle_structure(zip_path) 

138 

139 def test_missing_metadata(self): 

140 """Test validation fails when metadata.yaml is missing.""" 

141 zip_buffer = io.BytesIO() 

142 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: 

143 zf.writestr("assets_export_20240101/dashboards/test.yaml", "title: Test") 

144 

145 zip_buffer.seek(0) 

146 

147 with pytest.raises(ValueError, match="Missing metadata.yaml"): 

148 validate_assets_bundle_structure(zip_buffer) 

149 

150 def test_metadata_not_in_root_folder(self): 

151 """Test validation fails when metadata.yaml is not in root folder.""" 

152 zip_buffer = io.BytesIO() 

153 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: 

154 zf.writestr("metadata.yaml", "version: 1.0") # Not in root folder 

155 zf.writestr("dashboards/test.yaml", "title: Test") 

156 zf.writestr("charts/test.yaml", "title: Test") 

157 

158 zip_buffer.seek(0) 

159 

160 with pytest.raises(ValueError, match="Expected exactly one top-level folder"): 

161 validate_assets_bundle_structure(zip_buffer) 

162 

163 def test_multiple_root_folders(self): 

164 """Test validation fails with multiple top-level folders.""" 

165 zip_buffer = io.BytesIO() 

166 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: 

167 zf.writestr("folder1/metadata.yaml", "version: 1.0") 

168 zf.writestr("folder2/dashboards/test.yaml", "title: Test") 

169 

170 zip_buffer.seek(0) 

171 

172 with pytest.raises(ValueError, match="Expected exactly one top-level folder"): 

173 validate_assets_bundle_structure(zip_buffer) 

174 

175 def test_metadata_in_wrong_location(self): 

176 """Test validation fails when metadata.yaml is not at expected path.""" 

177 zip_buffer = io.BytesIO() 

178 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zf: 

179 # Also add another metadata in wrong place (should still pass) 

180 zf.writestr("assets_export_20240101/another/metadata.yaml", "version: 1.0") 

181 

182 zip_buffer.seek(0) 

183 

184 # This should still pass because we have metadata at the expected path 

185 with pytest.raises( 

186 ValueError, match="metadata.yaml not found at expected path" 

187 ): 

188 validate_assets_bundle_structure(zip_buffer) 

189 

190 def test_empty_zip(self): 

191 """Test validation fails with empty ZIP.""" 

192 zip_buffer = io.BytesIO() 

193 with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED): 

194 pass # Empty zip 

195 

196 zip_buffer.seek(0) 

197 

198 with pytest.raises(ValueError, match="Missing metadata.yaml"): 

199 validate_assets_bundle_structure(zip_buffer) 

200 

201 

202class TestGetVersion: 

203 def test_get_version_returns_version_string(self, monkeypatch): 

204 monkeypatch.setattr(metadata, "version", lambda name: "1.2.3") 

205 

206 assert get_version() == "1.2.3" 

207 

208 def test_get_version_returns_fallback_when_package_not_found(self, monkeypatch): 

209 def _raise(_name): 

210 raise metadata.PackageNotFoundError 

211 

212 monkeypatch.setattr(metadata, "version", _raise) 

213 

214 assert get_version() == "[not found] Use `uv sync` when developing!"