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

Increase hub unit tests 03-02

Browse source
This commit is contained in:
Jana Cupova 2023-03-28 18:31:01 +02:00
parent 6d01a3f63d
commit bbd841c0d4
36 changed files with 1434 additions and 356 deletions
Download patch file Download diff file Expand all files Collapse all files

35
tests/test_hub/test_create_external_repo.py
View file

@ -6,12 +6,15 @@ import mock
import koji
import kojihub
IP = kojihub.InsertProcessor
class TestCreateExternalRepo(unittest.TestCase):
def setUp(self):
self.get_external_repos = mock.patch('kojihub.kojihub.get_external_repos').start()
self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start()
self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start()
self.context = mock.patch('kojihub.kojihub.context').start()
# It seems MagicMock will not automatically handle attributes that
# start with "assert"
@ -20,6 +23,17 @@ class TestCreateExternalRepo(unittest.TestCase):
self.repo_url = 'http://path_to_ext_repo.com'
self.repo_name = 'test-repo'
self.repo_info = {'id': 1, 'name': self.repo_name, 'url': self.repo_url}
self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',
side_effect=self.getInsert).start()
self.inserts = []
self.context_db = mock.patch('kojihub.db.context').start()
self.context_db.session.assertLogin = mock.MagicMock()
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
self.inserts.append(insert)
return insert
def tearDown(self):
mock.patch.stopall()
@ -45,3 +59,24 @@ class TestCreateExternalRepo(unittest.TestCase):
with self.assertRaises(koji.GenericError) as cm:
kojihub.create_external_repo(self.repo_name, self.repo_url)
self.assertEqual(expected, str(cm.exception))
def test_valid(self):
self.verify_name_internal.return_value = None
self.get_external_repos.return_value = None
self.context_db.event_id = 42
self.context_db.session.user_id = 23
self.get_external_repo_id.return_value = self.repo_info['id']
repo_url = self.repo_url + '/'
result = kojihub.create_external_repo(self.repo_name, self.repo_url)
self.assertEqual(result,
{'id': self.repo_info['id'], 'name': self.repo_name, 'url': repo_url})
self.assertEqual(len(self.inserts), 1)
insert = self.inserts[0]
self.assertEqual(insert.table, 'external_repo_config')
self.assertEqual(insert.data, {
'create_event': 42,
'creator_id': 23,
'external_repo_id': self.repo_info['id'],
'url': repo_url
})
self.assertEqual(insert.rawdata, {})

35
tests/test_hub/test_delete_build_target.py
View file

@ -6,15 +6,46 @@ import koji
import kojihub
UP = kojihub.UpdateProcessor
class TestDeleteBuildTarget(unittest.TestCase):
def getUpdate(self, *args, **kwargs):
update = UP(*args, **kwargs)
update.execute = mock.MagicMock()
self.updates.append(update)
return update
def setUp(self):
self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start()
self.lookup_build_target = mock.patch('kojihub.kojihub.lookup_build_target').start()
self.exports = kojihub.RootExports()
self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
self.context_db = mock.patch('kojihub.db.context').start()
# It seems MagicMock will not automatically handle attributes that
# start with "assert"
self.context_db.session.assertLogin = mock.MagicMock()
def test_non_exist_target(self):
build_target = 'build-target'
self.lookup_name.return_value = None
self.lookup_build_target.return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.deleteBuildTarget(build_target)
self.assertEqual("No such build target: %s" % build_target, str(cm.exception))
self.assertEqual(len(self.updates), 0)
def test_valid(self):
build_target = 'build-target'
self.context_db.event_id = 42
self.context_db.session.user_id = 23
self.lookup_build_target.return_value = {'id': 123}
self.exports.deleteBuildTarget(build_target)
self.assertEqual(len(self.updates), 1)
update = self.updates[0]
self.assertEqual(update.table, 'build_target_config')
self.assertEqual(update.clauses, ["build_target_id = %(targetID)i", 'active = TRUE'])
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})
self.assertEqual(update.rawdata, {'active': 'NULL'})

55
tests/test_hub/test_delete_external_repo.py Normal file
View file

@ -0,0 +1,55 @@
import unittest
import mock
import kojihub
UP = kojihub.UpdateProcessor
class TestDeleteExternalRepo(unittest.TestCase):
def getUpdate(self, *args, **kwargs):
update = UP(*args, **kwargs)
update.execute = mock.MagicMock()
self.updates.append(update)
return update
def setUp(self):
self.context = mock.patch('kojihub.kojihub.context').start()
self.context.session.assertPerm = mock.MagicMock()
self.context_db = mock.patch('kojihub.db.context').start()
self.context_db.session.assertLogin = mock.MagicMock()
self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()
self.get_tag_external_repos = mock.patch('kojihub.kojihub.get_tag_external_repos').start()
self.remove_external_repo_from_tag = mock.patch(
'kojihub.kojihub.remove_external_repo_from_tag').start()
def tearDown(self):
mock.patch.stopall()
def test_valid(self):
tag_ids = [23, 25]
repo_id = 123
self.context_db.event_id = 42
self.context_db.session.user_id = 23
self.get_external_repo.return_value = {'id': repo_id}
self.get_tag_external_repos.return_value = [{'id': repo_id, 'tag_id': tag_ids[0]},
{'id': repo_id, 'tag_id': tag_ids[0]}]
self.remove_external_repo_from_tag.side_effect = [None, None]
kojihub.delete_external_repo(repo_id)
self.assertEqual(len(self.updates), 1)
update = self.updates[0]
self.assertEqual(update.table, 'external_repo_config')
self.assertEqual(update.clauses, ['external_repo_id = %(repo_id)i', 'active = TRUE'])
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})
self.assertEqual(update.rawdata, {'active': 'NULL'})
self.get_external_repo.assert_called_once_with(123, strict=True)
self.get_tag_external_repos.assert_called_once_with(repo_info=123)
self.remove_external_repo_from_tag.has_calls(
[mock.call(tag_info=tag_ids[0], repo_info=repo_id),
mock.call(tag_info=tag_ids[1], repo_info=repo_id)])

102
tests/test_hub/test_edit_external_repo.py
View file

@ -6,9 +6,33 @@ import mock
import koji
import kojihub
IP = kojihub.InsertProcessor
QP = kojihub.QueryProcessor
UP = kojihub.UpdateProcessor
class TestEditExternalRepo(unittest.TestCase):
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
query.singleValue = self.query_singleValue
self.queries.append(query)
return query
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
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 setUp(self):
self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()
self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start()
@ -20,6 +44,20 @@ class TestEditExternalRepo(unittest.TestCase):
self.repo_url = 'http://path_to_ext_repo.com'
self.repo_name = 'test-repo'
self.repo_info = {'id': 1, 'name': self.repo_name, 'url': self.repo_url}
self.context_db = mock.patch('kojihub.db.context').start()
# It seems MagicMock will not automatically handle attributes that
# start with "assert"
self.context_db.session.assertLogin = mock.MagicMock()
self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',
side_effect=self.getInsert).start()
self.inserts = []
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.query_singleValue = mock.MagicMock()
self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
def tearDown(self):
mock.patch.stopall()
@ -37,3 +75,67 @@ class TestEditExternalRepo(unittest.TestCase):
self.verify_name_internal.side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
kojihub.edit_external_repo(self.repo_name, repo_name_new, self.repo_url)
def test_edit_external_repo_existing_name(self):
repo_name_new = 'test-repo+'
self.get_external_repo.return_value = self.repo_info
self.verify_name_internal.side_effect = None
self.query_singleValue.return_value = 111
with self.assertRaises(koji.GenericError) as ex:
kojihub.edit_external_repo(self.repo_name, repo_name_new, self.repo_url)
self.assertEqual(f'name "{repo_name_new}" is already taken by external repo {111}',
str(ex.exception))
self.assertEqual(len(self.queries), 1)
self.assertEqual(len(self.inserts), 0)
self.assertEqual(len(self.updates), 0)
query = self.queries[0]
self.assertEqual(query.tables, ['external_repo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['name = %(name)s'])
self.assertEqual(query.columns, ['id'])
self.assertEqual(query.values, {'name': repo_name_new})
def test_edit_valid(self):
path_to_new_repo = 'http://path_to_new_ext_repo.com'
repo_name_new = 'test-repo+'
self.get_external_repo.return_value = self.repo_info
self.verify_name_internal.side_effect = None
self.query_singleValue.return_value = None
self.context_db.event_id = 42
self.context_db.session.user_id = 23
kojihub.edit_external_repo(self.repo_name, repo_name_new, path_to_new_repo)
self.assertEqual(len(self.queries), 1)
self.assertEqual(len(self.inserts), 1)
self.assertEqual(len(self.updates), 2)
query = self.queries[0]
self.assertEqual(query.tables, ['external_repo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['name = %(name)s'])
self.assertEqual(query.columns, ['id'])
self.assertEqual(query.values, {'name': repo_name_new})
insert = self.inserts[0]
self.assertEqual(insert.table, 'external_repo_config')
self.assertEqual(insert.data, {
'create_event': 42,
'creator_id': 23,
'external_repo_id': self.repo_info['id'],
'url': path_to_new_repo + '/'
})
self.assertEqual(insert.rawdata, {})
update = self.updates[0]
self.assertEqual(update.table, 'external_repo')
self.assertEqual(update.values, {'repo_id': self.repo_info['id']})
self.assertEqual(update.clauses, ['id = %(repo_id)i'])
self.assertEqual(update.data, {'name': repo_name_new})
self.assertEqual(update.rawdata, {})
update = self.updates[1]
self.assertEqual(update.table, 'external_repo_config')
self.assertEqual(update.clauses, ['external_repo_id = %(repo_id)i', 'active = TRUE'])
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})
self.assertEqual(update.rawdata, {'active': 'NULL'})

5
tests/test_hub/test_getPackageID.py
View file

@ -10,7 +10,7 @@ class TestGetPackageID(DBQueryTestCase):
maxDiff = None
def test_getPackageID(self):
self.qp_execute_return_value = [{'id': 1}]
self.qp_execute_one_return_value = {'id': 1}
rv = kojihub.RootExports().getPackageID('koji')
self.assertEqual(len(self.queries), 1)
self.assertLastQueryEqual(tables=['package'],
@ -30,8 +30,7 @@ class TestGetPackageID(DBQueryTestCase):
values={'name': 'invalidpkg',
'strict': True,
'self': mock.ANY})
self.assertEqual(cm.exception.args[0],
'No such package name: invalidpkg')
self.assertEqual(cm.exception.args[0], 'No such package name: invalidpkg')
def test_getPackageID_None(self):
rv = kojihub.RootExports().getPackageID('invalidpkg')

63
tests/test_hub/test_getRPM.py
View file

@ -1,33 +1,21 @@
import os.path
import shutil
import tempfile
import unittest
import mock
import unittest
import koji
import kojihub
QP = kojihub.QueryProcessor
from .utils import DBQueryTestCase
class TestGetRPM(unittest.TestCase):
class TestGetRPM(DBQueryTestCase):
def setUp(self):
super(TestGetRPM, self).setUp()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.kojihub.context').start()
self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start()
self.QueryProcessor = mock.patch('kojihub.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']
@ -37,7 +25,9 @@ class TestGetRPM(unittest.TestCase):
def test_rpm_info_int(self):
rpminfo = 123
kojihub.get_rpm(rpminfo)
self.qp_execute_one_return_value = {'rpminfo.id': 123}
result = kojihub.get_rpm(rpminfo)
self.assertEqual(result, {'rpminfo.id': 123})
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
@ -52,6 +42,45 @@ class TestGetRPM(unittest.TestCase):
['external_repo ON rpminfo.external_repo_id = external_repo.id'])
self.assertEqual(query.values, {'id': rpminfo})
def test_rpm_info_empty_query_result_without_strict(self):
rpminfo = 123
self.qp_execute_one_return_value = {}
result = kojihub.get_rpm(rpminfo)
self.assertEqual(result, None)
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.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_empty_query_result_with_strict(self):
rpminfo = 123
self.qp_execute_one_return_value = {}
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_rpm(rpminfo, strict=True)
self.assertEqual("No such rpm: {'id': 123}", str(ex.exception))
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.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)

