particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

rework DBQueryTestCase mocking

Browse source 
Mocking kojihub.kojihub.QueryProcessor doesn't affect other import
paths, e.g. in other kojihub modules
This commit is contained in:
Mike McLean 2024-03-29 18:44:48 -04:00 • committed by Tomas Kopecek
parent 4df957cb3d
commit a807137c3c
1 changed files with 25 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

43
tests/test_hub/utils.py
View file

@ -2,8 +2,27 @@ import mock
import unittest
import kojihub
import kojihub.db
QP = kojihub.QueryProcessor
def get_qp_init(testcase):
orig_qp_init = kojihub.db.QueryProcessor.__init__
def my_qp_init(_query, *a, **kw):
_query.execute = mock.MagicMock()
_query.execute.return_value = testcase.qp_execute_return_value
_query.execute.side_effect = testcase.qp_execute_side_effect
_query.executeOne = mock.MagicMock()
_query.executeOne.return_value = testcase.qp_execute_one_return_value
_query.executeOne.side_effect = testcase.qp_execute_one_side_effect
_query.singleValue = mock.MagicMock()
_query.singleValue.return_value = testcase.qp_single_value_return_value
_query.iterate = mock.MagicMock()
_query.iterate.return_value = testcase.qp_iterate_return_value
testcase.queries.append(_query)
return orig_qp_init(_query, *a, **kw)
return my_qp_init
class DBQueryTestCase(unittest.TestCase):
@ -16,8 +35,11 @@ class DBQueryTestCase(unittest.TestCase):
self.qp_execute_one_side_effect = None
self.qp_single_value_return_value = None
self.qp_iterate_return_value = None
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.get_query).start()
# patch init to catch queries regardless of how QP is imported
new_init = get_qp_init(self)
self.qp_init = mock.patch('kojihub.db.QueryProcessor.__init__', new=new_init).start()
self.queries = []
def tearDown(self):
@ -27,21 +49,6 @@ class DBQueryTestCase(unittest.TestCase):
def reset_query(self):
del self.queries[:]
def get_query(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.execute.return_value = self.qp_execute_return_value
query.execute.side_effect = self.qp_execute_side_effect
query.executeOne = mock.MagicMock()
query.executeOne.return_value = self.qp_execute_one_return_value
query.executeOne.side_effect = self.qp_execute_one_side_effect
query.singleValue = mock.MagicMock()
query.singleValue.return_value = self.qp_single_value_return_value
query.iterate = mock.MagicMock()
query.iterate.return_value = self.qp_iterate_return_value
self.queries.append(query)
return query
def assertQueryEqual(self, query, **kwargs):
for k, v in kwargs.items():
self.assertEqual(getattr(query, k, None), v)