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 2022-05-05 13:20:05 +02:00
parent ef0f366611
commit 83d4617493
13 changed files with 614 additions and 124 deletions
Download patch file Download diff file Expand all files Collapse all files

4
hub/kojihub.py
View file

@ -4673,8 +4673,8 @@ def get_rpm(rpminfo, strict=False, multi=False):
if 'id' in data:
clauses.append("rpminfo.id=%(id)s")
else:
clauses.append("""rpminfo.name=%(name)s AND version=%(version)s
AND release=%(release)s AND arch=%(arch)s""")
clauses.append("rpminfo.name=%(name)s AND version=%(version)s "
"AND release=%(release)s AND arch=%(arch)s")
retry = False
if 'location' in data:
data['external_repo_id'] = get_external_repo_id(data['location'], strict=True)

24
tests/test_hub/test_add_external_repo_to_tag.py
View file

@ -11,7 +11,11 @@ class TestAddExternalRepoToTag(unittest.TestCase):
self.tag_name = 'test-tag'
self.get_tag = mock.patch('kojihub.get_tag').start()
self.get_external_repo = mock.patch('kojihub.get_external_repo').start()
self.get_tag_external_repos = mock.patch('kojihub.get_tag_external_repos').start()
self.parse_arches = mock.patch('koji.parse_arches').start()
self.tag_info = {'id': 1, 'name': self.tag_name}
self.external_repo_info = {'id': 123, 'name': 'test-repo'}
self.priority = 11
def tearDown(self):
mock.patch.stopall()
@ -30,3 +34,23 @@ class TestAddExternalRepoToTag(unittest.TestCase):
kojihub.add_external_repo_to_tag(self.tag_name, 'repo', priority, merge_mode=None)
self.assertEqual(f"Invalid type for value '{priority}': {type(priority)}, "
f"expected type <class 'int'>", str(cm.exception))
def test_tag_asociated_with_ext_repo(self):
self.get_tag.return_value = self.tag_info
self.get_external_repo.return_value = self.external_repo_info
self.get_tag_external_repos.return_value = [{'external_repo_id': 234},
{'external_repo_id': 123}]
with self.assertRaises(koji.GenericError) as cm:
kojihub.add_external_repo_to_tag(self.tag_name, 'test-repo', self.priority)
self.assertEqual(f"tag {self.tag_info['name']} already associated with external "
f"repo {self.external_repo_info['name']}", str(cm.exception))
def test_tag_asociated_with_priority(self):
self.get_tag.return_value = self.tag_info
self.get_external_repo.return_value = self.external_repo_info
self.get_tag_external_repos.return_value = [{'external_repo_id': 234, 'priority': 12},
{'external_repo_id': 345, 'priority': 11}]
with self.assertRaises(koji.GenericError) as cm:
kojihub.add_external_repo_to_tag(self.tag_name, 'test-repo', self.priority)
self.assertEqual(f"tag {self.tag_info['name']} already associated "
f"with an external repo at priority {self.priority}", str(cm.exception))

199
tests/test_hub/test_add_host_to_channel.py
View file

