Moved sqllab tests from core_tests to sqllab_tests (#1502)

* Moved sqllab tests from core_tests to sqllab_tests

* Minor changes based on comments

tests/base_tests.py

@@ -176,3 +176,14 @@ def revoke_public_access(self, table_name):
             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'))

tests/core_tests.py

@@ -41,17 +41,9 @@ def setUpClass(cls):
     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 @@ def test_gamma(self):
         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', '[email protected]',
-                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 @@ def test_csv_endpoint(self):
         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')

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', '[email protected]',
+                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()