Coverage for tests / integration / test_dashboard_export_import.py: 16%

75 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-03-02 13:04 +0100

1""" 

2Integration tests for dashboard export and import functionality. 

3""" 

4 

5import io 

6import zipfile 

7 

8import pytest 

9import requests 

10 

11 

12@pytest.mark.integration 

13class TestDashboardExportImport: 

14 """Integration tests for dashboard export and import.""" 

15 

16 def test_get_dashboards(self, superset_client): 

17 """Test that we can retrieve the list of dashboards.""" 

18 dashboards = superset_client.get_dashboards() 

19 assert isinstance(dashboards, dict) 

20 # The response should have a 'result' key with dashboard list 

21 assert "result" in dashboards 

22 assert isinstance(dashboards["result"], list) 

23 

24 def test_export_and_import_dashboard(self, superset_client): 

25 """ 

26 Test exporting a dashboard and importing it back. 

27 

28 This test requires at least one dashboard to exist in Superset. 

29 If no dashboards exist, the test is skipped. 

30 """ 

31 # Get all dashboards 

32 dashboards = superset_client.get_dashboards() 

33 dashboard_list = dashboards.get("result", []) 

34 

35 if not dashboard_list: 

36 pytest.skip("No dashboards available for export/import test") 

37 

38 # Use the first dashboard 

39 dashboard = dashboard_list[0] 

40 dashboard_id = dashboard["id"] 

41 dashboard_title = dashboard["dashboard_title"] 

42 

43 print( 

44 f"Testing export/import with dashboard: {dashboard_title} (ID: {dashboard_id})" 

45 ) 

46 

47 # Export the dashboard 

48 zip_bytes, zip_file = superset_client.export_dashboard(dashboard_id) 

49 

50 # Verify the export is a valid ZIP file 

51 assert zip_bytes is not None 

52 assert len(zip_bytes) > 0 

53 assert zip_file.testzip() is None # No corrupted files 

54 

55 # Check the ZIP structure 

56 zip_names = zip_file.namelist() 

57 assert any(name.endswith("metadata.yaml") for name in zip_names) 

58 assert any("dashboards/" in name for name in zip_names) 

59 

60 # Import the dashboard back (with overwrite) 

61 import_response = superset_client.import_dashboard( 

62 zipfile_buffer=zip_bytes, 

63 overwrite=True, 

64 ) 

65 

66 # Verify import was successful 

67 assert import_response.status_code == 200 

68 import_result = import_response.json() 

69 

70 # The response should contain information about the imported assets 

71 assert "message" in import_result 

72 assert "changed_assets" in import_result.get("result", {}) 

73 

74 def test_export_invalid_dashboard(self, superset_client): 

75 """Test exporting a non-existent dashboard should fail.""" 

76 invalid_dashboard_id = 999999 

77 

78 # This should raise an exception 

79 with pytest.raises(requests.exceptions.HTTPError) as exc_info: 

80 superset_client.export_dashboard(invalid_dashboard_id) 

81 

82 # Should be a 404 or 400 error 

83 assert exc_info.value.response.status_code in (400, 404) 

84 

85 def test_import_invalid_zip(self, superset_client, tmp_path): 

86 """Test importing an invalid ZIP file should fail.""" 

87 invalid_zip = b"This is not a valid ZIP file" 

88 

89 with pytest.raises(requests.exceptions.HTTPError) as exc_info: 

90 superset_client.import_dashboard(zipfile_buffer=invalid_zip) 

91 

92 # Should be a 400 error 

93 assert exc_info.value.response.status_code == 400 

94 

95 def test_import_with_passwords(self, superset_client, sample_dashboard_zip): 

96 """Test importing a dashboard with password parameters.""" 

97 # This test requires a sample dashboard ZIP with database connections 

98 # that need passwords. Since we don't have that, we test the API call 

99 # but expect it to fail with a meaningful error. 

100 

101 passwords = {"some_database": "test_password"} 

102 

103 try: 

104 response = superset_client.import_dashboard( 

105 zipfile_buffer=sample_dashboard_zip, 

106 passwords=passwords, 

107 ) 

108 # If it succeeds, that's fine too 

109 assert response.status_code == 200 

110 except requests.exceptions.HTTPError as e: 

111 # The error should be about missing passwords or invalid bundle 

112 # Not a generic 500 error 

113 assert e.response.status_code in (400, 422) 

114 

115 def test_dashboard_roundtrip(self, superset_client): 

116 """ 

117 Full roundtrip test: export dashboard, modify something in memory, 

118 and verify we can work with the exported data. 

119 """ 

120 dashboards = superset_client.get_dashboards() 

121 dashboard_list = dashboards.get("result", []) 

122 

123 if not dashboard_list: 

124 pytest.skip("No dashboards available for roundtrip test") 

125 

126 dashboard = dashboard_list[0] 

127 dashboard_id = dashboard["id"] 

128 

129 # Export the dashboard 

130 zip_bytes, zip_file = superset_client.export_dashboard(dashboard_id) 

131 

132 # Extract and examine metadata 

133 with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zf: 

134 # Find metadata.yaml 

135 metadata_files = [ 

136 name for name in zf.namelist() if name.endswith("metadata.yaml") 

137 ] 

138 assert len(metadata_files) == 1 

139 

140 metadata_content = zf.read(metadata_files[0]).decode("utf-8") 

141 assert "version" in metadata_content 

142 assert "timestamp" in metadata_content 

143 

144 # Find dashboard YAML files 

145 dashboard_files = [ 

146 name 

147 for name in zf.namelist() 

148 if "dashboards/" in name and name.endswith(".yaml") 

149 ] 

150 assert len(dashboard_files) > 0 

151 

152 # Read one dashboard file 

153 dashboard_content = zf.read(dashboard_files[0]).decode("utf-8") 

154 assert "dashboard_title" in dashboard_content 

155 

156 # Create a modified version of the ZIP (just rename a file internally) 

157 # This is a simple test to ensure we can manipulate the ZIP 

158 with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zf_in: 

159 with io.BytesIO() as buffer: 

160 with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zf_out: 

161 for item in zf_in.infolist(): 

162 # Copy all files as-is 

163 zf_out.writestr(item, zf_in.read(item.filename)) 

164 

165 modified_zip = buffer.getvalue() 

166 

167 # Try to import the modified (but still valid) ZIP 

168 # This should work since we didn't change the structure 

169 response = superset_client.import_dashboard( 

170 zipfile_buffer=modified_zip, 

171 overwrite=True, 

172 ) 

173 

174 # Import should succeed 

175 assert response.status_code == 200