# pylint: disable=redefined-outer-name,missing-function-docstring,protected-access """ Tests on opening hours helpers """ import pytest from MySQLdb._exceptions import Error from mylib.mysql import MyDB class FakeMySQLdbCursor: """Fake MySQLdb 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 Error(f"{self}.execute({sql}, {params}): expected exception") if self.expected_just_try and not sql.lower().startswith("select "): assert False, f"{self}.execute({sql}, {params}) may not be executed in just try mode" # pylint: disable=consider-using-f-string assert ( sql == self.expected_sql ), "%s.execute(): Invalid SQL query:\n '%s'\nMay be:\n '%s'" % ( self, sql, self.expected_sql, ) # pylint: disable=consider-using-f-string assert ( params == self.expected_params ), "%s.execute(): Invalid params:\n %s\nMay be:\n %s" % ( self, params, self.expected_params, ) return self.expected_return @property def description(self): assert self.expected_return assert isinstance(self.expected_return, list) assert isinstance(self.expected_return[0], dict) return [(field, 1, 2, 3) for field in self.expected_return[0].keys()] def fetchall(self): if isinstance(self.expected_return, list): return ( list(row.values()) if isinstance(row, dict) else row for row in self.expected_return ) return self.expected_return def __repr__(self): return ( f"FakeMySQLdbCursor({self.expected_sql}, {self.expected_params}, " f"{self.expected_return}, {self.expected_just_try})" ) class FakeMySQLdb: """Fake MySQLdb 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 = { "db": str, "user": str, "passwd": (str, None), "host": str, "charset": str, "use_unicode": bool, } for arg, value in kwargs.items(): assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"' assert isinstance( value, allowed_kwargs[arg] ), f"Arg {arg} not a {allowed_kwargs[arg]} ({type(value)})" setattr(self, arg, value) def close(self): return self.expected_return def cursor(self): return FakeMySQLdbCursor( 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_mysqldb_connect(**kwargs): return FakeMySQLdb(**kwargs) def fake_mysqldb_connect_just_try(**kwargs): con = FakeMySQLdb(**kwargs) con.just_try = True return con @pytest.fixture def test_mydb(): return MyDB("127.0.0.1", "user", "password", "dbname") @pytest.fixture def fake_mydb(mocker): mocker.patch("MySQLdb.connect", fake_mysqldb_connect) return MyDB("127.0.0.1", "user", "password", "dbname") @pytest.fixture def fake_just_try_mydb(mocker): mocker.patch("MySQLdb.connect", fake_mysqldb_connect_just_try) return MyDB("127.0.0.1", "user", "password", "dbname", just_try=True) @pytest.fixture def fake_connected_mydb(fake_mydb): fake_mydb.connect() return fake_mydb @pytest.fixture def fake_connected_just_try_mydb(fake_just_try_mydb): fake_just_try_mydb.connect() return fake_just_try_mydb def generate_mock_args( expected_args=(), expected_kwargs={}, expected_return=True ): # pylint: disable=dangerous-default-value def mock_args(*args, **kwargs): # pylint: disable=consider-using-f-string assert args == expected_args, "Invalid call args:\n %s\nMay be:\n %s" % ( args, expected_args, ) # pylint: disable=consider-using-f-string 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={}, expected_return=True ): # pylint: disable=dangerous-default-value def mock_doSQL(self, sql, params=None): # pylint: disable=unused-argument # pylint: disable=consider-using-f-string assert sql == expected_sql, "Invalid generated SQL query:\n '%s'\nMay be:\n '%s'" % ( sql, expected_sql, ) # pylint: disable=consider-using-f-string assert params == expected_params, "Invalid generated params:\n %s\nMay be:\n %s" % ( params, expected_params, ) return expected_return return mock_doSQL # MyDB.doSelect() have same expected parameters as MyDB.doSQL() generate_mock_doSelect = generate_mock_doSQL mock_doSelect_just_try = mock_doSQL_just_try # # Test on MyDB helper methods # def test_combine_params_with_to_add_parameter(): assert MyDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2} def test_combine_params_with_kargs(): assert MyDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2} def test_combine_params_with_kargs_and_to_add_parameter(): assert MyDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == { "test1": 1, "test2": 2, "test3": 3, } def test_format_where_clauses_params_are_preserved(): args = ("test = test", {"test1": 1}) assert MyDB._format_where_clauses(*args) == args def test_format_where_clauses_raw(): assert MyDB._format_where_clauses("test = test") == ("test = test", {}) def test_format_where_clauses_tuple_clause_with_params(): where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", {"test1": 1, "test2": 2}) assert MyDB._format_where_clauses(where_clauses) == where_clauses def test_format_where_clauses_dict(): where_clauses = {"test1": 1, "test2": 2} assert MyDB._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", {"test2": 2}), {"test3": 3, "test4": 4}) assert MyDB._format_where_clauses(where_clauses) == ( "test1 = 1 AND test2 LIKE %(test2)s AND `test3` = %(test3)s AND `test4` = %(test4)s", {"test2": 2, "test3": 3, "test4": 4}, ) def test_format_where_clauses_with_where_op(): where_clauses = {"test1": 1, "test2": 2} assert MyDB._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 = {"test1": 1, "test2": 2} assert MyDB._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 = {"test1": 1, "test2": 2} params = {"fake1": 1} assert MyDB._add_where_clauses(sql, params.copy(), where_clauses) == ( sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s", {**where_clauses, **params}, ) def test_add_where_clauses_with_op(): sql = "SELECT * FROM table" where_clauses = ("test1=1", "test2=2") assert MyDB._add_where_clauses(sql, None, where_clauses, where_op="OR") == ( sql + " WHERE test1=1 OR test2=2", {}, ) def test_add_where_clauses_with_duplicated_field(): sql = "UPDATE table SET test1=%(test1)s" params = {"test1": "new_value"} where_clauses = {"test1": "where_value"} assert MyDB._add_where_clauses(sql, params, where_clauses) == ( sql + " WHERE `test1` = %(test1_1)s", {"test1": "new_value", "test1_1": "where_value"}, ) def test_quote_table_name(): assert MyDB._quote_table_name("mytable") == "`mytable`" assert MyDB._quote_table_name("myschema.mytable") == "`myschema`.`mytable`" def test_insert(mocker, test_mydb): values = {"test1": 1, "test2": 2} mocker.patch( "mylib.mysql.MyDB.doSQL", generate_mock_doSQL( "INSERT INTO `mytable` (`test1`, `test2`) VALUES (%(test1)s, %(test2)s)", values ), ) assert test_mydb.insert("mytable", values) def test_insert_just_try(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try) assert test_mydb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True) def test_update(mocker, test_mydb): values = {"test1": 1, "test2": 2} where_clauses = {"test3": 3, "test4": 4} mocker.patch( "mylib.mysql.MyDB.doSQL", generate_mock_doSQL( "UPDATE `mytable` SET `test1` = %(test1)s, `test2` = %(test2)s WHERE `test3` =" " %(test3)s AND `test4` = %(test4)s", {**values, **where_clauses}, ), ) assert test_mydb.update("mytable", values, where_clauses) def test_update_just_try(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try) assert test_mydb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True) def test_delete(mocker, test_mydb): where_clauses = {"test1": 1, "test2": 2} mocker.patch( "mylib.mysql.MyDB.doSQL", generate_mock_doSQL( "DELETE FROM `mytable` WHERE `test1` = %(test1)s AND `test2` = %(test2)s", where_clauses ), ) assert test_mydb.delete("mytable", where_clauses) def test_delete_just_try(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try) assert test_mydb.delete("mytable", None, just_try=True) def test_truncate(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSQL", generate_mock_doSQL("TRUNCATE TABLE `mytable`", None)) assert test_mydb.truncate("mytable") def test_truncate_just_try(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSelect_just_try) assert test_mydb.truncate("mytable", just_try=True) def test_select(mocker, test_mydb): fields = ("field1", "field2") where_clauses = {"test3": 3, "test4": 4} expected_return = [ {"field1": 1, "field2": 2}, {"field1": 2, "field2": 3}, ] order_by = "field1, DESC" mocker.patch( "mylib.mysql.MyDB.doSelect", generate_mock_doSQL( "SELECT `field1`, `field2` FROM `mytable` WHERE `test3` = %(test3)s AND `test4` =" " %(test4)s ORDER BY " + order_by, # nosec: B608 where_clauses, expected_return, ), ) assert test_mydb.select("mytable", where_clauses, fields, order_by=order_by) == expected_return def test_select_without_field_and_order_by(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSelect", generate_mock_doSQL("SELECT * FROM `mytable`")) assert test_mydb.select("mytable") def test_select_just_try(mocker, test_mydb): mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSelect_just_try) assert test_mydb.select("mytable", None, None, just_try=True) # # Tests on main methods # def test_connect(mocker, test_mydb): expected_kwargs = { "db": test_mydb._db, "user": test_mydb._user, "host": test_mydb._host, "passwd": test_mydb._pwd, "charset": test_mydb._charset, "use_unicode": True, } mocker.patch("MySQLdb.connect", generate_mock_args(expected_kwargs=expected_kwargs)) assert test_mydb.connect() def test_close(fake_mydb): assert fake_mydb.close() is None def test_close_connected(fake_connected_mydb): assert fake_connected_mydb.close() is None def test_doSQL(fake_connected_mydb): fake_connected_mydb._conn.expected_sql = "DELETE FROM table WHERE test1 = %(test1)s" fake_connected_mydb._conn.expected_params = {"test1": 1} fake_connected_mydb.doSQL( fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params ) def test_doSQL_without_params(fake_connected_mydb): fake_connected_mydb._conn.expected_sql = "DELETE FROM table" fake_connected_mydb.doSQL(fake_connected_mydb._conn.expected_sql) def test_doSQL_just_try(fake_connected_just_try_mydb): assert fake_connected_just_try_mydb.doSQL("DELETE FROM table") def test_doSQL_on_exception(fake_connected_mydb): fake_connected_mydb._conn.expected_exception = True assert fake_connected_mydb.doSQL("DELETE FROM table") is False def test_doSelect(fake_connected_mydb): fake_connected_mydb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_mydb._conn.expected_params = {"test1": 1} fake_connected_mydb._conn.expected_return = [{"test1": 1}] assert ( fake_connected_mydb.doSelect( fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params ) == fake_connected_mydb._conn.expected_return ) def test_doSelect_without_params(fake_connected_mydb): fake_connected_mydb._conn.expected_sql = "SELECT * FROM table" fake_connected_mydb._conn.expected_return = [{"test1": 1}] assert ( fake_connected_mydb.doSelect(fake_connected_mydb._conn.expected_sql) == fake_connected_mydb._conn.expected_return ) def test_doSelect_on_exception(fake_connected_mydb): fake_connected_mydb._conn.expected_exception = True assert fake_connected_mydb.doSelect("SELECT * FROM table") is False def test_doSelect_just_try(fake_connected_just_try_mydb): fake_connected_just_try_mydb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_just_try_mydb._conn.expected_params = {"test1": 1} fake_connected_just_try_mydb._conn.expected_return = [{"test1": 1}] assert ( fake_connected_just_try_mydb.doSelect( fake_connected_just_try_mydb._conn.expected_sql, fake_connected_just_try_mydb._conn.expected_params, ) == fake_connected_just_try_mydb._conn.expected_return )