From 334c46e252a112edc223279ae4239f22ea5abc42 Mon Sep 17 00:00:00 2001 From: Jana Cupova Date: Apr 20 2023 12:00:10 +0000 Subject: Increase hub unit tests --- diff --git a/tests/test_hub/test_add_archivetype.py b/tests/test_hub/test_add_archivetype.py index 4e8e095..68a0440 100644 --- a/tests/test_hub/test_add_archivetype.py +++ b/tests/test_hub/test_add_archivetype.py @@ -1,17 +1,16 @@ -import unittest - import mock import koji import kojihub import kojihub.kojihub +from .utils import DBQueryTestCase IP = kojihub.InsertProcessor -QP = kojihub.QueryProcessor -class TestAddArchiveType(unittest.TestCase): +class TestAddArchiveType(DBQueryTestCase): def setUp(self): + super(TestAddArchiveType, self).setUp() self.context = mock.patch('kojihub.kojihub.context').start() self.context.session.assertPerm = mock.MagicMock() self.exports = kojihub.RootExports() @@ -23,10 +22,6 @@ class TestAddArchiveType(unittest.TestCase): self.insert_execute = mock.MagicMock() self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start() self.get_archive_type = mock.patch('kojihub.kojihub.get_archive_type').start() - self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor', - side_effect=self.getQuery).start() - self.queries = [] - self.query_execute = mock.MagicMock() def tearDown(self): mock.patch.stopall() @@ -37,18 +32,19 @@ class TestAddArchiveType(unittest.TestCase): self.inserts.append(insert) return insert - def getQuery(self, *args, **kwargs): - query = QP(*args, **kwargs) - query.execute = self.query_execute - self.queries.append(query) - return query - def test_add_archive_type_valid_empty_compression_type(self): - self.query_execute.side_effect = [[]] + self.qp_execute_side_effect = [[]] self.verify_name_internal.return_value = None self.get_archive_type.return_value = None - kojihub.add_archive_type('deb', 'Debian package', 'deb') - + ext = 'deb' + rv = kojihub.add_archive_type('deb', 'Debian package', ext) + + self.assertEqual(len(self.queries), 1) + self.assertLastQueryEqual(tables=['archivetypes'], + columns=['id'], + clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], + values={}) + self.assertEqual(rv, None) self.assertEqual(len(self.inserts), 1) insert = self.inserts[0] self.assertEqual(insert.table, 'archivetypes') @@ -58,29 +54,41 @@ class TestAddArchiveType(unittest.TestCase): 'compression_type': None}) self.assertEqual(insert.rawdata, {}) self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_called_once_with(type_name='deb') def test_add_archive_type_valid_with_compression_type(self): - self.query_execute.side_effect = [[]] + self.qp_execute_side_effect = [[]] self.verify_name_internal.return_value = None self.get_archive_type.return_value = None - kojihub.add_archive_type('jar', 'Jar package', 'jar', compression_type='zip') - + ext = '.jar' + kojihub.add_archive_type('jar', 'Jar package', ext, compression_type='zip') + + self.assertEqual(len(self.queries), 1) + self.assertLastQueryEqual(tables=['archivetypes'], + columns=['id'], + clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], + values={}) self.assertEqual(len(self.inserts), 1) insert = self.inserts[0] self.assertEqual(insert.table, 'archivetypes') self.assertEqual(insert.data, {'name': 'jar', 'description': 'Jar package', - 'extensions': 'jar', + 'extensions': '.jar', 'compression_type': 'zip'}) self.assertEqual(insert.rawdata, {}) self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_called_once_with(type_name='jar') def test_add_archive_type_already_exists(self): self.get_archive_type.return_value = True - with self.assertRaises(koji.GenericError): - kojihub.add_archive_type('deb', 'Debian package', 'deb') + name = 'deb' + with self.assertRaises(koji.GenericError) as ex: + kojihub.add_archive_type(name, 'Debian package', '.deb') + self.assertEqual(f"archivetype {name} already exists", str(ex.exception)) self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.queries), 0) self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_called_once_with(type_name=name) def test_add_archive_type_invalid_value_type(self): self.verify_name_internal.return_value = None @@ -89,6 +97,10 @@ class TestAddArchiveType(unittest.TestCase): kojihub.add_archive_type('deb', description, 'deb') self.assertEqual(f"Invalid type for value '{description}': {type(description)}, " f"expected type ", str(ex.exception)) + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.queries), 0) + self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_not_called() def test_add_archive_type_invalid_value_extensions(self): extensions = ['deb'] @@ -96,6 +108,52 @@ class TestAddArchiveType(unittest.TestCase): kojihub.add_archive_type('deb', 'Debian package', extensions) self.assertEqual(f"Invalid type for value '{extensions}': {type(extensions)}, " f"expected type ", str(ex.exception)) + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.queries), 0) + self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_not_called() + + def test_add_archive_type_extension_exists(self): + ext = '.jar' + self.qp_execute_side_effect = [{'id': 123}] + self.get_archive_type.return_value = None + with self.assertRaises(koji.GenericError) as ex: + kojihub.add_archive_type('jar', 'Jar package', ext, compression_type='zip') + self.assertEqual(f'file extension {ext} already exists', str(ex.exception)) + + self.assertEqual(len(self.queries), 1) + self.assertLastQueryEqual(tables=['archivetypes'], + columns=['id'], + clauses=[f"extensions ~* E'(\\s|^){ext}(\\s|$)'"], + values={}) + self.assertEqual(len(self.inserts), 0) + self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_called_once_with(type_name='jar') + + def test_add_archive_type_unsupported_compression_type(self): + ext = '.jar' + compression_type = 'gzip' + self.qp_execute_side_effect = [{'id': 123}] + self.get_archive_type.return_value = None + with self.assertRaises(koji.GenericError) as ex: + kojihub.add_archive_type('jar', 'Jar package', ext, compression_type=compression_type) + self.assertEqual(f"Unsupported compression type {compression_type}", str(ex.exception)) + + self.assertEqual(len(self.queries), 0) + self.assertEqual(len(self.inserts), 0) + self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_not_called() + + def test_add_archive_type_not_alnum(self): + ext = '.jar@' + self.get_archive_type.return_value = None + with self.assertRaises(koji.GenericError) as ex: + kojihub.add_archive_type('jar', 'Jar package', ext, compression_type='zip') + self.assertEqual(f'No such {ext} file extension', str(ex.exception)) + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.queries), 0) + self.context.session.assertPerm.assert_called_with('admin') + self.get_archive_type.assert_called_once_with(type_name='jar') def test_add_archive_type_wrong_name_verify(self): # name is longer as expected @@ -103,8 +161,14 @@ class TestAddArchiveType(unittest.TestCase): self.verify_name_internal.side_effect = koji.GenericError with self.assertRaises(koji.GenericError): kojihub.add_archive_type(new_archive_type, 'Debian package', 'deb') + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.queries), 0) + self.get_archive_type.assert_not_called() # not except regex rules self.verify_name_internal.side_effect = koji.GenericError with self.assertRaises(koji.GenericError): kojihub.add_archive_type(new_archive_type, 'Debian package', 'deb') + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.queries), 0) + self.get_archive_type.assert_not_called() diff --git a/tests/test_hub/test_add_volume.py b/tests/test_hub/test_add_volume.py index 3bbaf80..de9023a 100644 --- a/tests/test_hub/test_add_volume.py +++ b/tests/test_hub/test_add_volume.py @@ -10,6 +10,9 @@ class TestAddVolume(unittest.TestCase): def setUp(self): self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start() + self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start() + self.isdir = mock.patch('os.path.isdir').start() + self.pathinfo_volumedir = mock.patch('koji.pathinfo.volumedir').start() self.context = mock.patch('kojihub.kojihub.context').start() # It seems MagicMock will not automatically handle attributes that # start with "assert" @@ -28,3 +31,36 @@ class TestAddVolume(unittest.TestCase): self.verify_name_internal.side_effect = koji.GenericError with self.assertRaises(koji.GenericError): kojihub.add_volume(volume_name) + + def test_non_exist_directory(self): + volume_name = 'test-volume' + self.isdir.return_value = False + self.pathinfo_volumedir.return_value = 'path/to/volume' + with self.assertRaises(koji.GenericError) as cm: + kojihub.add_volume(volume_name) + self.assertEqual("please create the volume directory first", str(cm.exception)) + self.verify_name_internal.assert_called_once_with(volume_name) + self.lookup_name.assert_not_called() + + def test_valid(self): + volume_name = 'test-volume' + volume_dict = {'id': 0, 'name': volume_name} + self.isdir.return_value = True + self.pathinfo_volumedir.return_value = 'path/to/volume' + self.lookup_name.return_value = volume_dict + rv = kojihub.add_volume(volume_name, strict=False) + self.assertEqual(rv, volume_dict) + self.verify_name_internal.assert_called_once_with(volume_name) + self.lookup_name.assert_called_once_with('volume', volume_name, strict=False, create=True) + + def test_volume_exists(self): + volume_name = 'test-volume' + volume_dict = {'id': 0, 'name': volume_name} + self.isdir.return_value = True + self.pathinfo_volumedir.return_value = 'path/to/volume' + self.lookup_name.return_value = volume_dict + with self.assertRaises(koji.GenericError) as cm: + kojihub.add_volume(volume_name, strict=True) + self.assertEqual(f'volume {volume_name} already exists', str(cm.exception)) + self.verify_name_internal.assert_called_once_with(volume_name) + self.lookup_name.assert_called_once_with('volume', volume_name, strict=False) diff --git a/tests/test_hub/test_build_notification.py b/tests/test_hub/test_build_notification.py new file mode 100644 index 0000000..a7d6cfb --- /dev/null +++ b/tests/test_hub/test_build_notification.py @@ -0,0 +1,104 @@ +import mock + +import koji +import unittest +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + + +class TestBuildNotification(unittest.TestCase): + def setUp(self): + self.context = mock.patch('kojihub.kojihub.context').start() + self.context.session.assertPerm = mock.MagicMock() + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + self._get_build_target = mock.patch('kojihub.kojihub._get_build_target').start() + self.get_notification_recipients = mock.patch( + 'kojihub.kojihub.get_notification_recipients').start() + self.make_task = mock.patch('kojihub.kojihub.make_task').start() + + def tearDown(self): + mock.patch.stopall() + + def test_disabled_notification(self): + self.context.opts = {'DisableNotifications': True} + rv = kojihub.build_notification(10, 1) + self.assertEqual(rv, None) + self.get_build.assert_not_called() + self._get_build_target.assert_not_called() + self.get_notification_recipients.assert_not_called() + self.make_task.assert_not_called() + + def test_not_complete_build(self): + build_id = 1 + task_id = 10 + self.context.opts = {'DisableNotifications': False} + buildinfo = {'state': 0, 'id': build_id} + target_info = {'dest_tag': 'test_dest_tag'} + self.get_build.return_value = buildinfo + self._get_build_target.return_value = target_info + + with self.assertRaises(koji.GenericError) as ex: + kojihub.build_notification(task_id, build_id) + self.assertEqual('never send notifications for incomplete builds', str(ex.exception)) + self.get_build.assert_called_once_with(build_id) + self._get_build_target.assert_called_once_with(task_id) + self.get_notification_recipients.assert_not_called() + self.make_task.assert_not_called() + + def test_valid(self): + build_id = 1 + task_id = 10 + self.context.opts = {'DisableNotifications': False} + buildinfo = {'state': 1, 'id': build_id} + target_info = {'dest_tag': 'test_dest_tag'} + recipients = ['email1@mail.com', 'email2@mail.com'] + weburl = 'http://localhost/koji' + self.get_build.return_value = buildinfo + self._get_build_target.return_value = target_info + self.get_notification_recipients.return_value = recipients + self.make_task.return_value = 11 + + rv = kojihub.build_notification(task_id, build_id) + self.assertEqual(rv, None) + self.get_build.assert_called_once_with(build_id) + self._get_build_target.assert_called_once_with(task_id) + self.get_notification_recipients.assert_called_once_with( + buildinfo, target_info['dest_tag'], buildinfo['state']) + self.make_task.assert_called_once_with('buildNotification', + [recipients, buildinfo, target_info, weburl]) + + +class TestGetBuildNotifications(DBQueryTestCase): + def setUp(self): + super(TestGetBuildNotifications, self).setUp() + + def tearDown(self): + mock.patch.stopall() + + def test_valid(self): + user_id = 1 + kojihub.get_build_notifications(user_id) + self.assertEqual(len(self.queries), 1) + self.assertLastQueryEqual(tables=['build_notifications'], + columns=['email', 'id', 'package_id', 'success_only', 'tag_id', + 'user_id'], + clauses=['user_id = %(user_id)i'], + values={'user_id': user_id}) + + +class TestGetBuildNotificationBlocks(DBQueryTestCase): + def setUp(self): + super(TestGetBuildNotificationBlocks, self).setUp() + + def tearDown(self): + mock.patch.stopall() + + def test_valid(self): + user_id = 1 + kojihub.get_build_notification_blocks(user_id) + self.assertEqual(len(self.queries), 1) + self.assertLastQueryEqual(tables=['build_notifications_block'], + columns=['id', 'package_id', 'tag_id', 'user_id'], + clauses=['user_id = %(user_id)i'], + values={'user_id': user_id}) diff --git a/tests/test_hub/test_cancel_build.py b/tests/test_hub/test_cancel_build.py new file mode 100644 index 0000000..87f108d --- /dev/null +++ b/tests/test_hub/test_cancel_build.py @@ -0,0 +1,132 @@ +# coding: utf-8 +import unittest + +import mock + +import koji +import kojihub + + +UP = kojihub.UpdateProcessor +DP = kojihub.DeleteProcessor + + +class TestCancelBuildRootExports(unittest.TestCase): + + def setUp(self): + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + self.cancel_build = mock.patch('kojihub.kojihub.cancel_build').start() + self.exports = kojihub.RootExports() + self.context = mock.patch('kojihub.kojihub.context').start() + # It seems MagicMock will not automatically handle attributes that + # start with "assert" + self.context.session.assertPerm = mock.MagicMock() + self.context.session.assertLogin = mock.MagicMock() + + def tearDown(self): + mock.patch.stopall() + + def test_cancel_build_no_build(self): + self.get_build.return_value = None + rv = self.exports.cancelBuild(1) + self.assertEqual(rv, False) + self.get_build.assert_called_once_with(1, False) + self.cancel_build.assert_not_called() + + def test_cancel_build_not_owner(self): + self.context.session.user_id = 24 + self.context.session.hasPerm.return_value = False + self.get_build.return_value = {'id': 1, 'owner_id': 23} + with self.assertRaises(koji.ActionNotAllowed) as cm: + self.exports.cancelBuild(1) + self.assertEqual('Cannot cancel build, not owner', str(cm.exception)) + self.get_build.assert_called_once_with(1, False) + self.cancel_build.assert_not_called() + + def test_cancel_build_valid(self): + self.context.session.user_id = 23 + self.cancel_build.return_value = True + self.get_build.return_value = {'id': 1, 'owner_id': 23} + rv = self.exports.cancelBuild(1) + self.assertEqual(rv, True) + self.get_build.assert_called_once_with(1, False) + self.cancel_build.assert_called_once_with(1) + + +class TestCancelBuildKojihub(unittest.TestCase): + + def setUp(self): + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + self.build_notification = mock.patch('kojihub.kojihub.build_notification').start() + self.run_callbacks = mock.patch('koji.plugin.run_callbacks').start() + self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor', + side_effect=self.getUpdate).start() + self.updates = [] + self.DeleteProcessor = mock.patch('kojihub.kojihub.DeleteProcessor', + side_effect=self.getDelete).start() + self.deletes = [] + + def tearDown(self): + mock.patch.stopall() + + def getUpdate(self, *args, **kwargs): + update = UP(*args, **kwargs) + update.execute = mock.MagicMock() + self.updates.append(update) + return update + + def getDelete(self, *args, **kwargs): + delete = DP(*args, **kwargs) + delete.execute = mock.MagicMock() + self.deletes.append(delete) + return delete + + def test_cancel_build_not_st_building(self): + self.get_build.return_value = {'id': 1, 'state': 2} + rv = kojihub.cancel_build(1) + self.assertEqual(rv, False) + self.run_callbacks.assert_not_called() + self.assertEqual(len(self.updates), 0) + self.get_build.assert_called_once_with(1, strict=True) + self.build_notification.assert_not_called() + + def test_cancel_build_not_st_canceled(self): + self.get_build.side_effect = [{'id': 1, 'state': 0}, {'id': 1, 'state': 2}] + rv = kojihub.cancel_build(1) + self.assertEqual(rv, False) + self.assertEqual(len(self.updates), 1) + self.assertEqual(len(self.deletes), 0) + update = self.updates[0] + self.assertEqual(update.table, 'build') + self.assertEqual(update.values, {'build_id': 1, 'st_building': 0}) + self.assertEqual(update.data, {'state': 4}) + self.assertEqual(update.rawdata, {'completion_time': 'NOW()'}) + self.assertEqual(update.clauses, ['id = %(build_id)i', 'state = %(st_building)i']) + self.get_build.assert_has_calls([mock.call(1, strict=True), mock.call(1)]) + self.build_notification.assert_not_called() + self.assertEqual(self.run_callbacks.call_count, 1) + + def test_cancel_build_valid(self): + self.get_build.side_effect = [{'id': 1, 'state': 0}, {'id': 1, 'state': 4, 'task_id': 1}, + {'id': 1, 'state': 4, 'task_id': 1}] + rv = kojihub.cancel_build(1, cancel_task=False) + self.assertEqual(rv, True) + + self.assertEqual(len(self.updates), 1) + self.assertEqual(len(self.deletes), 1) + update = self.updates[0] + self.assertEqual(update.table, 'build') + self.assertEqual(update.values, {'build_id': 1, 'st_building': 0}) + self.assertEqual(update.data, {'state': 4}) + self.assertEqual(update.rawdata, {'completion_time': 'NOW()'}) + self.assertEqual(update.clauses, ['id = %(build_id)i', 'state = %(st_building)i']) + + delete = self.deletes[0] + self.assertEqual(delete.table, 'build_reservations') + self.assertEqual(delete.clauses, ['build_id = %(build_id)i']) + self.assertEqual(delete.values, {'build_id': 1}) + + self.get_build.assert_has_calls([mock.call(1, strict=True), mock.call(1), + mock.call(1, strict=True)]) + self.build_notification.assert_called_once_with(1, 1) + self.assertEqual(self.run_callbacks.call_count, 2) diff --git a/tests/test_hub/test_grant_cg_access.py b/tests/test_hub/test_grant_cg_access.py new file mode 100644 index 0000000..4cb9e1a --- /dev/null +++ b/tests/test_hub/test_grant_cg_access.py @@ -0,0 +1,71 @@ +# coding: utf-8 +import unittest +import mock + +import koji +import kojihub + +IP = kojihub.InsertProcessor + + +class TestGrantCGAccess(unittest.TestCase): + + def getInsert(self, *args, **kwargs): + insert = IP(*args, **kwargs) + insert.execute = mock.MagicMock() + insert.make_create = mock.MagicMock() + insert.dup_check = self.ins_dup_check + self.inserts.append(insert) + return insert + + def setUp(self): + self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor', + side_effect=self.getInsert).start() + self.inserts = [] + self.ins_dup_check = mock.MagicMock() + self.get_user = mock.patch('kojihub.kojihub.get_user').start() + self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start() + self.context = mock.patch('kojihub.kojihub.context').start() + # It seems MagicMock will not automatically handle attributes that + # start with "assert" + self.context.session.assertPerm = mock.MagicMock() + + def tearDown(self): + mock.patch.stopall() + + def test_without_create(self): + self.ins_dup_check.return_value = False + cg = {'id': 11, 'name': 'test-cg-name'} + user = {'id': 123, 'name': 'testuser'} + self.get_user.return_value = user + self.lookup_name.return_value = cg + kojihub.grant_cg_access(user['name'], cg['name']) + + # check the insert + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'cg_users') + self.assertEqual(insert.data, {'cg_id': cg['id'], 'user_id': user['id']}) + self.assertEqual(insert.rawdata, {}) + self.get_user.assert_called_once_with(user['name'], strict=True) + self.lookup_name.assert_called_once_with('content_generator', cg['name'], strict=True) + + def test_with_create(self): + self.ins_dup_check.return_value = True + cg = {'id': 11, 'name': 'test-cg-name'} + user = {'id': 123, 'name': 'testuser'} + self.get_user.return_value = user + self.lookup_name.return_value = cg + with self.assertRaises(koji.GenericError) as ex: + kojihub.grant_cg_access(user['name'], cg['name'], create=True) + self.assertEqual(f"User already has access to content generator {cg['name']}", + str(ex.exception)) + + # check the insert + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'cg_users') + self.assertEqual(insert.data, {'cg_id': cg['id'], 'user_id': user['id']}) + self.assertEqual(insert.rawdata, {}) + self.get_user.assert_called_once_with(user['name'], strict=True) + self.lookup_name.assert_called_once_with('content_generator', cg['name'], create=True) diff --git a/tests/test_hub/test_list_cgs.py b/tests/test_hub/test_list_cgs.py new file mode 100644 index 0000000..b90ac79 --- /dev/null +++ b/tests/test_hub/test_list_cgs.py @@ -0,0 +1,31 @@ +import mock + +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + + +class TestListCGs(DBQueryTestCase): + def setUp(self): + super(TestListCGs, self).setUp() + + def tearDown(self): + mock.patch.stopall() + + def test_valid(self): + self.qp_iterate_return_value = [{'id': 1, 'name': 'cg_name_1', 'user_name': 'test-user'}, + {'id': 2, 'name': 'cg_name_2', 'user_name': 'test-user'}] + expected_result = {'cg_name_1': {'id': 1, 'users': ['test-user']}, + 'cg_name_2': {'id': 2, 'users': ['test-user']}} + rv = kojihub.list_cgs() + self.assertEqual(rv, expected_result) + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['cg_users']) + self.assertEqual(query.columns, ['content_generator.id', 'content_generator.name', + 'users.name']) + self.assertEqual(query.clauses, ['cg_users.active = TRUE']) + self.assertEqual(query.joins, ['content_generator ON content_generator.id = ' + 'cg_users.cg_id', 'users ON users.id = cg_users.user_id']) + self.assertEqual(query.values, {}) diff --git a/tests/test_hub/test_new_image_build.py b/tests/test_hub/test_new_image_build.py new file mode 100644 index 0000000..f28afce --- /dev/null +++ b/tests/test_hub/test_new_image_build.py @@ -0,0 +1,65 @@ +import mock + +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + +IP = kojihub.InsertProcessor + + +class TestNewImageBuild(DBQueryTestCase): + def setUp(self): + super(TestNewImageBuild, self).setUp() + self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor', + side_effect=self.getInsert).start() + self.inserts = [] + self.insert_execute = mock.MagicMock() + self.new_typed_build = mock.patch('kojihub.kojihub.new_typed_build').start() + self.build_info = { + 'id': 100, + 'name': 'test_name', + 'version': 'test_version', + 'release': 'test_release', + 'epoch': 'test_epoch', + 'owner': 'test_owner', + 'extra': {'extra_key': 'extra_value'}, + } + + def tearDown(self): + mock.patch.stopall() + + def getInsert(self, *args, **kwargs): + insert = IP(*args, **kwargs) + insert.execute = self.insert_execute + self.inserts.append(insert) + return insert + + def test_valid_new_image(self): + self.qp_execute_one_return_value = None + kojihub.new_image_build(self.build_info) + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['image_builds']) + self.assertEqual(query.columns, ['build_id']) + self.assertEqual(query.clauses, ['build_id = %(build_id)i']) + self.assertEqual(query.joins, None) + self.assertEqual(query.values, {'build_id': self.build_info['id']}) + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'image_builds') + self.assertEqual(insert.data, {'build_id': self.build_info['id']}) + self.assertEqual(insert.rawdata, {}) + self.new_typed_build.assert_called_once_with(self.build_info, 'image') + + def test_valid_existing_image(self): + self.qp_execute_one_return_value = {'build_id': 123} + kojihub.new_image_build(self.build_info) + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['image_builds']) + self.assertEqual(query.columns, ['build_id']) + self.assertEqual(query.clauses, ['build_id = %(build_id)i']) + self.assertEqual(query.joins, None) + self.assertEqual(query.values, {'build_id': self.build_info['id']}) + self.assertEqual(len(self.inserts), 0) + self.new_typed_build.assert_not_called() diff --git a/tests/test_hub/test_new_maven_build.py b/tests/test_hub/test_new_maven_build.py index 76c7cd7..d84e4fe 100644 --- a/tests/test_hub/test_new_maven_build.py +++ b/tests/test_hub/test_new_maven_build.py @@ -15,7 +15,7 @@ class TestNewMavenBuild(unittest.TestCase): self.inserts = [] self.insert_execute = mock.MagicMock() self.get_maven_build = mock.patch('kojihub.kojihub.get_maven_build').start() - self.get_maven_build.return_value = None + self.new_typed_build = mock.patch('kojihub.kojihub.new_typed_build').start() self.build_info = { 'id': 100, 'name': 'test_name', @@ -33,13 +33,17 @@ class TestNewMavenBuild(unittest.TestCase): return insert def test_empty_maven_info(self): + self.get_maven_build.return_value = None maven_info = {} with self.assertRaises(koji.GenericError) as cm: kojihub.new_maven_build(self.build_info, maven_info) self.assertEqual("Maven info is empty", str(cm.exception)) self.assertEqual(len(self.inserts), 0) + self.get_maven_build.assert_called_once_with(self.build_info) + self.new_typed_build.assert_not_called() def test_maven_info_without_some_key(self): + self.get_maven_build.return_value = None # maven_info without group_id maven_info = { 'artifact_id': '99', @@ -76,3 +80,41 @@ class TestNewMavenBuild(unittest.TestCase): kojihub.new_maven_build(self.build_info, maven_info) self.assertEqual('Invalid type for maven_info: %s' % type(maven_info), str(cm.exception)) self.assertEqual(len(self.inserts), 0) + self.get_maven_build.assert_not_called() + self.new_typed_build.assert_not_called() + + def test_valid_without_current_maven_info(self): + self.get_maven_build.return_value = None + maven_info = { + 'artifact_id': '99', + 'version': '33', + 'build_id': 100, + 'group_id': 5, + } + kojihub.new_maven_build(self.build_info, maven_info) + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'maven_builds') + self.assertEqual(insert.data, maven_info) + self.assertEqual(insert.rawdata, {}) + self.get_maven_build.assert_called_once_with(self.build_info) + self.new_typed_build.assert_called_once_with(self.build_info, 'maven') + + def test_mismatch_maven(self): + maven_info_api = { + 'artifact_id': '99', + 'version': '33', + 'group_id': 5, + } + maven_info = { + 'artifact_id': '99', + 'version': '34', + 'group_id': 5, + } + self.get_maven_build.return_value = maven_info_api + with self.assertRaises(koji.GenericError) as cm: + kojihub.new_maven_build(self.build_info, maven_info) + self.assertEqual('version mismatch (current: 33, new: 34)', str(cm.exception)) + self.assertEqual(len(self.inserts), 0) + self.get_maven_build.assert_called_once_with(self.build_info) + self.new_typed_build.assert_not_called() diff --git a/tests/test_hub/test_new_package.py b/tests/test_hub/test_new_package.py new file mode 100644 index 0000000..1d7e9a7 --- /dev/null +++ b/tests/test_hub/test_new_package.py @@ -0,0 +1,70 @@ +import mock + +import koji +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + +IP = kojihub.InsertProcessor + + +class TestNewPackage(DBQueryTestCase): + def setUp(self): + super(TestNewPackage, self).setUp() + self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor', + side_effect=self.getInsert).start() + self.inserts = [] + self.insert_execute = mock.MagicMock() + self.nextval = mock.patch('kojihub.kojihub.nextval').start() + self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start() + self.context = mock.patch('kojihub.kojihub.context').start() + self.pkg_name = 'test-pkg' + + def tearDown(self): + mock.patch.stopall() + + def getInsert(self, *args, **kwargs): + insert = IP(*args, **kwargs) + insert.execute = self.insert_execute + self.inserts.append(insert) + return insert + + def test_pkg_already_exist(self): + self.verify_name_internal.return_value = None + self.qp_single_value_return_value = 2 + with self.assertRaises(koji.GenericError) as cm: + kojihub.new_package(self.pkg_name, strict=True) + self.assertEqual("Package already exists [id 2]", str(cm.exception)) + self.verify_name_internal.assert_called_once_with(self.pkg_name) + self.nextval.assert_not_called() + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['package']) + self.assertEqual(query.columns, ['id']) + self.assertEqual(query.clauses, ['name=%(name)s']) + self.assertEqual(query.joins, None) + self.assertEqual(query.values, {'name': self.pkg_name}) + self.assertEqual(len(self.inserts), 0) + + def test_pkg_new(self): + self.verify_name_internal.return_value = None + self.qp_single_value_return_value = None + self.nextval.return_value = 3 + rv = kojihub.new_package(self.pkg_name, strict=True) + self.assertEqual(rv, 3) + self.verify_name_internal.assert_called_once_with(self.pkg_name) + self.nextval.assert_called_once_with('package_id_seq') + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['package']) + self.assertEqual(query.columns, ['id']) + self.assertEqual(query.clauses, ['name=%(name)s']) + self.assertEqual(query.joins, None) + self.assertEqual(query.values, {'name': self.pkg_name}) + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'package') + self.assertEqual(insert.data, {'id': 3, 'name': self.pkg_name}) + self.assertEqual(insert.rawdata, {}) diff --git a/tests/test_hub/test_new_win_build.py b/tests/test_hub/test_new_win_build.py index 7c61a6c..f62f9a6 100644 --- a/tests/test_hub/test_new_win_build.py +++ b/tests/test_hub/test_new_win_build.py @@ -6,6 +6,7 @@ import koji import kojihub IP = kojihub.InsertProcessor +UP = kojihub.UpdateProcessor class TestNewWinBuild(unittest.TestCase): @@ -14,6 +15,11 @@ class TestNewWinBuild(unittest.TestCase): side_effect=self.getInsert).start() self.inserts = [] self.insert_execute = mock.MagicMock() + self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor', + side_effect=self.getUpdate).start() + self.updates = [] + self.get_win_build = mock.patch('kojihub.kojihub.get_win_build').start() + self.new_typed_build = mock.patch('kojihub.kojihub.new_typed_build').start() self.build_info = { 'id': 100, @@ -31,12 +37,21 @@ class TestNewWinBuild(unittest.TestCase): self.inserts.append(insert) return insert + def getUpdate(self, *args, **kwargs): + update = UP(*args, **kwargs) + update.execute = mock.MagicMock() + self.updates.append(update) + return update + def test_empty_win_info(self): win_info = {} with self.assertRaises(koji.GenericError) as cm: kojihub.new_win_build(self.build_info, win_info) self.assertEqual("Windows info is empty", str(cm.exception)) self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.updates), 0) + self.get_win_build.assert_not_called() + self.new_typed_build.assert_not_called() def test_win_info_without_platform(self): win_info = { @@ -46,6 +61,9 @@ class TestNewWinBuild(unittest.TestCase): kojihub.new_win_build(self.build_info, win_info) self.assertEqual("Windows info doesn't have mandatory platform key", str(cm.exception)) self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.updates), 0) + self.get_win_build.assert_not_called() + self.new_typed_build.assert_not_called() def test_wrong_format_win_info(self): win_info = 'test-win-info' @@ -53,3 +71,39 @@ class TestNewWinBuild(unittest.TestCase): kojihub.new_win_build(self.build_info, win_info) self.assertEqual('Invalid type for win_info: %s' % type(win_info), str(cm.exception)) self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.updates), 0) + self.get_win_build.assert_not_called() + self.new_typed_build.assert_not_called() + + def test_valid_without_current_win_info(self): + win_info = { + 'platform': 'test-platform' + } + self.get_win_build.return_value = None + kojihub.new_win_build(self.build_info, win_info) + self.assertEqual(len(self.updates), 0) + self.assertEqual(len(self.inserts), 1) + insert = self.inserts[0] + self.assertEqual(insert.table, 'win_builds') + self.assertEqual(insert.data, + {'build_id': self.build_info['id'], 'platform': 'test-platform'}) + self.assertEqual(insert.rawdata, {}) + self.get_win_build.assert_called_once_with(self.build_info['id'], strict=False) + self.new_typed_build.assert_called_once_with(self.build_info, 'win') + + def test_valid_with_current_win_info(self): + win_info = { + 'platform': 'test-platform' + } + self.get_win_build.return_value = {'platform': 'test-platform-old'} + kojihub.new_win_build(self.build_info, win_info) + self.assertEqual(len(self.inserts), 0) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'win_builds') + self.assertEqual(update.values, {'build_id': self.build_info['id']}) + self.assertEqual(update.data, {'platform': 'test-platform'}) + self.assertEqual(update.rawdata, {}) + self.assertEqual(update.clauses, ['build_id=%(build_id)i']) + self.get_win_build.assert_called_once_with(self.build_info['id'], strict=False) + self.new_typed_build.assert_not_called() diff --git a/tests/test_hub/test_policy_tests.py b/tests/test_hub/test_policy_tests.py index 1a02cb9..1fab261 100644 --- a/tests/test_hub/test_policy_tests.py +++ b/tests/test_hub/test_policy_tests.py @@ -260,6 +260,12 @@ class TestHasTagTest(unittest.TestCase): def tearDown(self): mock.patch.stopall() + def test_build_not_in_data(self): + data = {'key': 'test'} + obj = kojihub.HasTagTest('hastag *-candidate') + self.assertFalse(obj.run(data)) + self.list_tags.assert_not_called() + def test_has_tag_simple(self): obj = kojihub.HasTagTest('hastag *-candidate') tags = ['foo-1.0', 'foo-2.0', 'foo-3.0-candidate'] @@ -427,3 +433,331 @@ class TestImportedTest(unittest.TestCase): self.get_build.assert_called_once_with('nvr-1-1', strict=True) self.list_rpms.assert_called_once_with(buildID=1) self.list_archives.assert_called_once_with(buildID=1) + + +class TestPolicyGetVersion(unittest.TestCase): + def setUp(self): + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + + def tearDown(self): + mock.patch.stopall() + + def test_policy_version(self): + data = {'version': 123} + rv = kojihub.policy_get_version(data) + self.assertEqual(rv, 123) + self.get_build.assert_not_called() + + def test_policy_build(self): + data = {'build': 123} + buildinfo = {'id': 123, 'version': 100} + self.get_build.return_value = buildinfo + rv = kojihub.policy_get_version(data) + self.assertEqual(rv, 100) + self.get_build.assert_called_once_with(data['build'], strict=True) + + def test_policy_version_error(self): + data = {'release': 123} + with self.assertRaises(koji.GenericError) as cm: + kojihub.policy_get_version(data) + self.assertEqual("policy requires version data", str(cm.exception)) + self.get_build.assert_not_called() + + +class TestPolicyGetRelease(unittest.TestCase): + def setUp(self): + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + + def tearDown(self): + mock.patch.stopall() + + def test_policy_release(self): + data = {'release': 123} + rv = kojihub.policy_get_release(data) + self.assertEqual(rv, 123) + self.get_build.assert_not_called() + + def test_policy_build(self): + data = {'build': 123} + buildinfo = {'id': 123, 'release': 100} + self.get_build.return_value = buildinfo + rv = kojihub.policy_get_release(data) + self.assertEqual(rv, 100) + self.get_build.assert_called_once_with(data['build'], strict=True) + + def test_policy_release_error(self): + data = {'version': 123} + with self.assertRaises(koji.GenericError) as cm: + kojihub.policy_get_release(data) + self.assertEqual("policy requires release data", str(cm.exception)) + self.get_build.assert_not_called() + + +class TestPolicyGetPkg(unittest.TestCase): + def setUp(self): + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + self.lookup_package = mock.patch('kojihub.kojihub.lookup_package').start() + + def tearDown(self): + mock.patch.stopall() + + def test_no_such_pkg(self): + data = {'package': 123} + self.lookup_package.return_value = None + with self.assertRaises(koji.GenericError) as cm: + kojihub.policy_get_pkg(data) + self.assertEqual(f"No such package: {data['package']}", str(cm.exception)) + self.lookup_package.assert_called_once_with(data['package'], strict=False) + self.get_build.assert_not_called() + + def test_with_pkg_not_found_lookup_pkg(self): + data = {'package': 'test-pkg'} + self.lookup_package.return_value = None + rv = kojihub.policy_get_pkg(data) + self.assertEqual(rv, {'id': None, 'name': data['package']}) + self.lookup_package.assert_called_once_with(data['package'], strict=False) + self.get_build.assert_not_called() + + def test_with_pkg(self): + data = {'package': 'test-pkg'} + self.lookup_package.return_value = {'id': 1, 'name': data['package']} + rv = kojihub.policy_get_pkg(data) + self.assertEqual(rv, {'id': 1, 'name': data['package']}) + self.lookup_package.assert_called_once_with(data['package'], strict=False) + self.get_build.assert_not_called() + + def test_build(self): + data = {'build': 123} + buildinfo = {'id': 123, 'package_id': 1, 'name': 'test-pkg'} + self.get_build.return_value = buildinfo + rv = kojihub.policy_get_pkg(data) + self.assertEqual(rv, {'id': buildinfo['package_id'], 'name': buildinfo['name']}) + self.lookup_package.assert_not_called() + self.get_build.assert_called_once_with(data['build'], strict=True) + + def test_wrong_data_key(self): + data = {'id': 123} + with self.assertRaises(koji.GenericError) as cm: + kojihub.policy_get_pkg(data) + self.assertEqual("policy requires package data", str(cm.exception)) + self.lookup_package.assert_not_called() + self.get_build.assert_not_called() + + +class TestVolumeTest(unittest.TestCase): + def setUp(self): + self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start() + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + + def tearDown(self): + mock.patch.stopall() + + def test_volume_data(self): + self.lookup_name.return_value = {'id': 159, 'name': 'test-volume'} + obj = kojihub.VolumeTest('volume - volume data') + data = {'volume': 'test-volume'} + obj.run(data) + self.get_build.assert_not_called() + self.lookup_name.assert_called_once_with('volume', data['volume'], strict=False) + + def test_build_data(self): + buildinfo = {'volume_id': 5, 'volume_name': 'test-volume', 'id': 2} + self.get_build.return_value = buildinfo + obj = kojihub.VolumeTest('volume - build data') + data = {'build': 'nvr-1-1'} + obj.run(data) + self.get_build.assert_called_once_with(data['build']) + self.lookup_name.assert_not_called() + + def test_not_volinfo(self): + data = {'key': 'test'} + obj = kojihub.VolumeTest('volume - volume none') + self.assertFalse(obj.run(data)) + self.get_build.assert_not_called() + self.lookup_name.assert_not_called() + + +class TestTagTest(unittest.TestCase): + def setUp(self): + self.get_tag = mock.patch('kojihub.kojihub.get_tag').start() + + def tearDown(self): + mock.patch.stopall() + + def test_tag_is_none(self): + data = {'key': 'test'} + obj = kojihub.TagTest('tagtest - tag none') + self.assertFalse(obj.run(data)) + self.get_tag.assert_not_called() + + def test_taginfo_is_none(self): + self.get_tag.return_value = None + data = {'tag': 'test-tag'} + obj = kojihub.TagTest('tagtest - taginfo none') + self.assertFalse(obj.run(data)) + self.get_tag.assert_called_once_with(data['tag'], strict=False) + + def test_taginfo_valid(self): + data = {'tag': 'test-tag'} + self.get_tag.return_value = {'name': data['tag']} + obj = kojihub.TagTest('tagtest - taginfo none') + self.assertIsNotNone(obj.run(data)) + self.get_tag.assert_called_once_with(data['tag'], strict=False) + + +class FromTagTest(unittest.TestCase): + def setUp(self): + self.get_tag = mock.patch('kojihub.kojihub.get_tag').start() + + def tearDown(self): + mock.patch.stopall() + + def test_tag_is_none(self): + data = {'key': 'test'} + obj = kojihub.FromTagTest('fromtagtest - tag none') + self.assertIsNone(obj.get_tag(data)) + self.get_tag.assert_not_called() + + def test_valid(self): + data = {'fromtag': 'test-tag'} + self.get_tag.return_value = {'name': data['fromtag']} + obj = kojihub.FromTagTest('fromtagtest - tag none') + self.assertEqual(obj.get_tag(data), {'name': data['fromtag']}) + self.get_tag.assert_called_once_with(data['fromtag'], strict=False) + + +class CGMatchAnyTest(unittest.TestCase): + def setUp(self): + self.policy_get_cgs = mock.patch('kojihub.kojihub.policy_get_cgs').start() + self.multi_fnmatch = mock.patch('kojihub.kojihub.multi_fnmatch').start() + + def tearDown(self): + mock.patch.stopall() + + def test_cgname_is_none(self): + data = {'key': 'test'} + self.policy_get_cgs.return_value = [None] + obj = kojihub.CGMatchAnyTest('cgmatchanytest - cg name none') + self.assertFalse(obj.run(data)) + + def test_cg_name_true(self): + data = {'key': 'test'} + self.policy_get_cgs.return_value = ['cgname'] + self.multi_fnmatch.return_value = True + obj = kojihub.CGMatchAnyTest('cgmatchanytest - cg name true') + self.assertTrue(obj.run(data)) + + +class CGMatchAllTest(unittest.TestCase): + def setUp(self): + self.policy_get_cgs = mock.patch('kojihub.kojihub.policy_get_cgs').start() + self.multi_fnmatch = mock.patch('kojihub.kojihub.multi_fnmatch').start() + + def tearDown(self): + mock.patch.stopall() + + def test_cgs_none(self): + data = {'key': 'test'} + self.policy_get_cgs.return_value = [] + obj = kojihub.CGMatchAllTest('cgmatchalltest - cg name none') + self.assertFalse(obj.run(data)) + + def test_cgname_is_none(self): + data = {'key': 'cgname'} + self.policy_get_cgs.return_value = [None] + obj = kojihub.CGMatchAllTest('cgmatchalltest - cg name none') + self.assertFalse(obj.run(data)) + + def test_cg_name_all_true(self): + data = {'key': 'cgname'} + self.policy_get_cgs.return_value = ['cgname'] + self.multi_fnmatch.return_value = True + obj = kojihub.CGMatchAllTest('cgmatchalltest - cg name true') + self.assertTrue(obj.run(data)) + + def test_cg_name_not_found(self): + data = {'key': 'cgname'} + self.policy_get_cgs.return_value = ['cgname'] + self.multi_fnmatch.return_value = False + obj = kojihub.CGMatchAllTest('cgmatchalltest - cg name true') + self.assertFalse(obj.run(data)) + + +class UserTest(unittest.TestCase): + def setUp(self): + self.policy_get_user = mock.patch('kojihub.kojihub.policy_get_user').start() + + def tearDown(self): + mock.patch.stopall() + + def test_not_user(self): + data = {'user': 'username'} + self.policy_get_user.return_value = None + obj = kojihub.UserTest('usertest - not user') + self.assertFalse(obj.run(data)) + + def test_valid(self): + data = {'user': 'username'} + self.policy_get_user.return_value = {'name': 'username'} + obj = kojihub.UserTest('usertest - valid') + self.assertIsNotNone(obj.run(data)) + + +class IsBuildOwnerTest(unittest.TestCase): + def setUp(self): + self.policy_get_user = mock.patch('kojihub.kojihub.policy_get_user').start() + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + self.get_user = mock.patch('kojihub.kojihub.get_user').start() + self.get_user_groups = mock.patch('kojihub.kojihub.get_user_groups').start() + + def tearDown(self): + mock.patch.stopall() + + def test_not_user(self): + data = {'build': 1} + self.get_build.return_value = {'build_id': data['build'], 'owner_id': 3} + self.get_user.return_value = {'id': 3, 'name': 'username'} + self.policy_get_user.return_value = None + obj = kojihub.IsBuildOwnerTest('isbuildownertest - not user') + self.assertFalse(obj.run(data)) + self.get_build.assert_called_once_with(data['build']) + self.get_user.assert_called_once_with(3) + self.policy_get_user.assert_called_once_with(data) + self.get_user_groups.assert_not_called() + + def test_owner_is_user(self): + data = {'build': 1} + self.get_build.return_value = {'build_id': data['build'], 'owner_id': 3} + self.get_user.return_value = {'id': 3, 'name': 'username'} + self.policy_get_user.return_value = {'id': 3, 'name': 'username'} + obj = kojihub.IsBuildOwnerTest('isbuildownertest - owner is user') + self.assertTrue(obj.run(data)) + self.get_build.assert_called_once_with(data['build']) + self.get_user.assert_called_once_with(3) + self.policy_get_user.assert_called_once_with(data) + self.get_user_groups.assert_not_called() + + def test_owner_is_group(self): + data = {'build': 1} + self.get_build.return_value = {'build_id': data['build'], 'owner_id': 3} + self.get_user.return_value = {'id': 2, 'name': 'testuser', 'usertype': 2} + self.policy_get_user.return_value = {'id': 3, 'name': 'username'} + self.get_user_groups.return_value = [2] + obj = kojihub.IsBuildOwnerTest('isbuildownertest - owner group') + self.assertTrue(obj.run(data)) + self.get_build.assert_called_once_with(data['build']) + self.get_user.assert_called_once_with(3) + self.policy_get_user.assert_called_once_with(data) + self.get_user_groups.assert_called_once_with(3) + + def test_owner_false(self): + data = {'build': 1} + self.get_build.return_value = {'build_id': data['build'], 'owner_id': 3} + self.get_user.return_value = {'id': 2, 'name': 'testuser', 'usertype': 1} + self.policy_get_user.return_value = {'id': 3, 'name': 'username'} + obj = kojihub.IsBuildOwnerTest('isbuildownertest - owner false') + self.assertFalse(obj.run(data)) + self.get_build.assert_called_once_with(data['build']) + self.get_user.assert_called_once_with(3) + self.policy_get_user.assert_called_once_with(data) + self.get_user_groups.assert_not_called() diff --git a/tests/test_hub/test_query_history.py b/tests/test_hub/test_query_history.py new file mode 100644 index 0000000..5d15602 --- /dev/null +++ b/tests/test_hub/test_query_history.py @@ -0,0 +1,504 @@ +import mock +import os +import time + +import koji +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + + +class TestQueryHistory(DBQueryTestCase): + def setUp(self): + super(TestQueryHistory, self).setUp() + self.maxDiff = None + self.context = mock.patch('kojihub.kojihub.context').start() + self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start() + self.get_build = mock.patch('kojihub.kojihub.get_build').start() + self.get_id = mock.patch('kojihub.kojihub.get_id').start() + self.get_package_id = mock.patch('kojihub.kojihub.get_package_id').start() + self.get_user = mock.patch('kojihub.kojihub.get_user').start() + self.get_perm_id = mock.patch('kojihub.kojihub.get_perm_id').start() + self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start() + self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start() + self.get_build_target_id = mock.patch('kojihub.kojihub.get_build_target_id').start() + self.get_group_id = mock.patch('kojihub.kojihub.get_group_id').start() + self.original_timezone = os.environ.get('TZ') + os.environ['TZ'] = 'UTC' + time.tzset() + + def tearDown(self): + if self.original_timezone is None: + del os.environ['TZ'] + else: + os.environ['TZ'] = self.original_timezone + time.tzset() + mock.patch.stopall() + + def test_not_exist_table(self): + table_name = 'test-table' + with self.assertRaises(koji.GenericError) as cm: + kojihub.query_history(tables=[table_name]) + self.assertEqual(f"No such history table: {table_name}", str(cm.exception)) + self.assertEqual(len(self.queries), 0) + + def test_valid(self): + self.get_tag_id.return_value = 12 + self.get_build.return_value = {'id': 13} + self.get_package_id.return_value = 15 + self.get_user.return_value = {'id': 357, 'name': 'test-editor'} + kojihub.query_history(tables=['user_perms', 'user_groups', 'cg_users', 'tag_inheritance', + 'tag_config', 'tag_extra', 'build_target_config', + 'external_repo_config', 'host_config', 'host_channels', + 'tag_external_repos', 'tag_listing', 'tag_packages', + 'tag_package_owners', 'group_config', 'group_req_listing', + 'group_package_listing'], + tag='test-tag', build='test-build-1.23-1', package='test-pkg', + active=True, editor='test-editor', afterEvent=555) + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['tag_listing']) + self.assertEqual(query.columns, ['tag_listing.create_event > %(afterEvent)i', + 'creator.id = %(editor)i', + 'tag_listing.revoke_event > %(afterEvent)i', + 'revoker.id = %(editor)i', 'tag_listing.active', + 'build.state', 'tag_listing.build_id', + 'tag_listing.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'tag_listing.creator_id', 'creator.name', 'build.epoch', + 'package.name', 'build.release', + 'tag_listing.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'tag_listing.revoker_id', 'revoker.name', 'tag.name', + 'tag_listing.tag_id', 'build.version']) + self.assertEqual(query.clauses, ['active = TRUE', 'build.id = %(build_id)i', + 'creator.id = %(editor)i OR revoker.id = %(editor)i', + 'package.id = %(pkg_id)i', 'tag.id = %(tag_id)i', + 'tag_listing.create_event > %(afterEvent)i OR ' + 'tag_listing.revoke_event > %(afterEvent)i']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'build ON build_id = build.id', + 'package ON build.pkg_id = package.id', + 'LEFT OUTER JOIN tag ON tag_id = tag.id']) + self.assertEqual(query.values, {'afterEvent': 555, 'build_id': 13, 'editor': 357, + 'pkg_id': 15, 'tag_id': 12}) + + def test_group_key(self): + self.get_group_id.return_value = 24 + kojihub.query_history(group='test-grp') + self.assertEqual(len(self.queries), 3) + query = self.queries[0] + self.assertEqual(query.tables, ['group_config']) + self.assertEqual(query.columns, ['group_config.active', 'group_config.biarchonly', + 'group_config.blocked', 'group_config.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'group_config.creator_id', 'creator.name', + 'group_config.description', 'group_config.display_name', + 'group_config.exported', 'groups.name', + 'group_config.group_id', 'group_config.is_default', + 'group_config.langonly', 'group_config.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'group_config.revoker_id', 'revoker.name', 'tag.name', + 'group_config.tag_id', 'group_config.uservisible']) + self.assertEqual(query.clauses, ['groups.id = %(group_id)i']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'groups ON group_id = groups.id', + 'LEFT OUTER JOIN tag ON tag_id = tag.id']) + self.assertEqual(query.values, {'group_id': 24}) + query = self.queries[1] + self.assertEqual(query.tables, ['group_package_listing']) + self.assertEqual(query.columns, ['group_package_listing.active', + 'group_package_listing.basearchonly', + 'group_package_listing.blocked', + 'group_package_listing.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'group_package_listing.creator_id', 'creator.name', + 'groups.name', 'group_package_listing.group_id', + 'group_package_listing.package', + 'group_package_listing.requires', + 'group_package_listing.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'group_package_listing.revoker_id', + 'revoker.name', 'tag.name', + 'group_package_listing.tag_id', + 'group_package_listing.type']) + self.assertEqual(query.clauses, ['groups.id = %(group_id)i']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'groups ON group_id = groups.id', + 'LEFT OUTER JOIN tag ON tag_id = tag.id']) + self.assertEqual(query.values, {'group_id': 24}) + + query = self.queries[2] + self.assertEqual(query.tables, ['group_req_listing']) + self.assertEqual(query.columns, ['group_req_listing.active', 'group_req_listing.blocked', + 'group_req_listing.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'group_req_listing.creator_id', 'creator.name', + 'groups.name', 'group_req_listing.group_id', + 'group_req_listing.is_metapkg', 'req.name', + 'group_req_listing.req_id', + 'group_req_listing.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'group_req_listing.revoker_id', 'revoker.name', + 'tag.name', 'group_req_listing.tag_id', + 'group_req_listing.type']) + self.assertEqual(query.clauses, ['req.id = %(group_id)i']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'groups ON group_id = groups.id', + 'LEFT OUTER JOIN tag ON tag_id = tag.id', + 'LEFT OUTER JOIN groups AS req ON req_id = req.id']) + self.assertEqual(query.values, {'group_id': 24}) + + def test_host_key_and_active_false(self): + self.get_id.return_value = 22 + kojihub.query_history(host='test-host', active=False) + self.assertEqual(len(self.queries), 2) + query = self.queries[0] + self.assertEqual(query.tables, ['host_channels']) + self.assertEqual(query.clauses, ['active IS NULL', 'host.id = %(host_id)i']) + self.assertEqual(query.columns, ['host_channels.active', 'host_channels.channel_id', + 'channels.name', 'host_channels.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'host_channels.creator_id', 'creator.name', 'host.name', + 'host_channels.host_id', 'host_channels.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'host_channels.revoker_id', 'revoker.name']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN host ON host_id = host.id', + 'LEFT OUTER JOIN channels ON channel_id = channels.id']) + self.assertEqual(query.values, {'host_id': 22}) + query = self.queries[1] + self.assertEqual(query.tables, ['host_config']) + self.assertEqual(query.columns, ['host_config.active', 'host_config.arches', + 'host_config.capacity', 'host_config.comment', + 'host_config.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'host_config.creator_id', 'creator.name', + 'host_config.description', 'host_config.enabled', + 'host.name', 'host_config.host_id', + 'host_config.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'host_config.revoker_id', 'revoker.name']) + self.assertEqual(query.clauses, ['active IS NULL', 'host.id = %(host_id)i']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN host ON host_id = host.id']) + self.assertEqual(query.values, {'host_id': 22}) + + def test_channel_and_before_event_key(self): + self.get_id.return_value = 22 + kojihub.query_history(channel='test-channel', beforeEvent=99) + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['host_channels']) + self.assertEqual(query.clauses, ['channels.id = %(channel_id)i', + 'host_channels.create_event < %(beforeEvent)i OR ' + 'host_channels.revoke_event < %(beforeEvent)i']) + self.assertEqual(query.columns, ['host_channels.create_event < %(beforeEvent)i', + 'host_channels.revoke_event < %(beforeEvent)i', + 'host_channels.active', 'host_channels.channel_id', + 'channels.name', 'host_channels.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'host_channels.creator_id', 'creator.name', 'host.name', + 'host_channels.host_id', 'host_channels.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'host_channels.revoker_id', 'revoker.name']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN host ON host_id = host.id', + 'LEFT OUTER JOIN channels ON channel_id = channels.id']) + self.assertEqual(query.values, {'beforeEvent': 99, 'channel_id': 22}) + + def test_user_and_before_key(self): + self.get_user.return_value = {'id': 159, 'name': 'test-user'} + kojihub.query_history(user='test-user', before=1681925020) + self.assertEqual(len(self.queries), 4) + query = self.queries[0] + self.assertEqual(query.tables, ['cg_users']) + self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s', + 'users.id = %(affected_user_id)i']) + self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s', + 'cg_users.active', 'cg_users.cg_id', + 'content_generator.name', 'cg_users.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'cg_users.creator_id', 'creator.name', + 'cg_users.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'cg_users.revoker_id', 'revoker.name', 'users.name', + 'cg_users.user_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN users ON user_id = users.id', + 'LEFT OUTER JOIN content_generator ON cg_id = content_generator.id']) + self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40'}) + + query = self.queries[1] + self.assertEqual(query.tables, ['tag_package_owners']) + self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s', + 'owner.id = %(affected_user_id)i']) + self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s', + 'tag_package_owners.active', + 'tag_package_owners.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'tag_package_owners.creator_id', 'creator.name', + 'tag_package_owners.owner', 'owner.name', 'package.name', + 'tag_package_owners.package_id', + 'tag_package_owners.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'tag_package_owners.revoker_id', 'revoker.name', + 'tag.name', 'tag_package_owners.tag_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN package ON package_id = package.id', + 'LEFT OUTER JOIN tag ON tag_id = tag.id', + 'LEFT OUTER JOIN users AS owner ON owner = owner.id']) + self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40'}) + + query = self.queries[2] + self.assertEqual(query.tables, ['user_groups']) + self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s', + 'usergroup.id = %(affected_user_id)i']) + self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s', + 'user_groups.active', 'user_groups.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'user_groups.creator_id', 'creator.name', + 'usergroup.name', 'user_groups.group_id', + 'user_groups.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'user_groups.revoker_id', 'revoker.name', + 'users.name', 'user_groups.user_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN users ON user_id = users.id', + 'users AS usergroup ON group_id = usergroup.id']) + self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40'}) + + query = self.queries[3] + self.assertEqual(query.tables, ['user_perms']) + self.assertEqual(query.clauses, ['ev1.time < %(before)s OR ev2.time < %(before)s', + 'users.id = %(affected_user_id)i']) + self.assertEqual(query.columns, ['ev1.time < %(before)s', 'ev2.time < %(before)s', + 'user_perms.active', 'user_perms.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'user_perms.creator_id', 'creator.name', + 'user_perms.perm_id', 'permission.name', + 'user_perms.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'user_perms.revoker_id', 'revoker.name', 'users.name', + 'user_perms.user_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN users ON user_id = users.id', + 'LEFT OUTER JOIN permissions AS permission ON perm_id = permission.id']) + self.assertEqual(query.values, {'affected_user_id': 159, 'before': '2023-04-19 17:23:40'}) + + def test_permission_and_after_key(self): + self.get_perm_id.return_value = 66 + kojihub.query_history(permission='test-perms', after=1681925020) + self.assertEqual(len(self.queries), 2) + query = self.queries[0] + self.assertEqual(query.tables, ['tag_config']) + self.assertEqual(query.clauses, ['ev1.time > %(after)s OR ev2.time > %(after)s', + 'permission.id = %(perm_id)i']) + self.assertEqual(query.columns, ['ev1.time > %(after)s', 'ev2.time > %(after)s', + 'tag_config.active', 'tag_config.arches', + 'tag_config.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'tag_config.creator_id', 'creator.name', + 'tag_config.locked', 'tag_config.maven_include_all', + 'tag_config.maven_support', 'tag_config.perm_id', + 'permission.name', 'tag_config.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'tag_config.revoker_id', 'revoker.name', 'tag.name', + 'tag_config.tag_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN tag ON tag_id = tag.id', + 'LEFT OUTER JOIN permissions AS permission ON perm_id = permission.id']) + self.assertEqual(query.values, {'after': '2023-04-19 17:23:40', 'perm_id': 66}) + + query = self.queries[1] + self.assertEqual(query.tables, ['user_perms']) + self.assertEqual(query.clauses, ['ev1.time > %(after)s OR ev2.time > %(after)s', + 'permission.id = %(perm_id)i']) + self.assertEqual(query.columns, ['ev1.time > %(after)s', 'ev2.time > %(after)s', + 'user_perms.active', 'user_perms.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'user_perms.creator_id', 'creator.name', + 'user_perms.perm_id', 'permission.name', + 'user_perms.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'user_perms.revoker_id', 'revoker.name', 'users.name', + 'user_perms.user_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN users ON user_id = users.id', + 'LEFT OUTER JOIN permissions AS permission ON perm_id = permission.id']) + self.assertEqual(query.values, {'after': '2023-04-19 17:23:40', 'perm_id': 66}) + + def test_cg_key(self): + self.lookup_name.return_value = {'id': 147} + kojihub.query_history(cg='test-cg') + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['cg_users']) + self.assertEqual(query.clauses, ['content_generator.id = %(cg_id)i']) + self.assertEqual(query.columns, ['cg_users.active', 'cg_users.cg_id', + 'content_generator.name', 'cg_users.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'cg_users.creator_id', 'creator.name', + 'cg_users.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'cg_users.revoker_id', 'revoker.name', 'users.name', + 'cg_users.user_id']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN users ON user_id = users.id', + 'LEFT OUTER JOIN content_generator ON cg_id = content_generator.id']) + self.assertEqual(query.values, {'cg_id': 147}) + + def test_external_repo_key(self): + self.get_external_repo_id.return_value = 49 + kojihub.query_history(external_repo='test-ext-repo') + self.assertEqual(len(self.queries), 2) + query = self.queries[0] + self.assertEqual(query.tables, ['external_repo_config']) + self.assertEqual(query.clauses, ['external_repo.id = %(external_repo_id)i']) + self.assertEqual(query.columns, ['external_repo_config.active', + 'external_repo_config.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'external_repo_config.creator_id', 'creator.name', + 'external_repo.name', + 'external_repo_config.external_repo_id', + 'external_repo_config.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'external_repo_config.revoker_id', 'revoker.name', + 'external_repo_config.url']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN external_repo ON external_repo_id = external_repo.id']) + self.assertEqual(query.values, {'external_repo_id': 49}) + + query = self.queries[1] + self.assertEqual(query.tables, ['tag_external_repos']) + self.assertEqual(query.clauses, ['external_repo.id = %(external_repo_id)i']) + self.assertEqual(query.columns, ['tag_external_repos.active', + 'tag_external_repos.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'tag_external_repos.creator_id', 'creator.name', + 'external_repo.name', + 'tag_external_repos.external_repo_id', + 'tag_external_repos.merge_mode', + 'tag_external_repos.priority', + 'tag_external_repos.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'tag_external_repos.revoker_id', 'revoker.name', + 'tag.name', 'tag_external_repos.tag_id' + ]) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN tag ON tag_id = tag.id', + 'LEFT OUTER JOIN external_repo ON external_repo_id = external_repo.id']) + self.assertEqual(query.values, {'external_repo_id': 49}) + + def test_build_target_key(self): + self.get_build_target_id.return_value = 55 + kojihub.query_history(build_target='test-target') + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['build_target_config']) + self.assertEqual(query.clauses, ['build_target.id = %(build_target_id)i']) + self.assertEqual(query.columns, ['build_target_config.active', + 'build_target_config.build_tag', 'build_tag.name', + 'build_target.name', + 'build_target_config.build_target_id', + 'build_target_config.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'build_target_config.creator_id', 'creator.name', + 'build_target_config.dest_tag', 'dest_tag.name', + 'build_target_config.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'build_target_config.revoker_id', 'revoker.name']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN build_target ON build_target_id = build_target.id', + 'LEFT OUTER JOIN tag AS build_tag ON build_tag = build_tag.id', + 'LEFT OUTER JOIN tag AS dest_tag ON dest_tag = dest_tag.id']) + self.assertEqual(query.values, {'build_target_id': 55}) + + def test_xkey_key(self): + kojihub.query_history(xkey='test-key') + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['tag_extra']) + self.assertEqual(query.clauses, ['tag_extra.key = %(key)s']) + self.assertEqual(query.columns, ['tag_extra.active', 'tag_extra.create_event', + "date_part('epoch', ev1.time) AS create_ts", + 'tag_extra.creator_id', 'creator.name', 'tag_extra.key', + 'tag_extra.revoke_event', + "date_part('epoch', ev2.time) AS revoke_ts", + 'tag_extra.revoker_id', 'revoker.name', 'tag.name', + 'tag_extra.tag_id', 'tag_extra.value']) + self.assertEqual(query.joins, + ["events AS ev1 ON ev1.id = create_event", + "LEFT OUTER JOIN events AS ev2 ON ev2.id = revoke_event", + "users AS creator ON creator.id = creator_id", + "LEFT OUTER JOIN users AS revoker ON revoker.id = revoker_id", + 'LEFT OUTER JOIN tag ON tag_id = tag.id']) + self.assertEqual(query.values, {'key': 'test-key'}) diff --git a/tests/test_hub/test_remove_volume.py b/tests/test_hub/test_remove_volume.py new file mode 100644 index 0000000..6bb18e6 --- /dev/null +++ b/tests/test_hub/test_remove_volume.py @@ -0,0 +1,70 @@ +import mock + +import koji +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + +DP = kojihub.DeleteProcessor + + +class TestRemoveVolume(DBQueryTestCase): + def setUp(self): + super(TestRemoveVolume, self).setUp() + self.DeleteProcessor = mock.patch('kojihub.kojihub.DeleteProcessor', + side_effect=self.getDelete).start() + self.deletes = [] + self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start() + self.context = mock.patch('kojihub.kojihub.context').start() + # It seems MagicMock will not automatically handle attributes that + # start with "assert" + self.context.session.assertPerm = mock.MagicMock() + + def tearDown(self): + mock.patch.stopall() + + def getDelete(self, *args, **kwargs): + delete = DP(*args, **kwargs) + delete.execute = mock.MagicMock() + self.deletes.append(delete) + return delete + + def test_with_references(self): + volume_name = 'test-volume' + volume_dict = {'id': 0, 'name': volume_name} + self.lookup_name.return_value = volume_dict + self.qp_execute_return_value = [{'id': 1}] + with self.assertRaises(koji.GenericError) as cm: + kojihub.remove_volume(volume_name) + self.assertEqual(f'volume {volume_name} has build references', str(cm.exception)) + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['build']) + self.assertEqual(query.columns, ['id']) + self.assertEqual(query.clauses, ['volume_id=%(id)i']) + self.assertEqual(query.joins, None) + self.assertEqual(query.values, volume_dict) + + self.assertEqual(len(self.deletes), 0) + + def test_valid(self): + volume_name = 'test-volume' + volume_dict = {'id': 0, 'name': volume_name} + self.lookup_name.return_value = volume_dict + self.qp_execute_return_value = None + kojihub.remove_volume(volume_name) + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['build']) + self.assertEqual(query.columns, ['id']) + self.assertEqual(query.clauses, ['volume_id=%(id)i']) + self.assertEqual(query.joins, None) + self.assertEqual(query.values, volume_dict) + + self.assertEqual(len(self.deletes), 1) + delete = self.deletes[0] + self.assertEqual(delete.table, 'volume') + self.assertEqual(delete.clauses, ['id=%(id)i']) + self.assertEqual(delete.values, volume_dict) diff --git a/tests/test_hub/test_revoke_cg_access.py b/tests/test_hub/test_revoke_cg_access.py new file mode 100644 index 0000000..0a9f50c --- /dev/null +++ b/tests/test_hub/test_revoke_cg_access.py @@ -0,0 +1,49 @@ +# coding: utf-8 +import unittest + +import mock +import kojihub + +UP = kojihub.UpdateProcessor + + +class TestRevokeCGAccess(unittest.TestCase): + + def getUpdate(self, *args, **kwargs): + update = UP(*args, **kwargs) + update.execute = mock.MagicMock() + update.make_revoke = mock.MagicMock() + self.updates.append(update) + return update + + def setUp(self): + self.maxDiff = None + self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor', + side_effect=self.getUpdate).start() + self.updates = [] + self.get_user = mock.patch('kojihub.kojihub.get_user').start() + self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start() + self.context = mock.patch('kojihub.kojihub.context').start() + # It seems MagicMock will not automatically handle attributes that + # start with "assert" + self.context.session.assertPerm = mock.MagicMock() + + def tearDown(self): + mock.patch.stopall() + + def test_valid(self): + cg = {'id': 11, 'name': 'test-cg-name'} + user = {'id': 123, 'name': 'testuser'} + self.get_user.return_value = user + self.lookup_name.return_value = cg + kojihub.revoke_cg_access(user['name'], cg['name']) + + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'cg_users') + self.assertEqual(update.values, {'user_id': user['id'], 'cg_id': cg['id']}) + self.assertEqual(update.data, {}) + self.assertEqual(update.rawdata, {}) + self.assertEqual(update.clauses, ["user_id = %(user_id)i", "cg_id = %(cg_id)i"]) + self.get_user.assert_called_once_with(user['name'], strict=True) + self.lookup_name.assert_called_once_with('content_generator', cg['name'], strict=True) diff --git a/tests/test_hub/test_set_user_status.py b/tests/test_hub/test_set_user_status.py new file mode 100644 index 0000000..3455d55 --- /dev/null +++ b/tests/test_hub/test_set_user_status.py @@ -0,0 +1,61 @@ +import mock +import unittest + +import koji +import kojihub + +UP = kojihub.UpdateProcessor + + +class TestSetUserStatus(unittest.TestCase): + + def getUpdate(self, *args, **kwargs): + update = UP(*args, **kwargs) + update.execute = self.update_execute + self.updates.append(update) + return update + + def setUp(self): + self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor', + side_effect=self.getUpdate).start() + self.updates = [] + self.context = mock.patch('kojihub.kojihub.context').start() + self.context.session.assertPerm = mock.MagicMock() + self.update_execute = mock.MagicMock() + + def test_wrong_status(self): + status = 111 + with self.assertRaises(koji.GenericError) as cm: + kojihub.set_user_status(1, status) + self.assertEqual(f'No such status: {status}', str(cm.exception)) + self.assertEqual(len(self.updates), 0) + + def test_status_is_setup(self): + rv = kojihub.set_user_status({'status': 1}, 1) + self.assertEqual(rv, None) + self.assertEqual(len(self.updates), 0) + + def test_valid(self): + self.update_execute.return_value = 1 + rv = kojihub.set_user_status({'status': 2, 'id': 123}, 1) + self.assertEqual(rv, None) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'users') + self.assertEqual(update.values, {'user_id': 123}) + self.assertEqual(update.data, {'status': 1}) + self.assertEqual(update.rawdata, {}) + self.assertEqual(update.clauses, ['id = %(user_id)i']) + + def test_user_not_exist(self): + self.update_execute.return_value = 0 + with self.assertRaises(koji.GenericError) as cm: + kojihub.set_user_status({'status': 2, 'id': 123}, 1) + self.assertEqual('No such user ID: 123', str(cm.exception)) + self.assertEqual(len(self.updates), 1) + update = self.updates[0] + self.assertEqual(update.table, 'users') + self.assertEqual(update.values, {'user_id': 123}) + self.assertEqual(update.data, {'status': 1}) + self.assertEqual(update.rawdata, {}) + self.assertEqual(update.clauses, ['id = %(user_id)i']) diff --git a/tests/test_hub/test_untagged_build.py b/tests/test_hub/test_untagged_build.py new file mode 100644 index 0000000..e6336ab --- /dev/null +++ b/tests/test_hub/test_untagged_build.py @@ -0,0 +1,59 @@ +import mock + +import kojihub +import kojihub.kojihub +from .utils import DBQueryTestCase + + +class TestUntaggedBuilds(DBQueryTestCase): + def setUp(self): + super(TestUntaggedBuilds, self).setUp() + + def tearDown(self): + mock.patch.stopall() + + def test_valid_name_none(self): + self.qp_iterate_return_value = [{'id': 1, 'name': 'pkg-name', 'version': '1', + 'release': '2'}, + {'id': 1, 'name': 'test-pkg', 'version': '2', + 'release': '3'}] + expected_result = [{'id': 1, 'name': 'pkg-name', 'version': '1', 'release': '2'}, + {'id': 1, 'name': 'test-pkg', 'version': '2', 'release': '3'}] + rv = kojihub.untagged_builds() + self.assertEqual(rv, expected_result) + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['build', 'package']) + self.assertEqual(query.columns, + ['build.id', 'package.name', 'build.release', 'build.version']) + self.assertEqual(query.aliases, ['id', 'name', 'release', 'version']) + self.assertEqual(query.clauses, ['NOT EXISTS\n' + ' (SELECT 1 FROM tag_listing\n' + ' WHERE tag_listing.build_id = build.id\n' + ' AND tag_listing.active IS TRUE)', + "build.state = %(st_complete)i", + "package.id = build.pkg_id"]) + self.assertEqual(query.joins, None) + + def test_valid_with_name(self): + self.qp_iterate_return_value = [{'id': 1, 'name': 'pkg-name', 'version': '1', + 'release': '2'}] + expected_result = [{'id': 1, 'name': 'pkg-name', 'version': '1', 'release': '2'}] + rv = kojihub.untagged_builds(name='pkg-name') + self.assertEqual(rv, expected_result) + + self.assertEqual(len(self.queries), 1) + query = self.queries[0] + self.assertEqual(query.tables, ['build', 'package']) + self.assertEqual(query.columns, + ['build.id', 'package.name', 'build.release', 'build.version']) + self.assertEqual(query.aliases, ['id', 'name', 'release', 'version']) + self.assertEqual(query.clauses, ['NOT EXISTS\n' + ' (SELECT 1 FROM tag_listing\n' + ' WHERE tag_listing.build_id = build.id\n' + ' AND tag_listing.active IS TRUE)', + "build.state = %(st_complete)i", + "package.id = build.pkg_id", + 'package.name = %(name)s']) + self.assertEqual(query.joins, None) diff --git a/tests/test_hub/utils.py b/tests/test_hub/utils.py index 4923cb3..4ac0b6c 100644 --- a/tests/test_hub/utils.py +++ b/tests/test_hub/utils.py @@ -15,6 +15,7 @@ class DBQueryTestCase(unittest.TestCase): self.qp_execute_one_return_value = [] 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() self.queries = [] @@ -36,6 +37,8 @@ class DBQueryTestCase(unittest.TestCase): 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