433 lines
14 KiB
Python
433 lines
14 KiB
Python
# pylint: disable=redefined-outer-name,missing-function-docstring,protected-access
|
|
""" Tests on opening hours helpers """
|
|
|
|
import pytest
|
|
|
|
from mylib.oracle import OracleDB
|
|
|
|
|
|
class FakeCXOracleCursor:
|
|
""" Fake cx_Oracle cursor """
|
|
|
|
def __init__(self, expected_sql, expected_params, expected_return, expected_just_try, expected_exception):
|
|
self.expected_sql = expected_sql
|
|
self.expected_params = expected_params
|
|
self.expected_return = expected_return
|
|
self.expected_just_try = expected_just_try
|
|
self.expected_exception = expected_exception
|
|
|
|
def execute(self, sql, **params):
|
|
if self.expected_exception:
|
|
raise Exception("%s.execute(%s, %s): expected exception" % (self, sql, params))
|
|
if self.expected_just_try and not sql.lower().startswith('select '):
|
|
assert False, "%s.execute(%s, %s) may not be executed in just try mode" % (self, sql, params)
|
|
assert sql == self.expected_sql, "%s.execute(): Invalid SQL query:\n '%s'\nMay be:\n '%s'" % (self, sql, self.expected_sql)
|
|
assert params == self.expected_params, "%s.execute(): Invalid params:\n %s\nMay be:\n %s" % (self, params, self.expected_params)
|
|
return self.expected_return
|
|
|
|
def fetchall(self):
|
|
return self.expected_return
|
|
|
|
def __repr__(self):
|
|
return "FakeCXOracleCursor(%s, %s, %s, %s)" % (
|
|
self.expected_sql, self.expected_params,
|
|
self.expected_return, self.expected_just_try
|
|
)
|
|
|
|
|
|
class FakeCXOracle:
|
|
""" Fake cx_Oracle connection """
|
|
|
|
expected_sql = None
|
|
expected_params = dict()
|
|
expected_return = True
|
|
expected_just_try = False
|
|
expected_exception = False
|
|
just_try = False
|
|
|
|
def __init__(self, **kwargs):
|
|
allowed_kwargs = dict(dsn=str, user=str, password=(str, None))
|
|
for arg, value in kwargs.items():
|
|
assert arg in allowed_kwargs, "Invalid arg %s='%s'" % (arg, value)
|
|
assert isinstance(value, allowed_kwargs[arg]), "Arg %s not a %s (%s)" % (arg, allowed_kwargs[arg], type(value))
|
|
setattr(self, arg, value)
|
|
|
|
def close(self):
|
|
return self.expected_return
|
|
|
|
def cursor(self):
|
|
return FakeCXOracleCursor(
|
|
self.expected_sql, self.expected_params,
|
|
self.expected_return, self.expected_just_try or self.just_try,
|
|
self.expected_exception
|
|
)
|
|
|
|
def commit(self):
|
|
self._check_just_try()
|
|
return self.expected_return
|
|
|
|
def rollback(self):
|
|
self._check_just_try()
|
|
return self.expected_return
|
|
|
|
def _check_just_try(self):
|
|
if self.just_try:
|
|
assert False, "May not be executed in just try mode"
|
|
|
|
|
|
def fake_cxoracle_connect(**kwargs):
|
|
return FakeCXOracle(**kwargs)
|
|
|
|
|
|
def fake_cxoracle_connect_just_try(**kwargs):
|
|
con = FakeCXOracle(**kwargs)
|
|
con.just_try = True
|
|
return con
|
|
|
|
|
|
@pytest.fixture
|
|
def test_oracledb():
|
|
return OracleDB('127.0.0.1/dbname', 'user', 'password')
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_oracledb(mocker):
|
|
mocker.patch('cx_Oracle.connect', fake_cxoracle_connect)
|
|
return OracleDB('127.0.0.1/dbname', 'user', 'password')
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_just_try_oracledb(mocker):
|
|
mocker.patch('cx_Oracle.connect', fake_cxoracle_connect_just_try)
|
|
return OracleDB('127.0.0.1/dbname', 'user', 'password', just_try=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_connected_oracledb(fake_oracledb):
|
|
fake_oracledb.connect()
|
|
return fake_oracledb
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_connected_just_try_oracledb(fake_just_try_oracledb):
|
|
fake_just_try_oracledb.connect()
|
|
return fake_just_try_oracledb
|
|
|
|
|
|
def generate_mock_args(expected_args=(), expected_kwargs=dict(), expected_return=True): # pylint: disable=dangerous-default-value
|
|
def mock_args(*args, **kwargs):
|
|
assert args == expected_args, "Invalid call args:\n %s\nMay be:\n %s" % (args, expected_args)
|
|
assert kwargs == expected_kwargs, "Invalid call kwargs:\n %s\nMay be:\n %s" % (kwargs, expected_kwargs)
|
|
return expected_return
|
|
return mock_args
|
|
|
|
|
|
def mock_doSQL_just_try(self, sql, params=None): # pylint: disable=unused-argument
|
|
assert False, "doSQL() may not be executed in just try mode"
|
|
|
|
|
|
def generate_mock_doSQL(expected_sql, expected_params=dict(), expected_return=True): # pylint: disable=dangerous-default-value
|
|
def mock_doSQL(self, sql, params=None): # pylint: disable=unused-argument
|
|
assert sql == expected_sql, "Invalid generated SQL query:\n '%s'\nMay be:\n '%s'" % (sql, expected_sql)
|
|
assert params == expected_params, "Invalid generated params:\n %s\nMay be:\n %s" % (params, expected_params)
|
|
return expected_return
|
|
return mock_doSQL
|
|
|
|
|
|
# OracleDB.doSelect() have same expected parameters as OracleDB.doSQL()
|
|
generate_mock_doSelect = generate_mock_doSQL
|
|
mock_doSelect_just_try = mock_doSQL_just_try
|
|
|
|
#
|
|
# Test on OracleDB helper methods
|
|
#
|
|
|
|
|
|
def test_combine_params_with_to_add_parameter():
|
|
assert OracleDB._combine_params(dict(test1=1), dict(test2=2)) == dict(
|
|
test1=1, test2=2
|
|
)
|
|
|
|
|
|
def test_combine_params_with_kargs():
|
|
assert OracleDB._combine_params(dict(test1=1), test2=2) == dict(
|
|
test1=1, test2=2
|
|
)
|
|
|
|
|
|
def test_combine_params_with_kargs_and_to_add_parameter():
|
|
assert OracleDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict(
|
|
test1=1, test2=2, test3=3
|
|
)
|
|
|
|
|
|
def test_format_where_clauses_params_are_preserved():
|
|
args = ('test = test', dict(test1=1))
|
|
assert OracleDB._format_where_clauses(*args) == args
|
|
|
|
|
|
def test_format_where_clauses_raw():
|
|
assert OracleDB._format_where_clauses('test = test') == (('test = test'), dict())
|
|
|
|
|
|
def test_format_where_clauses_tuple_clause_with_params():
|
|
where_clauses = (
|
|
'test1 = :test1 AND test2 = :test2',
|
|
dict(test1=1, test2=2)
|
|
)
|
|
assert OracleDB._format_where_clauses(where_clauses) == where_clauses
|
|
|
|
|
|
def test_format_where_clauses_dict():
|
|
where_clauses = dict(test1=1, test2=2)
|
|
assert OracleDB._format_where_clauses(where_clauses) == (
|
|
'"test1" = :test1 AND "test2" = :test2',
|
|
where_clauses
|
|
)
|
|
|
|
|
|
def test_format_where_clauses_combined_types():
|
|
where_clauses = (
|
|
'test1 = 1',
|
|
('test2 LIKE :test2', dict(test2=2)),
|
|
dict(test3=3, test4=4)
|
|
)
|
|
assert OracleDB._format_where_clauses(where_clauses) == (
|
|
'test1 = 1 AND test2 LIKE :test2 AND "test3" = :test3 AND "test4" = :test4',
|
|
dict(test2=2, test3=3, test4=4)
|
|
)
|
|
|
|
|
|
def test_format_where_clauses_with_where_op():
|
|
where_clauses = dict(test1=1, test2=2)
|
|
assert OracleDB._format_where_clauses(where_clauses, where_op='OR') == (
|
|
'"test1" = :test1 OR "test2" = :test2',
|
|
where_clauses
|
|
)
|
|
|
|
|
|
def test_add_where_clauses():
|
|
sql = "SELECT * FROM table"
|
|
where_clauses = dict(test1=1, test2=2)
|
|
assert OracleDB._add_where_clauses(sql, None, where_clauses) == (
|
|
sql + ' WHERE "test1" = :test1 AND "test2" = :test2',
|
|
where_clauses
|
|
)
|
|
|
|
|
|
def test_add_where_clauses_preserved_params():
|
|
sql = "SELECT * FROM table"
|
|
where_clauses = dict(test1=1, test2=2)
|
|
params = dict(fake1=1)
|
|
assert OracleDB._add_where_clauses(sql, params.copy(), where_clauses) == (
|
|
sql + ' WHERE "test1" = :test1 AND "test2" = :test2',
|
|
dict(**where_clauses, **params)
|
|
)
|
|
|
|
|
|
def test_add_where_clauses_with_op():
|
|
sql = "SELECT * FROM table"
|
|
where_clauses = ('test1=1', 'test2=2')
|
|
assert OracleDB._add_where_clauses(sql, None, where_clauses, where_op='OR') == (
|
|
sql + ' WHERE test1=1 OR test2=2',
|
|
dict()
|
|
)
|
|
|
|
|
|
def test_add_where_clauses_with_duplicated_field():
|
|
sql = "UPDATE table SET test1=:test1"
|
|
params = dict(test1='new_value')
|
|
where_clauses = dict(test1='where_value')
|
|
assert OracleDB._add_where_clauses(sql, params, where_clauses) == (
|
|
sql + ' WHERE "test1" = :test1_1',
|
|
dict(test1='new_value', test1_1='where_value')
|
|
)
|
|
|
|
|
|
def test_quote_table_name():
|
|
assert OracleDB._quote_table_name("mytable") == '"mytable"'
|
|
assert OracleDB._quote_table_name("myschema.mytable") == '"myschema"."mytable"'
|
|
|
|
|
|
def test_insert(mocker, test_oracledb):
|
|
values = dict(test1=1, test2=2)
|
|
mocker.patch(
|
|
'mylib.oracle.OracleDB.doSQL',
|
|
generate_mock_doSQL(
|
|
'INSERT INTO "mytable" ("test1", "test2") VALUES (:test1, :test2)',
|
|
values
|
|
)
|
|
)
|
|
|
|
assert test_oracledb.insert('mytable', values)
|
|
|
|
|
|
def test_insert_just_try(mocker, test_oracledb):
|
|
mocker.patch('mylib.oracle.OracleDB.doSQL', mock_doSQL_just_try)
|
|
assert test_oracledb.insert('mytable', dict(test1=1, test2=2), just_try=True)
|
|
|
|
|
|
def test_update(mocker, test_oracledb):
|
|
values = dict(test1=1, test2=2)
|
|
where_clauses = dict(test3=3, test4=4)
|
|
mocker.patch(
|
|
'mylib.oracle.OracleDB.doSQL',
|
|
generate_mock_doSQL(
|
|
'UPDATE "mytable" SET "test1" = :test1, "test2" = :test2 WHERE "test3" = :test3 AND "test4" = :test4',
|
|
dict(**values, **where_clauses)
|
|
)
|
|
)
|
|
|
|
assert test_oracledb.update('mytable', values, where_clauses)
|
|
|
|
|
|
def test_update_just_try(mocker, test_oracledb):
|
|
mocker.patch('mylib.oracle.OracleDB.doSQL', mock_doSQL_just_try)
|
|
assert test_oracledb.update('mytable', dict(test1=1, test2=2), None, just_try=True)
|
|
|
|
|
|
def test_delete(mocker, test_oracledb):
|
|
where_clauses = dict(test1=1, test2=2)
|
|
mocker.patch(
|
|
'mylib.oracle.OracleDB.doSQL',
|
|
generate_mock_doSQL(
|
|
'DELETE FROM "mytable" WHERE "test1" = :test1 AND "test2" = :test2',
|
|
where_clauses
|
|
)
|
|
)
|
|
|
|
assert test_oracledb.delete('mytable', where_clauses)
|
|
|
|
|
|
def test_delete_just_try(mocker, test_oracledb):
|
|
mocker.patch('mylib.oracle.OracleDB.doSQL', mock_doSQL_just_try)
|
|
assert test_oracledb.delete('mytable', None, just_try=True)
|
|
|
|
|
|
def test_truncate(mocker, test_oracledb):
|
|
mocker.patch(
|
|
'mylib.oracle.OracleDB.doSQL',
|
|
generate_mock_doSQL('TRUNCATE TABLE "mytable"', None)
|
|
)
|
|
|
|
assert test_oracledb.truncate('mytable')
|
|
|
|
|
|
def test_truncate_just_try(mocker, test_oracledb):
|
|
mocker.patch('mylib.oracle.OracleDB.doSQL', mock_doSelect_just_try)
|
|
assert test_oracledb.truncate('mytable', just_try=True)
|
|
|
|
|
|
def test_select(mocker, test_oracledb):
|
|
fields = ('field1', 'field2')
|
|
where_clauses = dict(test3=3, test4=4)
|
|
expected_return = [
|
|
dict(field1=1, field2=2),
|
|
dict(field1=2, field2=3),
|
|
]
|
|
order_by = "field1, DESC"
|
|
mocker.patch(
|
|
'mylib.oracle.OracleDB.doSelect',
|
|
generate_mock_doSQL(
|
|
'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = :test3 AND "test4" = :test4 ORDER BY ' + order_by,
|
|
where_clauses, expected_return
|
|
)
|
|
)
|
|
|
|
assert test_oracledb.select('mytable', where_clauses, fields, order_by=order_by) == expected_return
|
|
|
|
|
|
def test_select_without_field_and_order_by(mocker, test_oracledb):
|
|
mocker.patch(
|
|
'mylib.oracle.OracleDB.doSelect',
|
|
generate_mock_doSQL(
|
|
'SELECT * FROM "mytable"'
|
|
)
|
|
)
|
|
|
|
assert test_oracledb.select('mytable')
|
|
|
|
|
|
def test_select_just_try(mocker, test_oracledb):
|
|
mocker.patch('mylib.oracle.OracleDB.doSQL', mock_doSelect_just_try)
|
|
assert test_oracledb.select('mytable', None, None, just_try=True)
|
|
|
|
#
|
|
# Tests on main methods
|
|
#
|
|
|
|
|
|
def test_connect(mocker, test_oracledb):
|
|
expected_kwargs = dict(
|
|
dsn=test_oracledb._dsn,
|
|
user=test_oracledb._user,
|
|
password=test_oracledb._pwd
|
|
)
|
|
|
|
mocker.patch(
|
|
'cx_Oracle.connect',
|
|
generate_mock_args(
|
|
expected_kwargs=expected_kwargs
|
|
)
|
|
)
|
|
|
|
assert test_oracledb.connect()
|
|
|
|
|
|
def test_close(fake_oracledb):
|
|
assert fake_oracledb.close() is None
|
|
|
|
|
|
def test_close_connected(fake_connected_oracledb):
|
|
assert fake_connected_oracledb.close() is None
|
|
|
|
|
|
def test_doSQL(fake_connected_oracledb):
|
|
fake_connected_oracledb._conn.expected_sql = 'DELETE FROM table WHERE test1 = :test1'
|
|
fake_connected_oracledb._conn.expected_params = dict(test1=1)
|
|
fake_connected_oracledb.doSQL(fake_connected_oracledb._conn.expected_sql, fake_connected_oracledb._conn.expected_params)
|
|
|
|
|
|
def test_doSQL_without_params(fake_connected_oracledb):
|
|
fake_connected_oracledb._conn.expected_sql = 'DELETE FROM table'
|
|
fake_connected_oracledb.doSQL(fake_connected_oracledb._conn.expected_sql)
|
|
|
|
|
|
def test_doSQL_just_try(fake_connected_just_try_oracledb):
|
|
assert fake_connected_just_try_oracledb.doSQL('DELETE FROM table')
|
|
|
|
|
|
def test_doSQL_on_exception(fake_connected_oracledb):
|
|
fake_connected_oracledb._conn.expected_exception = True
|
|
assert fake_connected_oracledb.doSQL('DELETE FROM table') is False
|
|
|
|
|
|
def test_doSelect(fake_connected_oracledb):
|
|
fake_connected_oracledb._conn.expected_sql = 'SELECT * FROM table WHERE test1 = :test1'
|
|
fake_connected_oracledb._conn.expected_params = dict(test1=1)
|
|
fake_connected_oracledb._conn.expected_return = [dict(test1=1)]
|
|
assert fake_connected_oracledb.doSelect(
|
|
fake_connected_oracledb._conn.expected_sql,
|
|
fake_connected_oracledb._conn.expected_params) == fake_connected_oracledb._conn.expected_return
|
|
|
|
|
|
def test_doSelect_without_params(fake_connected_oracledb):
|
|
fake_connected_oracledb._conn.expected_sql = 'SELECT * FROM table'
|
|
fake_connected_oracledb._conn.expected_return = [dict(test1=1)]
|
|
assert fake_connected_oracledb.doSelect(fake_connected_oracledb._conn.expected_sql) == fake_connected_oracledb._conn.expected_return
|
|
|
|
|
|
def test_doSelect_on_exception(fake_connected_oracledb):
|
|
fake_connected_oracledb._conn.expected_exception = True
|
|
assert fake_connected_oracledb.doSelect('SELECT * FROM table') is False
|
|
|
|
|
|
def test_doSelect_just_try(fake_connected_just_try_oracledb):
|
|
fake_connected_just_try_oracledb._conn.expected_sql = 'SELECT * FROM table WHERE test1 = :test1'
|
|
fake_connected_just_try_oracledb._conn.expected_params = dict(test1=1)
|
|
fake_connected_just_try_oracledb._conn.expected_return = [dict(test1=1)]
|
|
assert fake_connected_just_try_oracledb.doSelect(
|
|
fake_connected_just_try_oracledb._conn.expected_sql,
|
|
fake_connected_just_try_oracledb._conn.expected_params
|
|
) == fake_connected_just_try_oracledb._conn.expected_return
|