mirror of https://github.com/apache/superset.git
551 lines
20 KiB
Python
551 lines
20 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"""
|
|
import copy
|
|
import json
|
|
import unittest
|
|
from random import random
|
|
|
|
from flask import escape
|
|
from sqlalchemy import func
|
|
from typing import Dict
|
|
|
|
import tests.test_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 superset.views import core as views
|
|
|
|
from .base_tests import SupersetTestCase
|
|
|
|
|
|
class DashboardTests(SupersetTestCase):
|
|
def __init__(self, *args, **kwargs):
|
|
super(DashboardTests, self).__init__(*args, **kwargs)
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
pass
|
|
|
|
def setUp(self):
|
|
pass
|
|
|
|
def tearDown(self):
|
|
pass
|
|
|
|
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_new_dashboard(self):
|
|
self.login(username="admin")
|
|
dash_count_before = db.session.query(func.count(Dashboard.id)).first()[0]
|
|
url = "/dashboard/new/"
|
|
resp = self.get_resp(url)
|
|
self.assertIn("[ untitled dashboard ]", resp)
|
|
dash_count_after = db.session.query(func.count(Dashboard.id)).first()[0]
|
|
self.assertEqual(dash_count_before + 1, dash_count_after)
|
|
|
|
def test_dashboard_modes(self):
|
|
self.login(username="admin")
|
|
dash = db.session.query(Dashboard).filter_by(slug="births").first()
|
|
url = dash.url
|
|
if dash.url.find("?") == -1:
|
|
url += "?"
|
|
else:
|
|
url += "&"
|
|
resp = self.get_resp(url + "edit=true&standalone=true")
|
|
self.assertIn("editMode": true", resp)
|
|
self.assertIn("standalone_mode": true", resp)
|
|
self.assertIn('<body class="standalone">', resp)
|
|
|
|
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,
|
|
}
|
|
url = "/superset/save_dash/{}/".format(dash.id)
|
|
resp = self.get_resp(url, data=dict(data=json.dumps(data)))
|
|
self.assertIn("SUCCESS", resp)
|
|
|
|
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,
|
|
}
|
|
|
|
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("region", new_url)
|
|
|
|
resp = self.get_resp(new_url)
|
|
self.assertIn("North America", resp)
|
|
|
|
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,
|
|
}
|
|
|
|
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)
|
|
|
|
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",
|
|
}
|
|
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)))
|
|
|
|
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,
|
|
}
|
|
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)))
|
|
|
|
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,
|
|
}
|
|
|
|
# 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"]:
|
|
self.assertEqual(slc[key], resp["slices"][index][key])
|
|
|
|
def test_set_dash_metadata(self, username="admin"):
|
|
self.login(username=username)
|
|
dash = db.session.query(Dashboard).filter_by(slug="world_health").first()
|
|
data = dash.data
|
|
positions = data["position_json"]
|
|
data.update({"positions": positions})
|
|
original_data = copy.deepcopy(data)
|
|
|
|
# add filter scopes
|
|
filter_slice = dash.slices[0]
|
|
immune_slices = dash.slices[2:]
|
|
filter_scopes = {
|
|
str(filter_slice.id): {
|
|
"region": {
|
|
"scope": ["ROOT_ID"],
|
|
"immune": [slc.id for slc in immune_slices],
|
|
}
|
|
}
|
|
}
|
|
data.update({"filter_scopes": json.dumps(filter_scopes)})
|
|
views.Superset._set_dash_metadata(dash, data)
|
|
updated_metadata = json.loads(dash.json_metadata)
|
|
self.assertEqual(updated_metadata["filter_scopes"], filter_scopes)
|
|
|
|
# remove a slice and change slice ids (as copy slices)
|
|
removed_slice = immune_slices.pop()
|
|
removed_component = [
|
|
key
|
|
for (key, value) in positions.items()
|
|
if isinstance(value, dict)
|
|
and value.get("type") == "CHART"
|
|
and value["meta"]["chartId"] == removed_slice.id
|
|
]
|
|
positions.pop(removed_component[0], None)
|
|
|
|
data.update({"positions": positions})
|
|
old_to_new_sliceids = {slc.id: slc.id + 10 for slc in dash.slices}
|
|
views.Superset._set_dash_metadata(dash, data, old_to_new_sliceids)
|
|
updated_metadata = json.loads(dash.json_metadata)
|
|
updated_filter_scopes = {
|
|
str(filter_slice.id + 10): {
|
|
"region": {
|
|
"scope": ["ROOT_ID"],
|
|
"immune": [slc.id + 10 for slc in immune_slices],
|
|
}
|
|
}
|
|
}
|
|
self.assertEqual(updated_metadata["filter_scopes"], updated_filter_scopes)
|
|
|
|
# reset dash to original data
|
|
views.Superset._set_dash_metadata(dash, original_data)
|
|
|
|
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()
|
|
|
|
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,
|
|
}
|
|
|
|
# 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)
|
|
|
|
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)
|
|
|
|
self.assertIn("Births", self.get_resp("/superset/dashboard/births/"))
|
|
|
|
# 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)
|
|
|
|
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()
|
|
|
|
assert "Births" in self.get_resp("/superset/dashboard/births/")
|
|
|
|
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")
|
|
|
|
def test_owners_can_view_empty_dashboard(self):
|
|
dash = db.session.query(Dashboard).filter_by(slug="empty_dashboard").first()
|
|
if not dash:
|
|
dash = Dashboard()
|
|
dash.dashboard_title = "Empty Dashboard"
|
|
dash.slug = "empty_dashboard"
|
|
else:
|
|
dash.slices = []
|
|
dash.owners = []
|
|
db.session.merge(dash)
|
|
db.session.commit()
|
|
|
|
gamma_user = security_manager.find_user("gamma")
|
|
self.login(gamma_user.username)
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertNotIn("/superset/dashboard/empty_dashboard/", resp)
|
|
|
|
def test_users_can_view_published_dashboard(self):
|
|
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)
|
|
|
|
hidden_dash_slug = f"hidden_dash_{random()}"
|
|
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 = published_dash_slug
|
|
published_dash.slices = [slice]
|
|
published_dash.published = True
|
|
|
|
hidden_dash = Dashboard()
|
|
hidden_dash.dashboard_title = "Hidden Dashboard"
|
|
hidden_dash.slug = hidden_dash_slug
|
|
hidden_dash.slices = [slice]
|
|
hidden_dash.published = False
|
|
|
|
db.session.merge(published_dash)
|
|
db.session.merge(hidden_dash)
|
|
db.session.commit()
|
|
|
|
resp = self.get_resp("/api/v1/dashboard/")
|
|
self.assertNotIn(f"/superset/dashboard/{hidden_dash_slug}/", resp)
|
|
self.assertIn(f"/superset/dashboard/{published_dash_slug}/", 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()
|