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

support batch operation in BulkInsertProcessor

Browse source
This commit is contained in:
Mike McLean 2019-11-12 16:12:34 -05:00
parent 116bf7adbc
commit 727de7d953
3 changed files with 61 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

11
tests/test_hub/test_complete_image_build.py
View file

@ -30,10 +30,11 @@ def make_insert_grabber(test):
def make_bulk_insert_grabber(test):
# test is the test class instance
def grab_insert(insert):
# insert is self for the InsertProcessor instance
# we are replacing execute()
info = [str(insert), copy.copy(insert.prepared_data)]
def grab_insert(insert, data):
# insert is self for the BulkInsertProcessor instance
# we are replacing _one_insert()
query, params = insert._get_insert(data)
info = [query, copy.copy(params)]
test.inserts.append(info)
return grab_insert
@ -78,7 +79,7 @@ class TestCompleteImageBuild(unittest.TestCase):
self.updates = []
mock.patch.object(kojihub.InsertProcessor, 'execute',
new=make_insert_grabber(self)).start()
mock.patch.object(kojihub.BulkInsertProcessor, 'execute',
mock.patch.object(kojihub.BulkInsertProcessor, '_one_insert',
new=make_bulk_insert_grabber(self)).start()
mock.patch.object(kojihub.UpdateProcessor, 'execute',
new=make_update_grabber(self)).start()

22
tests/test_hub/test_insert_processor.py
View file

@ -98,7 +98,7 @@ class TestBulkInsertProcessor(unittest.TestCase):
def test_basic_instantiation(self):
proc = kojihub.BulkInsertProcessor('sometable')
actual = str(proc)
expected = '-- incomplete update: no assigns'
expected = '-- incomplete insert: no data'
self.assertEquals(actual, expected)
def test_to_string_with_single_row(self):
@ -169,3 +169,23 @@ class TestBulkInsertProcessor(unittest.TestCase):
str(proc)
self.assertEquals(cm.exception.args[0], 'Missing value foo2 in BulkInsert')
@mock.patch('kojihub.context')
def test_batch_execution(self, context):
cursor = mock.MagicMock()
context.cnx.cursor.return_value = cursor
proc = kojihub.BulkInsertProcessor('sometable', data=[{'foo': 'bar1'}], batch=2)
proc.add_record(foo='bar2')
proc.add_record(foo='bar3')
proc.execute()
calls = cursor.execute.mock_calls
# list of (name, positional args, keyword args)
self.assertEquals(len(calls), 2)
self.assertEquals(
calls[0][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s), (%(foo1)s)',
{'foo0': 'bar1', 'foo1': 'bar2'}))
self.assertEquals(
calls[1][1],
('INSERT INTO sometable (foo) VALUES (%(foo0)s)',
{'foo0': 'bar3'}))