@ -28,34 +28,40 @@ class TestAddHostToChannel(unittest.TestCase):
self.context.session.user_id = 23
self.context.opts = {'HostPrincipalFormat': '-%s-'}
self.exports = kojihub.RootExports()
self.get_channel = mock.patch('kojihub.get_channel').start()
self.list_channels = mock.patch('kojihub.list_channels').start()
self.get_channel_id = mock.patch('kojihub.get_channel_id').start()
self.get_host = mock.patch('kojihub.get_host').start()
self.verify_name_internal = mock.patch('kojihub.verify_name_internal').start()
self.cname = 'channel_name'
self.name = 'hostname'
self.host_info = {'id': 123, 'name': self.name}
self.channel_id = 456
self.list_channels_dict = [{'id': 1, 'name': 'default'}]
self.channel_info = {'enabled': True}
def tearDown(self):
mock.patch.stopall()
@mock.patch('kojihub.get_channel')
@mock.patch('kojihub.list_channels')
@mock.patch('kojihub.get_channel_id')
@mock.patch('kojihub.get_host')
def test_valid(self, get_host, get_channel_id, list_channels, get_channel):
name = 'hostname'
cname = 'channel_name'
get_host.return_value = {'id': 123, 'name': name}
get_channel_id.return_value = 456
list_channels.return_value = [{'id': 1, 'name': 'default'}]
get_channel.return_value = {'enabled': True}
def test_valid(self):
self.get_host.return_value = self.host_info
self.get_channel_id.return_value = self.channel_id
self.list_channels.return_value = self.list_channels_dict
self.get_channel.return_value = self.channel_info
kojihub.add_host_to_channel(name, cname, create=False)
kojihub.add_host_to_channel(self.name, self.cname, create=False)
get_host.assert_called_once_with(name)
get_channel.assert_called_once_with(456)
get_channel_id.assert_called_once_with(cname, create=False)
list_channels.assert_called_once_with(123)
self.get_host.assert_called_once_with(self.name)
self.get_channel.assert_called_once_with(self.channel_id)
self.get_channel_id.assert_called_once_with(self.cname, create=False)
self.list_channels.assert_called_once_with(self.host_info['id'])
self.verify_name_internal.assert_not_called()
self.assertEqual(len(self.inserts), 1)
insert = self.inserts[0]
data = {
'host_id': 123,
'channel_id': 456,
'host_id': self.host_info['id'],
'channel_id': self.channel_id,
'creator_id': 23,
'create_event': 42,
}
@ -63,62 +69,48 @@ class TestAddHostToChannel(unittest.TestCase):
self.assertEqual(insert.data, data)
self.assertEqual(insert.rawdata, {})
@mock.patch('kojihub.list_channels')
@mock.patch('kojihub.get_channel_id')
@mock.patch('kojihub.get_host')
def test_no_host(self, get_host, get_channel_id, list_channels):
name = 'hostname'
cname = 'channel_name'
get_host.return_value = None
def test_no_host(self):
self.get_host.return_value = None
with self.assertRaises(koji.GenericError):
kojihub.add_host_to_channel(name, cname, create=False)
with self.assertRaises(koji.GenericError) as ex:
kojihub.add_host_to_channel(self.name, self.cname, create=False)
get_host.assert_called_once_with(name)
self.get_host.assert_called_once_with(self.name)
self.assertEqual(len(self.inserts), 0)
self.assertEqual(f"host does not exist: {self.name}", str(ex.exception))
@mock.patch('kojihub.get_channel_id')
@mock.patch('kojihub.get_host')
def test_no_channel(self, get_host, get_channel_id):
name = 'hostname'
cname = 'channel_name'
get_host.return_value = {'id': 123, 'name': name}
get_channel_id.return_value = None
def test_no_channel(self):
self.get_host.return_value = self.host_info
self.get_channel_id.return_value = None
with self.assertRaises(koji.GenericError):
kojihub.add_host_to_channel(name, cname, create=False)
with self.assertRaises(koji.GenericError) as ex:
kojihub.add_host_to_channel(self.name, self.cname, create=False)
get_host.assert_called_once_with(name)
get_channel_id.assert_called_once_with(cname, create=False)
self.get_host.assert_called_once_with(self.name)
self.get_channel_id.assert_called_once_with(self.cname, create=False)
self.assertEqual(len(self.inserts), 0)
self.assertEqual(f"channel does not exist: {self.cname}", str(ex.exception))
@mock.patch('kojihub.verify_name_internal')
@mock.patch('kojihub.get_channel')
@mock.patch('kojihub.list_channels')
@mock.patch('kojihub.get_channel_id')
@mock.patch('kojihub.get_host')
def test_no_channel_create(self, get_host, get_channel_id, list_channels, get_channel,
verify_name_internal):
name = 'hostname'
cname = 'channel_name'
get_host.return_value = {'id': 123, 'name': name}
get_channel_id.return_value = 456
list_channels.return_value = [{'id': 1, 'name': 'default'}]
get_channel.return_value = {'enabled': True}
verify_name_internal.return_value = None
def test_no_channel_create(self):
self.get_host.return_value = self.host_info
self.get_channel_id.return_value = self.channel_id
self.list_channels.return_value = self.list_channels_dict
self.get_channel.return_value = self.get_channel
self.verify_name_internal.return_value = None
kojihub.add_host_to_channel(name, cname, create=True)
kojihub.add_host_to_channel(self.name, self.cname, create=True)
get_host.assert_called_once_with(name)
get_channel.assert_called_once_with(456)
get_channel_id.assert_called_once_with(cname, create=True)
list_channels.assert_called_once_with(123)
self.get_host.assert_called_once_with(self.name)
self.get_channel.assert_called_once_with(self.channel_id)
self.get_channel_id.assert_called_once_with(self.cname, create=True)
self.list_channels.assert_called_once_with(self.host_info['id'])
self.verify_name_internal.assert_called_once_with(self.cname)
self.assertEqual(len(self.inserts), 1)
insert = self.inserts[0]
data = {
'host_id': 123,
'channel_id': 456,
'host_id': self.host_info['id'],
'channel_id': self.channel_id,
'creator_id': 23,
'create_event': 42,
}
@ -126,40 +118,75 @@ class TestAddHostToChannel(unittest.TestCase):
self.assertEqual(insert.data, data)
self.assertEqual(insert.rawdata, {})
@mock.patch('kojihub.get_channel')
@mock.patch('kojihub.list_channels')
@mock.patch('kojihub.get_channel_id')
@mock.patch('kojihub.get_host')
def test_exists(self, get_host, get_channel_id, list_channels, get_channel):
name = 'hostname'
cname = 'channel_name'
get_host.return_value = {'id': 123, 'name': name}
get_channel_id.return_value = 456
list_channels.return_value = [{'id': 456, 'name': cname}]
get_channel.return_value = {'enabled': True}
def test_exists(self):
self.get_host.return_value = self.host_info
self.get_channel_id.return_value = self.channel_id
self.list_channels.return_value = [{'id': self.channel_id, 'name': self.cname}]
self.get_channel.return_value = self.channel_info
with self.assertRaises(koji.GenericError):
kojihub.add_host_to_channel(name, cname, create=False)
with self.assertRaises(koji.GenericError) as ex:
kojihub.add_host_to_channel(self.name, self.cname, create=False)
get_host.assert_called_once_with(name)
get_channel.assert_called_once_with(456)
get_channel_id.assert_called_once_with(cname, create=False)
list_channels.assert_called_once_with(123)
self.get_host.assert_called_once_with(self.name)
self.get_channel.assert_called_once_with(self.channel_id)
self.get_channel_id.assert_called_once_with(self.cname, create=False)
self.list_channels.assert_called_once_with(self.host_info['id'])
self.verify_name_internal.assert_not_called()
self.assertEqual(len(self.inserts), 0)
self.assertEqual(f"host {self.name} is already subscribed to the {self.cname} channel",
str(ex.exception))
@mock.patch('kojihub.verify_name_internal')
@mock.patch('kojihub.get_host')
def test_channel_wrong_format(self, get_host, verify_name_internal):
name = 'hostname'
def test_channel_wrong_format(self):
channel_name = 'test-channel+'
get_host.return_value = {'id': 123, 'name': name}
self.get_host.return_value = self.host_info
# name is longer as expected
verify_name_internal.side_effect = koji.GenericError
self.verify_name_internal.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
kojihub.add_host_to_channel(name, channel_name, create=True)
kojihub.add_host_to_channel(self.name, channel_name, create=True)
# not except regex rules
verify_name_internal.side_effect = koji.GenericError
self.verify_name_internal.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
kojihub.add_host_to_channel(name, channel_name, create=True)
kojihub.add_host_to_channel(self.name, channel_name, create=True)
def test_channel_disabled_without_force(self):
self.get_host.return_value = self.host_info
self.get_channel_id.return_value = self.channel_id
self.get_channel.return_value = {'id': 456, 'enabled': False}
with self.assertRaises(koji.GenericError) as ex:
kojihub.add_host_to_channel(self.name, self.cname, create=False)
self.get_host.assert_called_once_with(self.name)
self.get_channel.assert_called_once_with(self.channel_id)
self.get_channel_id.assert_called_once_with(self.cname, create=False)
self.list_channels.assert_not_called()
self.verify_name_internal.assert_not_called()
self.assertEqual(len(self.inserts), 0)
self.assertEqual(f"channel {self.cname} is disabled", str(ex.exception))
def test_disable_with_force(self):
self.get_host.return_value = self.host_info
self.get_channel_id.return_value = self.channel_id
self.list_channels.return_value = self.list_channels_dict
kojihub.add_host_to_channel(self.name, self.cname, create=False, force=True)
self.get_host.assert_called_once_with(self.name)
self.get_channel.assert_not_called()
self.get_channel_id.assert_called_once_with(self.cname, create=False)
self.list_channels.assert_called_once_with(self.host_info['id'])
self.verify_name_internal.assert_not_called()
self.assertEqual(len(self.inserts), 1)
insert = self.inserts[0]
data = {
'host_id': self.host_info['id'],
'channel_id': self.channel_id,
'creator_id': 23,
'create_event': 42,
}
self.assertEqual(insert.table, 'host_channels')
self.assertEqual(insert.data, data)
self.assertEqual(insert.rawdata, {})

