Coverage for tests / unit / test_api.py: 100%
167 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-02 13:07 +0100
1"""
2Unit tests for the Superset API client.
3"""
5import io
6import zipfile
7from unittest.mock import Mock
9import pytest
10import requests
12from superset_io.api import SuperSetApiClient
15def _resp(*, status_code=200, text="", json_data=None, raise_exc=None):
16 r = Mock()
17 r.status_code = status_code
18 r.text = text
19 r.json = Mock(return_value=json_data if json_data is not None else {})
20 if raise_exc is None:
21 r.raise_for_status = Mock()
22 else:
23 r.raise_for_status = Mock(side_effect=raise_exc)
24 return r
27def _http_error(msg="http error", *, response_text=""):
28 err = requests.HTTPError(msg)
29 if response_text:
30 err.response = Mock()
31 err.response.text = response_text
32 return err
35class TestAPITestConnection:
36 """Lean unit tests for request-building logic in SuperSetApiClient."""
38 def test_connection_health_fails(self):
39 session = Mock()
40 session.base_url = "http://localhost:8088"
41 session.headers = {"X-CSRFToken": "csrf"} # shouldn't matter; should exit early
43 session.get = Mock(
44 side_effect=[
45 _resp(raise_exc=_http_error("health down", response_text="nope"))
46 ]
47 )
48 session.post = Mock()
50 client = SuperSetApiClient(session)
51 assert client.test_connection() is False
53 session.get.assert_called_once_with("/health")
54 session.post.assert_not_called()
56 def test_connection_log_fails(self):
57 session = Mock()
58 session.base_url = "http://localhost:8088"
59 session.headers = {"X-CSRFToken": "csrf"}
61 session.get = Mock(
62 side_effect=[
63 _resp(), # /health ok
64 _resp(
65 status_code=401, raise_exc=_http_error("unauthorized")
66 ), # /api/v1/log/ fails
67 ]
68 )
69 session.post = Mock()
71 client = SuperSetApiClient(session)
72 assert client.test_connection() is False
74 assert session.get.call_args_list[0].args[0] == "/health"
75 assert session.get.call_args_list[1].args[0] == "/api/v1/log/"
76 session.post.assert_not_called()
78 def test_connection_missing_csrf_header(self):
79 session = Mock()
80 session.base_url = "http://localhost:8088"
81 session.headers = {} # no X-CSRFToken
83 session.get = Mock(side_effect=[_resp(), _resp()])
84 session.post = Mock()
86 client = SuperSetApiClient(session)
87 assert client.test_connection() is False
89 session.post.assert_not_called()
91 def test_connection_succeeds_with_invalid_payload(self):
92 session = Mock()
93 session.base_url = "http://localhost:8088/"
94 session.headers = {"X-CSRFToken": "csrf"}
96 session.get = Mock(side_effect=[_resp(), _resp()])
98 post_res = _resp(
99 status_code=422,
100 text="unprocessable entity",
101 json_data={"errors": [{"error_type": "INVALID_PAYLOAD_FORMAT_ERROR"}]},
102 raise_exc=_http_error("422"),
103 )
104 session.post = Mock(return_value=post_res)
106 client = SuperSetApiClient(session)
107 assert client.test_connection() is True
109 def test_connection_error_type_not_expected(self):
110 """
111 This covers the 'POST fails and JSON doesn't match heuristic' => should return
112 False.
113 """
114 session = Mock()
115 session.base_url = "http://localhost:8088"
116 session.headers = {"X-CSRFToken": "csrf"}
118 session.get = Mock(side_effect=[_resp(), _resp()])
120 post_res = _resp(
121 status_code=403,
122 text="forbidden",
123 json_data={"errors": [{"error_type": "SOME_OTHER_ERROR"}]},
124 raise_exc=_http_error("403"),
125 )
126 session.post = Mock(return_value=post_res)
128 client = SuperSetApiClient(session)
130 # If your current code has the `e` NameError, this will raise.
131 # Once code is fixed, it should simply be False.
133 assert client.test_connection() is False
135 def test_connection_post_succeeds(self):
136 session = Mock()
137 session.base_url = "http://localhost:8088"
138 session.headers = {"X-CSRFToken": "csrf"}
140 session.get = Mock(side_effect=[_resp(), _resp()])
141 session.post = Mock(return_value=_resp(status_code=200))
143 client = SuperSetApiClient(session)
144 assert client.test_connection() is True
146 session.post.assert_called_once_with(
147 "/api/v1/assets/import/",
148 json={},
149 headers={"Referer": "http://localhost:8088/"},
150 )
153class TestUploadAssets:
154 @pytest.fixture
155 def asset_folder(self, tmp_path):
156 assets_root = tmp_path / "assets_folder"
157 (assets_root / "dashboards").mkdir(parents=True)
159 (assets_root / "metadata.yaml").write_text("version: 1.0", encoding="utf-8")
160 (assets_root / "dashboards" / "demo.yaml").write_text(
161 "dashboard_title: Demo", encoding="utf-8"
162 )
163 yield assets_root
165 @pytest.fixture
166 def client(self):
167 session = Mock()
168 session.base_url = "http://localhost:8088"
169 session.headers = {"X-CSRFToken": "csrf"}
170 return SuperSetApiClient(session)
172 def test_upload_assets_zip(self, tmp_path, asset_folder, client, monkeypatch):
173 # Create a real zip on disk from the fixture folder contents
174 zip_path = tmp_path / "assets_bundle.zip"
175 with zipfile.ZipFile(zip_path, "w") as zf:
176 for p in asset_folder.rglob("*"):
177 if p.is_file():
178 zf.write(p, arcname=p.relative_to(asset_folder.parent).as_posix())
180 post_mock = Mock()
181 monkeypatch.setattr(client, "_post_assets", post_mock)
183 client.upload_assets(zip_path)
185 post_mock.assert_called_once()
186 buf = post_mock.call_args.kwargs["zipfile_buffer"]
188 assert isinstance(buf, io.BytesIO)
189 assert buf.getvalue() == zip_path.read_bytes()
191 def test_upload_assets_folder(self, asset_folder, client, monkeypatch):
192 post_mock = Mock()
193 monkeypatch.setattr(client, "_post_assets", post_mock)
195 client.upload_assets(asset_folder)
197 post_mock.assert_called_once()
198 buf = post_mock.call_args.kwargs["zipfile_buffer"]
200 assert isinstance(buf, io.BytesIO)
201 # ensure zipped structure contains expected files
202 with zipfile.ZipFile(buf, "r") as zf:
203 names = zf.namelist()
204 assert "assets_folder/metadata.yaml" in names
205 assert "assets_folder/dashboards/demo.yaml" in names
208class TestDownloadAssets:
209 @pytest.fixture
210 def client(self):
211 session = Mock()
212 session.base_url = "http://localhost:8088"
213 session.headers = {"X-CSRFToken": "csrf"}
214 return SuperSetApiClient(session)
216 def _zip_bytes(self, files: dict[str, str]) -> bytes:
217 buf = io.BytesIO()
218 with zipfile.ZipFile(buf, "w") as zf:
219 for name, content in files.items():
220 zf.writestr(name, content)
221 return buf.getvalue()
223 def test_download_assets_writes_zip_file(self, tmp_path, client, monkeypatch):
224 zip_bytes = self._zip_bytes(
225 {
226 "assets_export_123/metadata.yaml": "version: 1.0",
227 "assets_export_123/dashboards/demo.yaml": "dashboard_title: Demo",
228 }
229 )
230 zip_file = zipfile.ZipFile(io.BytesIO(zip_bytes), "r")
231 monkeypatch.setattr(
232 client, "_get_assets_zip", Mock(return_value=(zip_bytes, zip_file))
233 )
235 out_zip = tmp_path / "out.zip"
236 client.download_assets(out_zip)
238 assert out_zip.exists()
239 assert out_zip.read_bytes() == zip_bytes
241 def test_download_assets_extracts_to_folder_and_moves_children(
242 self, tmp_path, client, monkeypatch
243 ):
244 zip_bytes = self._zip_bytes(
245 {
246 "assets_export_123/metadata.yaml": "version: 1.0",
247 "assets_export_123/dashboards/demo.yaml": "dashboard_title: Demo",
248 }
249 )
250 zip_file = zipfile.ZipFile(io.BytesIO(zip_bytes), "r")
251 monkeypatch.setattr(
252 client, "_get_assets_zip", Mock(return_value=(zip_bytes, zip_file))
253 )
255 out_dir = tmp_path / "out_folder" # no .zip suffix => folder mode
256 client.download_assets(out_dir)
258 assert out_dir.is_dir()
259 # children moved out of assets_export_123/ into out_dir
260 assert (out_dir / "metadata.yaml").exists()
261 assert (out_dir / "dashboards" / "demo.yaml").exists()
263 def test_download_assets_folder_destination_not_empty_raises(
264 self, tmp_path, client
265 ):
266 out_dir = tmp_path / "out_folder"
267 out_dir.mkdir(parents=True)
268 (out_dir / "already_there.txt").write_text("x", encoding="utf-8")
270 with pytest.raises(ValueError, match="is not empty"):
271 client.download_assets(out_dir)
273 def test_download_assets_raises_if_no_assets_export_folder_in_zip(
274 self, tmp_path, client, monkeypatch
275 ):
276 zip_bytes = self._zip_bytes(
277 {
278 "not_assets_export/metadata.yaml": "version: 1.0",
279 }
280 )
281 zip_file = zipfile.ZipFile(io.BytesIO(zip_bytes), "r")
282 monkeypatch.setattr(
283 client, "_get_assets_zip", Mock(return_value=(zip_bytes, zip_file))
284 )
286 out_dir = tmp_path / "out_folder"
287 with pytest.raises(
288 ValueError, match="Did not find a single `assets_export` folder"
289 ):
290 client.download_assets(out_dir)
292 def test_download_assets_raises_if_multiple_assets_export_folders_in_zip(
293 self, tmp_path, client, monkeypatch
294 ):
295 zip_bytes = self._zip_bytes(
296 {
297 "assets_export_111/metadata.yaml": "version: 1.0",
298 "assets_export_222/metadata.yaml": "version: 1.0",
299 }
300 )
301 zip_file = zipfile.ZipFile(io.BytesIO(zip_bytes), "r")
302 monkeypatch.setattr(
303 client, "_get_assets_zip", Mock(return_value=(zip_bytes, zip_file))
304 )
306 out_dir = tmp_path / "out_folder"
307 with pytest.raises(
308 ValueError, match="Did not find a single `assets_export` folder"
309 ):
310 client.download_assets(out_dir)