# 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. """Unit tests for Superset""" import json import unittest from flask import escape from sqlalchemy import func 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_new_dashboard(self): self.login(username='admin') dash_count_before = db.session.query(func.count(models.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(models.Dashboard.id)).first()[0] self.assertEquals(dash_count_before + 1, dash_count_after) 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) self.assertIn('', 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', 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', 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()