test_base.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # Licensed to Cloudera, Inc. under one
  4. # or more contributor license agreements. See the NOTICE file
  5. # distributed with this work for additional information
  6. # regarding copyright ownership. Cloudera, Inc. licenses this file
  7. # to you under the Apache License, Version 2.0 (the
  8. # "License"); you may not use this file except in compliance
  9. # with the License. You may obtain a copy of the License at
  10. #
  11. # http://www.apache.org/licenses/LICENSE-2.0
  12. #
  13. # Unless required by applicable law or agreed to in writing, software
  14. # distributed under the License is distributed on an "AS IS" BASIS,
  15. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16. # See the License for the specific language governing permissions and
  17. # limitations under the License.
  18. #
  19. """
  20. Common infrastructure for beeswax tests
  21. """
  22. import atexit
  23. import logging
  24. import pwd
  25. import os
  26. import re
  27. import subprocess
  28. import time
  29. import fb303.ttypes
  30. from nose.tools import assert_true, assert_false
  31. from desktop.lib.django_test_util import make_logged_in_client
  32. from hadoop import mini_cluster
  33. import hadoop.conf
  34. import beeswax.conf
  35. _INITIALIZED = False
  36. _SHARED_BEESWAX_SERVER_PROCESS = None
  37. BEESWAXD_TEST_PORT = 6969
  38. LOG = logging.getLogger(__name__)
  39. def _start_server(cluster):
  40. """
  41. Start beeswaxd and metastore
  42. """
  43. script = beeswax.conf.BEESWAX_SERVER_BIN.get()
  44. args = [
  45. script,
  46. '--beeswax',
  47. str(BEESWAXD_TEST_PORT),
  48. '--metastore',
  49. str(BEESWAXD_TEST_PORT + 1),
  50. '--superuser',
  51. pwd.getpwuid(os.getuid())[0],
  52. '--desktop-host',
  53. str('127.0.0.1'),
  54. '--desktop-port',
  55. str('42'), # Make up a port here. Tests don't start an actual server.
  56. ]
  57. env = {
  58. 'HADOOP_HOME': hadoop.conf.HADOOP_HOME.get(),
  59. 'HADOOP_CONF_DIR': cluster.config_dir,
  60. 'HIVE_CONF_DIR': beeswax.conf.BEESWAX_HIVE_CONF_DIR.get(),
  61. 'HADOOP_EXTRA_CLASSPATH_STRING': hadoop.conf.HADOOP_EXTRA_CLASSPATH_STRING.get()
  62. }
  63. if os.getenv("JAVA_HOME"):
  64. env["JAVA_HOME"] = os.getenv("JAVA_HOME")
  65. LOG.info("Executing %s, env %s, cwd %s" % (repr(args), repr(env), cluster.tmpdir))
  66. process = subprocess.Popen(args=args, env=env, cwd=cluster.tmpdir, stdin=subprocess.PIPE)
  67. return process
  68. def get_shared_beeswax_server():
  69. finish = (
  70. beeswax.conf.BEESWAX_SERVER_HOST.set_for_testing("localhost"),
  71. beeswax.conf.BEESWAX_SERVER_PORT.set_for_testing(BEESWAXD_TEST_PORT),
  72. beeswax.conf.BEESWAX_META_SERVER_HOST.set_for_testing("localhost"),
  73. beeswax.conf.BEESWAX_META_SERVER_PORT.set_for_testing(BEESWAXD_TEST_PORT + 1),
  74. # Use a bogus path to avoid loading the normal hive-site.xml
  75. beeswax.conf.BEESWAX_HIVE_CONF_DIR.set_for_testing('/my/bogus/path'),
  76. )
  77. cluster = mini_cluster.shared_cluster(conf=True)
  78. global _SHARED_BEESWAX_SERVER_PROCESS
  79. if _SHARED_BEESWAX_SERVER_PROCESS is None:
  80. p = _start_server(cluster)
  81. _SHARED_BEESWAX_SERVER_PROCESS = p
  82. def kill():
  83. LOG.info("Killing beeswax server (pid %d)." % p.pid)
  84. os.kill(p.pid, 9)
  85. p.wait()
  86. atexit.register(kill)
  87. # Wait for server to come up, by repeatedly trying.
  88. start = time.time()
  89. started = False
  90. sleep = 0.001
  91. while not started and time.time() - start < 20.0:
  92. try:
  93. client = beeswax.db_utils.db_client()
  94. meta_client = beeswax.db_utils.meta_client()
  95. client.echo("echo")
  96. if meta_client.getStatus() == fb303.ttypes.fb_status.ALIVE:
  97. started = True
  98. break
  99. time.sleep(sleep)
  100. sleep *= 2
  101. except:
  102. time.sleep(sleep)
  103. sleep *= 2
  104. pass
  105. if not started:
  106. raise Exception("Beeswax server took too long to come up.")
  107. # Make sure /tmp is 0777
  108. cluster.fs.setuser(cluster.superuser)
  109. if not cluster.fs.isdir('/tmp'):
  110. cluster.fs.mkdir('/tmp', 0777)
  111. else:
  112. cluster.fs.chmod('/tmp', 0777)
  113. def s():
  114. for f in finish:
  115. f()
  116. cluster.shutdown()
  117. return cluster, s
  118. REFRESH_RE = re.compile('<\s*meta\s+http-equiv="refresh"\s+content="\d*;([^"]*)"\s*/>', re.I)
  119. def wait_for_query_to_finish(client, response, max=30.0):
  120. logging.info(str(response.template) + ": " + str(response.content))
  121. start = time.time()
  122. sleep_time = 0.05
  123. # We don't check response.template == "watch_wait.mako" here,
  124. # because Django's response.template stuff is not thread-safe.
  125. while "Waiting for query..." in response.content:
  126. time.sleep(sleep_time)
  127. sleep_time = min(1.0, sleep_time * 2) # Capped exponential
  128. if (time.time() - start) > max:
  129. message = "Query took too long! %d seconds" % (time.time() - start,)
  130. LOG.warning(message)
  131. raise Exception(message)
  132. # Find out url to retry
  133. match = REFRESH_RE.search(response.content)
  134. if match is not None:
  135. url = match.group(1)
  136. url = url.lstrip('url=')
  137. else:
  138. url = response.request['PATH_INFO']
  139. response = client.get(url, follow=True)
  140. return response
  141. def make_query(client, query, submission_type="Execute",
  142. udfs=None, settings=None, resources=None,
  143. wait=False, name=None, desc=None, local=True,
  144. is_parameterized=True, **kwargs):
  145. """
  146. Prepares arguments for the execute view.
  147. If wait is True, waits for query to finish as well.
  148. """
  149. if settings is None:
  150. settings = []
  151. if local:
  152. # Tests run faster if not run against the real cluster.
  153. settings.append( ("mapred.job.tracker", "local") )
  154. # Prepares arguments for the execute view.
  155. parameters = {
  156. 'query-query': query,
  157. 'query-is_parameterized': is_parameterized and "on"
  158. }
  159. if submission_type == 'Execute':
  160. parameters['button-submit'] = 'Whatever'
  161. elif submission_type == 'Explain':
  162. parameters['button-explain'] = 'Whatever'
  163. elif submission_type == 'Save':
  164. parameters['saveform-save'] = 'True'
  165. if name:
  166. parameters['saveform-name'] = name
  167. if desc:
  168. parameters['saveform-desc'] = desc
  169. parameters["functions-next_form_id"] = str(len(udfs or []))
  170. for i, udf_pair in enumerate(udfs or []):
  171. name, klass = udf_pair
  172. parameters["functions-%d-name" % i] = name
  173. parameters["functions-%d-class_name" % i] = klass
  174. parameters["functions-%d-_exists" % i] = 'True'
  175. parameters["settings-next_form_id"] = str(len(settings))
  176. for i, settings_pair in enumerate(settings or []):
  177. key, value = settings_pair
  178. parameters["settings-%d-key" % i] = key
  179. parameters["settings-%d-value" % i] = value
  180. parameters["settings-%d-_exists" % i] = 'True'
  181. parameters["file_resources-next_form_id"] = str(len(resources or []))
  182. for i, resources_pair in enumerate(resources or []):
  183. type, path = resources_pair
  184. parameters["file_resources-%d-type" % i] = type
  185. parameters["file_resources-%d-path" % i] = path
  186. parameters["file_resources-%d-_exists" % i] = 'True'
  187. kwargs.setdefault('follow', True)
  188. response = client.post("/beeswax/execute", parameters, **kwargs)
  189. if wait:
  190. return wait_for_query_to_finish(client, response)
  191. return response
  192. def verify_history(client, fragment, design=None, reverse=False):
  193. """
  194. Verify that the query fragment and/or design are in the query history.
  195. If reverse is True, verify the opposite.
  196. Return the size of the history; -1 if we fail to determine it.
  197. """
  198. resp = client.get('/beeswax/query_history')
  199. my_assert = reverse and assert_false or assert_true
  200. my_assert(fragment in resp.content)
  201. if design:
  202. my_assert(design in resp.content)
  203. if resp.context:
  204. try:
  205. return len(resp.context['page'].object_list)
  206. except KeyError:
  207. pass
  208. # This could happen if we issue multiple requests in parallel.
  209. # The capturing of Django response context is not thread safe.
  210. # Also see:
  211. # http://docs.djangoproject.com/en/1.2/topics/testing/#testing-responses
  212. LOG.warn('Cannot find history size. Response context clobbered')
  213. return -1
  214. class BeeswaxSampleProvider(object):
  215. """
  216. Setup the test db and install sample data
  217. """
  218. @classmethod
  219. def setup_class(cls):
  220. cls.cluster, shutdown = get_shared_beeswax_server()
  221. cls.client = make_logged_in_client()
  222. # Weird redirection to avoid binding nonsense.
  223. cls.shutdown = [ shutdown ]
  224. cls.init_beeswax_db()
  225. @classmethod
  226. def teardown_class(cls):
  227. cls.cluster.fs.setuser(cls.cluster.superuser)
  228. try:
  229. cls.cluster.fs.rmtree('/tmp/beeswax')
  230. except IOError, ex:
  231. LOG.warn('Failed to cleanup /tmp/beeswax: %s' % (ex,))
  232. cls.shutdown[0]()
  233. @classmethod
  234. def init_beeswax_db(cls):
  235. """
  236. Install the common test tables (only once)
  237. """
  238. global _INITIALIZED
  239. if _INITIALIZED:
  240. return
  241. data_file = u'/tmp/beeswax/sample_data_échantillon_%d.tsv'
  242. # Create a "test_partitions" table.
  243. CREATE_PARTITIONED_TABLE = """
  244. CREATE TABLE test_partitions (foo INT, bar STRING)
  245. PARTITIONED BY (baz STRING, boom STRING)
  246. ROW FORMAT DELIMITED
  247. FIELDS TERMINATED BY '\t'
  248. LINES TERMINATED BY '\n'
  249. """
  250. make_query(cls.client, CREATE_PARTITIONED_TABLE, wait=True)
  251. cls._make_data_file(data_file % 1)
  252. LOAD_DATA = """
  253. LOAD DATA INPATH '%s'
  254. OVERWRITE INTO TABLE test_partitions
  255. PARTITION (baz='baz_one', boom='boom_two')
  256. """ % (data_file % 1,)
  257. make_query(cls.client, LOAD_DATA, wait=True, local=False)
  258. # Create a bunch of other tables
  259. CREATE_TABLE = """
  260. CREATE TABLE `%(name)s` (foo INT, bar STRING)
  261. COMMENT "%(comment)s"
  262. ROW FORMAT DELIMITED
  263. FIELDS TERMINATED BY '\t'
  264. LINES TERMINATED BY '\n'
  265. """
  266. # Create a "test" table.
  267. table_info = dict(name='test', comment='Test table')
  268. cls._make_data_file(data_file % 2)
  269. cls._make_table(table_info['name'], CREATE_TABLE % table_info, data_file % 2)
  270. # Create a "test_utf8" table.
  271. table_info = dict(name='test_utf8', comment=cls.get_i18n_table_comment())
  272. cls._make_i18n_data_file(data_file % 3, 'utf-8')
  273. cls._make_table(table_info['name'], CREATE_TABLE % table_info, data_file % 3)
  274. # Create a "test_latin1" table.
  275. table_info = dict(name='test_latin1', comment=cls.get_i18n_table_comment())
  276. cls._make_i18n_data_file(data_file % 4, 'latin1')
  277. cls._make_table(table_info['name'], CREATE_TABLE % table_info, data_file % 4)
  278. _INITIALIZED = True
  279. @staticmethod
  280. def get_i18n_table_comment():
  281. return u'en-hello pt-Olá ch-你好 ko-안녕 ru-Здравствуйте'
  282. @classmethod
  283. def _make_table(cls, table_name, create_ddl, filename):
  284. make_query(cls.client, create_ddl, wait=True)
  285. LOAD_DATA = """
  286. LOAD DATA INPATH '%s' OVERWRITE INTO TABLE %s
  287. """ % (filename, table_name)
  288. make_query(cls.client, LOAD_DATA, wait=True, local=False)
  289. @classmethod
  290. def _make_data_file(cls, filename):
  291. """
  292. Create data to be loaded into tables.
  293. Data contains two columns of:
  294. <num> 0x<hex_num>
  295. where <num> goes from 0 to 255 inclusive.
  296. """
  297. cls.cluster.fs.setuser(cls.cluster.superuser)
  298. f = cls.cluster.fs.open(filename, "w")
  299. for x in xrange(256):
  300. f.write("%d\t0x%x\n" % (x, x))
  301. f.close()
  302. @classmethod
  303. def _make_i18n_data_file(cls, filename, encoding):
  304. """
  305. Create i18n data to be loaded into tables.
  306. Data contains two columns of:
  307. <num> <unichr(num)>
  308. where <num> goes from 0 to 255 inclusive.
  309. """
  310. cls.cluster.fs.setuser(cls.cluster.superuser)
  311. f = cls.cluster.fs.open(filename, "w")
  312. for x in xrange(256):
  313. f.write("%d\t%s\n" % (x, unichr(x).encode(encoding)))
  314. f.close()