OmniSciDB  a987f07e93
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
test_connection.py
Go to the documentation of this file.
1 import os
2 import pytest
3 from heavydb.thrift.ttypes import TColumnType
4 from heavydb.common.ttypes import TTypeInfo
5 from heavydb import OperationalError, connect
6 from heavydb.connection import _parse_uri, ConnectionInfo
7 from heavydb.exceptions import Error
8 from heavydb._parsers import ColumnDetails, _extract_column_details
9 
10 heavydb_host = os.environ.get('HEAVYDB_HOST', 'localhost')
11 
12 
13 @pytest.mark.usefixtures("heavydb_server")
16  with pytest.raises(TypeError):
17  connect(user='foo')
18 
20  with pytest.raises(OperationalError):
21  connect(host=heavydb_host, protocol='binary', port=1234)
22 
23  def test_close(self):
24  conn = connect(
25  user='admin',
26  password='HyperInteractive',
27  host=heavydb_host,
28  dbname='heavyai',
29  )
30  assert conn.closed == 0
31  conn.close()
32  assert conn.closed == 1
33 
34  def test_commit_noop(self, con):
35  result = con.commit() # it worked
36  assert result is None
37 
38  def test_bad_protocol(self, mock_client):
39  with pytest.raises(ValueError) as m:
40  connect(
41  user='user',
42  host=heavydb_host,
43  dbname='dbname',
44  protocol='fake-proto',
45  )
46  assert m.match('fake-proto')
47 
49  conn = connect(
50  user='admin',
51  password='HyperInteractive',
52  host=heavydb_host,
53  dbname='heavyai',
54  )
55  sessionid = conn._session
56  connnew = connect(sessionid=sessionid, host=heavydb_host)
57  assert connnew._session == sessionid
58 
60  sessionid = 'ILoveDancingOnTables'
61  with pytest.raises(Error):
62  connect(sessionid=sessionid, host=heavydb_host, protocol='binary')
63 
65  with pytest.raises(TypeError):
66  connect(
67  user='admin',
68  host=heavydb_host,
69  dbname='heavyai',
70  protocol='http',
71  validate=False,
72  )
73 
74 
75 class TestURI:
76  def test_parse_uri(self):
77  uri = (
78  'heavydb://admin:HyperInteractive@{0}:6274/heavyai?'
79  'protocol=binary'.format(heavydb_host)
80  )
81  result = _parse_uri(uri)
82  expected = ConnectionInfo(
83  "admin",
84  "HyperInteractive",
85  heavydb_host,
86  6274,
87  "heavyai",
88  "binary",
89  None,
90  None,
91  )
92  assert result == expected
93 
94  def test_both_raises(self):
95  uri = (
96  'heavydb://admin:HyperInteractive@{0}:6274/heavyai?'
97  'protocol=binary'.format(heavydb_host)
98  )
99  with pytest.raises(TypeError):
100  connect(uri=uri, user='my user')
101 
102 
105  data = [
106  TColumnType(
107  col_name='date_',
108  col_type=TTypeInfo(
109  type=6,
110  encoding=4,
111  nullable=True,
112  is_array=False,
113  precision=0,
114  scale=0,
115  comp_param=32,
116  ),
117  is_reserved_keyword=False,
118  src_name='',
119  ),
120  TColumnType(
121  col_name='trans',
122  col_type=TTypeInfo(
123  type=6,
124  encoding=4,
125  nullable=True,
126  is_array=False,
127  precision=0,
128  scale=0,
129  comp_param=32,
130  ),
131  is_reserved_keyword=False,
132  src_name='',
133  ),
134  TColumnType(
135  col_name='symbol',
136  col_type=TTypeInfo(
137  type=6,
138  encoding=4,
139  nullable=True,
140  is_array=False,
141  precision=0,
142  scale=0,
143  comp_param=32,
144  ),
145  is_reserved_keyword=False,
146  src_name='',
147  ),
148  TColumnType(
149  col_name='qty',
150  col_type=TTypeInfo(
151  type=1,
152  encoding=0,
153  nullable=True,
154  is_array=False,
155  precision=0,
156  scale=0,
157  comp_param=0,
158  ),
159  is_reserved_keyword=False,
160  src_name='',
161  ),
162  TColumnType(
163  col_name='price',
164  col_type=TTypeInfo(
165  type=3,
166  encoding=0,
167  nullable=True,
168  is_array=False,
169  precision=0,
170  scale=0,
171  comp_param=0,
172  ),
173  is_reserved_keyword=False,
174  src_name='',
175  ),
176  TColumnType(
177  col_name='vol',
178  col_type=TTypeInfo(
179  type=3,
180  encoding=0,
181  nullable=True,
182  is_array=False,
183  precision=0,
184  scale=0,
185  comp_param=0,
186  ),
187  is_reserved_keyword=False,
188  src_name='',
189  ),
190  ]
191  result = _extract_column_details(data)
192 
193  expected = [
195  name='date_',
196  type='STR',
197  nullable=True,
198  precision=0,
199  scale=0,
200  comp_param=32,
201  encoding='DICT',
202  is_array=False,
203  ),
205  name='trans',
206  type='STR',
207  nullable=True,
208  precision=0,
209  scale=0,
210  comp_param=32,
211  encoding='DICT',
212  is_array=False,
213  ),
215  name='symbol',
216  type='STR',
217  nullable=True,
218  precision=0,
219  scale=0,
220  comp_param=32,
221  encoding='DICT',
222  is_array=False,
223  ),
225  name='qty',
226  type='INT',
227  nullable=True,
228  precision=0,
229  scale=0,
230  comp_param=0,
231  encoding='NONE',
232  is_array=False,
233  ),
235  name='price',
236  type='FLOAT',
237  nullable=True,
238  precision=0,
239  scale=0,
240  comp_param=0,
241  encoding='NONE',
242  is_array=False,
243  ),
245  name='vol',
246  type='FLOAT',
247  nullable=True,
248  precision=0,
249  scale=0,
250  comp_param=0,
251  encoding='NONE',
252  is_array=False,
253  ),
254  ]
255  assert result == expected
def _extract_column_details
Definition: _parsers.py:143
tuple ColumnDetails
Definition: _parsers.py:23