79
tests/test_hub/test_edit_build_target.py
View file

@ -11,35 +11,84 @@ class TestEditBuildTarget(unittest.TestCase):
def setUp(self):
self.lookup_build_target = mock.patch('kojihub.lookup_build_target').start()
self.verify_name_internal = mock.patch('kojihub.verify_name_internal').start()
self.get_tag = mock.patch('kojihub.get_tag').start()
self._singleValue = mock.patch('kojihub._singleValue').start()
self.exports = kojihub.RootExports()
self.target_name = 'build-target'
self.name = 'build-target-rename'
self.build_tag = 'tag'
self.dest_tag = 'dest-tag'
self.target_info = {'id': 123, 'name': self.target_name}
self.build_tag_info = {'id': 111, 'name': self.build_tag}
self.dest_tag_info = {'id': 112, 'name': self.dest_tag}
self.session = kojihub.context.session = mock.MagicMock()
self.session.assertPerm = mock.MagicMock()
def tearDown(self):
mock.patch.stopall()
def test_non_exist_build_target(self):
session = kojihub.context.session = mock.MagicMock()
session.assertPerm = mock.MagicMock()
self.verify_name_internal.return_value = None
target_name = 'build-target'
name = 'build-target-rename'
build_tag = 'tag'
dest_tag = 'dest-tag'
self.lookup_build_target.return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.editBuildTarget(target_name, name, build_tag, dest_tag)
self.assertEqual("No such build target: %s" % target_name,
str(cm.exception))
session.assertPerm.called_once_with('target')
self.exports.editBuildTarget(self.target_name, self.name, self.build_tag,
self.dest_tag)
self.assertEqual(f"No such build target: {self.target_name}", str(cm.exception))
self.session.assertPerm.called_once_with('target')
self.verify_name_internal.called_once_with(name=self.name)
self.lookup_build_target.called_once_with(self.target_name)
def test_target_wrong_format(self):
target_name = 'test-target'
name = 'build-target-rename+'
build_tag = 'tag'
dest_tag = 'dest-tag'
# name is longer as expected
self.verify_name_internal.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
self.exports.editBuildTarget(target_name, name, build_tag, dest_tag)
self.exports.editBuildTarget(self.target_name, name, self.build_tag, self.dest_tag)
# not except regex rules
self.verify_name_internal.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
self.exports.editBuildTarget(target_name, name, build_tag, dest_tag)
self.exports.editBuildTarget(self.target_name, name, self.build_tag, self.dest_tag)
def test_target_non_exist_build_tag(self):
self.verify_name_internal.return_value = None
self.lookup_build_target.return_value = self.target_info
self.get_tag.return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.editBuildTarget(self.target_name, self.name, self.build_tag,
self.dest_tag)
self.assertEqual(f"build tag '{self.build_tag}' does not exist", str(cm.exception))
self.session.assertPerm.called_once_with('target')
self.verify_name_internal.called_once_with(name=self.name)
self.lookup_build_target.called_once_with(self.target_name)
self.get_tag.called_once_with(self.build_tag)
def test_target_non_exist_dest_tag(self):
self.verify_name_internal.return_value = None
self.lookup_build_target.return_value = self.target_info
self.get_tag.side_effect = [self.build_tag_info, None]
with self.assertRaises(koji.GenericError) as cm:
self.exports.editBuildTarget(self.target_name, self.name, self.build_tag,
self.dest_tag)
self.assertEqual(f"destination tag '{self.dest_tag}' does not exist", str(cm.exception))
self.session.assertPerm.called_once_with('target')
self.verify_name_internal.called_once_with(name=self.name)
self.lookup_build_target.called_once_with(self.target_name)
self.get_tag.has_calls([mock.call(self.build_tag), mock.call(self.dest_tag)])
def test_target_exists(self):
self.verify_name_internal.return_value = None
self.lookup_build_target.return_value = self.target_info
self.get_tag.side_effect = [self.build_tag_info, self.dest_tag_info]
self._singleValue.return_value = 2
with self.assertRaises(koji.GenericError) as cm:
self.exports.editBuildTarget(self.target_name, self.name, self.build_tag,
self.dest_tag)
self.assertEqual(f'name "{self.name}" is already taken by build target 2',
str(cm.exception))
self.session.assertPerm.called_once_with('target')
self.verify_name_internal.called_once_with(name=self.name)
self.lookup_build_target.called_once_with(self.target_name)
self.get_tag.has_calls([mock.call(self.build_tag), mock.call(self.dest_tag)])
self._singleValue.called_once_with(self.name)

