OmniSciDB  72c90bc290
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
tests.conftest Namespace Reference

Functions

def _check_open
 
def heavydb_server
 
def con
 
def mock_client
 
def no_gpu
 
def gen_string
 
def tmp_table
 

Variables

tuple heavydb_host = os.environ.get('HEAVYDB_HOST', 'localhost')
 

Function Documentation

def tests.conftest._check_open ( )
private
Test to see if HeavyDB running on localhost and socket open

Definition at line 16 of file conftest.py.

Referenced by tests.conftest.heavydb_server().

16 
17 def _check_open():
18  """
19  Test to see if HeavyDB running on localhost and socket open
20  """
21  socket = TSocket.TSocket(heavydb_host, 6274)
22  transport = TTransport.TBufferedTransport(socket)
23 
24  try:
25  transport.open()
26  return True
27  except TTransportException:
28  return False
29 
30 
@pytest.fixture(scope='session')
def _check_open
Definition: conftest.py:16

+ Here is the caller graph for this function:

def tests.conftest.con (   heavydb_server)
Fixture to provide Connection for tests run against live HeavyDB instance

Definition at line 61 of file conftest.py.

References heavydb.connection.connect().

61 
62 def con(heavydb_server):
63  """
64  Fixture to provide Connection for tests run against live HeavyDB instance
65  """
66  return connect(
67  user="admin",
68  password='HyperInteractive',
69  host=heavydb_host,
70  port=6274,
71  protocol='binary',
72  dbname='heavyai',
73  )
74 
75 
@pytest.fixture

+ Here is the call graph for this function:

def tests.conftest.gen_string ( )
Generate a random string sequence for use in _tests_table_no_nulls

Definition at line 96 of file conftest.py.

References join().

96 
97 def gen_string():
98  """Generate a random string sequence for use in _tests_table_no_nulls"""
99  return ''.join(
100  [
101  random.choice(string.ascii_letters + string.digits)
102  for n in range(10)
103  ]
104  )
105 
106 
@pytest.fixture
std::string join(T const &container, std::string const &delim)
def gen_string
Definition: conftest.py:96

+ Here is the call graph for this function:

def tests.conftest.heavydb_server ( )
Ensure a HeavyDB server is running, optionally starting one if none

Definition at line 31 of file conftest.py.

References tests.conftest._check_open().

31 
32 def heavydb_server():
33  """Ensure a HeavyDB server is running, optionally starting one if none"""
34  if _check_open():
35  # already running before pytest started
36  pass
37  else:
38  # not yet running...
39  subprocess.check_output(
40  [
41  'docker',
42  'run',
43  '-d',
44  '--ipc=host',
45  '-v',
46  '/dev:/dev',
47  '-p',
48  '6274:6274',
49  '-p',
50  '9092:9092',
51  'heavyai/core-os-cpu:latest',
52  ]
53  )
54  # yield and stop afterwards?
55  assert _check_open()
56  # Takes some time to start up. Unfortunately even trying to connect
57  # will cause it to hang.
58  time.sleep(5)
59 
60 
@pytest.fixture(scope='session')
def _check_open
Definition: conftest.py:16
def heavydb_server
Definition: conftest.py:31

+ Here is the call graph for this function:

def tests.conftest.mock_client (   mocker)
A magicmock for heavydb.connection.Client

Definition at line 76 of file conftest.py.

76 
77 def mock_client(mocker):
78  """A magicmock for heavydb.connection.Client"""
79  return mocker.patch("heavydb.connection.Client")
80 
def mock_client
Definition: conftest.py:76
def tests.conftest.no_gpu ( )
Check for the required GPU dependencies

Definition at line 81 of file conftest.py.

81 
82 def no_gpu():
83  """Check for the required GPU dependencies"""
84  try:
85  from numba import cuda
86  import cudf # noqa
87 
88  try:
89  cuda.select_device(0)
90  except cuda.cudadrv.error.CudaDriverError:
91  return True
92  except ImportError:
93  return True
94  return False
95 
def tests.conftest.tmp_table (   con,
  str 
)

Definition at line 107 of file conftest.py.

108 def tmp_table(con) -> str:
109  table_name = 'table_{}'.format(uuid4().hex)
110  con.execute("drop table if exists {};".format(table_name))
111 
112  try:
113  yield table_name
114  finally:
115  con.execute("drop table if exists {};".format(table_name))

Variable Documentation

tuple tests.conftest.heavydb_host = os.environ.get('HEAVYDB_HOST', 'localhost')

Definition at line 13 of file conftest.py.