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