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

Increase hub unit tests

Browse source
This commit is contained in:
Jana Cupova 2023-04-16 10:45:06 +02:00 • committed by Tomas Kopecek
parent 5c0b9598bd
commit 3c8f309e47
17 changed files with 1771 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

106
tests/test_hub/test_add_archivetype.py
View file

@ -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 <class 'str'>", 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 <class 'str'>", 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()

36
tests/test_hub/test_add_volume.py
View file

@ -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)

104
tests/test_hub/test_build_notification.py Normal file
View file

@ -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})

132
tests/test_hub/test_cancel_build.py Normal file
View file

@ -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)

71
tests/test_hub/test_grant_cg_access.py Normal file
View file

@ -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)

31
tests/test_hub/test_list_cgs.py Normal file
View file

@ -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, {})

65
tests/test_hub/test_new_image_build.py Normal file
View file

@ -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()

44
tests/test_hub/test_new_maven_build.py
View file

@ -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()

70
tests/test_hub/test_new_package.py Normal file
View file

@ -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, {})

54
tests/test_hub/test_new_win_build.py
View file

@ -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()

334
tests/test_hub/test_policy_tests.py
View file

@ -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()

504
tests/test_hub/test_query_history.py Normal file
View file

@ -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'})

70
tests/test_hub/test_remove_volume.py Normal file
View file

@ -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)

49
tests/test_hub/test_revoke_cg_access.py Normal file
View file

@ -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)

61
tests/test_hub/test_set_user_status.py Normal file
View file

@ -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'])

59
tests/test_hub/test_untagged_build.py Normal file
View file

@ -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)

3
tests/test_hub/utils.py
View file

@ -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