77
tests/test_hub/test_getRPM.py
View file

@ -7,15 +7,92 @@ import mock
import koji
import kojihub
QP = kojihub.QueryProcessor
class TestGetRPM(unittest.TestCase):
def setUp(self):
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.get_external_repo_id = mock.patch('kojihub.get_external_repo_id').start()
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
self.queries.append(query)
return query
def tearDown(self):
mock.patch.stopall()
def test_wrong_type_rpminfo(self):
rpminfo = ['test-user']
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_rpm(rpminfo)
self.assertEqual(f"Invalid type for rpminfo: {type(rpminfo)}", str(cm.exception))
def test_rpm_info_int(self):
rpminfo = 123
kojihub.get_rpm(rpminfo)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['rpminfo'])
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',
'size', 'buildtime', 'metadata_only', 'extra']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses, ['external_repo_id = 0', "rpminfo.id=%(id)s"])
self.assertEqual(query.joins,
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
self.assertEqual(query.values, {'id': rpminfo})
def test_rpm_info_str(self):
rpminfo = 'testrpm-1.23-4.x86_64.rpm'
kojihub.get_rpm(rpminfo, multi=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['rpminfo'])
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',
'size', 'buildtime', 'metadata_only', 'extra']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses, ["rpminfo.name=%(name)s AND version=%(version)s "
"AND release=%(release)s AND arch=%(arch)s"])
self.assertEqual(query.joins,
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
self.assertEqual(query.values, {'arch': 'x86_64', 'epoch': '', 'name': 'testrpm',
'release': '4', 'src': False, 'version': '1.23'})
def test_rpm_info_dict_location(self):
rpminfo = {'id': 123, 'name': 'testrpm-1.23-4.x86_64.rpm', 'location': 'test-location'}
self.get_external_repo_id.return_value = 125
rpminfo_data = rpminfo.copy()
rpminfo_data['external_repo_id'] = 125
kojihub.get_rpm(rpminfo, multi=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['rpminfo'])
columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',
'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',
'size', 'buildtime', 'metadata_only', 'extra']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses,
["external_repo_id = %(external_repo_id)i", "rpminfo.id=%(id)s"])
self.assertEqual(query.joins,
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
self.assertEqual(query.values, rpminfo_data)
class TestGetRPMHeaders(unittest.TestCase):

48
tests/test_hub/test_get_archive.py
View file

@ -10,6 +10,9 @@ class TestGetArchive(unittest.TestCase):
def setUp(self):
self.maxDiff = None
self.list_archives = mock.patch('kojihub.list_archives').start()
self.get_maven_archive = mock.patch('kojihub.get_maven_archive').start()
self.get_win_archive = mock.patch('kojihub.get_win_archive').start()
self.get_image_archive = mock.patch('kojihub.get_image_archive').start()
def tearDown(self):
mock.patch.stopall()
@ -26,3 +29,48 @@ class TestGetArchive(unittest.TestCase):
self.list_archives.return_value = []
rv = kojihub.get_archive(archive_id)
self.assertEqual(rv, None)
def test_valid(self):
archive_id = 1
self.list_archives.return_value = [{'archive_id': 1, 'name': 'test-archive-1'}]
self.get_maven_archive.return_value = {'archive_id': 2, 'name': 'test-archive-maven'}
self.get_win_archive.return_value = {'archive_id': 3, 'name': 'test-archive-win'}
self.get_image_archive.return_value = {'archive_id': 5, 'name': 'test-archive-image'}
result = kojihub.get_archive(archive_id)
self.assertEqual({'archive_id': 1, 'name': 'test-archive-image'}, result)
def test_maven_archive_only(self):
archive_id = 1
self.list_archives.return_value = [{}]
self.get_maven_archive.return_value = {'archive_id': 2, 'name': 'test-archive-maven'}
self.get_win_archive.return_value = None
self.get_image_archive.return_value = None
result = kojihub.get_archive(archive_id)
self.assertEqual({'name': 'test-archive-maven'}, result)
def test_win_archive_only(self):
archive_id = 1
self.list_archives.return_value = [{}]
self.get_maven_archive.return_value = None
self.get_win_archive.return_value = {'archive_id': 3, 'name': 'test-archive-win'}
self.get_image_archive.return_value = None
result = kojihub.get_archive(archive_id)
self.assertEqual({'name': 'test-archive-win'}, result)
def test_image_archive_only(self):
archive_id = 1
self.list_archives.return_value = [{}]
self.get_maven_archive.return_value = None
self.get_win_archive.return_value = None
self.get_image_archive.return_value = {'archive_id': 5, 'name': 'test-archive-image'}
result = kojihub.get_archive(archive_id)
self.assertEqual({'name': 'test-archive-image'}, result)
def test_default_archive_only(self):
archive_id = 1
self.list_archives.return_value = [{'archive_id': 1, 'name': 'test-archive-1'}]
self.get_maven_archive.return_value = None
self.get_win_archive.return_value = None
self.get_image_archive.return_value = None
result = kojihub.get_archive(archive_id)
self.assertEqual({'archive_id': 1, 'name': 'test-archive-1'}, result)

14
tests/test_hub/test_get_build_target.py
View file

@ -11,24 +11,26 @@ class TestGetBuildTarget(unittest.TestCase):
def setUp(self):
self.get_build_targets = mock.patch('kojihub.get_build_targets').start()
self.exports = kojihub.RootExports()
self.build_target = 'build-target'
def tearDown(self):
mock.patch.stopall()
def test_non_exist_build_target(self):
build_target = 'build-target'
self.get_build_targets.return_value = []
with self.assertRaises(koji.GenericError) as cm:
self.exports.getBuildTarget(build_target, strict=True)
self.assertEqual("No such build target: %s" % build_target, str(cm.exception))
self.exports.getBuildTarget(self.build_target, strict=True)
self.assertEqual(f"No such build target: {self.build_target}", str(cm.exception))
def test_wrong_type_build_target(self):
build_target = {'info_key': 'info_value'}
expected = "Invalid type for lookup: %s" % type(build_target)
expected = f"Invalid type for lookup: {build_target}"
self.get_build_targets.side_effect = koji.GenericError(expected)
with self.assertRaises(koji.GenericError) as cm:
self.exports.getBuildTarget(build_target, strict=True)
self.assertEqual(expected, str(cm.exception))
def test_more_targets_without_strict(self):
build_target = 'build-target'
target_info = [{'build_tag': 123,
'build_tag_name': 'test-tag',
'dest_tag': 124,
@ -43,7 +45,7 @@ class TestGetBuildTarget(unittest.TestCase):
'name': 'test-build-target-5678'}
]
self.get_build_targets.return_value = target_info
rv = self.exports.getBuildTarget(build_target, strict=False)
rv = self.exports.getBuildTarget(self.build_target, strict=False)
self.assertEqual(None, rv)
def test_once_target(self):

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

@ -0,0 +1,49 @@
import unittest
import mock
import kojihub
QP = kojihub.QueryProcessor
class TestGetBuildTargets(unittest.TestCase):
def setUp(self):
self.maxDiff = None
self.name_or_id_clause = mock.patch('kojihub.name_or_id_clause').start()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.cursor = mock.MagicMock()
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.build_target = 'build-target'
self.build_tag = 'tag'
self.dest_tag = 'dest-tag'
def tearDown(self):
mock.patch.stopall()
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
def test_get_build_targets(self):
self.name_or_id_clause.return_value = '(build_target.name = %(build_target_name)s)', \
{'build_target_name': 'build-target-url'}
kojihub.get_build_targets(self.build_target, buildTagID=self.build_tag,
destTagID=self.dest_tag)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build_target_config'])
self.assertEqual(query.joins,
['build_target ON build_target_config.build_target_id = build_target.id',
'tag AS tag1 ON build_target_config.build_tag = tag1.id',
'tag AS tag2 ON build_target_config.dest_tag = tag2.id'])
self.assertEqual(query.clauses,
['(active = TRUE)', '(build_target.name = %(build_target_name)s)',
'build_tag = %(buildTagID)i', 'dest_tag = %(destTagID)i'])

51
tests/test_hub/test_get_tag_external_repos.py Normal file
View file

@ -0,0 +1,51 @@
import unittest
import mock
import kojihub
QP = kojihub.QueryProcessor
class TestGetTagExternalRepos(unittest.TestCase):
def setUp(self):
self.maxDiff = None
self.get_tag = mock.patch('kojihub.get_tag').start()
self.get_external_repo = mock.patch('kojihub.get_external_repo').start()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.cursor = mock.MagicMock()
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.build_tag = 'tag'
self.repo = 'repo_name'
self.build_tag_info = {'id': 111, 'name': self.build_tag}
self.repo_info = {'id': 123, 'name': self.repo}
def tearDown(self):
mock.patch.stopall()
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
def test_valid(self):
self.get_tag.return_value = self.build_tag_info
self.get_external_repo.return_value = self.repo_info
kojihub.get_tag_external_repos(tag_info=self.build_tag, repo_info=self.repo)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_external_repos'])
self.assertEqual(query.joins, [
'tag ON tag_external_repos.tag_id = tag.id',
'external_repo ON tag_external_repos.external_repo_id = external_repo.id',
'external_repo_config ON external_repo.id = external_repo_config.external_repo_id'])
self.assertEqual(query.clauses,
['(external_repo_config.active = TRUE)',
'(tag_external_repos.active = TRUE)',
'external_repo.id = %(repo_id)i', 'tag.id = %(tag_id)i'])

61
tests/test_hub/test_get_user.py
View file

@ -5,38 +5,87 @@ import mock
import koji
import kojihub
QP = kojihub.QueryProcessor
class TestGetUser(unittest.TestCase):
def setUp(self):
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
self.queries.append(query)
return query
def tearDown(self):
mock.patch.stopall()
def test_wrong_format_user_info(self):
userinfo = ['test-user']
with self.assertRaises(koji.GenericError) as cm:
self.exports.getUser(userinfo)
self.assertEqual("Invalid type for userInfo: %s" % type(userinfo), str(cm.exception))
self.assertEqual(f"Invalid type for userInfo: {type(userinfo)}", str(cm.exception))
def test_wrong_format_userid(self):
userinfo = {'id': '123456'}
with self.assertRaises(koji.GenericError) as cm:
self.exports.getUser(userinfo)
self.assertEqual("Invalid type for userid: %s" % type(userinfo['id']), str(cm.exception))
self.assertEqual(f"Invalid type for userid: {type(userinfo['id'])}", str(cm.exception))
def test_wrong_format_username(self):
userinfo = {'name': 57896}
with self.assertRaises(koji.GenericError) as cm:
self.exports.getUser(userinfo)
self.assertEqual("Invalid type for username: %s" % type(userinfo['name']),
str(cm.exception))
self.assertEqual(f"Invalid type for username: {type(userinfo['name'])}", str(cm.exception))
def test_wrong_format_krb_principal(self):
userinfo = {'krb_principal': 57896}
with self.assertRaises(koji.GenericError) as cm:
self.exports.getUser(userinfo)
self.assertEqual("Invalid type for krb_principal: %s" % type(userinfo['krb_principal']),
self.assertEqual(f"Invalid type for krb_principal: {type(userinfo['krb_principal'])}",
str(cm.exception))
def test_not_logged_user(self):
self.context.session.user_id = None
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_user(userInfo=None)
self.assertEqual("No user provided", str(cm.exception))
def test_userinfo_string(self):
userinfo = 'test-user'
kojihub.get_user(userinfo, krb_princs=False)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['users'])
columns = ['id', 'name', 'status', 'usertype']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses, ['krb_principal = %(info)s OR name = %(info)s'])
self.assertEqual(query.joins, ['LEFT JOIN user_krb_principals ON '
'users.id = user_krb_principals.user_id'])
self.assertEqual(query.values, {'info': userinfo})
def test_userinfo_dict(self):
userinfo = {'id': 123456, 'name': 'test-user', 'krb_principal': 'test-krb@krb.com'}
kojihub.get_user(userinfo, krb_princs=False)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['users'])
columns = ['id', 'name', 'status', 'usertype']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses, ['user_krb_principals.krb_principal = %(krb_principal)s',
'users.id = %(id)i', 'users.name = %(name)s'])
self.assertEqual(query.joins, ['LEFT JOIN user_krb_principals ON '
'users.id = user_krb_principals.user_id'])
self.assertEqual(query.values, userinfo)
class TestGetUserByKrbPrincipal(unittest.TestCase):
def setUp(self):
@ -46,7 +95,7 @@ class TestGetUserByKrbPrincipal(unittest.TestCase):
krb_principal = ['test-user']
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_user_by_krb_principal(krb_principal)
self.assertEqual("Invalid type for krb_principal: %s" % type(krb_principal),
self.assertEqual(f"Invalid type for krb_principal: {type(krb_principal)}",
str(cm.exception))
def test_krb_principal_none(self):

7
tests/test_hub/test_group_operations.py
View file

@ -236,9 +236,10 @@ class TestGrouplist(unittest.TestCase):
# db
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
q = "SELECT blocked FROM group_config WHERE (active = TRUE) AND (group_id=%(grp_id)s) AND (tag_id=%(tag_id)s) FOR UPDATE"
actual = ' '.join(str(query).split())
self.assertEqual(actual, q)
self.assertEqual(query.tables, ['group_config'])
self.assertEqual(query.joins,None)
self.assertEqual(query.clauses,
['active = TRUE', 'group_id=%(grp_id)s', 'tag_id=%(tag_id)s'])
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.inserts), 0)

