# pylint: disable=redefined-outer-name,missing-function-docstring,protected-access """ Tests on opening hours helpers """ import pytest from mylib.pgsql import PgDB class FakePsycopg2Cursor: """ Fake Psycopg2 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=None): 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 "FakePsycopg2Cursor(%s, %s, %s, %s)" % ( self.expected_sql, self.expected_params, self.expected_return, self.expected_just_try ) class FakePsycopg2: """ Fake Psycopg2 connection """ expected_sql = None expected_params = None expected_return = True expected_just_try = False expected_exception = False just_try = False def __init__(self, **kwargs): allowed_kwargs = dict(dbname=str, user=str, password=(str, None), host=str) 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 set_client_encoding(self, *arg): self._check_just_try() assert len(arg) == 1 and isinstance(arg[0], str) if self.expected_exception: raise Exception("set_client_encoding(%s): Expected exception" % arg[0]) return self.expected_return def cursor(self): return FakePsycopg2Cursor( 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_psycopg2_connect(**kwargs): return FakePsycopg2(**kwargs) def fake_psycopg2_connect_just_try(**kwargs): con = FakePsycopg2(**kwargs) con.just_try = True return con @pytest.fixture def test_pgdb(): return PgDB('127.0.0.1', 'user', 'password', 'dbname') @pytest.fixture def fake_pgdb(mocker): mocker.patch('psycopg2.connect', fake_psycopg2_connect) return PgDB('127.0.0.1', 'user', 'password', 'dbname') @pytest.fixture def fake_just_try_pgdb(mocker): mocker.patch('psycopg2.connect', fake_psycopg2_connect_just_try) return PgDB('127.0.0.1', 'user', 'password', 'dbname', just_try=True) @pytest.fixture def fake_connected_pgdb(fake_pgdb): fake_pgdb.connect() return fake_pgdb @pytest.fixture def fake_connected_just_try_pgdb(fake_just_try_pgdb): fake_just_try_pgdb.connect() return fake_just_try_pgdb 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 # PgDB.doSelect() have same expected parameters as PgDB.doSQL() generate_mock_doSelect = generate_mock_doSQL mock_doSelect_just_try = mock_doSQL_just_try # # Test on PgDB helper methods # def test_combine_params_with_to_add_parameter(): assert PgDB._combine_params(dict(test1=1), dict(test2=2)) == dict( test1=1, test2=2 ) def test_combine_params_with_kargs(): assert PgDB._combine_params(dict(test1=1), test2=2) == dict( test1=1, test2=2 ) def test_combine_params_with_kargs_and_to_add_parameter(): assert PgDB._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 PgDB._format_where_clauses(*args) == args def test_format_where_clauses_raw(): assert PgDB._format_where_clauses('test = test') == (('test = test'), dict()) def test_format_where_clauses_tuple_clause_with_params(): where_clauses = ( 'test1 = %(test1)s AND test2 = %(test2)s', dict(test1=1, test2=2) ) assert PgDB._format_where_clauses(where_clauses) == where_clauses def test_format_where_clauses_dict(): where_clauses = dict(test1=1, test2=2) assert PgDB._format_where_clauses(where_clauses) == ( '"test1" = %(test1)s AND "test2" = %(test2)s', where_clauses ) def test_format_where_clauses_combined_types(): where_clauses = ( 'test1 = 1', ('test2 LIKE %(test2)s', dict(test2=2)), dict(test3=3, test4=4) ) assert PgDB._format_where_clauses(where_clauses) == ( 'test1 = 1 AND test2 LIKE %(test2)s AND "test3" = %(test3)s AND "test4" = %(test4)s', dict(test2=2, test3=3, test4=4) ) def test_format_where_clauses_with_where_op(): where_clauses = dict(test1=1, test2=2) assert PgDB._format_where_clauses(where_clauses, where_op='OR') == ( '"test1" = %(test1)s OR "test2" = %(test2)s', where_clauses ) def test_add_where_clauses(): sql = "SELECT * FROM table" where_clauses = dict(test1=1, test2=2) assert PgDB._add_where_clauses(sql, None, where_clauses) == ( sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s', 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 PgDB._add_where_clauses(sql, params.copy(), where_clauses) == ( sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s', dict(**where_clauses, **params) ) def test_add_where_clauses_with_op(): sql = "SELECT * FROM table" where_clauses = ('test1=1', 'test2=2') assert PgDB._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)s" params = dict(test1='new_value') where_clauses = dict(test1='where_value') assert PgDB._add_where_clauses(sql, params, where_clauses) == ( sql + ' WHERE "test1" = %(test1_1)s', dict(test1='new_value', test1_1='where_value') ) def test_quote_table_name(): assert PgDB._quote_table_name("mytable") == '"mytable"' assert PgDB._quote_table_name("myschema.mytable") == '"myschema"."mytable"' def test_insert(mocker, test_pgdb): values = dict(test1=1, test2=2) mocker.patch( 'mylib.pgsql.PgDB.doSQL', generate_mock_doSQL( 'INSERT INTO "mytable" ("test1", "test2") VALUES (%(test1)s, %(test2)s)', values ) ) assert test_pgdb.insert('mytable', values) def test_insert_just_try(mocker, test_pgdb): mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSQL_just_try) assert test_pgdb.insert('mytable', dict(test1=1, test2=2), just_try=True) def test_update(mocker, test_pgdb): values = dict(test1=1, test2=2) where_clauses = dict(test3=3, test4=4) mocker.patch( 'mylib.pgsql.PgDB.doSQL', generate_mock_doSQL( 'UPDATE "mytable" SET "test1" = %(test1)s, "test2" = %(test2)s WHERE "test3" = %(test3)s AND "test4" = %(test4)s', dict(**values, **where_clauses) ) ) assert test_pgdb.update('mytable', values, where_clauses) def test_update_just_try(mocker, test_pgdb): mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSQL_just_try) assert test_pgdb.update('mytable', dict(test1=1, test2=2), None, just_try=True) def test_delete(mocker, test_pgdb): where_clauses = dict(test1=1, test2=2) mocker.patch( 'mylib.pgsql.PgDB.doSQL', generate_mock_doSQL( 'DELETE FROM "mytable" WHERE "test1" = %(test1)s AND "test2" = %(test2)s', where_clauses ) ) assert test_pgdb.delete('mytable', where_clauses) def test_delete_just_try(mocker, test_pgdb): mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSQL_just_try) assert test_pgdb.delete('mytable', None, just_try=True) def test_truncate(mocker, test_pgdb): mocker.patch( 'mylib.pgsql.PgDB.doSQL', generate_mock_doSQL('TRUNCATE "mytable"', None) ) assert test_pgdb.truncate('mytable') def test_truncate_just_try(mocker, test_pgdb): mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSelect_just_try) assert test_pgdb.truncate('mytable', just_try=True) def test_select(mocker, test_pgdb): 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.pgsql.PgDB.doSelect', generate_mock_doSQL( 'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = %(test3)s AND "test4" = %(test4)s ORDER BY ' + order_by, where_clauses, expected_return ) ) assert test_pgdb.select('mytable', where_clauses, fields, order_by=order_by) == expected_return def test_select_without_field_and_order_by(mocker, test_pgdb): mocker.patch( 'mylib.pgsql.PgDB.doSelect', generate_mock_doSQL( 'SELECT * FROM "mytable"' ) ) assert test_pgdb.select('mytable') def test_select_just_try(mocker, test_pgdb): mocker.patch('mylib.pgsql.PgDB.doSQL', mock_doSelect_just_try) assert test_pgdb.select('mytable', None, None, just_try=True) # # Tests on main methods # def test_connect(mocker, test_pgdb): expected_kwargs = dict( dbname=test_pgdb._db, user=test_pgdb._user, host=test_pgdb._host, password=test_pgdb._pwd ) mocker.patch( 'psycopg2.connect', generate_mock_args( expected_kwargs=expected_kwargs ) ) assert test_pgdb.connect() def test_close(fake_pgdb): assert fake_pgdb.close() is None def test_close_connected(fake_connected_pgdb): assert fake_connected_pgdb.close() is None def test_setEncoding(fake_connected_pgdb): assert fake_connected_pgdb.setEncoding('utf8') def test_setEncoding_not_connected(fake_pgdb): assert fake_pgdb.setEncoding('utf8') is False def test_setEncoding_on_exception(fake_connected_pgdb): fake_connected_pgdb._conn.expected_exception = True assert fake_connected_pgdb.setEncoding('utf8') is False def test_doSQL(fake_connected_pgdb): fake_connected_pgdb._conn.expected_sql = 'DELETE FROM table WHERE test1 = %(test1)s' fake_connected_pgdb._conn.expected_params = dict(test1=1) fake_connected_pgdb.doSQL(fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params) def test_doSQL_without_params(fake_connected_pgdb): fake_connected_pgdb._conn.expected_sql = 'DELETE FROM table' fake_connected_pgdb.doSQL(fake_connected_pgdb._conn.expected_sql) def test_doSQL_just_try(fake_connected_just_try_pgdb): assert fake_connected_just_try_pgdb.doSQL('DELETE FROM table') def test_doSQL_on_exception(fake_connected_pgdb): fake_connected_pgdb._conn.expected_exception = True assert fake_connected_pgdb.doSQL('DELETE FROM table') is False def test_doSelect(fake_connected_pgdb): fake_connected_pgdb._conn.expected_sql = 'SELECT * FROM table WHERE test1 = %(test1)s' fake_connected_pgdb._conn.expected_params = dict(test1=1) fake_connected_pgdb._conn.expected_return = [dict(test1=1)] assert fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params) == fake_connected_pgdb._conn.expected_return def test_doSelect_without_params(fake_connected_pgdb): fake_connected_pgdb._conn.expected_sql = 'SELECT * FROM table' fake_connected_pgdb._conn.expected_return = [dict(test1=1)] assert fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql) == fake_connected_pgdb._conn.expected_return def test_doSelect_on_exception(fake_connected_pgdb): fake_connected_pgdb._conn.expected_exception = True assert fake_connected_pgdb.doSelect('SELECT * FROM table') is False def test_doSelect_just_try(fake_connected_just_try_pgdb): fake_connected_just_try_pgdb._conn.expected_sql = 'SELECT * FROM table WHERE test1 = %(test1)s' fake_connected_just_try_pgdb._conn.expected_params = dict(test1=1) fake_connected_just_try_pgdb._conn.expected_return = [dict(test1=1)] assert fake_connected_just_try_pgdb.doSelect( fake_connected_just_try_pgdb._conn.expected_sql, fake_connected_just_try_pgdb._conn.expected_params ) == fake_connected_just_try_pgdb._conn.expected_return