| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # from superset import db |
| # from superset.models.dashboard import Dashboard |
| import urllib.request |
| from unittest import skipUnless |
| from unittest.mock import patch |
| |
| from flask_testing import LiveServerTestCase |
| from sqlalchemy.sql import func |
| |
| from superset import db, is_feature_enabled, security_manager, thumbnail_cache |
| from superset.extensions import machine_auth_provider_factory |
| from superset.models.dashboard import Dashboard |
| from superset.models.slice import Slice |
| from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot |
| from superset.utils.urls import get_url_path |
| from tests.test_app import app |
| |
| from .base_tests import SupersetTestCase |
| |
| |
| class TestThumbnailsSeleniumLive(LiveServerTestCase): |
| def create_app(self): |
| return app |
| |
| def url_open_auth(self, username: str, url: str): |
| admin_user = security_manager.find_user(username=username) |
| cookies = machine_auth_provider_factory.instance.get_auth_cookies(admin_user) |
| opener = urllib.request.build_opener() |
| opener.addheaders.append(("Cookie", f"session={cookies['session']}")) |
| return opener.open(f"{self.get_server_url()}/{url}") |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_async_dashboard_screenshot(self): |
| """ |
| Thumbnails: Simple get async dashboard screenshot |
| """ |
| dashboard = db.session.query(Dashboard).all()[0] |
| with patch("superset.dashboards.api.DashboardRestApi.get") as mock_get: |
| response = self.url_open_auth( |
| "admin", |
| f"api/v1/dashboard/{dashboard.id}/thumbnail/{dashboard.digest}/", |
| ) |
| self.assertEqual(response.getcode(), 202) |
| |
| |
| class TestThumbnails(SupersetTestCase): |
| |
| mock_image = b"bytes mock image" |
| |
| def test_dashboard_thumbnail_disabled(self): |
| """ |
| Thumbnails: Dashboard thumbnail disabled |
| """ |
| if is_feature_enabled("THUMBNAILS"): |
| return |
| dashboard = db.session.query(Dashboard).all()[0] |
| self.login(username="admin") |
| uri = f"api/v1/dashboard/{dashboard.id}/thumbnail/{dashboard.digest}/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 404) |
| |
| def test_chart_thumbnail_disabled(self): |
| """ |
| Thumbnails: Chart thumbnail disabled |
| """ |
| if is_feature_enabled("THUMBNAILS"): |
| return |
| chart = db.session.query(Slice).all()[0] |
| self.login(username="admin") |
| uri = f"api/v1/chart/{chart}/thumbnail/{chart.digest}/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 404) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_async_dashboard_screenshot(self): |
| """ |
| Thumbnails: Simple get async dashboard screenshot |
| """ |
| dashboard = db.session.query(Dashboard).all()[0] |
| self.login(username="admin") |
| uri = f"api/v1/dashboard/{dashboard.id}/thumbnail/{dashboard.digest}/" |
| with patch( |
| "superset.tasks.thumbnails.cache_dashboard_thumbnail.delay" |
| ) as mock_task: |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 202) |
| mock_task.assert_called_with(dashboard.id, force=True) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_async_dashboard_notfound(self): |
| """ |
| Thumbnails: Simple get async dashboard not found |
| """ |
| max_id = db.session.query(func.max(Dashboard.id)).scalar() |
| self.login(username="admin") |
| uri = f"api/v1/dashboard/{max_id + 1}/thumbnail/1234/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 404) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_async_dashboard_not_allowed(self): |
| """ |
| Thumbnails: Simple get async dashboard not allowed |
| """ |
| dashboard = db.session.query(Dashboard).all()[0] |
| self.login(username="gamma") |
| uri = f"api/v1/dashboard/{dashboard.id}/thumbnail/{dashboard.digest}/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 404) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_async_chart_screenshot(self): |
| """ |
| Thumbnails: Simple get async chart screenshot |
| """ |
| chart = db.session.query(Slice).all()[0] |
| self.login(username="admin") |
| uri = f"api/v1/chart/{chart.id}/thumbnail/{chart.digest}/" |
| with patch( |
| "superset.tasks.thumbnails.cache_chart_thumbnail.delay" |
| ) as mock_task: |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 202) |
| mock_task.assert_called_with(chart.id, force=True) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_async_chart_notfound(self): |
| """ |
| Thumbnails: Simple get async chart not found |
| """ |
| max_id = db.session.query(func.max(Slice.id)).scalar() |
| self.login(username="admin") |
| uri = f"api/v1/chart/{max_id + 1}/thumbnail/1234/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 404) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_cached_chart_wrong_digest(self): |
| """ |
| Thumbnails: Simple get chart with wrong digest |
| """ |
| chart = db.session.query(Slice).all()[0] |
| chart_url = get_url_path("Superset.slice", slice_id=chart.id, standalone="true") |
| # Cache a test "image" |
| screenshot = ChartScreenshot(chart_url, chart.digest) |
| thumbnail_cache.set(screenshot.cache_key, self.mock_image) |
| self.login(username="admin") |
| uri = f"api/v1/chart/{chart.id}/thumbnail/1234/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 302) |
| self.assertRedirects(rv, f"api/v1/chart/{chart.id}/thumbnail/{chart.digest}/") |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_cached_dashboard_screenshot(self): |
| """ |
| Thumbnails: Simple get cached dashboard screenshot |
| """ |
| dashboard = db.session.query(Dashboard).all()[0] |
| dashboard_url = get_url_path("Superset.dashboard", dashboard_id=dashboard.id) |
| # Cache a test "image" |
| screenshot = DashboardScreenshot(dashboard_url, dashboard.digest) |
| thumbnail_cache.set(screenshot.cache_key, self.mock_image) |
| self.login(username="admin") |
| uri = f"api/v1/dashboard/{dashboard.id}/thumbnail/{dashboard.digest}/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 200) |
| self.assertEqual(rv.data, self.mock_image) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_cached_chart_screenshot(self): |
| """ |
| Thumbnails: Simple get cached chart screenshot |
| """ |
| chart = db.session.query(Slice).all()[0] |
| chart_url = get_url_path("Superset.slice", slice_id=chart.id, standalone="true") |
| # Cache a test "image" |
| screenshot = ChartScreenshot(chart_url, chart.digest) |
| thumbnail_cache.set(screenshot.cache_key, self.mock_image) |
| self.login(username="admin") |
| uri = f"api/v1/chart/{chart.id}/thumbnail/{chart.digest}/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 200) |
| self.assertEqual(rv.data, self.mock_image) |
| |
| @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature") |
| def test_get_cached_dashboard_wrong_digest(self): |
| """ |
| Thumbnails: Simple get dashboard with wrong digest |
| """ |
| dashboard = db.session.query(Dashboard).all()[0] |
| dashboard_url = get_url_path("Superset.dashboard", dashboard_id=dashboard.id) |
| # Cache a test "image" |
| screenshot = DashboardScreenshot(dashboard_url, dashboard.digest) |
| thumbnail_cache.set(screenshot.cache_key, self.mock_image) |
| self.login(username="admin") |
| uri = f"api/v1/dashboard/{dashboard.id}/thumbnail/1234/" |
| rv = self.client.get(uri) |
| self.assertEqual(rv.status_code, 302) |
| self.assertRedirects( |
| rv, f"api/v1/dashboard/{dashboard.id}/thumbnail/{dashboard.digest}/" |
| ) |