76
tests/test_hub/test_list_tags.py
View file

@ -5,43 +5,59 @@ import mock
import koji
import kojihub
QP = kojihub.QueryProcessor
class TestListTags(unittest.TestCase):
def setUp(self):
self.maxDiff = None
self.get_build = mock.patch('kojihub.get_build').start()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.cursor = mock.MagicMock()
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
def tearDown(self):
mock.patch.stopall()
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
def test_non_exist_build(self):
build_id = 999
self.get_build.return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTags(build=build_id)
self.assertEqual("No such build: %s" % build_id, str(cm.exception))
self.assertEqual(f"No such build: {build_id}", str(cm.exception))
build_name = 'test-build-1-23'
self.get_build.return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTags(build=build_name)
self.assertEqual("No such build: %s" % build_name, str(cm.exception))
self.assertEqual(f"No such build: {build_name}", str(cm.exception))
@mock.patch('kojihub.lookup_package')
def test_non_exist_package(self, lookup_package):
self.cursor.fetchone.return_value = None
self.context.cnx.cursor.return_value = self.cursor
lookup_package.side_effect = koji.GenericError('Expected error')
lookup_package.return_value = None
package_id = 999
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTags(package=package_id)
self.assertEqual('Expected error', str(cm.exception))
self.assertEqual(f'No such package: {package_id}', str(cm.exception))
package_name = 'test-pkg'
with self.assertRaises(koji.GenericError) as cm:
self.exports.listTags(package=package_name)
self.assertEqual("Expected error", str(cm.exception))
self.assertEqual(f'No such package: {package_name}', str(cm.exception))
def test_build_package_not_none(self):
build_id = 999
@ -51,3 +67,53 @@ class TestListTags(unittest.TestCase):
self.exports.listTags(build=build_id, package=package_id)
self.assertEqual("only one of build and package may be specified", str(cm.exception))
self.get_build.assert_not_called()
@mock.patch('kojihub.lookup_package')
def test_exist_package_and_perms(self, lookup_package):
package_info = {'id': 123, 'name': 'package-name'}
self.cursor.fetchone.return_value = None
self.context.cnx.cursor.return_value = self.cursor
lookup_package.return_value = package_info
self.exports.listTags(package=package_info['id'], perms=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_config'])
self.assertEqual(query.joins,
['tag ON tag.id = tag_config.tag_id',
'LEFT OUTER JOIN permissions ON tag_config.perm_id = permissions.id',
'tag_packages ON tag.id = tag_packages.tag_id',
'tag_package_owners ON\n '
'tag_packages.tag_id = tag_package_owners.tag_id '
'AND\n tag_packages.package_id = tag_package_owners.package_id AND\n'
' tag_package_owners.active IS TRUE',
'users ON tag_package_owners.owner = users.id'])
self.assertEqual(query.clauses, [
'tag_config.active = true',
'tag_packages.active = true',
'tag_packages.package_id = %(packageID)i',
])
def test_exist_build_and_pattern(self):
build_info = {'id': 123, 'name': 'package-build-1.23.0'}
pattern = 'package-build'
self.cursor.fetchone.return_value = None
self.context.cnx.cursor.return_value = self.cursor
self.get_build.return_value = build_info
self.exports.listTags(build=build_info['id'], pattern=pattern)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_config'])
self.assertEqual(query.joins,
['tag ON tag.id = tag_config.tag_id',
'LEFT OUTER JOIN permissions ON tag_config.perm_id = permissions.id',
'tag_listing ON tag.id = tag_listing.tag_id'])
self.assertEqual(query.clauses, [
'tag.name ILIKE %(pattern)s',
'tag_config.active = true',
'tag_listing.active = true',
'tag_listing.build_id = %(buildID)i',
])

49
tests/test_hub/test_list_user_krb_principals.py
View file

@ -5,14 +5,61 @@ import mock
import koji
import kojihub
QP = kojihub.QueryProcessor
class TestListUserKrbPrincipals(unittest.TestCase):
def setUp(self):
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.context').start()
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
self.queries.append(query)
return query
def tearDown(self):
mock.patch.stopall()
def test_wrong_format_user_info(self):
userinfo = ['test-user']
with self.assertRaises(koji.GenericError) as cm:
kojihub.list_user_krb_principals(userinfo)
self.assertEqual("Invalid type for user_info: %s" % type(userinfo), str(cm.exception))
self.assertEqual(f"Invalid type for user_info: {type(userinfo)}", str(cm.exception))
def test_not_logged_user(self):
self.context.session.user_id = None
with self.assertRaises(koji.GenericError) as cm:
kojihub.list_user_krb_principals(user_info=None)
self.assertEqual("No user provided", str(cm.exception))
def test_userinfo_string(self):
userinfo = 'test-user'
kojihub.list_user_krb_principals(userinfo)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['user_krb_principals'])
columns = ['krb_principal']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses, ['name = %(info)s'])
self.assertEqual(query.joins, ['users ON users.id = user_krb_principals.user_id'])
self.assertEqual(query.values, {'info': userinfo})
def test_userinfo_int(self):
userinfo = 123
kojihub.list_user_krb_principals(userinfo)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
self.assertEqual(query.tables, ['user_krb_principals'])
columns = ['krb_principal']
self.assertEqual(set(query.columns), set(columns))
self.assertEqual(query.clauses, ['user_id = %(info)i'])
self.assertEqual(query.joins, [])
self.assertEqual(query.values, {'info': userinfo})