37
tests/test_hub/test_getRPMDeps.py
View file

@ -6,49 +6,50 @@ import kojihub
class TestGetRPMDeps(unittest.TestCase):
def setUp(self):
self.exports = kojihub.RootExports()
self.get_rpm = mock.patch('kojihub.kojihub.get_rpm').start()
self.get_build = mock.patch('kojihub.kojihub.get_build').start()
@mock.patch('kojihub.kojihub.get_rpm')
def test_getRPMDeps_no_rpminfo(self, get_rpm):
def test_getRPMDeps_no_rpminfo(self):
def mock_get_rpm(rpmID, strict=False):
if strict:
raise koji.GenericError('msg')
else:
return None
get_rpm.side_effect = mock_get_rpm
re = kojihub.RootExports().getRPMDeps(1)
self.get_rpm.side_effect = mock_get_rpm
re = self.exports.getRPMDeps(1)
self.assertEqual(re, [])
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMDeps(1, strict=True)
self.exports.getRPMDeps(1, strict=True)
self.assertEqual(cm.exception.args[0], 'msg')
@mock.patch('kojihub.kojihub.get_rpm', return_value={'id': 1, 'build_id': None})
def test_getRPMDeps_external_rpm(self, get_rpm):
re = kojihub.RootExports().getRPMDeps(1)
def test_getRPMDeps_external_rpm(self):
self.get_rpm.return_value = {'id': 1, 'build_id': None}
re = self.exports.getRPMDeps(1)
self.assertEqual(re, [])
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMDeps(1, strict=True)
self.exports.getRPMDeps(1, strict=True)
self.assertEqual(cm.exception.args[0],
'Can not get dependencies, because RPM: 1 is not internal')
@mock.patch('kojihub.kojihub.get_rpm', return_value={'id': 1, 'build_id': 1})
@mock.patch('kojihub.kojihub.get_build', return_value={'id': 1})
@mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
@mock.patch('koji.pathinfo.rpm', return_value='fakerpmrelpath')
@mock.patch('os.path.exists', return_value=False)
def test_getRPMDeps_no_rpmfile(self, ope, pr, pb, get_build, get_rpm):
re = kojihub.RootExports().getRPMDeps(1)
def test_getRPMDeps_no_rpmfile(self, ope, pr, pb):
self.get_rpm.return_value = {'id': 1, 'build_id': 1}
self.get_build.return_value = {'id': 1}
re = self.exports.getRPMDeps(1)
self.assertEqual(re, [])
with self.assertRaises(koji.GenericError) as cm:
kojihub.RootExports().getRPMDeps(1, strict=True)
self.exports.getRPMDeps(1, strict=True)
self.assertEqual(cm.exception.args[0], "RPM file of 1 doesn't exist")
@mock.patch('kojihub.kojihub.get_rpm')
@mock.patch('kojihub.kojihub.get_build')
@mock.patch('koji.pathinfo')
def test_getRPMDeps(self, pi, build, rpm):
def test_getRPMDeps(self, pi):
pi.build.return_value = os.path.join(os.path.dirname(__file__), '../test_lib/data/rpms')
pi.rpm.return_value = 'test-deps-1-1.fc24.x86_64.rpm'
getRPMDeps = kojihub.RootExports().getRPMDeps
getRPMDeps = self.exports.getRPMDeps
res = getRPMDeps('')
# limit test for rpm < 4.12
if any(koji.SUPPORTED_OPT_DEP_HDRS.values()):

2
tests/test_hub/test_get_active_repos.py
View file

@ -25,7 +25,7 @@ class TestGetActiveRepos(unittest.TestCase):
def test_get_active_repos(self):
# currently not really a lot of parameters to test
result = kojihub.get_active_repos()
kojihub.get_active_repos()
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
# make sure the following does not error

188
tests/test_hub/test_get_archive_type.py
View file

