diff --git a/tests/base_tests.py b/tests/base_tests.py index 45abf8433f..0b2671826e 100644 --- a/tests/base_tests.py +++ b/tests/base_tests.py @@ -176,3 +176,14 @@ class CaravelTestCase(unittest.TestCase): if (perm.permission.name == 'datasource_access' and perm.view_menu and table_name in perm.view_menu.name): appbuilder.sm.del_permission_role(public_role, perm) + + def run_sql(self, sql, user_name, client_id): + self.login(username=(user_name if user_name else 'admin')) + dbid = self.get_main_database(db.session).id + resp = self.client.post( + '/caravel/sql_json/', + data=dict(database_id=dbid, sql=sql, select_as_create_as=False, + client_id=client_id), + ) + self.logout() + return json.loads(resp.data.decode('utf-8')) diff --git a/tests/core_tests.py b/tests/core_tests.py index e65522ea1d..1d58dde3df 100644 --- a/tests/core_tests.py +++ b/tests/core_tests.py @@ -41,17 +41,9 @@ class CoreTests(CaravelTestCase): def setUp(self): db.session.query(models.Query).delete() db.session.query(models.DatasourceAccessRequest).delete() - self.login('admin') - self.run_sql("SELECT * FROM ab_user", 'admin', client_id='client_id_1') - self.run_sql("SELECT * FROM NO_TABLE", 'admin', client_id='client_id_3') - self.logout() - - self.login('gamma') - self.run_sql("SELECT * FROM ab_permission", 'gamma', client_id='client_id_2') - self.logout() def tearDown(self): - pass + db.session.query(models.Query).delete() def test_welcome(self): self.login() @@ -271,51 +263,6 @@ class CoreTests(CaravelTestCase): assert "List Slice" in self.get_resp('/slicemodelview/list/') assert "List Dashboard" in self.get_resp('/dashboardmodelview/list/') - def run_sql(self, sql, user_name, client_id): - self.login(username=user_name) - dbid = self.get_main_database(db.session).id - resp = self.client.post( - '/caravel/sql_json/', - data=dict(database_id=dbid, sql=sql, select_as_create_as=False, - client_id=client_id), - ) - self.logout() - return json.loads(resp.data.decode('utf-8')) - - def test_sql_json(self): - data = self.run_sql('SELECT * FROM ab_user', 'admin', "1") - assert len(data['data']) > 0 - - data = self.run_sql('SELECT * FROM unexistant_table', 'admin', "2") - assert len(data['error']) > 0 - - def test_sql_json_has_access(self): - main_db = self.get_main_database(db.session) - utils.merge_perm(sm, 'database_access', main_db.perm) - db.session.commit() - main_db_permission_view = ( - db.session.query(ab_models.PermissionView) - .join(ab_models.ViewMenu) - .filter(ab_models.ViewMenu.name == '[main].(id:1)') - .first() - ) - astronaut = sm.add_role("Astronaut") - sm.add_permission_role(astronaut, main_db_permission_view) - # Astronaut role is Gamma + main db permissions - for gamma_perm in sm.find_role('Gamma').permissions: - sm.add_permission_role(astronaut, gamma_perm) - - gagarin = appbuilder.sm.find_user('gagarin') - if not gagarin: - appbuilder.sm.add_user( - 'gagarin', 'Iurii', 'Gagarin', 'gagarin@cosmos.ussr', - appbuilder.sm.find_role('Astronaut'), - password='general') - data = self.run_sql('SELECT * FROM ab_user', 'gagarin', "3") - db.session.query(models.Query).delete() - db.session.commit() - assert len(data['data']) > 0 - def test_csv_endpoint(self): sql = """ SELECT first_name, last_name @@ -334,105 +281,6 @@ class CoreTests(CaravelTestCase): self.assertEqual(list(expected_data), list(data)) self.logout() - def test_queries_endpoint(self): - resp = self.client.get('/caravel/queries/{}'.format(0)) - self.assertEquals(403, resp.status_code) - - self.login('admin') - data = self.get_json_resp('/caravel/queries/{}'.format(0)) - self.assertEquals(2, len(data)) - self.logout() - - self.run_sql("SELECT * FROM ab_user1", 'admin', client_id='client_id_4') - self.run_sql("SELECT * FROM ab_user2", 'admin', client_id='client_id_5') - self.login('admin') - data = self.get_json_resp('/caravel/queries/{}'.format(0)) - self.assertEquals(4, len(data)) - - query = db.session.query(models.Query).filter_by( - sql='SELECT * FROM ab_user1').first() - query.changed_on = utils.EPOCH - db.session.commit() - - data = self.get_json_resp('/caravel/queries/{}'.format(123456000)) - self.assertEquals(3, len(data)) - - self.logout() - resp = self.client.get('/caravel/queries/{}'.format(0)) - self.assertEquals(403, resp.status_code) - - def test_search_query_on_db_id(self): - self.login('admin') - # Test search queries on database Id - resp = self.get_resp('/caravel/search_queries?database_id=1') - data = json.loads(resp) - self.assertEquals(3, len(data)) - db_ids = [data[k]['dbId'] for k in data] - self.assertEquals([1, 1, 1], db_ids) - - resp = self.get_resp('/caravel/search_queries?database_id=-1') - data = json.loads(resp) - self.assertEquals(0, len(data)) - self.logout() - - def test_search_query_on_user(self): - self.login('admin') - # Test search queries on user Id - user = appbuilder.sm.find_user('admin') - resp = self.get_resp('/caravel/search_queries?user_id={}'.format(user.id)) - data = json.loads(resp) - self.assertEquals(2, len(data)) - user_ids = [data[k]['userId'] for k in data] - self.assertEquals([user.id, user.id], user_ids) - - user = appbuilder.sm.find_user('gamma') - resp = self.get_resp('/caravel/search_queries?user_id={}'.format(user.id)) - data = json.loads(resp) - self.assertEquals(1, len(data)) - self.assertEquals(list(data.values())[0]['userId'] , user.id) - self.logout() - - def test_search_query_on_status(self): - self.login('admin') - # Test search queries on status - resp = self.get_resp('/caravel/search_queries?status=success') - data = json.loads(resp) - self.assertEquals(2, len(data)) - states = [data[k]['state'] for k in data] - self.assertEquals(['success', 'success'], states) - - resp = self.get_resp('/caravel/search_queries?status=failed') - data = json.loads(resp) - self.assertEquals(1, len(data)) - self.assertEquals(list(data.values())[0]['state'], 'failed') - self.logout() - - def test_search_query_on_text(self): - self.login('admin') - resp = self.get_resp('/caravel/search_queries?search_text=permission') - data = json.loads(resp) - self.assertEquals(1, len(data)) - self.assertIn('permission', list(data.values())[0]['sql']) - self.logout() - - def test_search_query_on_time(self): - self.login('admin') - first_query_time = db.session.query(models.Query).filter_by( - sql='SELECT * FROM ab_user').first().start_time - second_query_time = db.session.query(models.Query).filter_by( - sql='SELECT * FROM ab_permission').first().start_time - # Test search queries on time filter - from_time = 'from={}'.format(int(first_query_time)) - to_time = 'to={}'.format(int(second_query_time)) - params = [from_time, to_time] - resp = self.get_resp('/caravel/search_queries?'+'&'.join(params)) - data = json.loads(resp) - self.assertEquals(2, len(data)) - for _, v in data.items(): - self.assertLess(int(first_query_time), v['startDttm']) - self.assertLess(v['startDttm'], int(second_query_time)) - self.logout() - def test_public_user_dashboard_access(self): # Try access before adding appropriate permissions. self.revoke_public_access('birth_names') diff --git a/tests/sqllab_tests.py b/tests/sqllab_tests.py new file mode 100644 index 0000000000..6323410c12 --- /dev/null +++ b/tests/sqllab_tests.py @@ -0,0 +1,167 @@ +"""Unit tests for Sql Lab""" +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function +from __future__ import unicode_literals + +import csv +import json +import io +import unittest + +from flask_appbuilder.security.sqla import models as ab_models +from caravel import db, models, utils, appbuilder, sm +from .base_tests import CaravelTestCase + + +class SqlLabTests(CaravelTestCase): + """Testings for Sql Lab""" + + def __init__(self, *args, **kwargs): + super(SqlLabTests, self).__init__(*args, **kwargs) + + def setUp(self): + db.session.query(models.Query).delete() + self.run_sql("SELECT * FROM ab_user", 'admin', client_id='client_id_1') + self.run_sql("SELECT * FROM NO_TABLE", 'admin', client_id='client_id_3') + self.run_sql("SELECT * FROM ab_permission", 'gamma', client_id='client_id_2') + + def tearDown(self): + db.session.query(models.Query).delete() + + def test_sql_json(self): + data = self.run_sql('SELECT * FROM ab_user', 'admin', "1") + self.assertLess(0, len(data['data'])) + + data = self.run_sql('SELECT * FROM unexistant_table', 'admin', "2") + self.assertLess(0, len(data['error'])) + + def test_sql_json_has_access(self): + main_db = self.get_main_database(db.session) + utils.merge_perm(sm, 'database_access', main_db.perm) + db.session.commit() + main_db_permission_view = ( + db.session.query(ab_models.PermissionView) + .join(ab_models.ViewMenu) + .filter(ab_models.ViewMenu.name == '[main].(id:1)') + .first() + ) + astronaut = sm.add_role("Astronaut") + sm.add_permission_role(astronaut, main_db_permission_view) + # Astronaut role is Gamma + main db permissions + for gamma_perm in sm.find_role('Gamma').permissions: + sm.add_permission_role(astronaut, gamma_perm) + + gagarin = appbuilder.sm.find_user('gagarin') + if not gagarin: + appbuilder.sm.add_user( + 'gagarin', 'Iurii', 'Gagarin', 'gagarin@cosmos.ussr', + appbuilder.sm.find_role('Astronaut'), + password='general') + data = self.run_sql('SELECT * FROM ab_user', 'gagarin', "3") + db.session.query(models.Query).delete() + db.session.commit() + self.assertLess(0, len(data['data'])) + + def test_queries_endpoint(self): + resp = self.client.get('/caravel/queries/{}'.format(0)) + self.assertEquals(403, resp.status_code) + + self.login('admin') + data = self.get_json_resp('/caravel/queries/{}'.format(0)) + self.assertEquals(2, len(data)) + self.logout() + + self.run_sql("SELECT * FROM ab_user1", 'admin', client_id='client_id_4') + self.run_sql("SELECT * FROM ab_user2", 'admin', client_id='client_id_5') + self.login('admin') + data = self.get_json_resp('/caravel/queries/{}'.format(0)) + self.assertEquals(4, len(data)) + + query = db.session.query(models.Query).filter_by( + sql='SELECT * FROM ab_user1').first() + query.changed_on = utils.EPOCH + db.session.commit() + + data = self.get_json_resp('/caravel/queries/{}'.format(123456000)) + self.assertEquals(3, len(data)) + + self.logout() + resp = self.client.get('/caravel/queries/{}'.format(0)) + self.assertEquals(403, resp.status_code) + + def test_search_query_on_db_id(self): + self.login('admin') + # Test search queries on database Id + resp = self.get_resp('/caravel/search_queries?database_id=1') + data = json.loads(resp) + self.assertEquals(3, len(data)) + db_ids = [data[k]['dbId'] for k in data] + self.assertEquals([1, 1, 1], db_ids) + + resp = self.get_resp('/caravel/search_queries?database_id=-1') + data = json.loads(resp) + self.assertEquals(0, len(data)) + self.logout() + + def test_search_query_on_user(self): + self.login('admin') + # Test search queries on user Id + user = appbuilder.sm.find_user('admin') + resp = self.get_resp('/caravel/search_queries?user_id={}'.format(user.id)) + data = json.loads(resp) + self.assertEquals(2, len(data)) + user_ids = [data[k]['userId'] for k in data] + self.assertEquals([user.id, user.id], user_ids) + + user = appbuilder.sm.find_user('gamma') + resp = self.get_resp('/caravel/search_queries?user_id={}'.format(user.id)) + data = json.loads(resp) + self.assertEquals(1, len(data)) + self.assertEquals(list(data.values())[0]['userId'] , user.id) + self.logout() + + def test_search_query_on_status(self): + self.login('admin') + # Test search queries on status + resp = self.get_resp('/caravel/search_queries?status=success') + data = json.loads(resp) + self.assertEquals(2, len(data)) + states = [data[k]['state'] for k in data] + self.assertEquals(['success', 'success'], states) + + resp = self.get_resp('/caravel/search_queries?status=failed') + data = json.loads(resp) + self.assertEquals(1, len(data)) + self.assertEquals(list(data.values())[0]['state'], 'failed') + self.logout() + + def test_search_query_on_text(self): + self.login('admin') + resp = self.get_resp('/caravel/search_queries?search_text=permission') + data = json.loads(resp) + self.assertEquals(1, len(data)) + self.assertIn('permission', list(data.values())[0]['sql']) + self.logout() + + def test_search_query_on_time(self): + self.login('admin') + first_query_time = db.session.query(models.Query).filter_by( + sql='SELECT * FROM ab_user').first().start_time + second_query_time = db.session.query(models.Query).filter_by( + sql='SELECT * FROM ab_permission').first().start_time + # Test search queries on time filter + from_time = 'from={}'.format(int(first_query_time)) + to_time = 'to={}'.format(int(second_query_time)) + params = [from_time, to_time] + resp = self.get_resp('/caravel/search_queries?'+'&'.join(params)) + data = json.loads(resp) + self.assertEquals(2, len(data)) + for _, v in data.items(): + self.assertLess(int(first_query_time), v['startDttm']) + self.assertLess(v['startDttm'], int(second_query_time)) + self.logout() + + +if __name__ == '__main__': + unittest.main()