mirror of https://github.com/apache/superset.git
554 lines
22 KiB
Python
554 lines
22 KiB
Python
# 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.
|
|
# isort:skip_file
|
|
"""Unit tests for Superset"""
|
|
from datetime import datetime
|
|
import json
|
|
import re
|
|
import unittest
|
|
from random import random
|
|
|
|
import pytest
|
|
from flask import escape, url_for
|
|
from sqlalchemy import func
|
|
|
|
from tests.integration_tests.test_app import app
|
|
from superset import db, security_manager
|
|
from superset.connectors.sqla.models import SqlaTable
|
|
from superset.models import core as models
|
|
from superset.models.dashboard import Dashboard
|
|
from superset.models.slice import Slice
|
|
from tests.integration_tests.fixtures.birth_names_dashboard import (
|
|
load_birth_names_dashboard_with_slices,
|
|
load_birth_names_data,
|
|
)
|
|
from tests.integration_tests.fixtures.energy_dashboard import (
|
|
load_energy_table_with_slice,
|
|
load_energy_table_data,
|
|
)
|
|
from tests.integration_tests.fixtures.public_role import public_role_like_gamma
|
|
from tests.integration_tests.fixtures.unicode_dashboard import (
|
|
load_unicode_dashboard_with_position,
|
|
load_unicode_data,
|
|
)
|
|
from tests.integration_tests.fixtures.world_bank_dashboard import (
|
|
load_world_bank_dashboard_with_slices,
|
|
load_world_bank_data,
|
|
)
|
|
|
|
from .base_tests import SupersetTestCase
|
|
|
|
|
|
class TestDashboard(SupersetTestCase):
|
|
@pytest.fixture
|
|
def cleanup_copied_dash(self):
|
|
with app.app_context():
|
|
original_dashboard = (
|
|
db.session.query(Dashboard).filter_by(slug="births").first()
|
|
)
|
|
original_dashboard_id = original_dashboard.id
|
|
yield
|
|
copied_dashboard = (
|
|
db.session.query(Dashboard)
|
|
.filter(
|
|
Dashboard.dashboard_title == "Copy Of Births",
|
|
Dashboard.id != original_dashboard_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
db.session.merge(original_dashboard)
|
|
if copied_dashboard:
|
|
db.session.delete(copied_dashboard)
|
|
db.session.commit()
|
|
|
|
@pytest.fixture
|
|
def load_dashboard(self):
|
|
with app.app_context():
|
|
table = (
|
|
db.session.query(SqlaTable).filter_by(table_name="energy_usage").one()
|
|
)
|
|
# get a slice from the allowed table
|
|
slice = db.session.query(Slice).filter_by(slice_name="Energy Sankey").one()
|
|
|
|
self.grant_public_access_to_table(table)
|
|
|
|
pytest.hidden_dash_slug = f"hidden_dash_{random()}"
|
|
pytest.published_dash_slug = f"published_dash_{random()}"
|
|
|
|
# Create a published and hidden dashboard and add them to the database
|
|
published_dash = Dashboard()
|
|
published_dash.dashboard_title = "Published Dashboard"
|
|
published_dash.slug = pytest.published_dash_slug
|
|
published_dash.slices = [slice]
|
|
published_dash.published = True
|
|
|
|
hidden_dash = Dashboard()
|
|
hidden_dash.dashboard_title = "Hidden Dashboard"
|
|
hidden_dash.slug = pytest.hidden_dash_slug
|
|
hidden_dash.slices = [slice]
|
|
hidden_dash.published = False
|
|
|
|
db.session.merge(published_dash)
|
|
db.session.merge(hidden_dash)
|
|
yield db.session.commit()
|
|
|
|
self.revoke_public_access_to_table(table)
|
|
db.session.delete(published_dash)
|
|
db.session.delete(hidden_dash)
|
|
db.session.commit()
|
|
|
|
def get_mock_positions(self, dash):
|
|
positions = {"DASHBOARD_VERSION_KEY": "v2"}
|
|
for i, slc in enumerate(dash.slices):
|
|
id = "DASHBOARD_CHART_TYPE-{}".format(i)
|
|
d = {
|
|
"type": "CHART",
|
|
"id": id,
|
|
"children": [],
|
|
"meta": {"width": 4, "height": 50, "chartId": slc.id},
|
|
}
|
|
positions[id] = d
|
|
return positions
|
|
|
|
def test_dashboard(self):
|
|
self.login(username="admin")
|
|
urls = {}
|
|
for dash in db.session.query(Dashboard).all():
|
|
urls[dash.dashboard_title] = dash.url
|
|
for title, url in urls.items():
|
|
assert escape(title) in self.client.get(url).data.decode("utf-8")
|
|
|
|
def test_superset_dashboard_url(self):
|
|
url_for("Superset.dashboard", dashboard_id_or_slug=1)
|
|
|
|
def test_new_dashboard(self):
|
|
self.login(username="admin")
|
|
dash_count_before = db.session.query(func.count(Dashboard.id)).first()[0]
|
|
url = "/dashboard/new/"
|
|
response = self.client.get(url, follow_redirects=False)
|
|
dash_count_after = db.session.query(func.count(Dashboard.id)).first()[0]
|
|
self.assertEqual(dash_count_before + 1, dash_count_after)
|
|
group = re.match(
|
|
r"http:\/\/localhost\/superset\/dashboard\/([0-9]*)\/\?edit=true",
|
|
response.headers["Location"],
|
|
)
|
|
assert group is not None
|
|
|
|
# Cleanup
|
|
created_dashboard_id = int(group[1])
|
|
created_dashboard = db.session.query(Dashboard).get(created_dashboard_id)
|
|
db.session.delete(created_dashboard)
|
|
db.session.commit()
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
def test_save_dash(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
positions = self.get_mock_positions(dash)
|
|
data = {
|
|
"css": "",
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": dash.dashboard_title,
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
url = "/superset/save_dash/{}/".format(dash.id)
|
|
resp = self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
self.assertIn("SUCCESS", resp)
|
|
|
|
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
|
|
def test_save_dash_with_filter(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="world_health").first()
|
|
|
|
positions = self.get_mock_positions(dash)
|
|
filters = {str(dash.slices[0].id): {"region": ["North America"]}}
|
|
default_filters = json.dumps(filters)
|
|
data = {
|
|
"css": "",
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": dash.dashboard_title,
|
|
"default_filters": default_filters,
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
|
|
url = "/superset/save_dash/{}/".format(dash.id)
|
|
resp = self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
self.assertIn("SUCCESS", resp)
|
|
|
|
updatedDash = db.session.query(Dashboard).filter_by(slug="world_health").first()
|
|
new_url = updatedDash.url
|
|
self.assertIn("world_health", new_url)
|
|
self.assertNotIn("preselect_filters", new_url)
|
|
|
|
@pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
|
|
def test_save_dash_with_invalid_filters(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="world_health").first()
|
|
|
|
# add an invalid filter slice
|
|
positions = self.get_mock_positions(dash)
|
|
filters = {str(99999): {"region": ["North America"]}}
|
|
default_filters = json.dumps(filters)
|
|
data = {
|
|
"css": "",
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": dash.dashboard_title,
|
|
"default_filters": default_filters,
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
|
|
url = "/superset/save_dash/{}/".format(dash.id)
|
|
resp = self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
self.assertIn("SUCCESS", resp)
|
|
|
|
updatedDash = db.session.query(Dashboard).filter_by(slug="world_health").first()
|
|
new_url = updatedDash.url
|
|
self.assertNotIn("region", new_url)
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
def test_save_dash_with_dashboard_title(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
origin_title = dash.dashboard_title
|
|
positions = self.get_mock_positions(dash)
|
|
data = {
|
|
"css": "",
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": "new title",
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
url = "/superset/save_dash/{}/".format(dash.id)
|
|
self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
updatedDash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
self.assertEqual(updatedDash.dashboard_title, "new title")
|
|
# bring back dashboard original title
|
|
data["dashboard_title"] = origin_title
|
|
self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
def test_save_dash_with_colors(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
positions = self.get_mock_positions(dash)
|
|
new_label_colors = {"data value": "random color"}
|
|
data = {
|
|
"css": "",
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": dash.dashboard_title,
|
|
"color_namespace": "Color Namespace Test",
|
|
"color_scheme": "Color Scheme Test",
|
|
"label_colors": new_label_colors,
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
url = "/superset/save_dash/{}/".format(dash.id)
|
|
self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
updatedDash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
self.assertIn("color_namespace", updatedDash.json_metadata)
|
|
self.assertIn("color_scheme", updatedDash.json_metadata)
|
|
self.assertIn("label_colors", updatedDash.json_metadata)
|
|
# bring back original dashboard
|
|
del data["color_namespace"]
|
|
del data["color_scheme"]
|
|
del data["label_colors"]
|
|
self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
|
|
@pytest.mark.usefixtures(
|
|
"load_birth_names_dashboard_with_slices",
|
|
"cleanup_copied_dash",
|
|
"load_unicode_dashboard_with_position",
|
|
)
|
|
def test_copy_dash(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
positions = self.get_mock_positions(dash)
|
|
new_label_colors = {"data value": "random color"}
|
|
data = {
|
|
"css": "",
|
|
"duplicate_slices": False,
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": "Copy Of Births",
|
|
"color_namespace": "Color Namespace Test",
|
|
"color_scheme": "Color Scheme Test",
|
|
"label_colors": new_label_colors,
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
|
|
# Save changes to Births dashboard and retrieve updated dash
|
|
dash_id = dash.id
|
|
url = "/superset/save_dash/{}/".format(dash_id)
|
|
self.client.post(url, data=dict(data=json.dumps(data)))
|
|
dash = db.session.query(Dashboard).filter_by(id=dash_id).first()
|
|
orig_json_data = dash.data
|
|
|
|
# Verify that copy matches original
|
|
url = "/superset/copy_dash/{}/".format(dash_id)
|
|
resp = self.get_json_resp(url, data=dict(data=json.dumps(data)))
|
|
self.assertEqual(resp["dashboard_title"], "Copy Of Births")
|
|
self.assertEqual(resp["position_json"], orig_json_data["position_json"])
|
|
self.assertEqual(resp["metadata"], orig_json_data["metadata"])
|
|
# check every attribute in each dashboard's slices list,
|
|
# exclude modified and changed_on attribute
|
|
for index, slc in enumerate(orig_json_data["slices"]):
|
|
for key in slc:
|
|
if key not in ["modified", "changed_on", "changed_on_humanized"]:
|
|
self.assertEqual(slc[key], resp["slices"][index][key])
|
|
|
|
@pytest.mark.usefixtures(
|
|
"load_energy_table_with_slice", "load_birth_names_dashboard_with_slices"
|
|
)
|
|
def test_add_slices(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
new_slice = (
|
|
db.session.query(Slice).filter_by(slice_name="Energy Force Layout").first()
|
|
)
|
|
existing_slice = (
|
|
db.session.query(Slice).filter_by(slice_name="Girl Name Cloud").first()
|
|
)
|
|
data = {
|
|
"slice_ids": [new_slice.data["slice_id"], existing_slice.data["slice_id"]]
|
|
}
|
|
url = "/superset/add_slices/{}/".format(dash.id)
|
|
resp = self.client.post(url, data=dict(data=json.dumps(data)))
|
|
assert "SLICES ADDED" in resp.data.decode("utf-8")
|
|
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
new_slice = (
|
|
db.session.query(Slice).filter_by(slice_name="Energy Force Layout").first()
|
|
)
|
|
assert new_slice in dash.slices
|
|
assert len(set(dash.slices)) == len(dash.slices)
|
|
|
|
# cleaning up
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
dash.slices = [o for o in dash.slices if o.slice_name != "Energy Force Layout"]
|
|
db.session.commit()
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
def test_remove_slices(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
origin_slices_length = len(dash.slices)
|
|
|
|
positions = self.get_mock_positions(dash)
|
|
# remove one chart
|
|
chart_keys = []
|
|
for key in positions.keys():
|
|
if key.startswith("DASHBOARD_CHART_TYPE"):
|
|
chart_keys.append(key)
|
|
positions.pop(chart_keys[0])
|
|
|
|
data = {
|
|
"css": "",
|
|
"expanded_slices": {},
|
|
"positions": positions,
|
|
"dashboard_title": dash.dashboard_title,
|
|
# set a further modified_time for unit test
|
|
"last_modified_time": datetime.now().timestamp() + 1000,
|
|
}
|
|
|
|
# save dash
|
|
dash_id = dash.id
|
|
url = "/superset/save_dash/{}/".format(dash_id)
|
|
self.client.post(url, data=dict(data=json.dumps(data)))
|
|
dash = db.session.query(Dashboard).filter_by(id=dash_id).first()
|
|
|
|
# verify slices data
|
|
data = dash.data
|
|
self.assertEqual(len(data["slices"]), origin_slices_length - 1)
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
@pytest.mark.usefixtures("public_role_like_gamma")
|
|
def test_public_user_dashboard_access(self):
|
|
table = db.session.query(SqlaTable).filter_by(table_name="birth_names").one()
|
|
|
|
# Make the births dash published so it can be seen
|
|
births_dash = db.session.query(Dashboard).filter_by(slug="births").one()
|
|
births_dash.published = True
|
|
|
|
db.session.merge(births_dash)
|
|
db.session.commit()
|
|
|
|
# Try access before adding appropriate permissions.
|
|
self.revoke_public_access_to_table(table)
|
|
self.logout()
|
|
|
|
resp = self.get_resp("/api/v1/chart/")
|
|
self.assertNotIn("birth_names", resp)
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertNotIn("/superset/dashboard/births/", resp)
|
|
|
|
self.grant_public_access_to_table(table)
|
|
|
|
# Try access after adding appropriate permissions.
|
|
self.assertIn("birth_names", self.get_resp("/api/v1/chart/"))
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertIn("/superset/dashboard/births/", resp)
|
|
|
|
# Confirm that public doesn't have access to other datasets.
|
|
resp = self.get_resp("/api/v1/chart/")
|
|
self.assertNotIn("wb_health_population", resp)
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertNotIn("/superset/dashboard/world_health/", resp)
|
|
|
|
# Cleanup
|
|
self.revoke_public_access_to_table(table)
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
@pytest.mark.usefixtures("public_role_like_gamma")
|
|
def test_dashboard_with_created_by_can_be_accessed_by_public_users(self):
|
|
self.logout()
|
|
table = db.session.query(SqlaTable).filter_by(table_name="birth_names").one()
|
|
self.grant_public_access_to_table(table)
|
|
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
dash.owners = [security_manager.find_user("admin")]
|
|
dash.created_by = security_manager.find_user("admin")
|
|
db.session.merge(dash)
|
|
db.session.commit()
|
|
|
|
# this asserts a non-4xx response
|
|
self.get_resp("/superset/dashboard/births/")
|
|
# Cleanup
|
|
self.revoke_public_access_to_table(table)
|
|
|
|
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
|
|
def test_only_owners_can_save(self):
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
dash.owners = []
|
|
db.session.merge(dash)
|
|
db.session.commit()
|
|
self.test_save_dash("admin")
|
|
|
|
self.logout()
|
|
self.assertRaises(Exception, self.test_save_dash, "alpha")
|
|
|
|
alpha = security_manager.find_user("alpha")
|
|
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
dash.owners = [alpha]
|
|
db.session.merge(dash)
|
|
db.session.commit()
|
|
self.test_save_dash("alpha")
|
|
|
|
@pytest.mark.usefixtures("load_energy_table_with_slice", "load_dashboard")
|
|
def test_users_can_list_published_dashboard(self):
|
|
self.login("alpha")
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
assert f"/superset/dashboard/{pytest.hidden_dash_slug}/" not in resp
|
|
assert f"/superset/dashboard/{pytest.published_dash_slug}/" in resp
|
|
|
|
def test_users_can_view_own_dashboard(self):
|
|
user = security_manager.find_user("gamma")
|
|
my_dash_slug = f"my_dash_{random()}"
|
|
not_my_dash_slug = f"not_my_dash_{random()}"
|
|
|
|
# Create one dashboard I own and another that I don't
|
|
dash = Dashboard()
|
|
dash.dashboard_title = "My Dashboard"
|
|
dash.slug = my_dash_slug
|
|
dash.owners = [user]
|
|
dash.slices = []
|
|
|
|
hidden_dash = Dashboard()
|
|
hidden_dash.dashboard_title = "Not My Dashboard"
|
|
hidden_dash.slug = not_my_dash_slug
|
|
hidden_dash.slices = []
|
|
hidden_dash.owners = []
|
|
|
|
db.session.merge(dash)
|
|
db.session.merge(hidden_dash)
|
|
db.session.commit()
|
|
|
|
self.login(user.username)
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertIn(f"/superset/dashboard/{my_dash_slug}/", resp)
|
|
self.assertNotIn(f"/superset/dashboard/{not_my_dash_slug}/", resp)
|
|
|
|
def test_users_can_view_favorited_dashboards(self):
|
|
user = security_manager.find_user("gamma")
|
|
fav_dash_slug = f"my_favorite_dash_{random()}"
|
|
regular_dash_slug = f"regular_dash_{random()}"
|
|
|
|
favorite_dash = Dashboard()
|
|
favorite_dash.dashboard_title = "My Favorite Dashboard"
|
|
favorite_dash.slug = fav_dash_slug
|
|
|
|
regular_dash = Dashboard()
|
|
regular_dash.dashboard_title = "A Plain Ol Dashboard"
|
|
regular_dash.slug = regular_dash_slug
|
|
|
|
db.session.merge(favorite_dash)
|
|
db.session.merge(regular_dash)
|
|
db.session.commit()
|
|
|
|
dash = db.session.query(Dashboard).filter_by(slug=fav_dash_slug).first()
|
|
|
|
favorites = models.FavStar()
|
|
favorites.obj_id = dash.id
|
|
favorites.class_name = "Dashboard"
|
|
favorites.user_id = user.id
|
|
|
|
db.session.merge(favorites)
|
|
db.session.commit()
|
|
|
|
self.login(user.username)
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertIn(f"/superset/dashboard/{fav_dash_slug}/", resp)
|
|
|
|
def test_user_can_not_view_unpublished_dash(self):
|
|
admin_user = security_manager.find_user("admin")
|
|
gamma_user = security_manager.find_user("gamma")
|
|
slug = f"admin_owned_unpublished_dash_{random()}"
|
|
|
|
# Create a dashboard owned by admin and unpublished
|
|
dash = Dashboard()
|
|
dash.dashboard_title = "My Dashboard"
|
|
dash.slug = slug
|
|
dash.owners = [admin_user]
|
|
dash.slices = []
|
|
dash.published = False
|
|
db.session.merge(dash)
|
|
db.session.commit()
|
|
|
|
# list dashboards as a gamma user
|
|
self.login(gamma_user.username)
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertNotIn(f"/superset/dashboard/{slug}/", resp)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|