OmniSciDB  471d68cefb
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
test_exceptions.py
Go to the documentation of this file.
1 #cython: language_level=3
2 
3 #import sys
4 import pytest
5 import omniscidbe as dbe
6 import ctypes
7 import pandas
8 import pyarrow
9 import numpy
10 import shutil
11 ctypes._dlopen('libDBEngine.so', ctypes.RTLD_GLOBAL)
12 
13 data_path = "dbe_test_data"
14 
15 #######################Check init with wrong parameter
17  global engine
18  try:
19  shutil.rmtree(data_path)
20  except FileNotFoundError:
21  pass
22 
23  with pytest.raises(RuntimeError) as excinfo:
24  engine = dbe.PyDbEngine(data='/' + data_path, calcite_port=9091)
25  assert "Permission denied" in str(excinfo.value)
26 
27 ######################Check init with right parameters
29  global engine
30  engine = dbe.PyDbEngine(data=data_path, calcite_port=9091)
31  assert bool(engine.closed) == False
32 
33 engine = None
34 
35 ######################Check DDL statement
37  engine.executeDDL("drop table if exists test")
38  engine.executeDDL("create table test (x int not null, w tinyint, y int, z text)")
39  assert engine.get_tables() == ['test']
40 
41 #######################Check creating a duplicate table
43  with pytest.raises(RuntimeError) as excinfo:
44  engine.executeDDL("create table test (x int not null, w tinyint, y int, z text)")
45  assert "already exists" in str(excinfo.value)
46 
47 #######################Check right DML statement
49  engine.executeDML("insert into test values(55,5,3,'la-la-la')")
50  engine.executeDML("insert into test values(66,6,1, 'aa')")
51  engine.executeDML("insert into test values(77,7,0,'bb')")
52  dframe = engine.select_df("select * from test")
53  dforig = pandas.DataFrame({'x': [55,66,77], 'w': [5,6,7], 'y': [3,1,0], 'z': ['la-la-la', 'aa', 'bb']})
54  dforig = dforig.astype(dtype= {"x":"int32", "w":"int8","y":"int32", "z":"category"})
55  assert dframe.equals(dforig)
56 
57 #######################Check wrong DML statement
59  with pytest.raises(RuntimeError) as excinfo:
60  cursor = engine.executeDML("selectTT * from test")
61  assert "SQL Error" in str(excinfo.value)
62 
63 #######################Check zero division exception
65  with pytest.raises(RuntimeError) as excinfo:
66  cursor = engine.executeDML("SELECT x / (x - x) FROM test")
67  assert "Division by zero" in str(excinfo.value)
68 
69 #######################Check double init exception
71  with pytest.raises(RuntimeError) as excinfo:
72  engine = dbe.PyDbEngine()
73  assert "already initialized" in str(excinfo.value)
74 
75 
76 if __name__ == "__main__":
77  pytest.main(["-v", __file__])
def test_failed_DML
Check wrong DML statement.
def test_success_DDL
Check DDL statement.
def test_success_DML
Check right DML statement.
def test_failed_DDL
Check creating a duplicate table.
def test_success_init
Check init with right parameters.
def test_zero_division
Check zero division exception.
def test_failed_init
Check init with wrong parameter.
def test_double_init
Check double init exception.