"""Unit tests for Caravel""" from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals import csv import doctest import imp import json import io import random import unittest from flask import escape from flask_appbuilder.security.sqla import models as ab_models import caravel from caravel import app, db, models, utils, appbuilder, sm from caravel.source_registry import SourceRegistry from caravel.models import DruidDatasource from .base_tests import CaravelTestCase BASE_DIR = app.config.get("BASE_DIR") cli = imp.load_source('cli', BASE_DIR + "/bin/caravel") class CoreTests(CaravelTestCase): def __init__(self, *args, **kwargs): # Load examples first, so that we setup proper permission-view # relations for all example data sources. super(CoreTests, self).__init__(*args, **kwargs) @classmethod def setUpClass(cls): cli.load_examples(load_test_data=True) utils.init(caravel) cls.table_ids = {tbl.table_name: tbl.id for tbl in ( db.session .query(models.SqlaTable) .all() )} def setUp(self): db.session.query(models.Query).delete() db.session.query(models.DatasourceAccessRequest).delete() def tearDown(self): pass def test_admin_only_permissions(self): def assert_admin_permission_in(role_name, assert_func): role = sm.find_role(role_name) permissions = [p.permission.name for p in role.permissions] assert_func('can_sync_druid_source', permissions) assert_func('can_approve', permissions) assert_admin_permission_in('Admin', self.assertIn) assert_admin_permission_in('Alpha', self.assertNotIn) assert_admin_permission_in('Gamma', self.assertNotIn) def test_admin_only_menu_views(self): def assert_admin_view_menus_in(role_name, assert_func): role = sm.find_role(role_name) view_menus = [p.view_menu.name for p in role.permissions] assert_func('ResetPasswordView', view_menus) assert_func('RoleModelView', view_menus) assert_func('Security', view_menus) assert_func('UserDBModelView', view_menus) assert_func('SQL Lab alpha', view_menus) assert_func('AccessRequestsModelView', view_menus) assert_admin_view_menus_in('Admin', self.assertIn) assert_admin_view_menus_in('Alpha', self.assertNotIn) assert_admin_view_menus_in('Gamma', self.assertNotIn) def test_save_slice(self): self.login(username='admin') slc = ( db.session.query(models.Slice.id) .filter_by(slice_name="Energy Sankey") .first()) slice_id = slc.id copy_name = "Test Sankey Save" tbl_id = self.table_ids.get('energy_usage') url = ( "/caravel/explore/table/{}/?viz_type=sankey&groupby=source&" "groupby=target&metric=sum__value&row_limit=5000&where=&having=&" "flt_col_0=source&flt_op_0=in&flt_eq_0=&slice_id={}&slice_name={}&" "collapsed_fieldsets=&action={}&datasource_name=energy_usage&" "datasource_id=1&datasource_type=table&previous_viz_type=sankey") db.session.commit() resp = self.client.get( url.format(tbl_id, slice_id, copy_name, 'save'), follow_redirects=True) assert copy_name in resp.data.decode('utf-8') resp = self.client.get( url.format(tbl_id, slice_id, copy_name, 'overwrite'), follow_redirects=True) assert 'Energy' in resp.data.decode('utf-8') def test_slices(self): # Testing by hitting the two supported end points for all slices self.login(username='admin') Slc = models.Slice urls = [] for slc in db.session.query(Slc).all(): urls += [ (slc.slice_name, 'slice_url', slc.slice_url), (slc.slice_name, 'json_endpoint', slc.viz.json_endpoint), (slc.slice_name, 'csv_endpoint', slc.viz.csv_endpoint), (slc.slice_name, 'slice_id_url', "/caravel/{slc.datasource_type}/{slc.datasource_id}/{slc.id}/".format(slc=slc)), ] for name, method, url in urls: print("[{name}]/[{method}]: {url}".format(**locals())) self.client.get(url) 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_doctests(self): modules = [utils, models] for mod in modules: failed, tests = doctest.testmod(mod) if failed: raise Exception("Failed a doctest") def test_misc(self): assert self.client.get('/health').data.decode('utf-8') == "OK" assert self.client.get('/ping').data.decode('utf-8') == "OK" def test_testconn(self): database = ( db.session .query(models.Database) .filter_by(database_name='main') .first() ) # validate that the endpoint works with the password-masked sqlalchemy uri data = json.dumps({ 'uri': database.safe_sqlalchemy_uri(), 'name': 'main' }) response = self.client.post('/caravel/testconn', data=data, content_type='application/json') assert response.status_code == 200 # validate that the endpoint works with the decrypted sqlalchemy uri data = json.dumps({ 'uri': database.sqlalchemy_uri_decrypted, 'name': 'main' }) response = self.client.post('/caravel/testconn', data=data, content_type='application/json') assert response.status_code == 200 def test_warm_up_cache(self): slice = db.session.query(models.Slice).first() resp = self.client.get( '/caravel/warm_up_cache?slice_id={}'.format(slice.id), follow_redirects=True) data = json.loads(resp.data.decode('utf-8')) assert data == [{'slice_id': slice.id, 'slice_name': slice.slice_name}] resp = self.client.get( '/caravel/warm_up_cache?table_name=energy_usage&db_name=main', follow_redirects=True) data = json.loads(resp.data.decode('utf-8')) assert len(data) == 3 def test_shortner(self): self.login(username='admin') data = ( "//caravel/explore/table/1/?viz_type=sankey&groupby=source&" "groupby=target&metric=sum__value&row_limit=5000&where=&having=&" "flt_col_0=source&flt_op_0=in&flt_eq_0=&slice_id=78&slice_name=" "Energy+Sankey&collapsed_fieldsets=&action=&datasource_name=" "energy_usage&datasource_id=1&datasource_type=table&" "previous_viz_type=sankey" ) resp = self.client.post('/r/shortner/', data=data) assert '/r/' in resp.data.decode('utf-8') def test_save_dash(self, username='admin'): self.login(username=username) dash = db.session.query(models.Dashboard).filter_by( slug="births").first() positions = [] for i, slc in enumerate(dash.slices): d = { 'col': 0, 'row': i * 4, 'size_x': 4, 'size_y': 4, 'slice_id': '{}'.format(slc.id)} positions.append(d) data = { 'css': '', 'expanded_slices': {}, 'positions': positions, } url = '/caravel/save_dash/{}/'.format(dash.id) resp = self.client.post(url, data=dict(data=json.dumps(data))) assert "SUCCESS" in resp.data.decode('utf-8') 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="Mapbox Long/Lat").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 = '/caravel/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="Mapbox Long/Lat").first() assert new_slice in dash.slices assert len(set(dash.slices)) == len(dash.slices) def test_approve(self): session = db.session sm.add_role('table_role') self.login('admin') def prepare_request(ds_type, ds_name, role): ds_class = SourceRegistry.sources[ds_type] # TODO: generalize datasource names if ds_type == 'table': ds = session.query(ds_class).filter( ds_class.table_name == ds_name).first() else: ds = session.query(ds_class).filter( ds_class.datasource_name == ds_name).first() ds_perm_view = sm.find_permission_view_menu( 'datasource_access', ds.perm) sm.add_permission_role(sm.find_role(role), ds_perm_view) access_request = models.DatasourceAccessRequest( datasource_id=ds.id, datasource_type=ds_type, created_by_fk=sm.find_user(username='gamma').id, ) session.add(access_request) session.commit() return access_request EXTEND_ROLE_REQUEST = ( '/caravel/approve?datasource_type={}&datasource_id={}&' 'created_by={}&role_to_extend={}') GRANT_ROLE_REQUEST = ( '/caravel/approve?datasource_type={}&datasource_id={}&' 'created_by={}&role_to_grant={}') # Case 1. Grant new role to the user. access_request1 = prepare_request( 'table', 'unicode_test', 'table_role') ds_1_id = access_request1.datasource_id self.client.get(GRANT_ROLE_REQUEST.format( 'table', ds_1_id, 'gamma', 'table_role')) access_requests = self.get_access_requests('gamma', 'table', ds_1_id) # request was removed self.assertFalse(access_requests) # user was granted table_role user_roles = [r.name for r in sm.find_user('gamma').roles] self.assertIn('table_role', user_roles) # Case 2. Extend the role to have access to the table access_request2 = prepare_request('table', 'long_lat', 'table_role') ds_2_id = access_request2.datasource_id long_lat_perm = access_request2.datasource.perm self.client.get(EXTEND_ROLE_REQUEST.format( 'table', access_request2.datasource_id, 'gamma', 'table_role')) access_requests = self.get_access_requests('gamma', 'table', ds_2_id) # request was removed self.assertFalse(access_requests) # table_role was extended to grant access to the long_lat table/ table_role = sm.find_role('table_role') perm_view = sm.find_permission_view_menu( 'datasource_access', long_lat_perm) self.assertIn(perm_view, table_role.permissions) # Case 3. Grant new role to the user to access the druid datasource. sm.add_role('druid_role') access_request3 = prepare_request('druid', 'druid_ds_1', 'druid_role') self.client.get(GRANT_ROLE_REQUEST.format( 'druid', access_request3.datasource_id, 'gamma', 'druid_role')) # user was granted table_role user_roles = [r.name for r in sm.find_user('gamma').roles] self.assertIn('druid_role', user_roles) # Case 4. Extend the role to have access to the druid datasource access_request4 = prepare_request('druid', 'druid_ds_2', 'druid_role') druid_ds_2_perm = access_request4.datasource.perm self.client.get(EXTEND_ROLE_REQUEST.format( 'druid', access_request4.datasource_id, 'gamma', 'druid_role')) # druid_role was extended to grant access to the druid_access_ds_2 druid_role = sm.find_role('druid_role') perm_view = sm.find_permission_view_menu( 'datasource_access', druid_ds_2_perm) self.assertIn(perm_view, druid_role.permissions) # cleanup gamma_user = sm.find_user(username='gamma') gamma_user.roles.remove(sm.find_role('druid_role')) gamma_user.roles.remove(sm.find_role('table_role')) session.delete(sm.find_role('druid_role')) session.delete(sm.find_role('table_role')) session.commit() def test_request_access(self): session = db.session self.login(username='gamma') gamma_user = sm.find_user(username='gamma') sm.add_role('dummy_role') gamma_user.roles.append(sm.find_role('dummy_role')) session.commit() ACCESS_REQUEST = ( '/caravel/request_access?datasource_type={}&datasource_id={}') ROLE_EXTEND_LINK = ( 'Extend {} Role') ROLE_GRANT_LINK = ( 'Grant {} Role') # Case 1. Request table access, there are no roles have this table. table1 = session.query(models.SqlaTable).filter_by( table_name='random_time_series').first() table_1_id = table1.id # request access to the table self.client.get(ACCESS_REQUEST.format('table', table_1_id)) access_request1 = self.get_access_requests( 'gamma', 'table', table_1_id)[0] approve_link_1 = ROLE_EXTEND_LINK.format( 'table', table_1_id, 'gamma', 'dummy_role', 'dummy_role') self.assertEqual( access_request1.user_roles, '