OmniSciDB  8fa3bf436f
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros 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 = "test_dbe_data"
14 #######################Check init with wrong parameter
16  global engine
17  try:
18  shutil.rmtree(data_path)
19  except FileNotFoundError:
20  pass
21 
22  with pytest.raises(RuntimeError) as excinfo:
23  engine = dbe.PyDbEngine(data='/'+data_path, calcite_port=9091)
24  assert "Permission denied" in str(excinfo.value)
25 
26 ######################Check init with right parameters
28  global engine
29  engine = dbe.PyDbEngine(data=data_path, calcite_port=9091)
30  assert bool(engine.closed) == False
31 
32 engine = None
33 
34 ######################Check DDL statement
36  engine.executeDDL("drop table if exists test")
37  engine.executeDDL("create table test (x int not null, w tinyint, y int, z text)")
38  assert engine.get_tables() == ['test']
39 
40 #######################Check creating a duplicate table
42  with pytest.raises(RuntimeError) as excinfo:
43  engine.executeDDL("create table test (x int not null, w tinyint, y int, z text)")
44  assert "already exists" in str(excinfo.value)
45 
46 #######################Check right DML statement
48  engine.executeDML("insert into test values(55,5,3,'la-la-la')")
49  engine.executeDML("insert into test values(66,6,1, 'aa')")
50  engine.executeDML("insert into test values(77,7,0,'bb')")
51  dframe = engine.select_df("select * from test")
52  dforig = pandas.DataFrame({'x': [55,66,77], 'w': [5,6,7], 'y': [3,1,0], 'z': ['la-la-la', 'aa', 'bb']})
53  dforig = dforig.astype(dtype= {"x":"int32", "w":"int8","y":"int32", "z":"category"})
54  assert dframe.equals(dforig)
55 
56 #######################Check wrong DML statement
58  with pytest.raises(RuntimeError) as excinfo:
59  cursor = engine.executeDML("selectTT * from test")
60  assert "SQL Error" in str(excinfo.value)
61 
62 #######################Check zero division exception
64  with pytest.raises(RuntimeError) as excinfo:
65  cursor = engine.executeDML("SELECT x / (x - x) FROM test")
66  assert "Division by zero" in str(excinfo.value)
67 
68 #######################Check double init exception
70  with pytest.raises(RuntimeError) as excinfo:
71  engine = dbe.PyDbEngine()
72  assert "already initialized" in str(excinfo.value)
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.