tests.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. from builtins import object
  18. import json
  19. import logging
  20. import re
  21. from nose.plugins.skip import SkipTest
  22. from nose.tools import assert_true, assert_equal, assert_false, assert_raises
  23. from mock import patch, Mock
  24. from django.urls import reverse
  25. import desktop.conf as desktop_conf
  26. from desktop.conf import ENABLE_ORGANIZATIONS
  27. from desktop.lib.django_test_util import make_logged_in_client
  28. from desktop.lib.exceptions_renderable import PopupException
  29. from desktop.lib.test_utils import add_to_group
  30. from desktop.models import Document
  31. from hadoop.pseudo_hdfs4 import get_db_prefix, is_live_cluster
  32. from beeswax import data_export
  33. from beeswax.design import hql_query
  34. from beeswax.data_export import download
  35. from beeswax.models import SavedQuery, QueryHistory
  36. from beeswax.server import dbms
  37. from beeswax.test_base import get_query_server_config, wait_for_query_to_finish, fetch_query_result_data
  38. from beeswax.tests import _make_query
  39. from impala import conf
  40. from impala.dbms import ImpalaDbms
  41. if ENABLE_ORGANIZATIONS.get():
  42. from useradmin.models import User
  43. else:
  44. from django.contrib.auth.models import User
  45. LOG = logging.getLogger(__name__)
  46. class MockDbms(object):
  47. def get_databases(self):
  48. return ['db1', 'db2']
  49. def get_tables(self, database):
  50. return ['table1', 'table2']
  51. class TestMockedImpala(object):
  52. def setUp(self):
  53. self.client = make_logged_in_client()
  54. # Mock DB calls as we don't need the real ones
  55. self.prev_dbms = dbms.get
  56. dbms.get = lambda a, b: MockDbms()
  57. def tearDown(self):
  58. # Remove monkey patching
  59. dbms.get = self.prev_dbms
  60. def test_basic_flow(self):
  61. response = self.client.get("/impala/")
  62. assert_true(re.search('Impala', response.content), response.content)
  63. assert_true('Query Editor' in response.content)
  64. response = self.client.get("/impala/execute/")
  65. assert_true('Query Editor' in response.content)
  66. def test_saved_queries(self):
  67. user = User.objects.get(username='test')
  68. response = self.client.get("/impala/list_designs")
  69. assert_equal(len(response.context[0]['page'].object_list), 0)
  70. try:
  71. beewax_query = create_saved_query('beeswax', user)
  72. response = self.client.get("/impala/list_designs")
  73. assert_equal(len(response.context[0]['page'].object_list), 0)
  74. impala_query = create_saved_query('impala', user)
  75. response = self.client.get("/impala/list_designs")
  76. assert_equal(len(response.context[0]['page'].object_list), 1)
  77. # Test my query page
  78. QueryHistory.objects.create(owner=user, design=impala_query, query='', last_state=QueryHistory.STATE.available.value)
  79. resp = self.client.get('/impala/my_queries')
  80. assert_equal(len(resp.context[0]['q_page'].object_list), 1)
  81. assert_equal(resp.context[0]['h_page'].object_list[0].design.name, 'create_saved_query')
  82. finally:
  83. if beewax_query is not None:
  84. beewax_query.delete()
  85. if impala_query is not None:
  86. impala_query.delete()
  87. def test_invalidate(self):
  88. with patch('impala.dbms.ImpalaDbms._get_different_tables') as get_different_tables:
  89. with patch('desktop.models.ClusterConfig.get_hive_metastore_interpreters') as get_hive_metastore_interpreters:
  90. ddms = ImpalaDbms(Mock(query_server={'server_name': ''}), None)
  91. get_different_tables.return_value = ['customers']
  92. get_hive_metastore_interpreters.return_value = []
  93. assert_raises(PopupException, ddms.invalidate, 'default') # No hive/metastore configured
  94. get_hive_metastore_interpreters.return_value = ['hive']
  95. ddms.invalidate('default')
  96. ddms.client.query.assert_called_once_with(ddms.client.query.call_args[0][0])
  97. assert_true('customers' in ddms.client.query.call_args[0][0].hql_query) # diff of 1 table
  98. get_different_tables.return_value = ['customers','','','','','','','','','','']
  99. assert_raises(PopupException, ddms.invalidate, 'default') # diff of 11 tables. Limit is 10.
  100. ddms.invalidate('default', 'customers')
  101. assert_true(ddms.client.query.call_count == 2) # Second call
  102. assert_true('customers' in ddms.client.query.call_args[0][0].hql_query) # invalidate 1 table
  103. ddms.invalidate()
  104. assert_true(ddms.client.query.call_count == 3) # Third call
  105. assert_true('customers' not in ddms.client.query.call_args[0][0].hql_query) # Full invalidate
  106. class TestImpalaIntegration(object):
  107. integration = True
  108. @classmethod
  109. def setup_class(cls):
  110. cls.finish = []
  111. if not is_live_cluster():
  112. raise SkipTest
  113. cls.client = make_logged_in_client()
  114. cls.user = User.objects.get(username='test')
  115. add_to_group('test')
  116. cls.db = dbms.get(cls.user, get_query_server_config(name='impala'))
  117. cls.DATABASE = get_db_prefix(name='impala')
  118. queries = ["""
  119. DROP TABLE IF EXISTS %(db)s.tweets;
  120. """ % {'db': cls.DATABASE}, """
  121. DROP DATABASE IF EXISTS %(db)s CASCADE;
  122. """ % {'db': cls.DATABASE}, """
  123. CREATE DATABASE %(db)s;
  124. """ % {'db': cls.DATABASE}]
  125. for query in queries:
  126. resp = _make_query(cls.client, query, database='default', local=False, server_name='impala')
  127. resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
  128. content = json.loads(resp.content)
  129. assert_true(content['status'] == 0, resp.content)
  130. queries = ["""
  131. CREATE TABLE tweets (row_num INTEGER, id_str STRING, text STRING) STORED AS PARQUET;
  132. """, """
  133. INSERT INTO TABLE tweets VALUES (1, "531091827395682000", "My dad looks younger than costa");
  134. """, """
  135. 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.");
  136. """, """
  137. INSERT INTO TABLE tweets VALUES (3, "531091827768979000", "@Mustang_Sally83 and they need to get into you :))))");
  138. """, """
  139. INSERT INTO TABLE tweets VALUES (4, "531091827114668000", "@RachelZJohnson thank you rach!xxx");
  140. """, """
  141. 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");
  142. """]
  143. for query in queries:
  144. resp = _make_query(cls.client, query, database=cls.DATABASE, local=False, server_name='impala')
  145. resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
  146. content = json.loads(resp.content)
  147. assert_true(content['status'] == 0, resp.content)
  148. @classmethod
  149. def teardown_class(cls):
  150. # We need to drop tables before dropping the database
  151. queries = ["""
  152. DROP TABLE IF EXISTS %(db)s.tweets;
  153. """ % {'db': cls.DATABASE}, """
  154. DROP DATABASE %(db)s CASCADE;
  155. """ % {'db': cls.DATABASE}]
  156. for query in queries:
  157. resp = _make_query(cls.client, query, database='default', local=False, server_name='impala')
  158. resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
  159. # Check the cleanup
  160. databases = cls.db.get_databases()
  161. assert_false(cls.DATABASE in databases)
  162. assert_false('%(db)s_other' % {'db': cls.DATABASE} in databases)
  163. for f in cls.finish:
  164. f()
  165. def test_basic_flow(self):
  166. dbs = self.db.get_databases()
  167. assert_true('_impala_builtins' in dbs, dbs)
  168. assert_true(self.DATABASE in dbs, dbs)
  169. tables = self.db.get_tables(database=self.DATABASE)
  170. assert_true('tweets' in tables, tables)
  171. QUERY = """
  172. SELECT * FROM tweets ORDER BY row_num;
  173. """
  174. response = _make_query(self.client, QUERY, database=self.DATABASE, local=False, server_name='impala')
  175. content = json.loads(response.content)
  176. query_history = QueryHistory.get(content['id'])
  177. response = wait_for_query_to_finish(self.client, response, max=180.0)
  178. results = []
  179. # Check that we multiple fetches get all the result set
  180. while len(results) < 5:
  181. content = fetch_query_result_data(self.client, response, n=len(results), server_name='impala') # We get less than 5 results most of the time, so increase offset
  182. results += content['results']
  183. assert_equal([1, 2, 3, 4, 5], [col[0] for col in results])
  184. # Check start over
  185. results_start_over = []
  186. while len(results_start_over) < 5:
  187. content = fetch_query_result_data(self.client, response, n=len(results_start_over), server_name='impala')
  188. results_start_over += content['results']
  189. assert_equal(results_start_over, results)
  190. # Check cancel query
  191. resp = self.client.post(reverse('impala:api_cancel_query', kwargs={'query_history_id': query_history.id}))
  192. content = json.loads(resp.content)
  193. assert_equal(0, content['status'])
  194. def test_data_download(self):
  195. hql = 'SELECT * FROM tweets %(limit)s'
  196. FETCH_SIZE = data_export.FETCH_SIZE
  197. data_export.FETCH_SIZE = 2 # Decrease fetch size to validate last fetch logic
  198. try:
  199. query = hql_query(hql % {'limit': ''})
  200. handle = self.db.execute_and_wait(query)
  201. # Get the result in csv. Should have 5 + 1 header row.
  202. csv_resp = download(handle, 'csv', self.db)
  203. csv_content = ''.join(csv_resp.streaming_content)
  204. assert_equal(len(csv_content.strip().split('\n')), 5 + 1)
  205. query = hql_query(hql % {'limit': 'LIMIT 0'})
  206. handle = self.db.execute_and_wait(query)
  207. csv_resp = download(handle, 'csv', self.db)
  208. csv_content = ''.join(csv_resp.streaming_content)
  209. assert_equal(len(csv_content.strip().split('\n')), 1)
  210. query = hql_query(hql % {'limit': 'LIMIT 1'})
  211. handle = self.db.execute_and_wait(query)
  212. csv_resp = download(handle, 'csv', self.db)
  213. csv_content = ''.join(csv_resp.streaming_content)
  214. assert_equal(len(csv_content.strip().split('\n')), 1 + 1)
  215. query = hql_query(hql % {'limit': 'LIMIT 2'})
  216. handle = self.db.execute_and_wait(query)
  217. csv_resp = download(handle, 'csv', self.db)
  218. csv_content = ''.join(csv_resp.streaming_content)
  219. assert_equal(len(csv_content.strip().split('\n')), 1 + 2)
  220. finally:
  221. data_export.FETCH_SIZE = FETCH_SIZE
  222. def test_explain(self):
  223. QUERY = """
  224. SELECT * FROM tweets ORDER BY row_num;
  225. """
  226. response = _make_query(self.client, QUERY, database=self.DATABASE, local=False, server_name='impala', submission_type='Explain')
  227. json_response = json.loads(response.content)
  228. assert_true('MERGING-EXCHANGE' in json_response['explanation'], json_response)
  229. assert_true('SCAN HDFS' in json_response['explanation'], json_response)
  230. def test_get_table_sample(self):
  231. client = make_logged_in_client()
  232. resp = client.get(reverse('impala:get_sample_data', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
  233. data = json.loads(resp.content)
  234. assert_equal(0, data['status'], data)
  235. assert_equal([u'row_num', u'id_str', u'text'], data['headers'], data)
  236. assert_true(len(data['rows']), data)
  237. def test_get_session(self):
  238. session = None
  239. try:
  240. # Create open session
  241. session = self.db.open_session(self.user)
  242. resp = self.client.get(reverse("impala:api_get_session"))
  243. data = json.loads(resp.content)
  244. assert_true('properties' in data)
  245. assert_true(data['properties'].get('http_addr'))
  246. assert_true('session' in data, data)
  247. assert_true('id' in data['session'], data['session'])
  248. finally:
  249. if session is not None:
  250. try:
  251. self.db.close_session(session)
  252. except Exception:
  253. pass
  254. def test_get_settings(self):
  255. resp = self.client.get(reverse("impala:get_settings"))
  256. json_resp = json.loads(resp.content)
  257. assert_equal(0, json_resp['status'])
  258. assert_true('QUERY_TIMEOUT_S' in json_resp['settings'])
  259. def test_invalidate_tables(self):
  260. # Helper function to get Impala and Beeswax (HMS) tables
  261. def get_impala_beeswax_tables():
  262. impala_resp = self.client.get(reverse('impala:api_autocomplete_tables', kwargs={'database': self.DATABASE}))
  263. impala_tables_meta = json.loads(impala_resp.content)['tables_meta']
  264. impala_tables = [table['name'] for table in impala_tables_meta]
  265. beeswax_resp = self.client.get(reverse('beeswax:api_autocomplete_tables', kwargs={'database': self.DATABASE}))
  266. beeswax_tables_meta = json.loads(beeswax_resp.content)['tables_meta']
  267. beeswax_tables = [table['name'] for table in beeswax_tables_meta]
  268. return impala_tables, beeswax_tables
  269. impala_tables, beeswax_tables = get_impala_beeswax_tables()
  270. assert_equal(impala_tables, beeswax_tables,
  271. "\ntest_invalidate_tables: `%s`\nImpala Tables: %s\nBeeswax Tables: %s" % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
  272. hql = """
  273. CREATE TABLE new_table (a INT);
  274. """
  275. resp = _make_query(self.client, hql, wait=True, local=False, max=180.0, database=self.DATABASE)
  276. impala_tables, beeswax_tables = get_impala_beeswax_tables()
  277. # New table is not found by Impala
  278. assert_true('new_table' in beeswax_tables, beeswax_tables)
  279. assert_false('new_table' in impala_tables, impala_tables)
  280. resp = self.client.post(reverse('impala:invalidate'), {'database': self.DATABASE})
  281. impala_tables, beeswax_tables = get_impala_beeswax_tables()
  282. # Invalidate picks up new table
  283. assert_equal(impala_tables, beeswax_tables,
  284. "\ntest_invalidate_tables: `%s`\nImpala Tables: %s\nBeeswax Tables: %s" % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
  285. def test_refresh_table(self):
  286. # Helper function to get Impala and Beeswax (HMS) columns
  287. def get_impala_beeswax_columns():
  288. impala_resp = self.client.get(reverse('impala:api_autocomplete_columns', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
  289. impala_columns = json.loads(impala_resp.content)['columns']
  290. beeswax_resp = self.client.get(reverse('beeswax:api_autocomplete_columns', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
  291. beeswax_columns = json.loads(beeswax_resp.content)['columns']
  292. return impala_columns, beeswax_columns
  293. impala_columns, beeswax_columns = get_impala_beeswax_columns()
  294. assert_equal(impala_columns, beeswax_columns,
  295. "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s" % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
  296. hql = """
  297. ALTER TABLE tweets ADD COLUMNS (new_column INT);
  298. """
  299. resp = _make_query(self.client, hql, wait=True, local=False, max=180.0, database=self.DATABASE)
  300. impala_columns, beeswax_columns = get_impala_beeswax_columns()
  301. # New column is not found by Impala
  302. assert_true('new_column' in beeswax_columns, beeswax_columns)
  303. assert_false('new_column' in impala_columns, impala_columns)
  304. resp = self.client.post(reverse('impala:refresh_table', kwargs={'database': self.DATABASE, 'table': 'tweets'}))
  305. impala_columns, beeswax_columns = get_impala_beeswax_columns()
  306. # Invalidate picks up new column
  307. assert_equal(impala_columns, beeswax_columns,
  308. "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s" % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
  309. def test_get_exec_summary(self):
  310. query = """
  311. SELECT COUNT(1) FROM tweets;
  312. """
  313. response = _make_query(self.client, query, database=self.DATABASE, local=False, server_name='impala')
  314. content = json.loads(response.content)
  315. query_history = QueryHistory.get(content['id'])
  316. wait_for_query_to_finish(self.client, response, max=180.0)
  317. resp = self.client.post(reverse('impala:get_exec_summary', kwargs={'query_history_id': query_history.id}))
  318. data = json.loads(resp.content)
  319. assert_equal(0, data['status'], data)
  320. assert_true('nodes' in data['summary'], data)
  321. assert_true(len(data['summary']['nodes']) > 0, data['summary']['nodes'])
  322. # Attempt to call get_exec_summary on a closed query
  323. resp = self.client.post(reverse('impala:get_exec_summary', kwargs={'query_history_id': query_history.id}))
  324. data = json.loads(resp.content)
  325. assert_equal(0, data['status'], data)
  326. assert_true('nodes' in data['summary'], data)
  327. assert_true(len(data['summary']['nodes']) > 0, data['summary']['nodes'])
  328. def test_get_runtime_profile(self):
  329. query = """
  330. SELECT COUNT(1) FROM tweets;
  331. """
  332. response = _make_query(self.client, query, database=self.DATABASE, local=False, server_name='impala')
  333. content = json.loads(response.content)
  334. query_history = QueryHistory.get(content['id'])
  335. wait_for_query_to_finish(self.client, response, max=180.0)
  336. resp = self.client.post(reverse('impala:get_runtime_profile', kwargs={'query_history_id': query_history.id}))
  337. data = json.loads(resp.content)
  338. assert_equal(0, data['status'], data)
  339. assert_true('Execution Profile' in data['profile'], data)
  340. # Could be refactored with SavedQuery.create_empty()
  341. def create_saved_query(app_name, owner):
  342. query_type = SavedQuery.TYPES_MAPPING[app_name]
  343. design = SavedQuery(owner=owner, type=query_type)
  344. design.name = 'create_saved_query'
  345. design.desc = ''
  346. design.data = hql_query('show $tables', database='db1').dumps()
  347. design.is_auto = False
  348. design.save()
  349. Document.objects.link(design, owner=design.owner, extra=design.type, name=design.name, description=design.desc)
  350. return design
  351. def test_ssl_cacerts():
  352. for desktop_kwargs, conf_kwargs, expected in [
  353. ({'present': False}, {'present': False}, ''),
  354. ({'present': False}, {'data': 'local-cacerts.pem'}, 'local-cacerts.pem'),
  355. ({'data': 'global-cacerts.pem'}, {'present': False}, 'global-cacerts.pem'),
  356. ({'data': 'global-cacerts.pem'}, {'data': 'local-cacerts.pem'}, 'local-cacerts.pem'),
  357. ]:
  358. resets = [
  359. desktop_conf.SSL_CACERTS.set_for_testing(**desktop_kwargs),
  360. conf.SSL.CACERTS.set_for_testing(**conf_kwargs),
  361. ]
  362. try:
  363. assert_equal(conf.SSL.CACERTS.get(), expected,
  364. 'desktop:%s conf:%s expected:%s got:%s' % (desktop_kwargs, conf_kwargs, expected, conf.SSL.CACERTS.get()))
  365. finally:
  366. for reset in resets:
  367. reset()
  368. def test_ssl_validate():
  369. for desktop_kwargs, conf_kwargs, expected in [
  370. ({'present': False}, {'present': False}, True),
  371. ({'present': False}, {'data': False}, False),
  372. ({'present': False}, {'data': True}, True),
  373. ({'data': False}, {'present': False}, False),
  374. ({'data': False}, {'data': False}, False),
  375. ({'data': False}, {'data': True}, True),
  376. ({'data': True}, {'present': False}, True),
  377. ({'data': True}, {'data': False}, False),
  378. ({'data': True}, {'data': True}, True),
  379. ]:
  380. resets = [
  381. desktop_conf.SSL_VALIDATE.set_for_testing(**desktop_kwargs),
  382. conf.SSL.VALIDATE.set_for_testing(**conf_kwargs),
  383. ]
  384. try:
  385. assert_equal(conf.SSL.VALIDATE.get(), expected,
  386. 'desktop:%s conf:%s expected:%s got:%s' % (desktop_kwargs, conf_kwargs, expected, conf.SSL.VALIDATE.get()))
  387. finally:
  388. for reset in resets:
  389. reset()
  390. class TestImpalaDbms(object):
  391. def test_get_impala_nested_select(self):
  392. assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'id', None), ('id', '`default`.`customers`'))
  393. assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'email_preferences', 'categories/promos/'),
  394. ('email_preferences.categories.promos', '`default`.`customers`'))
  395. assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'addresses', 'key'),
  396. ('key', '`default`.`customers`.`addresses`'))
  397. assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'addresses', 'value/street_1/'),
  398. ('street_1', '`default`.`customers`.`addresses`'))
  399. assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'orders', 'item/order_date'),
  400. ('order_date', '`default`.`customers`.`orders`'))
  401. assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'orders', 'item/items/item/product_id'),
  402. ('product_id', '`default`.`customers`.`orders`.`items`'))