Prechádzať zdrojové kódy

[impala] Add an impala test command that runs against a real Impalad

e.g.
build/env/bin/hue test impala [hostname]

hostname will default to the Impala hostname in hue.ini.
These live Impala tests are executed only when running impala specific tests.
Tests are re-entrant

Had to refactor the Hive test base to allow to point to an Impala.
Romain Rigaux 10 rokov pred
rodič
commit
11bc1df1b8

+ 7 - 7
apps/beeswax/src/beeswax/test_base.py

@@ -205,11 +205,11 @@ def is_finished(response):
   return 'error' in status or status.get('isSuccess') or status.get('isFailure')
 
 
-def fetch_query_result_data(client, status_response):
+def fetch_query_result_data(client, status_response, n=0):
   # Take a wait_for_query_to_finish() response in input
   status = json.loads(status_response.content)
 
-  response = client.get("/beeswax/results/%s/0?format=json" % status.get('id'))
+  response = client.get("/beeswax/results/%(id)s/%(n)s?format=json" % {'id': status.get('id'), 'n': n})
   content = json.loads(response.content)
 
   return content
@@ -217,7 +217,7 @@ def fetch_query_result_data(client, status_response):
 def make_query(client, query, submission_type="Execute",
                udfs=None, settings=None, resources=None,
                wait=False, name=None, desc=None, local=True,
-               is_parameterized=True, max=30.0, database='default', email_notify=False, params=None, **kwargs):
+               is_parameterized=True, max=30.0, database='default', email_notify=False, params=None, server_name='beeswax', **kwargs):
   """
   Prepares arguments for the execute view.
 
@@ -276,12 +276,12 @@ def make_query(client, query, submission_type="Execute",
     parameters["parameterization-%s" % name] = value
 
   kwargs.setdefault('follow', True)
-  execute_url = reverse("beeswax:api_execute")
+  execute_url = reverse("%(server_name)s:api_execute" % {'server_name': server_name})
 
   if submission_type == 'Explain':
     execute_url += "?explain=true"
   if submission_type == 'Save':
-    execute_url = reverse("beeswax:api_save_design")
+    execute_url = reverse("%(server_name)s:api_save_design" % {'server_name': server_name})
 
   response = client.post(execute_url, parameters, **kwargs)
 
@@ -291,13 +291,13 @@ def make_query(client, query, submission_type="Execute",
   return response
 
 
-def verify_history(client, fragment, design=None, reverse=False):
+def verify_history(client, fragment, design=None, reverse=False, server_name='beeswax'):
   """
   Verify that the query fragment and/or design are in the query history.
   If reverse is True, verify the opposite.
   Return the size of the history; -1 if we fail to determine it.
   """
-  resp = client.get('/beeswax/query_history')
+  resp = client.get('/%(server_name)s/query_history' % {'server_name': server_name})
   my_assert = reverse and assert_false or assert_true
   my_assert(fragment in resp.content)
   if design:

+ 3 - 3
apps/beeswax/src/beeswax/tests.py

@@ -78,16 +78,16 @@ LOG = logging.getLogger(__name__)
 def _make_query(client, query, submission_type="Execute",
                 udfs=None, settings=None, resources=[],
                 wait=False, name=None, desc=None, local=True,
-                is_parameterized=True, max=30.0, database='default', email_notify=False, params=None, **kwargs):
+                is_parameterized=True, max=30.0, database='default', email_notify=False, params=None, server_name='beeswax', **kwargs):
 
   res = make_query(client, query, submission_type,
                    udfs, settings, resources,
-                   wait, name, desc, local, is_parameterized, max, database, email_notify, params, **kwargs)
+                   wait, name, desc, local, is_parameterized, max, database, email_notify, params, server_name, **kwargs)
 
   # Should be in the history if it's submitted.
   if submission_type == 'Execute':
     fragment = collapse_whitespace(smart_str(escape(query[:20])))
-    verify_history(client, fragment=fragment)
+    verify_history(client, fragment=fragment, server_name=server_name)
 
   return res
 

+ 82 - 5
apps/impala/src/impala/tests.py

@@ -16,15 +16,22 @@
 # limitations under the License.
 
 import re
+import sys
 
+from nose.plugins.skip import SkipTest
 from nose.tools import assert_true, assert_equal, assert_false
+
 from django.contrib.auth.models import User
 
 from desktop.lib.django_test_util import make_logged_in_client
+from desktop.lib.test_utils import grant_access, add_to_group
+from desktop.models import Document
+
+from beeswax.design import hql_query
 from beeswax.models import SavedQuery, QueryHistory
 from beeswax.server import dbms
-from beeswax.design import hql_query
-from desktop.models import Document
+from beeswax.test_base import get_query_server_config, wait_for_query_to_finish, fetch_query_result_data
+from beeswax.tests import _make_query
 
 
 class MockDbms:
@@ -55,7 +62,6 @@ class TestMockedImpala:
     response = self.client.get("/impala/execute/")
     assert_true('Query Editor' in response.content)
 
-
   def test_saved_queries(self):
     user = User.objects.get(username='test')
 
@@ -76,7 +82,7 @@ class TestMockedImpala:
 
       resp = self.client.get('/impala/my_queries')
       assert_equal(len(resp.context['q_page'].object_list), 1)
-      assert_equal(len(resp.context['h_page'].object_list), 1)
+      assert_equal(resp.context['h_page'].object_list[0].design.name, 'create_saved_query')
     finally:
       if beewax_query is not None:
         beewax_query.delete()
@@ -84,11 +90,82 @@ class TestMockedImpala:
         impala_query.delete()
 
 
+class TestImpalaIntegration:
+  DATABASE = 'test_hue_impala'
+
+  def setUp(self):
+    # We need a real Impala cluster currently
+    if not 'impala' in sys.argv:
+      raise SkipTest
+
+    self.client = make_logged_in_client()
+    self.user = User.objects.get(username='test')
+    add_to_group('test')
+    self.db = dbms.get(self.user, get_query_server_config(name='impala'))
+
+    hql = """
+      DROP TABLE IF EXISTS %(db)s.tweets;
+      DROP DATABASE IF EXISTS %(db)s;
+      CREATE DATABASE %(db)s;
+
+      USE %(db)s;
+    """ % {'db': self.DATABASE}
+
+    resp = _make_query(self.client, hql, local=False, server_name='impala') # Does not point to the database yet
+    resp = wait_for_query_to_finish(self.client, resp, max=30.0)
+
+    hql = """
+      CREATE TABLE tweets (row_num INTEGER, id_str STRING, text STRING) STORED AS PARQUET;
+
+      INSERT INTO TABLE tweets VALUES (1, "531091827395682000", "My dad looks younger than costa");
+      INSERT INTO TABLE tweets VALUES (2, "531091827781550000", "There is a thin line between your partner being vengeful and you reaping the consequences of your bad actions towards your partner.");
+      INSERT INTO TABLE tweets VALUES (3, "531091827768979000", "@Mustang_Sally83 and they need to get into you :))))");
+      INSERT INTO TABLE tweets VALUES (4, "531091827114668000", "@RachelZJohnson thank you rach!xxx");
+      INSERT INTO TABLE tweets VALUES (5, "531091827949309000", "i think @WWERollins was robbed of the IC title match this week on RAW also i wonder if he will get a rematch i hope so @WWE");
+    """
+
+    resp = _make_query(self.client, hql, database=self.DATABASE, local=False, server_name='impala')
+    resp = wait_for_query_to_finish(self.client, resp, max=30.0)
+
+  def test_basic_flow(self):
+    dbs = self.db.get_databases()
+    assert_true('_impala_builtins' in dbs, dbs)
+    assert_true(self.DATABASE in dbs, dbs)
+
+    tables = self.db.get_tables(database=self.DATABASE)
+    assert_true('tweets' in tables, tables)
+
+    QUERY = """
+      SELECT * FROM tweets ORDER BY row_num;
+    """
+    response = _make_query(self.client, QUERY, database=self.DATABASE, local=False, server_name='impala')
+
+    response = wait_for_query_to_finish(self.client, response, max=180.0)
+
+    results = []
+
+    # Check that we multiple fetches get all the result set
+    while len(results) < 5:
+      content = fetch_query_result_data(self.client, response, n=len(results)) # We get less than 5 results most of the time, so increase offset
+      results += content['results']
+
+    assert_equal([1, 2, 3, 4, 5], [col[0] for col in results])
+
+    # Check start over
+    results_start_over = []
+
+    while len(results_start_over) < 5:
+      content = fetch_query_result_data(self.client, response, n=len(results_start_over))
+      results_start_over += content['results']
+
+    assert_equal(results_start_over, results)
+
+
 # Could be refactored with SavedQuery.create_empty()
 def create_saved_query(app_name, owner):
     query_type = SavedQuery.TYPES_MAPPING[app_name]
     design = SavedQuery(owner=owner, type=query_type)
-    design.name = SavedQuery.DEFAULT_NEW_DESIGN_NAME
+    design.name = 'create_saved_query'
     design.desc = ''
     design.data = hql_query('show $tables', database='db1').dumps()
     design.is_auto = False

+ 7 - 0
desktop/core/src/desktop/management/commands/test.py

@@ -52,6 +52,8 @@ class Command(BaseCommand):
 
       windmill   Runs windmill tests
 
+      impala [hostname]   Runs Impala tests only against a pre-installed Impalad on hostname.
+
     Common useful extra arguments for nose:
       --nologcapture
       --nocapture (-s)
@@ -85,6 +87,11 @@ class Command(BaseCommand):
 
     if args[0] == "all":
       nose_args = args + all_apps
+    elif args[0] == "impala":
+      if len(args) > 1:
+        from impala.conf import SERVER_HOST
+        SERVER_HOST.set_for_testing(args[1])
+      nose_args = args + ["impala"]
     elif args[0] == "fast":
       test_apps = [ app.module.__name__ for app in appmanager.DESKTOP_MODULES ]
       nose_args = args + all_apps + ["-a", "!requires_hadoop"]