@ -1,20 +1,190 @@
import unittest
import koji
import kojihub
import mock
from .utils import DBQueryTestCase
class TestGetArchiveType(unittest.TestCase):
class TestGetArchiveType(DBQueryTestCase):
def setUp(self):
super(TestGetArchiveType, self).setUp()
self.maxDiff = None
self.archive_info = {'id': 1, 'name': 'archive-type-1',
'description': 'archive-desc', 'extensions': 'ext',
'compression_type': 'cmptype'}
def test_get_archive_wrong_type_filename(self):
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_wrong_type_filename(
self, get_archive_type_by_id, get_archive_type_by_name):
filename = ['test-filename']
with self.assertRaises(koji.ParameterError) as cm:
with self.assertRaises(koji.ParameterError) as ex:
kojihub.get_archive_type(filename=filename)
self.assertEqual(f"Invalid type for value '{filename}': {type(filename)}, "
f"expected type <class 'str'>", str(cm.exception))
f"expected type <class 'str'>", str(ex.exception))
self.assertEqual(len(self.queries), 0)
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_not_called()
def test_get_archive_without_opt(self):
with self.assertRaises(koji.GenericError) as cm:
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_without_opt(self, get_archive_type_by_id, get_archive_type_by_name):
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_archive_type()
self.assertEqual("one of filename, type_name, or type_id must be specified",
str(cm.exception))
str(ex.exception))
self.assertEqual(len(self.queries), 0)
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_not_called()
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_type_id(self, get_archive_type_by_id, get_archive_type_by_name):
get_archive_type_by_id.return_value = self.archive_info
kojihub.get_archive_type(type_id=1)
self.assertEqual(len(self.queries), 0)
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_called_once_with(1, False)
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_type_name(self, get_archive_type_by_id, get_archive_type_by_name):
get_archive_type_by_name.return_value = self.archive_info
kojihub.get_archive_type(type_name='archive-type-1')
self.assertEqual(len(self.queries), 0)
get_archive_type_by_name.assert_called_once_with('archive-type-1', False)
get_archive_type_by_id.assert_not_called()
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_more_files_extension(
self, get_archive_type_by_id, get_archive_type_by_name):
archive_info = [{'id': 1, 'name': 'archive-type-1', 'extensions': 'ext'},
{'id': 2, 'name': 'archive-type-2', 'extensions': 'ext'}]
filename = 'test-filename.ext'
self.qp_execute_return_value = archive_info
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_archive_type(filename=filename)
self.assertEqual("multiple matches for file extension: ext", str(ex.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_not_called()
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_no_extensions_with_strict(
self, get_archive_type_by_id, get_archive_type_by_name):
archive_info = []
filename = 'test-filename.ext'
self.qp_execute_return_value = archive_info
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_archive_type(filename=filename, strict=True)
self.assertEqual(f'unsupported file extension: {filename}', str(ex.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_not_called()
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_no_extensions_without_strict(
self, get_archive_type_by_id, get_archive_type_by_name):
archive_info = []
filename = 'test-filename.ext'
self.qp_execute_return_value = archive_info
result = kojihub.get_archive_type(filename=filename, strict=False)
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_not_called()
@mock.patch('kojihub.kojihub._get_archive_type_by_name')
@mock.patch('kojihub.kojihub._get_archive_type_by_id')
def test_get_archive_valid(self, get_archive_type_by_id, get_archive_type_by_name):
filename = 'test-filename.ext'
self.qp_execute_return_value = [self.archive_info]
result = kojihub.get_archive_type(filename=filename)
self.assertEqual(result, self.archive_info)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
get_archive_type_by_name.assert_not_called()
get_archive_type_by_id.assert_not_called()
def test_get_archive_type_by_id_empty_without_strict(self):
self.qp_execute_one_return_value = None
result = kojihub._get_archive_type_by_id(1, False)
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['id = %(type_id)i'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
self.assertEqual(query.values, {'type_id': 1})
def test_get_archive_type_by_id_empty_with_strict(self):
self.qp_execute_one_side_effect = koji.GenericError('query returned no rows')
with self.assertRaises(koji.GenericError) as ex:
kojihub._get_archive_type_by_id(1, True)
self.assertEqual('query returned no rows', str(ex.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['id = %(type_id)i'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
self.assertEqual(query.values, {'type_id': 1})
def test_get_archive_type_by_name_without_strict(self):
self.qp_execute_one_return_value = None
result = kojihub._get_archive_type_by_name('archive-type', False)
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['name = %(name)s'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
self.assertEqual(query.values, {'name': 'archive-type'})
def test_get_archive_type_by_name_with_strict(self):
self.qp_execute_one_side_effect = koji.GenericError('query returned no rows')
with self.assertRaises(koji.GenericError) as ex:
kojihub._get_archive_type_by_name('archive-type', True)
self.assertEqual('query returned no rows', str(ex.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['archivetypes'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['name = %(name)s'])
self.assertEqual(query.columns,
['compression_type', 'description', 'extensions', 'id', 'name'])
self.assertEqual(query.values, {'name': 'archive-type'})

116
tests/test_hub/test_get_build.py
View file

@ -10,22 +10,69 @@ class TestGetBuild(DBQueryTestCase):
def setUp(self):
super(TestGetBuild, self).setUp()
self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()
self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start()
def test_non_exist_build_string(self):
def test_non_exist_build_string_with_strict(self):
build = 'build-1-23'
self.find_build_id.side_effect = koji.GenericError('No such build: %s' % build)
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_build(build, strict=True)
self.assertEqual('No such build: %s' % build, str(cm.exception))
self.find_build_id.assert_called_once_with(build, strict=True)
self.assertEqual(len(self.queries), 0)
def test_non_exist_build_int(self):
def test_non_exist_build_int_without_result_with_strict(self):
build = 11
self.find_build_id.return_value = build
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_build(build, strict=True)
self.assertEqual('No such build: %s' % build, str(cm.exception))
self.find_build_id.assert_called_once_with(build, strict=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build'])
self.assertEqual(query.joins, ['events ON build.create_event = events.id',
'package on build.pkg_id = package.id',
'volume on build.volume_id = volume.id',
'users on build.owner = users.id'])
self.assertEqual(query.clauses, ['build.id = %(buildID)i'])
self.assertEqual(query.columns,
['build.id', 'build.cg_id', 'build.completion_time',
"date_part('epoch', build.completion_time)", 'events.id', 'events.time',
"date_part('epoch', events.time)", 'build.epoch', 'build.extra',
'build.id', 'package.name',
"package.name || '-' || build.version || '-' || build.release",
'users.id', 'users.name', 'package.id', 'package.name', 'build.release',
'build.source', 'build.start_time',
"date_part('epoch', build.start_time)", 'build.state', 'build.task_id',
'build.version', 'volume.id', 'volume.name'])
def test_non_exist_build_dict(self):
def test_non_exist_build_int_without_result_without_strict(self):
build = 11
self.find_build_id.return_value = build
result = kojihub.get_build(build, strict=False)
self.assertEqual(result, None)
self.find_build_id.assert_called_once_with(build, strict=False)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build'])
self.assertEqual(query.joins, ['events ON build.create_event = events.id',
'package on build.pkg_id = package.id',
'volume on build.volume_id = volume.id',
'users on build.owner = users.id'])
self.assertEqual(query.clauses, ['build.id = %(buildID)i'])
self.assertEqual(query.columns,
['build.id', 'build.cg_id', 'build.completion_time',
"date_part('epoch', build.completion_time)", 'events.id', 'events.time',
"date_part('epoch', events.time)", 'build.epoch', 'build.extra',
'build.id', 'package.name',
"package.name || '-' || build.version || '-' || build.release",
'users.id', 'users.name', 'package.id', 'package.name', 'build.release',
'build.source', 'build.start_time',
"date_part('epoch', build.start_time)", 'build.state', 'build.task_id',
'build.version', 'volume.id', 'volume.name'])
def test_non_exist_build_dict_with_strict(self):
build = {
'name': 'test_name',
'version': 'test_version',
@ -35,3 +82,66 @@ class TestGetBuild(DBQueryTestCase):
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_build(build, strict=True)
self.assertEqual('No such build: %s' % build['name'], str(cm.exception))
self.find_build_id.assert_called_once_with(build, strict=True)
self.assertEqual(len(self.queries), 0)
def test_build_none_exist_without_strict(self):
build = 'build-1-23'
self.find_build_id.return_value = None
result = kojihub.get_build(build, strict=False)
self.assertEqual(result, None)
self.find_build_id.assert_called_once_with(build, strict=False)
self.assertEqual(len(self.queries), 0)
def test_result_with_cg_id_none(self):
build = 11
self.find_build_id.return_value = build
self.qp_execute_one_return_value = {'cg_id': None}
result = kojihub.get_build(build, strict=True)
self.assertEqual(result, {'cg_id': None, 'cg_name': None})
self.find_build_id.assert_called_once_with(build, strict=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build'])
self.assertEqual(query.joins, ['events ON build.create_event = events.id',
'package on build.pkg_id = package.id',
'volume on build.volume_id = volume.id',
'users on build.owner = users.id'])
self.assertEqual(query.clauses, ['build.id = %(buildID)i'])
self.assertEqual(query.columns,
['build.id', 'build.cg_id', 'build.completion_time',
"date_part('epoch', build.completion_time)", 'events.id', 'events.time',
"date_part('epoch', events.time)", 'build.epoch', 'build.extra',
'build.id', 'package.name',
"package.name || '-' || build.version || '-' || build.release",
'users.id', 'users.name', 'package.id', 'package.name', 'build.release',
'build.source', 'build.start_time',
"date_part('epoch', build.start_time)", 'build.state', 'build.task_id',
'build.version', 'volume.id', 'volume.name'])
def test_result_with_cg_id(self):
build = 11
self.find_build_id.return_value = build
self.lookup_name.return_value = {'name': 'cg_name'}
self.qp_execute_one_return_value = {'cg_id': 1}
result = kojihub.get_build(build, strict=True)
self.assertEqual(result, {'cg_id': 1, 'cg_name': 'cg_name'})
self.find_build_id.assert_called_once_with(build, strict=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build'])
self.assertEqual(query.joins, ['events ON build.create_event = events.id',
'package on build.pkg_id = package.id',
'volume on build.volume_id = volume.id',
'users on build.owner = users.id'])
self.assertEqual(query.clauses, ['build.id = %(buildID)i'])
self.assertEqual(query.columns,
['build.id', 'build.cg_id', 'build.completion_time',
"date_part('epoch', build.completion_time)", 'events.id', 'events.time',
"date_part('epoch', events.time)", 'build.epoch', 'build.extra',
'build.id', 'package.name',
"package.name || '-' || build.version || '-' || build.release",
'users.id', 'users.name', 'package.id', 'package.name', 'build.release',
'build.source', 'build.start_time',
"date_part('epoch', build.start_time)", 'build.state', 'build.task_id',
'build.version', 'volume.id', 'volume.name'])

48
tests/test_hub/test_get_build_notification.py
View file

@ -1,22 +1,52 @@
import mock
import unittest
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetBuildNotification(unittest.TestCase):
class TestGetBuildNotification(DBQueryTestCase):
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()
self.query = self.QueryProcessor.return_value
super(TestGetBuildNotification, self).setUp()
self.exports = kojihub.RootExports()
def tearDown(self):
mock.patch.stopall()
def test_empty_result_with_strict(self):
notif_id = 1
self.query.executeOne.return_value = None
self.qp_execute_one_return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.getBuildNotification(notif_id, strict=True)
self.assertEqual(f"No notification with ID {notif_id} found", str(cm.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build_notifications'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['id = %(id)i'])
self.assertEqual(query.columns,
['email', 'id', 'package_id', 'success_only', 'tag_id', 'user_id'])
def test_empty_result_without_strict(self):
notif_id = 1
self.qp_execute_one_return_value = None
result = self.exports.getBuildNotification(notif_id, strict=False)
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build_notifications'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['id = %(id)i'])
self.assertEqual(query.columns,
['email', 'id', 'package_id', 'success_only', 'tag_id', 'user_id'])
def test_valid_result(self):
qp_result = {'id': 1, 'user_id': 2, 'package_id': 3, 'tag_id': 4, 'success_only': True,
'email': 'test-mail'}
notif_id = 1
self.qp_execute_one_return_value = qp_result
result = self.exports.getBuildNotification(notif_id, strict=True)
self.assertEqual(result, qp_result)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build_notifications'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['id = %(id)i'])
self.assertEqual(query.columns,
['email', 'id', 'package_id', 'success_only', 'tag_id', 'user_id'])

29
tests/test_hub/test_get_build_notification_block.py
View file

@ -1,14 +1,13 @@
import mock
import unittest
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetBuildNotificationBlock(unittest.TestCase):
class TestGetBuildNotificationBlock(DBQueryTestCase):
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()
self.query = self.QueryProcessor.return_value
super(TestGetBuildNotificationBlock, self).setUp()
self.exports = kojihub.RootExports()
def tearDown(self):
@ -16,7 +15,27 @@ class TestGetBuildNotificationBlock(unittest.TestCase):
def test_empty_result_with_strict(self):
notif_id = 1
self.query.executeOne.return_value = None
self.qp_execute_one_return_value = None
with self.assertRaises(koji.GenericError) as cm:
self.exports.getBuildNotificationBlock(notif_id, strict=True)
self.assertEqual(f"No notification block with ID {notif_id} found", str(cm.exception))
def test_empty_result_without_strict(self):
notif_id = 1
self.qp_execute_one_return_value = None
result = self.exports.getBuildNotificationBlock(notif_id, strict=False)
self.assertEqual(result, None)
def test_valid_result(self):
qp_result = {'id': 1, 'user_id': 2, 'package_id': 3, 'tag_id': 4}
notif_id = 1
self.qp_execute_one_return_value = qp_result
result = self.exports.getBuildNotificationBlock(notif_id, strict=True)
self.assertEqual(result, qp_result)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build_notifications_block'])
self.assertEqual(query.joins, None)
self.assertEqual(query.values, {'id': 1})
self.assertEqual(query.clauses, ['id = %(id)i'])
self.assertEqual(query.columns, ['id', 'package_id', 'tag_id', 'user_id'])

21
tests/test_hub/test_get_build_targets.py
View file

@ -1,40 +1,25 @@
import unittest
import mock
import kojihub
QP = kojihub.QueryProcessor
from .utils import DBQueryTestCase
class TestGetBuildTargets(unittest.TestCase):
class TestGetBuildTargets(DBQueryTestCase):
def setUp(self):
super(TestGetBuildTargets, self).setUp()
self.maxDiff = None
self.name_or_id_clause = mock.patch('kojihub.kojihub.name_or_id_clause').start()
self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.kojihub.context').start()
self.cursor = mock.MagicMock()
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.build_target = 'build-target'
self.build_tag_name = 'tag'
self.dest_tag_name = 'dest-tag'
self.build_tag_id = 1
self.dest_tag_id = 2
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_strings(self):
self.name_or_id_clause.return_value = '(build_target.name = %(build_target_name)s)', \
{'build_target_name': 'build-target-url'}

22
tests/test_hub/test_get_build_type.py
View file

@ -1,33 +1,19 @@
import mock
import unittest
import kojihub
QP = kojihub.QueryProcessor
from .utils import DBQueryTestCase
class TestGetBuildType(unittest.TestCase):
class TestGetBuildType(DBQueryTestCase):
def setUp(self):
super(TestGetBuildType, self).setUp()
self.maxDiff = None
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.query_execute = mock.MagicMock()
self.get_build = mock.patch('kojihub.kojihub.get_build').start()
self.get_maven_build = mock.patch('kojihub.kojihub.get_maven_build').start()
self.get_win_build = mock.patch('kojihub.kojihub.get_win_build').start()
self.get_image_build = mock.patch('kojihub.kojihub.get_image_build').start()
def tearDown(self):
mock.patch.stopall()
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = self.query_execute
self.queries.append(query)
return query
def test_no_build(self):
self.get_build.return_value = None
@ -47,7 +33,7 @@ class TestGetBuildType(unittest.TestCase):
self.get_win_build.return_value = typeinfo['win']
self.get_image_build.return_value = typeinfo['image']
self.query_execute.return_value = [['new_type']]
self.qp_execute_return_value = [['new_type']]
ret = kojihub.get_build_type('mytestbuild-1-1', strict=True)
assert ret == typeinfo

25
tests/test_hub/test_get_channel.py
View file

@ -1,33 +1,17 @@
import unittest
import mock
import koji
import kojihub
from .utils import DBQueryTestCase
QP = kojihub.QueryProcessor
class TestGetChannel(unittest.TestCase):
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
self.queries.append(query)
return query
class TestGetChannel(DBQueryTestCase):
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
super(TestGetChannel, self).setUp()
self.context = mock.patch('kojihub.kojihub.context').start()
self.exports = kojihub.RootExports()
def tearDown(self):
mock.patch.stopall()
def test_wrong_type_channelInfo(self):
# dict
channel_info = {'channel': 'val'}
@ -54,7 +38,6 @@ class TestGetChannel(unittest.TestCase):
self.assertEqual(set(query.clauses), set(clauses))
self.assertEqual(query.values, values)
def test_query_by_id(self):
self.exports.getChannel(12345)
self.assertEqual(len(self.queries), 1)
@ -67,7 +50,7 @@ class TestGetChannel(unittest.TestCase):
self.assertEqual(query.values, values)
def test_query_by_dict(self):
self.exports.getChannel({'id':12345, 'name': 'whatever'})
self.exports.getChannel({'id': 12345, 'name': 'whatever'})
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
clauses = ['(channels.id = %(channels_id)s)']

34
tests/test_hub/test_get_external_repo_list.py Normal file
View file

@ -0,0 +1,34 @@
import unittest
import mock
import kojihub
class TestGetExternalRepo(unittest.TestCase):
def setUp(self):
self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()
self.read_full_inheritance = mock.patch('kojihub.kojihub.readFullInheritance').start()
self.get_tag_external_repos = mock.patch('kojihub.kojihub.get_tag_external_repos').start()
def tearDown(self):
mock.patch.stopall()
def test_valid(self):
self.get_tag.return_value = {'id': 123}
self.read_full_inheritance.return_value = [{'parent_id': 111}, {'parent_id': 112}]
self.get_tag_external_repos.side_effect = [
[{'external_repo_id': 10, 'tag_id': 123}],
[{'external_repo_id': 11, 'tag_id': 111}, {'external_repo_id': 12, 'tag_id': 111}],
[{'external_repo_id': 13, 'tag_id': 112}]]
result = kojihub.get_external_repo_list('test-tag')
self.assertEqual(result, [
{'external_repo_id': 10, 'tag_id': 123},
{'external_repo_id': 11, 'tag_id': 111}, {'external_repo_id': 12, 'tag_id': 111},
{'external_repo_id': 13, 'tag_id': 112}])
self.get_tag.assert_called_once_with('test-tag', strict=True, event=None)
self.read_full_inheritance.assert_called_once_with(123, None)
self.get_tag_external_repos.assert_has_calls(
[mock.call(tag_info=123, event=None), mock.call(tag_info=111, event=None),
mock.call(tag_info=112, event=None)])

23
tests/test_hub/test_get_group_members.py
View file

@ -1,16 +1,14 @@
import unittest
import mock
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetGroupMembers(unittest.TestCase):
class TestGetGroupMembers(DBQueryTestCase):
def setUp(self):
self.context = mock.patch('kojihub.kojihub.context').start()
self.context.session.assertPerm = mock.MagicMock()
super(TestGetGroupMembers, self).setUp()
self.get_user = mock.patch('kojihub.kojihub.get_user').start()
self.exports = kojihub.RootExports()
@ -20,3 +18,18 @@ class TestGetGroupMembers(unittest.TestCase):
with self.assertRaises(koji.GenericError) as cm:
self.exports.getGroupMembers(group)
self.assertEqual("No such group: %s" % group, str(cm.exception))
self.assertEqual(len(self.queries), 0)
def test_valid(self):
group = 'test-group'
self.get_user.return_value = {'id': 23, 'usertype': 2}
self.exports.getGroupMembers(group)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['user_groups'])
self.assertEqual(query.joins, ['JOIN users ON user_groups.user_id = users.id',
'LEFT JOIN user_krb_principals'
' ON users.id = user_krb_principals.user_id'])
self.assertEqual(query.clauses, ['(active = TRUE)', 'group_id = %(group_id)i'])
self.assertEqual(query.values, {'group_id': 23})
self.assertEqual(query.columns, ['id', 'array_agg(krb_principal)', 'name', 'usertype'])

41
tests/test_hub/test_get_image_archive.py Normal file
View file

@ -0,0 +1,41 @@
import kojihub
from .utils import DBQueryTestCase
class TestGetImageArchive(DBQueryTestCase):
def setUp(self):
super(TestGetImageArchive, self).setUp()
self.maxDiff = None
def test_not_exist_image_archive(self):
self.qp_execute_one_return_value = {}
result = kojihub.get_image_archive(123)
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['image_archives'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])
self.assertEqual(query.values, {'archive_id': 123})
self.assertEqual(query.columns, ['arch', 'archive_id'])
def test_valid(self):
self.qp_execute_one_side_effect = [{'archive_id': 123, 'arch': 'arch'},
{'rpm_id': 1}]
result = kojihub.get_image_archive(123)
self.assertEqual(result, {'archive_id': 123, 'arch': 'arch', 'rootid': True})
self.assertEqual(len(self.queries), 2)
query = self.queries[0]
self.assertEqual(query.tables, ['image_archives'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])
self.assertEqual(query.values, {'archive_id': 123})
self.assertEqual(query.columns, ['arch', 'archive_id'])
query = self.queries[1]
self.assertEqual(query.tables, ['archive_rpm_components'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])
self.assertEqual(query.values, {'archive_id': 123})
self.assertEqual(query.columns, ['rpm_id'])

64
tests/test_hub/test_get_image_build.py Normal file
View file

@ -0,0 +1,64 @@
import mock
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetImageBuild(DBQueryTestCase):
def setUp(self):
super(TestGetImageBuild, self).setUp()
self.maxDiff = None
self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()
def test_build_id_not_found(self):
self.find_build_id.return_value = None
result = kojihub.get_image_build('test-build.1-23.1')
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 0)
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)
def test_valid(self):
self.find_build_id.return_value = 123
self.qp_execute_one_return_value = {'build_id': 123}
kojihub.get_image_build('test-build.1-23.1')
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['image_builds'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(build_id)i'])
self.assertEqual(query.values, {'build_id': 123})
self.assertEqual(query.columns, ['build_id'])
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)
def test_without_result_without_strict(self):
self.find_build_id.return_value = 123
self.qp_execute_one_return_value = {}
result = kojihub.get_image_build('test-build.1-23.1')
self.assertEqual(result, {})
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['image_builds'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(build_id)i'])
self.assertEqual(query.values, {'build_id': 123})
self.assertEqual(query.columns, ['build_id'])
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)
def test_without_result_with_strict(self):
self.find_build_id.return_value = 123
self.qp_execute_one_return_value = {}
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_image_build('test-build.1-23.1', strict=True)
self.assertEqual('no such image build: test-build.1-23.1', str(ex.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['image_builds'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(build_id)i'])
self.assertEqual(query.values, {'build_id': 123})
self.assertEqual(query.columns, ['build_id'])
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=True)

17
tests/test_hub/test_get_last_event.py
View file

@ -1,12 +1,12 @@
import unittest
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetLastEvent(unittest.TestCase):
class TestGetLastEvent(DBQueryTestCase):
def setUp(self):
super(TestGetLastEvent, self).setUp()
self.exports = kojihub.RootExports()
def test_wrong_type_before(self):
@ -14,3 +14,14 @@ class TestGetLastEvent(unittest.TestCase):
with self.assertRaises(koji.GenericError) as cm:
self.exports.getLastEvent(before)
self.assertEqual("Invalid type for before: %s" % type(before), str(cm.exception))
def test_valid(self):
before = 123
self.exports.getLastEvent(before)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['events'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ["date_part('epoch', time) < %(before)r"])
self.assertEqual(query.values, {'before': 123})
self.assertEqual(query.columns, ['id', "date_part('epoch', time)"])

28
tests/test_hub/test_get_last_host_update.py
View file

@ -1,26 +1,15 @@
import unittest
import mock
import datetime
import sys
import kojihub
QP = kojihub.QueryProcessor
from .utils import DBQueryTestCase
class TestGetLastHostUpdate(unittest.TestCase):
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.singleValue = self.query_singleValue
self.queries.append(query)
return query
class TestGetLastHostUpdate(DBQueryTestCase):
def setUp(self):
super(TestGetLastHostUpdate, self).setUp()
self.exports = kojihub.RootExports()
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.query_singleValue = mock.MagicMock()
def test_valid_ts(self):
expected = 1615875554.862938
@ -30,9 +19,16 @@ class TestGetLastHostUpdate(unittest.TestCase):
else:
dt = datetime.datetime.strptime(
"2021-03-16T06:19:14.862938+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")
self.query_singleValue.return_value = dt
self.qp_single_value_return_value = dt
rv = self.exports.getLastHostUpdate(1, ts=True)
self.assertEqual(rv, expected)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['sessions'])
self.assertEqual(query.joins, ['host ON sessions.user_id = host.user_id'])
self.assertEqual(query.clauses, ['host.id = %(hostID)i'])
self.assertEqual(query.values, {'hostID': 1})
self.assertEqual(query.columns, ['update_time'])
def test_valid_datetime(self):
if sys.version_info[1] <= 6:
@ -41,6 +37,6 @@ class TestGetLastHostUpdate(unittest.TestCase):
else:
dt = datetime.datetime.strptime(
"2021-03-16T06:19:14.862938+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")
self.query_singleValue.return_value = dt
self.qp_single_value_return_value = dt
rv = self.exports.getLastHostUpdate(1)
self.assertEqual(rv, dt)

19
tests/test_hub/test_get_maven_archive.py Normal file
View file

@ -0,0 +1,19 @@
import kojihub
from .utils import DBQueryTestCase
class TestGetMavenArchive(DBQueryTestCase):
def setUp(self):
super(TestGetMavenArchive, self).setUp()
self.maxDiff = None
def test_valid(self):
kojihub.get_maven_archive(123)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['maven_archives'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])
self.assertEqual(query.values, {'archive_id': 123})
self.assertEqual(query.columns, ['archive_id', 'artifact_id', 'group_id', 'version'])

33
tests/test_hub/test_get_next_release.py
View file

@ -1,31 +1,37 @@
import mock
import unittest
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetNextRelease(unittest.TestCase):
class TestGetNextRelease(DBQueryTestCase):
def setUp(self):
super(TestGetNextRelease, self).setUp()
self.maxDiff = None
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()
self.get_build = mock.patch('kojihub.kojihub.get_build').start()
self._dml = mock.patch('kojihub.kojihub._dml').start()
self.query = self.QueryProcessor.return_value
self.binfo = {'name': 'name', 'version': 'version'}
def tearDown(self):
mock.patch.stopall()
def test_get_next_release_new(self):
# no previous build
self.query.executeOne.return_value = None
self.qp_execute_one_return_value = None
result = kojihub.get_next_release(self.binfo)
self.assertEqual(result, '1')
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['build'])
self.assertEqual(query.joins, ['package ON build.pkg_id = package.id'])
self.assertEqual(query.clauses,
['name = %(name)s', 'state in %(states)s', 'version = %(version)s'])
self.assertEqual(query.values, {'name': self.binfo['name'],
'version': self.binfo['version'],
'states': (1, 2, 0)
})
self.assertEqual(query.columns, ['build.id', 'release'])
def test_get_next_release_int(self):
for n in [1, 2, 3, 5, 8, 13, 21, 34, 55]:
self.query.executeOne.return_value = {'release': str(n)}
self.qp_execute_one_return_value = {'release': str(n)}
result = kojihub.get_next_release(self.binfo)
self.assertEqual(result, str(n + 1))
@ -38,7 +44,7 @@ class TestGetNextRelease(unittest.TestCase):
['20211105.nightly.7', '20211105.nightly.8'],
]
for a, b in data:
self.query.executeOne.return_value = {'release': a}
self.qp_execute_one_return_value = {'release': a}
result = kojihub.get_next_release(self.binfo)
self.assertEqual(result, b)
@ -52,9 +58,10 @@ class TestGetNextRelease(unittest.TestCase):
"1.2.fc23",
]
for val in data:
self.query.executeOne.return_value = {'release': val}
with self.assertRaises(koji.BuildError):
self.qp_execute_one_return_value = {'release': val}
with self.assertRaises(koji.BuildError) as ex:
kojihub.get_next_release(self.binfo)
self.assertEqual(f'Unable to increment release value: {val}', str(ex.exception))
def test_get_next_release_bad_incr(self):
data = [

100
tests/test_hub/test_get_notification_recipients.py
View file

@ -5,16 +5,9 @@ import koji
import kojihub
QP = kojihub.QueryProcessor
IP = kojihub.InsertProcessor
UP = kojihub.UpdateProcessor
class TestGetNotificationRecipients(unittest.TestCase):
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
self.inserts.append(insert)
return insert
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
@ -22,12 +15,6 @@ class TestGetNotificationRecipients(unittest.TestCase):
self.queries.append(query)
return query
def getUpdate(self, *args, **kwargs):
update = UP(*args, **kwargs)
update.execute = mock.MagicMock()
self.updates.append(update)
return update
def setUp(self):
self.context = mock.patch('kojihub.kojihub.context').start()
self.context.opts = {
@ -38,12 +25,6 @@ class TestGetNotificationRecipients(unittest.TestCase):
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',
side_effect=self.getInsert).start()
self.inserts = []
self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
self.readPackageList = mock.patch('kojihub.kojihub.readPackageList').start()
self.get_user = mock.patch('kojihub.kojihub.get_user').start()
@ -75,15 +56,6 @@ class TestGetNotificationRecipients(unittest.TestCase):
self.assertEqual(q.values['state'], state)
self.assertEqual(q.values['build'], build)
self.assertEqual(q.values['tag_id'], tag_id)
'''
q = self.queries[1]
self.assertEqual(q.columns, ['user_id'])
self.assertEqual(q.tables, ['build_notifications_block'])
self.assertEqual(q.clauses, ['user_id IN %(user_ids)s'])
self.assertEqual(q.joins, [])
self.assertEqual(q.values['user_ids'], None)
'''
self.readPackageList.assert_not_called()
def test_get_notification_recipients_build_without_tag(self):
@ -128,6 +100,34 @@ class TestGetNotificationRecipients(unittest.TestCase):
self.readPackageList.assert_not_called()
def test_get_notification_not_recipients(self):
### without build, without tag, result not recipients
tag_id = None
state = koji.BUILD_STATES['CANCELED']
build = None
self.queries = []
self.set_queries([[]])
result = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(result, [])
# there should be only query to watchers
self.assertEqual(len(self.queries), 1)
q = self.queries[0]
self.assertEqual(q.columns, ['email', 'user_id'])
self.assertEqual(q.tables, ['build_notifications'])
self.assertEqual(q.clauses, ['package_id IS NULL',
'status = %(users_status)i',
'success_only = FALSE',
'tag_id IS NULL',
'usertype IN %(users_usertypes)s'])
self.assertEqual(q.joins, ['JOIN users ON build_notifications.user_id = users.id'])
self.assertEqual(q.values['state'], state)
self.assertEqual(q.values['build'], build)
self.assertEqual(q.values['tag_id'], tag_id)
self.readPackageList.assert_not_called()
def test_get_notification_recipients_tag_without_build(self):
### with tag without build makes no sense
build = None
@ -135,8 +135,9 @@ class TestGetNotificationRecipients(unittest.TestCase):
state = koji.BUILD_STATES['CANCELED']
self.queries = []
with self.assertRaises(koji.GenericError):
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual('Invalid call', str(ex.exception))
self.assertEqual(self.queries, [])
self.readPackageList.assert_not_called()
@ -256,3 +257,44 @@ class TestGetNotificationRecipients(unittest.TestCase):
])
emails = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(emails, ['owner_name@test.domain.com'])
def test_get_notification_recipients_without_build(self):
### without build, without tag, result not recipients
tag_id = None
state = koji.BUILD_STATES['CANCELED']
build = None
self.queries = []
self.set_queries([
[{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
[]
])
result = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(result, ['owner_name@test.domain.com'])
# there should be only query to watchers
self.assertEqual(len(self.queries), 2)
q = self.queries[0]
self.assertEqual(q.columns, ['email', 'user_id'])
self.assertEqual(q.tables, ['build_notifications'])
self.assertEqual(q.clauses, ['package_id IS NULL',
'status = %(users_status)i',
'success_only = FALSE',
'tag_id IS NULL',
'usertype IN %(users_usertypes)s'])
self.assertEqual(q.joins, ['JOIN users ON build_notifications.user_id = users.id'])
self.assertEqual(q.values['state'], state)
self.assertEqual(q.values['build'], build)
self.assertEqual(q.values['tag_id'], tag_id)
q = self.queries[1]
self.assertEqual(q.columns, ['user_id'])
self.assertEqual(q.tables, ['build_notifications_block'])
self.assertEqual(q.clauses, ['package_id IS NULL',
'tag_id IS NULL',
'user_id IN %(user_ids)s',
])
self.assertEqual(q.joins, None)
self.assertEqual(q.values['user_ids'], [5])
self.readPackageList.assert_not_called()

64
tests/test_hub/test_get_ready_hosts.py Normal file
View file

@ -0,0 +1,64 @@
import kojihub
import mock
import unittest
QP = kojihub.QueryProcessor
class TestGetReadyHosts(unittest.TestCase):
def setUp(self):
self.maxDiff = None
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()
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = self.query_execute
self.queries.append(query)
return query
def test_valid(self):
hosts = [{'host.id': 1, 'name': 'hostname', 'arches': 'arch123', 'task_load': 0,
'capacity': 3},
{'host.id': 2, 'name': 'hostname-2', 'arches': 'arch123', 'task_load': 0,
'capacity': 3}]
expected_res = [{'host.id': 1, 'name': 'hostname', 'arches': 'arch123', 'task_load': 0,
'capacity': 3, 'channels': [1]},
{'host.id': 2, 'name': 'hostname-2', 'arches': 'arch123', 'task_load': 0,
'capacity': 3, 'channels': [2, 3]}
]
self.query_execute.side_effect = [hosts, [{'channel_id': 1}],
[{'channel_id': 2}, {'channel_id': 3}]]
result = kojihub.get_ready_hosts()
self.assertEqual(result, expected_res)
self.assertEqual(len(self.queries), 3)
query = self.queries[0]
self.assertEqual(query.tables, ['host'])
self.assertEqual(query.joins, ['sessions USING (user_id)',
'host_config ON host.id = host_config.host_id'])
self.assertEqual(query.clauses, ['active IS TRUE', 'enabled IS TRUE', 'expired IS FALSE',
'master IS NULL', 'ready IS TRUE',
"update_time > NOW() - '5 minutes'::interval"])
self.assertEqual(query.values, {})
self.assertEqual(query.columns, ['arches', 'capacity', 'host.id', 'name', 'task_load'])
query = self.queries[1]
self.assertEqual(query.tables, ['host_channels'])
self.assertEqual(query.joins, ['channels ON host_channels.channel_id = channels.id'])
self.assertEqual(query.clauses, ['active IS TRUE', 'enabled IS TRUE', 'host_id=%(id)s'])
self.assertEqual(query.values, hosts[0])
self.assertEqual(query.columns, ['channel_id'])
query = self.queries[2]
self.assertEqual(query.tables, ['host_channels'])
self.assertEqual(query.joins, ['channels ON host_channels.channel_id = channels.id'])
self.assertEqual(query.clauses, ['active IS TRUE', 'enabled IS TRUE', 'host_id=%(id)s'])
self.assertEqual(query.values, hosts[1])
self.assertEqual(query.columns, ['channel_id'])

24
tests/test_hub/test_get_session_info.py
View file

@ -1,34 +1,19 @@
import mock
import unittest
import koji
import kojihub
QP = kojihub.QueryProcessor
import mock
from .utils import DBQueryTestCase
class TestGetSessionInfo(unittest.TestCase):
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
self.queries.append(query)
return query
class TestGetSessionInfo(DBQueryTestCase):
def setUp(self):
super(TestGetSessionInfo, self).setUp()
self.context = mock.patch('kojihub.kojihub.context').start()
self.exports = kojihub.RootExports()
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.context.session.hasPerm = mock.MagicMock()
self.get_user = mock.patch('kojihub.kojihub.get_user').start()
self.userinfo = {'id': 123, 'name': 'testuser'}
self.exports.getLoggedInUser = mock.MagicMock()
def tearDown(self):
mock.patch.stopall()
def test_get_session_info_not_logged(self):
self.context.session.logged_in = False
result = self.exports.getSessionInfo()
@ -94,6 +79,7 @@ class TestGetSessionInfo(unittest.TestCase):
def test_get_session_info_details(self):
self.context.session.logged_in = True
self.context.session.hasPerm.return_value = True
self.qp_execute_one_return_value = {'hostip': '10.0.0.0', 'id': 123}
self.exports.getSessionInfo(details=True)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]

21
tests/test_hub/test_get_tag_external_repos.py
View file

@ -1,39 +1,24 @@
import unittest
import mock
import kojihub
QP = kojihub.QueryProcessor
from .utils import DBQueryTestCase
class TestGetTagExternalRepos(unittest.TestCase):
class TestGetTagExternalRepos(DBQueryTestCase):
def setUp(self):
super(TestGetTagExternalRepos, self).setUp()
self.maxDiff = None
self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()
self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.kojihub.context').start()
self.cursor = mock.MagicMock()
self.QueryProcessor = mock.patch('kojihub.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

28
tests/test_hub/test_get_task_children.py
View file

@ -1,34 +1,22 @@
import mock
import unittest
import koji
import kojihub
from .utils import DBQueryTestCase
QP = kojihub.QueryProcessor
class TestGetTaskChildren(unittest.TestCase):
class TestGetTaskChildren(DBQueryTestCase):
def setUp(self):
super(TestGetTaskChildren, self).setUp()
self.exports = kojihub.RootExports()
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = mock.MagicMock()
query.executeOne = mock.MagicMock()
query.singleValue = mock.MagicMock()
self.queries.append(query)
return query
def tearDown(self):
mock.patch.stopall()
def test_get_task_children_non_existing(self):
q = self.getQuery()
q.execute.return_value = []
self.QueryProcessor.side_effect = [q]
self.qp_execute_return_value = []
r = self.exports.getTaskChildren(1000)
@ -36,18 +24,14 @@ class TestGetTaskChildren(unittest.TestCase):
def test_get_task_children_non_existing_strict(self):
# get task info
q = self.getQuery()
q.executeOne.side_effect = koji.GenericError
self.QueryProcessor.side_effect = [q]
self.qp_execute_one_side_effect = koji.GenericError
with self.assertRaises(koji.GenericError):
self.exports.getTaskChildren(1000, strict=True)
def test_get_task_children(self):
children = [{'id': 1}]
q = self.getQuery()
q.execute.return_value = children
self.QueryProcessor.side_effect = [q]
self.qp_execute_return_value = children
r = self.exports.getTaskChildren(1000)

49
tests/test_hub/test_get_user.py
View file

@ -1,30 +1,19 @@
import unittest
import mock
import unittest
import koji
import kojihub
QP = kojihub.QueryProcessor
from .utils import DBQueryTestCase
class TestGetUser(unittest.TestCase):
class TestGetUser(DBQueryTestCase):
def setUp(self):
super(TestGetUser, self).setUp()
self.exports = kojihub.RootExports()
self.context = mock.patch('kojihub.kojihub.context').start()
self.QueryProcessor = mock.patch('kojihub.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()
self.list_user_krb_principals = mock.patch(
'kojihub.kojihub.list_user_krb_principals').start()
def test_wrong_format_user_info(self):
userinfo = ['test-user']
@ -71,9 +60,14 @@ class TestGetUser(unittest.TestCase):
'users.id = user_krb_principals.user_id'])
self.assertEqual(query.values, {'info': userinfo})
def test_userinfo_dict(self):
def test_userinfo_dict_with_krbs(self):
userinfo = {'id': 123456, 'name': 'test-user', 'krb_principal': 'test-krb@krb.com'}
kojihub.get_user(userinfo, krb_princs=False)
self.qp_execute_one_return_value = {'id': 123456, 'name': 'test-user',
'status': 1, 'usertype': 1}
self.list_user_krb_principals.return_value = 'test-krb@krb.com'
result = kojihub.get_user(userinfo, krb_princs=True)
self.assertEqual(result, {'id': 123456, 'krb_principals': 'test-krb@krb.com',
'name': 'test-user', 'status': 1, 'usertype': 1})
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
str(query)
@ -86,6 +80,23 @@ class TestGetUser(unittest.TestCase):
'users.id = user_krb_principals.user_id'])
self.assertEqual(query.values, userinfo)
def test_userinfo_int_user_not_exist_and_strict(self):
userinfo = {'id': 123456}
self.qp_execute_one_return_value = {}
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_user(userinfo['id'], strict=True, krb_princs=False)
self.assertEqual(f"No such user: {userinfo['id']}", str(cm.exception))
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, ['users.id = %(id)i'])
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):

4
tests/test_hub/test_get_verify_class.py
View file

@ -9,8 +9,9 @@ from koji.util import adler32_constructor
class TestGetVerifyClass(unittest.TestCase):
def test_get_verify_class_generic_error(self):
with self.assertRaises(GenericError):
with self.assertRaises(GenericError) as ex:
kojihub.get_verify_class('not_a_real_value')
self.assertEqual('Unsupported verify type: not_a_real_value', str(ex.exception))
def test_get_verify_class_is_none(self):
kojihub.get_verify_class(None) is None
@ -23,4 +24,3 @@ class TestGetVerifyClass(unittest.TestCase):
def test_get_verify_class_is_sha256(self):
kojihub.get_verify_class('sha256') is hashlib.sha256

2
tests/test_hub/test_get_volume.py
View file

@ -18,7 +18,7 @@ class TestGetVolume(unittest.TestCase):
self.exports.getVolume(volume, strict=True)
self.assertEqual("No such volume: %s" % volume, str(cm.exception))
def test_non_exist_volume_with_strict(self):
def test_non_exist_volume_without_strict(self):
volume = ['test-volume']
self.lookup_name.return_value = None
result = self.exports.getVolume(volume)

19
tests/test_hub/test_get_win_archive.py Normal file
View file

@ -0,0 +1,19 @@
import kojihub
from .utils import DBQueryTestCase
class TestGetWinArchive(DBQueryTestCase):
def setUp(self):
super(TestGetWinArchive, self).setUp()
self.maxDiff = None
def test_valid(self):
kojihub.get_win_archive(123)
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['win_archives'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])
self.assertEqual(query.values, {'archive_id': 123})
self.assertEqual(query.columns, ['archive_id', 'flags', 'platforms', 'relpath'])

64
tests/test_hub/test_get_win_build.py Normal file
View file

@ -0,0 +1,64 @@
import mock
import koji
import kojihub
from .utils import DBQueryTestCase
class TestGetWinBuild(DBQueryTestCase):
def setUp(self):
super(TestGetWinBuild, self).setUp()
self.maxDiff = None
self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()
def test_build_id_not_found(self):
self.find_build_id.return_value = None
result = kojihub.get_win_build('test-build.1-23.1')
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 0)
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)
def test_valid(self):
self.find_build_id.return_value = 123
self.qp_execute_one_return_value = {'build_id': 123}
kojihub.get_win_build('test-build.1-23.1')
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['win_builds'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(build_id)i'])
self.assertEqual(query.values, {'build_id': 123})
self.assertEqual(query.columns, ['build_id', 'platform'])
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)
def test_without_result_without_strict(self):
self.find_build_id.return_value = 123
self.qp_execute_one_return_value = {}
result = kojihub.get_win_build('test-build.1-23.1')
self.assertEqual(result, {})
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['win_builds'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(build_id)i'])
self.assertEqual(query.values, {'build_id': 123})
self.assertEqual(query.columns, ['build_id', 'platform'])
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)
def test_without_result_with_strict(self):
self.find_build_id.return_value = 123
self.qp_execute_one_return_value = {}
with self.assertRaises(koji.GenericError) as ex:
kojihub.get_win_build('test-build.1-23.1', strict=True)
self.assertEqual('no such Windows build: test-build.1-23.1', str(ex.exception))
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['win_builds'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(build_id)i'])
self.assertEqual(query.values, {'build_id': 123})
self.assertEqual(query.columns, ['build_id', 'platform'])
self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=True)

290
tests/test_hub/test_recycle_build.py
View file

@ -6,18 +6,26 @@ import kojihub
QP = kojihub.QueryProcessor
UP = kojihub.UpdateProcessor
DP = kojihub.DeleteProcessor
class TestRecycleBuild(unittest.TestCase):
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()
self.maxDiff = None
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.getQuery).start()
self.queries = []
self.query_execute = mock.MagicMock()
self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self._dml = mock.patch('kojihub.kojihub._dml').start()
self.run_callbacks = mock.patch('koji.plugin.run_callbacks').start()
self.rmtree = mock.patch('koji.util.rmtree').start()
self.exists = mock.patch('os.path.exists').start()
self.updates = []
self.DeleteProcessor = mock.patch('kojihub.kojihub.DeleteProcessor',
side_effect=self.getDelete).start()
self.deletes = []
self.get_build = mock.patch('kojihub.kojihub.get_build').start()
def tearDown(self):
mock.patch.stopall()
@ -28,9 +36,21 @@ class TestRecycleBuild(unittest.TestCase):
self.updates.append(update)
return update
def getQuery(self, *args, **kwargs):
query = QP(*args, **kwargs)
query.execute = self.query_execute
self.queries.append(query)
return query
def getDelete(self, *args, **kwargs):
delete = DP(*args, **kwargs)
delete.execute = mock.MagicMock()
self.deletes.append(delete)
return delete
# Basic old and new build infos
old = {'id': 2,
'state': 0,
'state': 3,
'task_id': None,
'epoch': None,
'name': 'GConf2',
@ -44,7 +64,7 @@ class TestRecycleBuild(unittest.TestCase):
'cg_id': None,
'volume_id': 0,
'volume_name': 'DEFAULT'}
new = {'state': 0,
new = {'state': 3,
'name': 'GConf2',
'version': '3.2.6',
'release': '15.fc23',
@ -58,97 +78,201 @@ class TestRecycleBuild(unittest.TestCase):
'cg_id': None,
'volume_id': 0}
def test_recycle_building(self):
new = self.new.copy()
old = self.old.copy()
old['state'] = new['state'] = koji.BUILD_STATES['BUILDING']
old['task_id'] = new['task_id'] = 137
kojihub.recycle_build(old, new)
self.UpdateProcessor.assert_not_called()
self.QueryProcessor.assert_not_called()
self._dml.assert_not_called()
self.run_callbacks.assert_not_called()
def run_fail(self, old, new):
try:
kojihub.recycle_build(old, new)
except koji.GenericError:
pass
else:
raise Exception("expected koji.GenericError")
self.UpdateProcessor.assert_not_called()
self._dml.assert_not_called()
self.run_callbacks.assert_not_called()
def test_recycle_building_bad(self):
def test_build_already_in_progress(self):
new = self.new.copy()
old = self.old.copy()
old['state'] = new['state'] = koji.BUILD_STATES['BUILDING']
old['task_id'] = 137
new['task_id'] = 200
self.run_fail(old, new)
self.QueryProcessor.assert_not_called()
with self.assertRaises(koji.GenericError) as ex:
kojihub.recycle_build(old, new)
self.assertEqual(f"Build already in progress (task {old['task_id']})", str(ex.exception))
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.deletes), 0)
def test_recycle_states_good(self):
for state in 'FAILED', 'CANCELED':
self.check_recycle_states_good, koji.BUILD_STATES[state]
self.get_build.assert_not_called()
self.run_callbacks.assert_not_called()
def check_recycle_states_good(self, state):
def test_build_already_in_progress_same_task_id(self):
new = self.new.copy()
old = self.old.copy()
old['state'] = state
new['state'] = koji.BUILD_STATES['BUILDING']
old['task_id'] = 99
new['task_id'] = 137
query = self.QueryProcessor.return_value
# for all checks
query.execute.return_value = []
# for getBuild
query.executeOne.return_value = old
self.run_pass(old, new)
old['state'] = new['state'] = koji.BUILD_STATES['BUILDING']
old['task_id'] = new['task_id'] = 137
result = kojihub.recycle_build(old, new)
self.assertEqual(result, None)
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.deletes), 0)
self.get_build.assert_not_called()
self.run_callbacks.assert_not_called()
def test_not_in_failed_or_canceled_state(self):
old = self.old.copy()
old['state'] = koji.BUILD_STATES['COMPLETE']
with self.assertRaises(koji.GenericError) as ex:
kojihub.recycle_build(old, self.new)
self.assertEqual(f"Build already exists (id={old['id']}, state=COMPLETE): {self.new}",
str(ex.exception))
self.assertEqual(len(self.queries), 0)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.deletes), 0)
self.get_build.assert_not_called()
self.run_callbacks.assert_not_called()
def test_tag_activity_already_exists(self):
old = self.old.copy()
old['task_id'] = 137
self.query_execute.return_value = [{'tag_id': 123}]
with self.assertRaises(koji.GenericError) as ex:
kojihub.recycle_build(old, self.new)
self.assertEqual("Build already exists. Unable to recycle, has tag history",
str(ex.exception))
self.assertEqual(len(self.queries), 1)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.deletes), 0)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_listing'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['tag_id'])
self.get_build.assert_not_called()
self.run_callbacks.assert_not_called()
def test_rpm_activity_already_exists(self):
old = self.old.copy()
old['task_id'] = 137
self.query_execute.side_effect = [[], [{'id': 1}]]
with self.assertRaises(koji.GenericError) as ex:
kojihub.recycle_build(old, self.new)
self.assertEqual("Build already exists. Unable to recycle, has rpm data",
str(ex.exception))
self.assertEqual(len(self.queries), 2)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.deletes), 0)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_listing'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['tag_id'])
query = self.queries[1]
self.assertEqual(query.tables, ['rpminfo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['id'])
self.get_build.assert_not_called()
self.run_callbacks.assert_not_called()
def test_archive_activity_already_exists(self):
old = self.old.copy()
old['task_id'] = 137
self.query_execute.side_effect = [[], [], [{'id': 11}]]
with self.assertRaises(koji.GenericError) as ex:
kojihub.recycle_build(old, self.new)
self.assertEqual("Build already exists. Unable to recycle, has archive data",
str(ex.exception))
self.assertEqual(len(self.queries), 3)
self.assertEqual(len(self.updates), 0)
self.assertEqual(len(self.deletes), 0)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_listing'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['tag_id'])
query = self.queries[1]
self.assertEqual(query.tables, ['rpminfo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['id'])
query = self.queries[2]
self.assertEqual(query.tables, ['archiveinfo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['id'])
self.get_build.assert_not_called()
self.run_callbacks.assert_not_called()
def test_valid(self):
old = self.old.copy()
new = self.new.copy()
old['task_id'] = new['task_id'] = 137
self.query_execute.side_effect = [[], [], []]
self.get_build.return_value = {'build_id': 2, 'name': 'GConf2', 'version': '3.2.6',
'release': '15.fc23'}
def run_pass(self, old, new):
kojihub.recycle_build(old, new)
self.UpdateProcessor.assert_called_once()
self.assertEqual(len(self.queries), 3)
self.assertEqual(len(self.updates), 1)
self.assertEqual(len(self.deletes), 4)
query = self.queries[0]
self.assertEqual(query.tables, ['tag_listing'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['tag_id'])
query = self.queries[1]
self.assertEqual(query.tables, ['rpminfo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['id'])
query = self.queries[2]
self.assertEqual(query.tables, ['archiveinfo'])
self.assertEqual(query.joins, None)
self.assertEqual(query.clauses, ['build_id = %(id)s'])
self.assertEqual(query.values, old)
self.assertEqual(query.columns, ['id'])
delete = self.deletes[0]
self.assertEqual(delete.table, 'maven_builds')
self.assertEqual(delete.clauses, ['build_id = %(id)i'])
self.assertEqual(delete.values, old)
delete = self.deletes[1]
self.assertEqual(delete.table, 'win_builds')
self.assertEqual(delete.clauses, ['build_id = %(id)i'])
self.assertEqual(delete.values, old)
delete = self.deletes[2]
self.assertEqual(delete.table, 'image_builds')
self.assertEqual(delete.clauses, ['build_id = %(id)i'])
self.assertEqual(delete.values, old)
delete = self.deletes[3]
self.assertEqual(delete.table, 'build_types')
self.assertEqual(delete.clauses, ['build_id = %(id)i'])
self.assertEqual(delete.values, old)
update = self.updates[0]
assert update.table == 'build'
self.assertEqual(update.table, 'build')
self.assertEqual(update.values, new)
for key in ['state', 'task_id', 'owner', 'start_time',
'completion_time', 'epoch']:
assert update.data[key] == new[key]
assert update.rawdata == {'create_event': 'get_event()'}
assert update.clauses == ['id=%(id)s']
assert update.values['id'] == old['id']
self.assertEqual(update.rawdata, {'create_event': 'get_event()'})
self.assertEqual(update.clauses, ['id=%(id)s'])
def test_recycle_states_bad(self):
for state in 'BUILDING', 'COMPLETE', 'DELETED':
self.check_recycle_states_bad, koji.BUILD_STATES[state]
def check_recycle_states_bad(self, state):
new = self.new.copy()
old = self.old.copy()
old['state'] = state
new['state'] = koji.BUILD_STATES['BUILDING']
old['task_id'] = 99
new['task_id'] = 137
self.run_fail(old, new)
self.QueryProcessor.assert_not_called()
def test_recycle_query_bad(self):
vlists = [
[[], [], True],
[True, [], []],
[[], True, []],
]
for values in vlists:
self.check_recycle_query_bad, values
def check_recycle_query_bad(self, values):
new = self.new.copy()
old = self.old.copy()
old['state'] = koji.BUILD_STATES['FAILED']
new['state'] = koji.BUILD_STATES['BUILDING']
old['task_id'] = 99
new['task_id'] = 137
query = self.QueryProcessor.return_value
query.execute.side_effect = values
self.run_fail(old, new)
self.get_build.assert_called_once_with(new['id'], strict=True)
self.assertEqual(self.run_callbacks.call_count, 2)

72
tests/test_hub/test_remove_external_repo_from_tag.py Normal file
View file

@ -0,0 +1,72 @@
import unittest
import mock
import koji
import kojihub
UP = kojihub.UpdateProcessor
class TestRemoveExternalRepoFromTag(unittest.TestCase):
def getUpdate(self, *args, **kwargs):
update = UP(*args, **kwargs)
update.execute = mock.MagicMock()
self.updates.append(update)
return update
def setUp(self):
self.context = mock.patch('kojihub.kojihub.context').start()
self.context.session.assertPerm = mock.MagicMock()
self.context_db = mock.patch('kojihub.db.context').start()
self.context_db.session.assertLogin = mock.MagicMock()
self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()
self.get_tag_external_repos = mock.patch('kojihub.kojihub.get_tag_external_repos').start()
self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()
def tearDown(self):
mock.patch.stopall()
def test_not_exist_tag_external_repo(self):
tag_info = {'id': 22, 'name': 'test-tag'}
repo_info = {'id': 1, 'name': 'test-repo'}
self.get_tag.return_value = tag_info
self.get_external_repo.return_value = repo_info
self.get_tag_external_repos.return_value = None
with self.assertRaises(koji.GenericError) as ex:
kojihub.remove_external_repo_from_tag(tag_info['name'], repo_info['name'])
self.assertEqual(
f'external repo {repo_info["name"]} not associated with tag {tag_info["name"]}',
str(ex.exception))
self.assertEqual(len(self.updates), 0)
self.get_tag.assert_called_once_with(tag_info['name'], strict=True)
self.get_external_repo.assert_called_once_with(repo_info['name'], strict=True)
self.get_tag_external_repos.assert_called_once_with(
tag_info=tag_info['id'], repo_info=repo_info['id'])
def test_valid(self):
tag_info = {'id': 22, 'name': 'test-tag'}
repo_info = {'id': 1, 'name': 'test-repo'}
self.context_db.event_id = 42
self.context_db.session.user_id = 23
self.get_tag.return_value = tag_info
self.get_external_repo.return_value = repo_info
self.get_tag_external_repos.return_value = [{'id': repo_info['id'],
'tag_id': tag_info['id']}]
kojihub.remove_external_repo_from_tag(tag_info['name'], repo_info['name'])
self.assertEqual(len(self.updates), 1)
update = self.updates[0]
self.assertEqual(update.table, 'tag_external_repos')
self.assertEqual(update.clauses, ["tag_id = %(tag_id)i", "external_repo_id = %(repo_id)i",
'active = TRUE'])
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})
self.assertEqual(update.rawdata, {'active': 'NULL'})
self.get_tag.assert_called_once_with(tag_info['name'], strict=True)
self.get_external_repo.assert_called_once_with(repo_info['name'], strict=True)
self.get_tag_external_repos.assert_called_once_with(
tag_info=tag_info['id'], repo_info=repo_info['id'])

11
tests/test_hub/utils.py
View file

@ -12,8 +12,11 @@ class DBQueryTestCase(unittest.TestCase):
mock.patch.stopall()
self.qp_execute_return_value = []
self.qp_execute_side_effect = None
self.qp_execute_one_return_value = []
self.qp_execute_one_side_effect = None
self.qp_single_value_return_value = None
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
side_effect=self.get_query).start()
side_effect=self.get_query).start()
self.queries = []
def tearDown(self):
@ -28,6 +31,11 @@ class DBQueryTestCase(unittest.TestCase):
query.execute = mock.MagicMock()
query.execute.return_value = self.qp_execute_return_value
query.execute.side_effect = self.qp_execute_side_effect
query.executeOne = mock.MagicMock()
query.executeOne.return_value = self.qp_execute_one_return_value
query.executeOne.side_effect = self.qp_execute_one_side_effect
query.singleValue = mock.MagicMock()
query.singleValue.return_value = self.qp_single_value_return_value
self.queries.append(query)
return query
@ -42,4 +50,3 @@ class DBQueryTestCase(unittest.TestCase):
def assertQueriesEqual(self, arglist):
for i, query in enumerate(self.queries):
self.assertQueryEqual(query, **arglist[i])