OmniSciDB  a987f07e93
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
conftest.py
Go to the documentation of this file.
1 import os
2 import subprocess
3 import time
4 from uuid import uuid4
5 
6 import pytest
7 from thrift.transport import TSocket, TTransport
8 from thrift.transport.TSocket import TTransportException
9 from heavydb import connect
10 import random
11 import string
12 
13 heavydb_host = os.environ.get('HEAVYDB_HOST', 'localhost')
14 
15 
17  """
18  Test to see if HeavyDB running on localhost and socket open
19  """
20  socket = TSocket.TSocket(heavydb_host, 6274)
21  transport = TTransport.TBufferedTransport(socket)
22 
23  try:
24  transport.open()
25  return True
26  except TTransportException:
27  return False
28 
29 
30 @pytest.fixture(scope='session')
32  """Ensure a HeavyDB server is running, optionally starting one if none"""
33  if _check_open():
34  # already running before pytest started
35  pass
36  else:
37  # not yet running...
38  subprocess.check_output(
39  [
40  'docker',
41  'run',
42  '-d',
43  '--ipc=host',
44  '-v',
45  '/dev:/dev',
46  '-p',
47  '6274:6274',
48  '-p',
49  '9092:9092',
50  'heavyai/core-os-cpu:latest',
51  ]
52  )
53  # yield and stop afterwards?
54  assert _check_open()
55  # Takes some time to start up. Unfortunately even trying to connect
56  # will cause it to hang.
57  time.sleep(5)
58 
59 
60 @pytest.fixture(scope='session')
61 def con(heavydb_server):
62  """
63  Fixture to provide Connection for tests run against live HeavyDB instance
64  """
65  return connect(
66  user="admin",
67  password='HyperInteractive',
68  host=heavydb_host,
69  port=6274,
70  protocol='binary',
71  dbname='heavyai',
72  )
73 
74 
75 @pytest.fixture
76 def mock_client(mocker):
77  """A magicmock for heavydb.connection.Client"""
78  return mocker.patch("heavydb.connection.Client")
79 
80 
81 def no_gpu():
82  """Check for the required GPU dependencies"""
83  try:
84  from numba import cuda
85  import cudf # noqa
86 
87  try:
88  cuda.select_device(0)
89  except cuda.cudadrv.error.CudaDriverError:
90  return True
91  except ImportError:
92  return True
93  return False
94 
95 
96 def gen_string():
97  """Generate a random string sequence for use in _tests_table_no_nulls"""
98  return ''.join(
99  [
100  random.choice(string.ascii_letters + string.digits)
101  for n in range(10)
102  ]
103  )
104 
105 
106 @pytest.fixture
107 def tmp_table(con) -> str:
108  table_name = 'table_{}'.format(uuid4().hex)
109  con.execute("drop table if exists {};".format(table_name))
110 
111  try:
112  yield table_name
113  finally:
114  con.execute("drop table if exists {};".format(table_name))
std::string join(T const &container, std::string const &delim)
def gen_string
Definition: conftest.py:96
def _check_open
Definition: conftest.py:16
def heavydb_server
Definition: conftest.py:31
def mock_client
Definition: conftest.py:76