mirror of
https://github.com/apache/superset.git
synced 2024-09-19 12:09:42 -04:00
91b758f3bc
* reduce data loaded before loading tests * make cypress only load needed tests
384 lines
13 KiB
Python
384 lines
13 KiB
Python
"""Unit tests for Superset"""
|
|
import json
|
|
import unittest
|
|
|
|
from flask import escape
|
|
|
|
from superset import db, security_manager
|
|
from superset.connectors.sqla.models import SqlaTable
|
|
from superset.models import core as models
|
|
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': 'DASHBOARD_CHART_TYPE',
|
|
'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(models.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_dashboard_modes(self):
|
|
self.login(username='admin')
|
|
dash = (
|
|
db.session.query(models.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)
|
|
|
|
def test_save_dash(self, username='admin'):
|
|
self.login(username=username)
|
|
dash = db.session.query(models.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(models.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(models.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(models.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(models.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(models.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(models.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_copy_dash(self, username='admin'):
|
|
self.login(username=username)
|
|
dash = db.session.query(models.Dashboard).filter_by(
|
|
slug='births').first()
|
|
positions = self.get_mock_positions(dash)
|
|
data = {
|
|
'css': '',
|
|
'duplicate_slices': False,
|
|
'expanded_slices': {},
|
|
'positions': positions,
|
|
'dashboard_title': 'Copy Of Births',
|
|
}
|
|
|
|
# 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(models.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_add_slices(self, username='admin'):
|
|
self.login(username=username)
|
|
dash = db.session.query(models.Dashboard).filter_by(
|
|
slug='births').first()
|
|
new_slice = db.session.query(models.Slice).filter_by(
|
|
slice_name='Energy Force Layout').first()
|
|
existing_slice = db.session.query(models.Slice).filter_by(
|
|
slice_name='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(models.Dashboard).filter_by(
|
|
slug='births').first()
|
|
new_slice = db.session.query(models.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(models.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(models.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(models.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()
|
|
)
|
|
# Try access before adding appropriate permissions.
|
|
self.revoke_public_access_to_table(table)
|
|
self.logout()
|
|
|
|
resp = self.get_resp('/chart/list/')
|
|
self.assertNotIn('birth_names</a>', resp)
|
|
|
|
resp = self.get_resp('/dashboard/list/')
|
|
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('/chart/list/'))
|
|
|
|
resp = self.get_resp('/dashboard/list/')
|
|
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('/chart/list/')
|
|
self.assertNotIn('wb_health_population</a>', resp)
|
|
|
|
resp = self.get_resp('/dashboard/list/')
|
|
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(models.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(models.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(models.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(models.Dashboard)
|
|
.filter_by(slug='empty_dashboard')
|
|
.first()
|
|
)
|
|
if not dash:
|
|
dash = models.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('/dashboard/list/')
|
|
self.assertNotIn('/superset/dashboard/empty_dashboard/', resp)
|
|
|
|
dash = (
|
|
db.session
|
|
.query(models.Dashboard)
|
|
.filter_by(slug='empty_dashboard')
|
|
.first()
|
|
)
|
|
dash.owners = [gamma_user]
|
|
db.session.merge(dash)
|
|
db.session.commit()
|
|
|
|
resp = self.get_resp('/dashboard/list/')
|
|
self.assertIn('/superset/dashboard/empty_dashboard/', resp)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|