PR#3745: Fix pkglist_add when extra_arches is None
Merges #3745 https://pagure.io/koji/pull-request/3745 Fixes: #3744 https://pagure.io/koji/issue/3744 [hub] pkgListAdd rewrites extra_arches value None to empty string
This commit is contained in:
Tomas Kopecek
commit
d5eea0c6a5
6 changed files with 1421 additions and 205 deletions
|
|
@ -1001,11 +1001,13 @@ def _pkglist_owner_add(tag_id, pkg_id, owner):
|
|||
def _pkglist_add(tag_id, pkg_id, owner, block, extra_arches):
|
||||
# revoke old entry (if present)
|
||||
_pkglist_remove(tag_id, pkg_id)
|
||||
if extra_arches is not None:
|
||||
extra_arches = koji.parse_arches(extra_arches, strict=True, allow_none=True)
|
||||
data = {
|
||||
'tag_id': tag_id,
|
||||
'package_id': pkg_id,
|
||||
'blocked': block,
|
||||
'extra_arches': koji.parse_arches(extra_arches, strict=True, allow_none=True)
|
||||
'extra_arches': extra_arches
|
||||
}
|
||||
insert = InsertProcessor('tag_packages', data=data)
|
||||
insert.make_create() # XXX user_id?
|
||||
|
|
@ -1045,7 +1047,8 @@ def _direct_pkglist_add(taginfo, pkginfo, owner, block, extra_arches, force,
|
|||
if not pkg:
|
||||
pkg = lookup_package(pkginfo, create=True)
|
||||
# validate arches before running callbacks
|
||||
extra_arches = koji.parse_arches(extra_arches, strict=True, allow_none=True)
|
||||
if extra_arches is not None:
|
||||
extra_arches = koji.parse_arches(extra_arches, strict=True, allow_none=True)
|
||||
user = get_user(context.session.user_id)
|
||||
# first check to see if package is:
|
||||
# already present (via inheritance)
|
||||
|
|
|
|||
145
tests/test_hub/test_get_tag_groups.py
Normal file
145
tests/test_hub/test_get_tag_groups.py
Normal file
|
|
@ -0,0 +1,145 @@
|
|||
import mock
|
||||
import unittest
|
||||
|
||||
import kojihub
|
||||
|
||||
QP = kojihub.QueryProcessor
|
||||
|
||||
|
||||
class TestGetTagGroups(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.exports = kojihub.RootExports()
|
||||
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
|
||||
side_effect=self.getQuery).start()
|
||||
self.queries = []
|
||||
self.query_execute = mock.MagicMock()
|
||||
self.maxDiff = None
|
||||
self.context = mock.patch('kojihub.kojihub.context').start()
|
||||
self.context_db = mock.patch('kojihub.db.context').start()
|
||||
self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start()
|
||||
self.read_full_inheritance = mock.patch('kojihub.kojihub.readFullInheritance').start()
|
||||
|
||||
def getQuery(self, *args, **kwargs):
|
||||
query = QP(*args, **kwargs)
|
||||
query.execute = self.query_execute
|
||||
query.executeOne = mock.MagicMock()
|
||||
query.singleValue = mock.MagicMock()
|
||||
self.queries.append(query)
|
||||
return query
|
||||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
|
||||
def test_valid(self):
|
||||
tag_name = 'test-tag'
|
||||
self.get_tag_id.return_value = 11
|
||||
self.read_full_inheritance.return_value = [
|
||||
{
|
||||
'child_id': 123,
|
||||
'currdepth': 1,
|
||||
'filter': [],
|
||||
'intransitive': False,
|
||||
'maxdepth': None,
|
||||
'name': tag_name,
|
||||
'nextdepth': None,
|
||||
'noconfig': False,
|
||||
'parent_id': 1234,
|
||||
'pkg_filter': '',
|
||||
'priority': 0
|
||||
}
|
||||
]
|
||||
self.query_execute.side_effect = [
|
||||
[{'group_id': 1, 'blocked': True}], [{'group_id': 2, 'blocked': False}],
|
||||
[{'group_id': 1, 'package': 'pkg'}], [{'group_id': 2, 'package': 'pkg-2'}],
|
||||
[{'group_id': 1, 'req_id': 1}], [{'group_id': 2, 'req_id': 2}]]
|
||||
kojihub.get_tag_groups(tag_name)
|
||||
|
||||
self.assertEqual(len(self.queries), 6)
|
||||
clauses = ['(active = TRUE)', 'tag_id = %(tagid)s']
|
||||
values = {'tagid': 11}
|
||||
|
||||
for i in range(0, 1):
|
||||
query = self.queries[i]
|
||||
columns = ['biarchonly', 'blocked', 'description', 'display_name', 'exported',
|
||||
'group_id', 'is_default', 'langonly', 'name', 'tag_id', 'uservisible']
|
||||
self.assertEqual(query.tables, ['group_config'])
|
||||
self.assertEqual(query.joins, ['groups ON group_id = id'])
|
||||
self.assertEqual(query.values, values)
|
||||
self.assertEqual(query.columns, columns)
|
||||
self.assertEqual(query.clauses, clauses)
|
||||
|
||||
for i in range(2, 3):
|
||||
query = self.queries[i]
|
||||
columns = \
|
||||
['basearchonly', 'blocked', 'group_id', 'package', 'requires', 'tag_id', 'type']
|
||||
self.assertEqual(query.tables, ['group_package_listing'])
|
||||
self.assertEqual(query.joins, None)
|
||||
self.assertEqual(query.values, values)
|
||||
self.assertEqual(query.columns, columns)
|
||||
self.assertEqual(query.clauses, clauses)
|
||||
|
||||
for i in range(4, 5):
|
||||
query = self.queries[i]
|
||||
columns = ['blocked', 'group_id', 'is_metapkg', 'name', 'req_id', 'tag_id', 'type']
|
||||
self.assertEqual(query.tables, ['group_req_listing'])
|
||||
self.assertEqual(query.joins, ['groups on req_id = id'])
|
||||
self.assertEqual(query.values, values)
|
||||
self.assertEqual(query.columns, columns)
|
||||
self.assertEqual(query.clauses, clauses)
|
||||
|
||||
def test_valid_group_not_in_groups(self):
|
||||
tag_name = 'test-tag'
|
||||
self.get_tag_id.return_value = 11
|
||||
self.read_full_inheritance.return_value = [
|
||||
{
|
||||
'child_id': 123,
|
||||
'currdepth': 1,
|
||||
'filter': [],
|
||||
'intransitive': False,
|
||||
'maxdepth': None,
|
||||
'name': tag_name,
|
||||
'nextdepth': None,
|
||||
'noconfig': False,
|
||||
'parent_id': 1234,
|
||||
'pkg_filter': '',
|
||||
'priority': 0
|
||||
}
|
||||
]
|
||||
self.query_execute.side_effect = [
|
||||
[{'group_id': 1, 'blocked': True}], [{'group_id': 2, 'blocked': False}],
|
||||
[{'group_id': 2, 'package': 'pkg'}], [{'group_id': 3, 'package': 'pkg-2'}],
|
||||
[{'group_id': 4, 'req_id': 6}], [{'group_id': 2, 'req_id': 7}]]
|
||||
kojihub.get_tag_groups(tag_name)
|
||||
|
||||
self.assertEqual(len(self.queries), 6)
|
||||
clauses = ['(active = TRUE)', 'tag_id = %(tagid)s']
|
||||
values = {'tagid': 11}
|
||||
|
||||
for i in range(0, 1):
|
||||
query = self.queries[i]
|
||||
columns = ['biarchonly', 'blocked', 'description', 'display_name', 'exported',
|
||||
'group_id', 'is_default', 'langonly', 'name', 'tag_id', 'uservisible']
|
||||
self.assertEqual(query.tables, ['group_config'])
|
||||
self.assertEqual(query.joins, ['groups ON group_id = id'])
|
||||
self.assertEqual(query.values, values)
|
||||
self.assertEqual(query.columns, columns)
|
||||
self.assertEqual(query.clauses, clauses)
|
||||
|
||||
for i in range(2, 3):
|
||||
query = self.queries[i]
|
||||
columns = \
|
||||
['basearchonly', 'blocked', 'group_id', 'package', 'requires', 'tag_id', 'type']
|
||||
self.assertEqual(query.tables, ['group_package_listing'])
|
||||
self.assertEqual(query.joins, None)
|
||||
self.assertEqual(query.values, values)
|
||||
self.assertEqual(query.columns, columns)
|
||||
self.assertEqual(query.clauses, clauses)
|
||||
|
||||
for i in range(4, 5):
|
||||
query = self.queries[i]
|
||||
columns = ['blocked', 'group_id', 'is_metapkg', 'name', 'req_id', 'tag_id', 'type']
|
||||
self.assertEqual(query.tables, ['group_req_listing'])
|
||||
self.assertEqual(query.joins, ['groups on req_id = id'])
|
||||
self.assertEqual(query.values, values)
|
||||
self.assertEqual(query.columns, columns)
|
||||
self.assertEqual(query.clauses, clauses)
|
||||
|
|
@ -12,6 +12,7 @@ class TestGrouplist(unittest.TestCase):
|
|||
def getQuery(self, *args, **kwargs):
|
||||
query = QP(*args, **kwargs)
|
||||
query.execute = mock.MagicMock()
|
||||
query.singleValue = self.query_singleValue
|
||||
self.queries.append(query)
|
||||
return query
|
||||
|
||||
|
|
@ -54,6 +55,7 @@ class TestGrouplist(unittest.TestCase):
|
|||
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
|
||||
side_effect=self.getQuery).start()
|
||||
self.queries = []
|
||||
self.query_singleValue = mock.MagicMock()
|
||||
self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',
|
||||
side_effect=self.getInsert).start()
|
||||
self.inserts = []
|
||||
|
|
@ -68,7 +70,7 @@ class TestGrouplist(unittest.TestCase):
|
|||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
|
||||
def test_grplist_add(self):
|
||||
def test_grplist_add_valid(self):
|
||||
self.get_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {}
|
||||
|
|
@ -108,6 +110,140 @@ class TestGrouplist(unittest.TestCase):
|
|||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
def test_grplist_add_previous_blocked(self):
|
||||
self.get_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {2: {'blocked': True}}
|
||||
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grplist_add(self.tag, self.group)
|
||||
self.assertEqual(f"group {self.group} is blocked in tag {self.tag}", str(ex.exception))
|
||||
|
||||
# what was called
|
||||
self.context.session.assertPerm.assert_called_once_with('tag')
|
||||
self.get_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, create=True)
|
||||
self.get_tag_groups.assert_called_with(self.taginfo['id'], inherit=True, incl_pkgs=False,
|
||||
incl_reqs=False)
|
||||
# db
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
def test_grplist_add_previous_not_changed(self):
|
||||
self.get_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {2: {'exported': 'exp', 'uservisible': True,
|
||||
'description': 'desc', 'langonly': False,
|
||||
'biarchonly': False, 'blocked': False,
|
||||
'display_name': 'test-group', 'is_default': True}}
|
||||
|
||||
kojihub.grplist_add(self.tag, self.group)
|
||||
|
||||
# what was called
|
||||
self.context.session.assertPerm.assert_called_once_with('tag')
|
||||
self.get_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, create=True)
|
||||
self.get_tag_groups.assert_called_with(self.taginfo['id'], inherit=True, incl_pkgs=False,
|
||||
incl_reqs=False)
|
||||
# db
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
def test_grplist_add_previous_all_changed(self):
|
||||
self.get_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {2: {'exported': False, 'uservisible': True,
|
||||
'description': 'desc', 'langonly': False,
|
||||
'biarchonly': False, 'blocked': False,
|
||||
'display_name': 'test-group', 'is_default': True}}
|
||||
self.context_db.event_id = 42
|
||||
self.context_db.session.user_id = 24
|
||||
|
||||
kojihub.grplist_add(self.tag, self.group, block=True, description='test-desc',
|
||||
exported=True, uservisible=False, langonly=True, biarchonly=True,
|
||||
display_name='test-group-new', is_default=False)
|
||||
|
||||
# what was called
|
||||
self.context.session.assertPerm.assert_called_once_with('tag')
|
||||
self.get_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, create=True)
|
||||
self.get_tag_groups.assert_called_with(self.taginfo['id'], inherit=True, incl_pkgs=False,
|
||||
incl_reqs=False)
|
||||
# db
|
||||
# revoke
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_config')
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
# insert new group
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'display_name': 'test-group-new',
|
||||
'biarchonly': True,
|
||||
'exported': True,
|
||||
'uservisible': False,
|
||||
'create_event': 42,
|
||||
'creator_id': 24,
|
||||
'tag_id': self.taginfo['id'],
|
||||
'group_id': self.groupinfo['id'],
|
||||
'blocked': True,
|
||||
'description': 'test-desc',
|
||||
'langonly': True,
|
||||
'is_default': False,
|
||||
}
|
||||
self.assertEqual(insert.table, 'group_config')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
def test_grplist_add_previous_few_changed(self):
|
||||
self.get_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {2: {'exported': False, 'uservisible': True,
|
||||
'description': 'desc', 'langonly': False,
|
||||
'biarchonly': False, 'blocked': False,
|
||||
'display_name': 'test-group', 'is_default': True}}
|
||||
self.context_db.event_id = 42
|
||||
self.context_db.session.user_id = 24
|
||||
|
||||
kojihub.grplist_add(self.tag, self.group, block=True, description='test-desc',
|
||||
exported=True)
|
||||
|
||||
# what was called
|
||||
self.context.session.assertPerm.assert_called_once_with('tag')
|
||||
self.get_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, create=True)
|
||||
self.get_tag_groups.assert_called_with(self.taginfo['id'], inherit=True, incl_pkgs=False,
|
||||
incl_reqs=False)
|
||||
# db
|
||||
# revoke
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_config')
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
# insert new group
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'display_name': 'test-group',
|
||||
'biarchonly': False,
|
||||
'exported': True,
|
||||
'uservisible': True,
|
||||
'create_event': 42,
|
||||
'creator_id': 24,
|
||||
'tag_id': self.taginfo['id'],
|
||||
'group_id': self.groupinfo['id'],
|
||||
'blocked': True,
|
||||
'description': 'test-desc',
|
||||
'langonly': False,
|
||||
'is_default': True,
|
||||
}
|
||||
self.assertEqual(insert.table, 'group_config')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
def test_grplist_add_no_admin(self):
|
||||
self.context.session.assertPerm.side_effect = koji.GenericError
|
||||
with self.assertRaises(koji.GenericError):
|
||||
|
|
@ -201,27 +337,27 @@ class TestGrouplist(unittest.TestCase):
|
|||
self.assertEqual(len(self.queries), 1)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(cm.exception.args[0],
|
||||
'No group: group found for tag: tag')
|
||||
self.assertEqual(cm.exception.args[0], 'No group: group found for tag: tag')
|
||||
|
||||
# force = True
|
||||
self.reset_db_processors()
|
||||
with mock.patch('kojihub.kojihub.QueryProcessor',
|
||||
side_effect=self.getEmptyQuery):
|
||||
with mock.patch('kojihub.kojihub.QueryProcessor', side_effect=self.getEmptyQuery):
|
||||
kojihub.grplist_remove(self.tag, self.group, force=True)
|
||||
|
||||
self.assertEqual(len(self.queries), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
|
||||
def test_grplist_unblock(self):
|
||||
def test_grplist_unblock_without_blocked_groups(self):
|
||||
# identical with test_grplist_add except blocked=True
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.query_singleValue.return_value = None
|
||||
|
||||
# will fail for non-blocked group
|
||||
with self.assertRaises(koji.GenericError):
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grplist_unblock(self.tag, self.group)
|
||||
self.assertEqual(f"group {self.group} is NOT blocked in tag {self.tag}", str(ex.exception))
|
||||
|
||||
# what was called
|
||||
self.context.session.assertPerm.assert_called_once_with('tag')
|
||||
|
|
@ -238,6 +374,37 @@ class TestGrouplist(unittest.TestCase):
|
|||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
def test_grplist_unblock_with_blocked_groups(self):
|
||||
# identical with test_grplist_add except blocked=True
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.query_singleValue.return_value = 333
|
||||
self.context_db.event_id = 42
|
||||
self.context_db.session.user_id = 24
|
||||
|
||||
kojihub.grplist_unblock(self.tag, self.group)
|
||||
|
||||
# what was called
|
||||
self.context.session.assertPerm.assert_called_once_with('tag')
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
|
||||
# db
|
||||
self.assertEqual(len(self.queries), 1)
|
||||
query = self.queries[0]
|
||||
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), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_config')
|
||||
self.assertEqual(update.clauses,
|
||||
['group_id=%(grp_id)s', 'tag_id=%(tag_id)s', 'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
def test_readTagGroups_empty(self):
|
||||
self.get_tag_groups.return_value = {}
|
||||
|
||||
|
|
|
|||
283
tests/test_hub/test_group_pkg_operations.py
Normal file
283
tests/test_hub/test_group_pkg_operations.py
Normal file
|
|
@ -0,0 +1,283 @@
|
|||
import mock
|
||||
import unittest
|
||||
import koji
|
||||
import kojihub
|
||||
|
||||
QP = kojihub.QueryProcessor
|
||||
IP = kojihub.InsertProcessor
|
||||
UP = kojihub.UpdateProcessor
|
||||
|
||||
|
||||
class TestGroupPkglist(unittest.TestCase):
|
||||
def getQuery(self, *args, **kwargs):
|
||||
query = QP(*args, **kwargs)
|
||||
query.execute = 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.maxDiff = None
|
||||
self.context = mock.patch('kojihub.kojihub.context').start()
|
||||
self.context_db = mock.patch('kojihub.db.context').start()
|
||||
self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start()
|
||||
self.get_group_id = mock.patch('kojihub.kojihub.get_group_id').start()
|
||||
self.lookup_tag = mock.patch('kojihub.kojihub.lookup_tag').start()
|
||||
self.lookup_group = mock.patch('kojihub.kojihub.lookup_group').start()
|
||||
self.get_tag_groups = mock.patch('kojihub.kojihub.get_tag_groups').start()
|
||||
# It seems MagicMock will not automatically handle attributes that
|
||||
# start with "assert"
|
||||
self.context.session.assertPerm = mock.MagicMock()
|
||||
self.context_db.session.assertLogin = mock.MagicMock()
|
||||
|
||||
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
|
||||
side_effect=self.getQuery).start()
|
||||
self.queries = []
|
||||
self.query_singleValue = mock.MagicMock()
|
||||
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.tag = 'tag'
|
||||
self.group = 'group'
|
||||
self.pkg = 'test-pkg'
|
||||
self.taginfo = {'name': self.tag, 'id': 1}
|
||||
self.groupinfo = {'name': self.group, 'id': 2}
|
||||
self.pkginfo = {'name': self.pkg, 'id': 3}
|
||||
self.context_db.event_id = 42
|
||||
self.context_db.session.user_id = 24
|
||||
|
||||
def test_grp_pkg_add_previous_changed(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {'test-pkg': {'basearchonly': None, 'blocked': False,
|
||||
'group_id': 3, 'requires': None, 'tag_id': 1,
|
||||
'type': 'mandatory'}},
|
||||
'grouplist': {}}}
|
||||
kojihub.grp_pkg_add(self.tag, self.group, self.pkg, block=True, type='test-type')
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_package_listing')
|
||||
self.assertEqual(update.clauses,
|
||||
['group_id=%(group_id)s', 'tag_id=%(tag_id)s', 'package=%(package)s',
|
||||
'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
# insert new group
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'basearchonly': None,
|
||||
'blocked': True,
|
||||
'create_event': 42,
|
||||
'creator_id': 24,
|
||||
'group_id': self.groupinfo['id'],
|
||||
'package': self.pkg,
|
||||
'requires': None,
|
||||
'tag_id': self.taginfo['id'],
|
||||
'type': 'test-type'
|
||||
}
|
||||
self.assertEqual(insert.table, 'group_package_listing')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=True, incl_reqs=False)
|
||||
|
||||
def test_grp_pkg_add_previous_not_changed(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {'test-pkg': {'basearchonly': None, 'blocked': False,
|
||||
'group_id': 3, 'requires': None, 'tag_id': 1,
|
||||
'type': 'mandatory'}},
|
||||
'grouplist': {}}}
|
||||
kojihub.grp_pkg_add(self.tag, self.group, self.pkg, block=False, type='mandatory')
|
||||
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=True, incl_reqs=False)
|
||||
|
||||
def test_grp_pkg_add_not_previous(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {},
|
||||
'grouplist': {}}}
|
||||
kojihub.grp_pkg_add(self.tag, self.group, self.pkg)
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_package_listing')
|
||||
self.assertEqual(update.clauses,
|
||||
['group_id=%(group_id)s', 'tag_id=%(tag_id)s', 'package=%(package)s',
|
||||
'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
# insert new group
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'blocked': False,
|
||||
'create_event': 42,
|
||||
'creator_id': 24,
|
||||
'group_id': self.groupinfo['id'],
|
||||
'package': self.pkg,
|
||||
'tag_id': self.taginfo['id'],
|
||||
'type': 'mandatory'
|
||||
}
|
||||
self.assertEqual(insert.table, 'group_package_listing')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=True, incl_reqs=False)
|
||||
|
||||
def test_grp_pkg_add_previous_blocked(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {'test-pkg': {'basearchonly': None, 'blocked': True, 'group_id': 3,
|
||||
'requires': None, 'tag_id': 1, 'type': 'mandatory'}},
|
||||
'grouplist': {}}}
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_pkg_add(self.tag, self.group, self.pkg)
|
||||
self.assertEqual(f"package {self.pkg} blocked in group {self.group}, tag {self.tag}",
|
||||
str(ex.exception))
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=True, incl_reqs=False)
|
||||
|
||||
def test_grp_pkg_add_not_group_in_tag(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {
|
||||
111: {'blocked': False, 'display_name': 'group-1', 'group_id': 111, 'tag_id': 1,
|
||||
'packagelist': {'pkg': {'basearchonly': None, 'blocked': False, 'group_id': 3,
|
||||
'requires': None, 'tag_id': 1, 'type': 'mandatory'}},
|
||||
'grouplist': {}}}
|
||||
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_pkg_add(self.tag, self.group, self.pkg)
|
||||
self.assertEqual(f"group {self.group} not present in tag {self.tag}", str(ex.exception))
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=True, incl_reqs=False)
|
||||
|
||||
def test_grp_pkg_add_group_blocked(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.return_value = self.groupinfo
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': True,
|
||||
'packagelist': {self.pkg: {'basearchonly': None, 'blocked': False, 'group_id': 3,
|
||||
'requires': None, 'tag_id': 1, 'type': 'mandatory'}},
|
||||
'grouplist': {}}}
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_pkg_add(self.tag, self.group, self.pkg)
|
||||
self.assertEqual(f"group {self.group} is blocked in tag {self.tag}", str(ex.exception))
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_called_once_with(self.group, strict=True)
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=True, incl_reqs=False)
|
||||
|
||||
def test_grp_req_remove(self):
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.return_value = self.groupinfo['id']
|
||||
kojihub.grp_pkg_remove(self.taginfo, self.groupinfo, self.pkg)
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_package_listing')
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.clauses, ['package=%(pkg_name)s', 'tag_id=%(tag_id)s',
|
||||
'group_id = %(grp_id)s', 'active = TRUE'])
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_called_once_with(self.groupinfo, strict=True)
|
||||
|
||||
@mock.patch('kojihub.kojihub.grp_pkg_add')
|
||||
def test_grp_pkg_block(self, grp_pkg_add):
|
||||
grp_pkg_add.return_value = None
|
||||
rv = kojihub.grp_pkg_block(self.taginfo, self.groupinfo, self.pkg)
|
||||
self.assertEqual(rv, None)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
grp_pkg_add.assert_called_once_with(self.taginfo, self.groupinfo, self.pkg, block=True)
|
||||
|
||||
def test_grp_pkg_unblock_not_blocked_groups(self):
|
||||
self.query_singleValue.return_value = None
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.return_value = self.groupinfo['id']
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_pkg_unblock(self.taginfo, self.groupinfo, self.pkg)
|
||||
self.assertEqual(f"package {self.pkg} is NOT blocked in group {self.groupinfo['id']}, "
|
||||
f"tag {self.taginfo['id']}", str(ex.exception))
|
||||
|
||||
self.assertEqual(len(self.queries), 1)
|
||||
query = self.queries[0]
|
||||
self.assertEqual(query.tables, ['group_package_listing'])
|
||||
self.assertEqual(query.joins, None)
|
||||
self.assertEqual(query.clauses,
|
||||
['active = TRUE', 'group_id=%(grp_id)s', 'package = %(pkg_name)s',
|
||||
'tag_id=%(tag_id)s'])
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_called_once_with(self.groupinfo, strict=True)
|
||||
|
||||
def test_grp_pkg_unblock_valid(self):
|
||||
self.query_singleValue.return_value = 123
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.return_value = self.groupinfo['id']
|
||||
kojihub.grp_pkg_unblock(self.taginfo, self.groupinfo, self.pkg)
|
||||
|
||||
self.assertEqual(len(self.queries), 1)
|
||||
query = self.queries[0]
|
||||
self.assertEqual(query.tables, ['group_package_listing'])
|
||||
self.assertEqual(query.joins, None)
|
||||
self.assertEqual(query.clauses,
|
||||
['active = TRUE', 'group_id=%(grp_id)s', 'package = %(pkg_name)s',
|
||||
'tag_id=%(tag_id)s'])
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_package_listing')
|
||||
self.assertEqual(update.clauses, ['group_id=%(grp_id)s', 'tag_id=%(tag_id)s',
|
||||
'package = %(pkg_name)s', 'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_called_once_with(self.groupinfo, strict=True)
|
||||
332
tests/test_hub/test_group_req_operations.py
Normal file
332
tests/test_hub/test_group_req_operations.py
Normal file
|
|
@ -0,0 +1,332 @@
|
|||
import mock
|
||||
import unittest
|
||||
import koji
|
||||
import kojihub
|
||||
|
||||
QP = kojihub.QueryProcessor
|
||||
IP = kojihub.InsertProcessor
|
||||
UP = kojihub.UpdateProcessor
|
||||
|
||||
|
||||
class TestGroupReqlist(unittest.TestCase):
|
||||
def getQuery(self, *args, **kwargs):
|
||||
query = QP(*args, **kwargs)
|
||||
query.execute = 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.maxDiff = None
|
||||
self.context = mock.patch('kojihub.kojihub.context').start()
|
||||
self.context_db = mock.patch('kojihub.db.context').start()
|
||||
self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start()
|
||||
self.get_group_id = mock.patch('kojihub.kojihub.get_group_id').start()
|
||||
self.lookup_tag = mock.patch('kojihub.kojihub.lookup_tag').start()
|
||||
self.lookup_group = mock.patch('kojihub.kojihub.lookup_group').start()
|
||||
self.get_tag_groups = mock.patch('kojihub.kojihub.get_tag_groups').start()
|
||||
# It seems MagicMock will not automatically handle attributes that
|
||||
# start with "assert"
|
||||
self.context.session.assertPerm = mock.MagicMock()
|
||||
self.context_db.session.assertLogin = mock.MagicMock()
|
||||
|
||||
self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
|
||||
side_effect=self.getQuery).start()
|
||||
self.queries = []
|
||||
self.query_singleValue = mock.MagicMock()
|
||||
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.tag = 'tag'
|
||||
self.group = 'group'
|
||||
self.group_req = 'group_req'
|
||||
self.taginfo = {'name': self.tag, 'id': 1}
|
||||
self.groupinfo = {'name': self.group, 'id': 2}
|
||||
self.reqinfo = {'name': self.group_req, 'id': 3}
|
||||
self.context_db.event_id = 42
|
||||
self.context_db.session.user_id = 24
|
||||
|
||||
def test_grp_req_add_not_previous(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.side_effect = [self.groupinfo, self.reqinfo]
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {},
|
||||
'grouplist': {}}}
|
||||
|
||||
kojihub.grp_req_add(self.tag, self.group, self.group_req, block=True)
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_req_listing')
|
||||
self.assertEqual(update.clauses,
|
||||
['group_id=%(group_id)s', 'tag_id=%(tag_id)s', 'req_id=%(req_id)s',
|
||||
'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
# insert new group
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'blocked': True,
|
||||
'create_event': 42,
|
||||
'creator_id': 24,
|
||||
'group_id': self.groupinfo['id'],
|
||||
'req_id': self.reqinfo['id'],
|
||||
'tag_id': self.taginfo['id'],
|
||||
'type': 'mandatory'
|
||||
}
|
||||
self.assertEqual(insert.table, 'group_req_listing')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_has_calls([
|
||||
mock.call(self.group, create=False, strict=True),
|
||||
mock.call(self.group_req, create=False, strict=True),
|
||||
])
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=False, incl_reqs=True)
|
||||
|
||||
def test_grp_req_add_previous(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.side_effect = [self.groupinfo, self.reqinfo]
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {},
|
||||
'grouplist': {3: {'blocked': False, 'group_id': 2, 'is_metapkg': False,
|
||||
'req_id': 3, 'tag_id': 1, 'type': 'mandatory'}}}}
|
||||
|
||||
kojihub.grp_req_add(self.tag, self.group, self.group_req, block=True, is_metapkg=True)
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_req_listing')
|
||||
self.assertEqual(update.clauses,
|
||||
['group_id=%(group_id)s', 'tag_id=%(tag_id)s', 'req_id=%(req_id)s',
|
||||
'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
# insert new group
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'blocked': True,
|
||||
'create_event': 42,
|
||||
'creator_id': 24,
|
||||
'group_id': self.groupinfo['id'],
|
||||
'is_metapkg': True,
|
||||
'req_id': self.reqinfo['id'],
|
||||
'tag_id': self.taginfo['id'],
|
||||
'type': 'mandatory'
|
||||
}
|
||||
self.assertEqual(insert.table, 'group_req_listing')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_has_calls([
|
||||
mock.call(self.group, create=False, strict=True),
|
||||
mock.call(self.group_req, create=False, strict=True),
|
||||
])
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=False, incl_reqs=True)
|
||||
|
||||
def test_grp_req_add_previous_not_changed(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.side_effect = [self.groupinfo, self.reqinfo]
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {},
|
||||
'grouplist': {3: {'blocked': False, 'group_id': 2, 'is_metapkg': False,
|
||||
'req_id': 3, 'tag_id': 1, 'type': 'mandatory'}}}}
|
||||
|
||||
kojihub.grp_req_add(self.tag, self.group, self.group_req, block=False, is_metapkg=False)
|
||||
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_has_calls([
|
||||
mock.call(self.group, create=False, strict=True),
|
||||
mock.call(self.group_req, create=False, strict=True),
|
||||
])
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=False, incl_reqs=True)
|
||||
|
||||
def test_grp_req_add_previous_blocked(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.side_effect = [self.groupinfo, self.reqinfo]
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': False, 'display_name': 'group-1', 'group_id': 2, 'tag_id': 1,
|
||||
'packagelist': {},
|
||||
'grouplist': {3: {'blocked': True}}}}
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_req_add(self.tag, self.group, self.group_req)
|
||||
self.assertEqual(f"requirement on group {self.group_req} blocked in group {self.group}, "
|
||||
f"tag {self.tag}", str(ex.exception))
|
||||
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_has_calls([
|
||||
mock.call(self.group, create=False, strict=True),
|
||||
mock.call(self.group_req, create=False, strict=True),
|
||||
])
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=False, incl_reqs=True)
|
||||
|
||||
def test_grp_req_add_not_group_in_tag(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.side_effect = [self.groupinfo, self.reqinfo]
|
||||
self.get_tag_groups.return_value = {
|
||||
111: {'blocked': False, 'display_name': 'group-1', 'group_id': 111, 'tag_id': 1,
|
||||
'packagelist': {},
|
||||
'grouplist': {8: {'blocked': False, 'group_id': 111, 'is_metapkg': False,
|
||||
'req_id': 8, 'tag_id': 1, 'type': 'mandatory'}}}}
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_req_add(self.tag, self.group, self.group_req)
|
||||
self.assertEqual(f"group {self.group} not present in tag {self.tag}", str(ex.exception))
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_has_calls([
|
||||
mock.call(self.group, create=False, strict=True),
|
||||
mock.call(self.group_req, create=False, strict=True),
|
||||
])
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=False, incl_reqs=True)
|
||||
|
||||
def test_grp_req_add_group_blocked(self):
|
||||
self.lookup_tag.return_value = self.taginfo
|
||||
self.lookup_group.side_effect = [self.groupinfo, self.reqinfo]
|
||||
self.get_tag_groups.return_value = {
|
||||
2: {'blocked': True,
|
||||
'grouplist': {8: {'blocked': False, 'group_id': 2}}}}
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_req_add(self.tag, self.group, self.group_req)
|
||||
self.assertEqual(f"group {self.group} is blocked in tag {self.tag}", str(ex.exception))
|
||||
|
||||
self.lookup_tag.assert_called_once_with(self.tag, strict=True)
|
||||
self.lookup_group.assert_has_calls([
|
||||
mock.call(self.group, create=False, strict=True),
|
||||
mock.call(self.group_req, create=False, strict=True),
|
||||
])
|
||||
self.get_tag_groups.assert_called_once_with(
|
||||
self.taginfo['id'], inherit=True, incl_pkgs=False, incl_reqs=True)
|
||||
|
||||
def test_grp_req_remove_force(self):
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.side_effect = [self.groupinfo['id'], self.reqinfo['id']]
|
||||
with mock.patch('kojihub.kojihub.logger') as logger:
|
||||
kojihub.grp_req_remove(self.taginfo, self.groupinfo, self.reqinfo, force=True)
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_req_listing')
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.clauses, ['req_id=%(req_id)s', 'tag_id=%(tag_id)s',
|
||||
'group_id = %(grp_id)s', 'active = TRUE'])
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
|
||||
logger.warning.assert_called_once()
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_has_calls([
|
||||
mock.call(self.groupinfo, strict=True),
|
||||
mock.call(self.reqinfo, strict=True),
|
||||
])
|
||||
|
||||
def test_grp_req_remove_without_force(self):
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.side_effect = [self.groupinfo['id'], self.reqinfo['id']]
|
||||
with mock.patch('kojihub.kojihub.logger') as logger:
|
||||
kojihub.grp_req_remove(self.taginfo, self.groupinfo, self.reqinfo)
|
||||
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_req_listing')
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.clauses, ['req_id=%(req_id)s', 'tag_id=%(tag_id)s',
|
||||
'group_id = %(grp_id)s', 'active = TRUE'])
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
logger.warning.assert_not_called()
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_has_calls([
|
||||
mock.call(self.groupinfo, strict=True),
|
||||
mock.call(self.reqinfo, strict=True),
|
||||
])
|
||||
|
||||
@mock.patch('kojihub.kojihub.grp_req_add')
|
||||
def test_grp_req_block(self, grp_req_add):
|
||||
grp_req_add.return_value = None
|
||||
rv = kojihub.grp_req_block(self.taginfo, self.groupinfo, self.reqinfo)
|
||||
self.assertEqual(rv, None)
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
grp_req_add.assert_called_once_with(self.taginfo, self.groupinfo, self.reqinfo, block=True)
|
||||
|
||||
def test_grp_req_unblock_not_blocked_groups(self):
|
||||
self.query_singleValue.return_value = None
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.side_effect = [self.groupinfo['id'], self.reqinfo['id']]
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.grp_req_unblock(self.taginfo, self.groupinfo, self.reqinfo)
|
||||
self.assertEqual(f"group req {self.reqinfo['id']} is NOT blocked in group "
|
||||
f"{self.groupinfo['id']}, tag {self.taginfo['id']}", str(ex.exception))
|
||||
|
||||
self.assertEqual(len(self.queries), 1)
|
||||
query = self.queries[0]
|
||||
self.assertEqual(query.tables, ['group_req_listing'])
|
||||
self.assertEqual(query.joins, None)
|
||||
self.assertEqual(query.clauses,
|
||||
['active = TRUE', 'group_id=%(grp_id)s', 'req_id = %(req_id)s',
|
||||
'tag_id=%(tag_id)s'])
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_has_calls([
|
||||
mock.call(self.groupinfo, strict=True),
|
||||
mock.call(self.reqinfo, strict=True),
|
||||
])
|
||||
|
||||
def test_grp_req_unblock_valid(self):
|
||||
self.query_singleValue.return_value = 123
|
||||
self.get_tag_id.return_value = self.taginfo['id']
|
||||
self.get_group_id.side_effect = [self.groupinfo['id'], self.reqinfo['id']]
|
||||
kojihub.grp_req_unblock(self.taginfo, self.groupinfo, self.reqinfo)
|
||||
|
||||
self.assertEqual(len(self.queries), 1)
|
||||
query = self.queries[0]
|
||||
self.assertEqual(query.tables, ['group_req_listing'])
|
||||
self.assertEqual(query.joins, None)
|
||||
self.assertEqual(query.clauses,
|
||||
['active = TRUE', 'group_id=%(grp_id)s', 'req_id = %(req_id)s',
|
||||
'tag_id=%(tag_id)s'])
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'group_req_listing')
|
||||
self.assertEqual(update.clauses, ['group_id=%(grp_id)s', 'tag_id=%(tag_id)s',
|
||||
'req_id = %(req_id)s', 'active = TRUE'])
|
||||
self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 24})
|
||||
self.assertEqual(update.rawdata, {'active': 'NULL'})
|
||||
|
||||
self.get_tag_id.assert_called_once_with(self.taginfo, strict=True)
|
||||
self.get_group_id.assert_has_calls([
|
||||
mock.call(self.groupinfo, strict=True),
|
||||
mock.call(self.reqinfo, strict=True),
|
||||
])
|
||||
|
|
@ -4,6 +4,10 @@ import unittest
|
|||
import koji
|
||||
import kojihub
|
||||
|
||||
QP = kojihub.QueryProcessor
|
||||
IP = kojihub.InsertProcessor
|
||||
UP = kojihub.UpdateProcessor
|
||||
|
||||
|
||||
def get_user_factory(data):
|
||||
def get_user(userInfo, strict=False):
|
||||
|
|
@ -21,6 +25,21 @@ def get_user_factory(data):
|
|||
|
||||
|
||||
class TestPkglist(unittest.TestCase):
|
||||
|
||||
def getInsert(self, *args, **kwargs):
|
||||
insert = IP(*args, **kwargs)
|
||||
insert.execute = mock.MagicMock()
|
||||
insert.make_create = mock.MagicMock()
|
||||
self.inserts.append(insert)
|
||||
return insert
|
||||
|
||||
def getUpdate(self, *args, **kwargs):
|
||||
update = UP(*args, **kwargs)
|
||||
update.execute = mock.MagicMock()
|
||||
update.make_revoke = mock.MagicMock()
|
||||
self.updates.append(update)
|
||||
return update
|
||||
|
||||
def setUp(self):
|
||||
self.context = mock.patch('kojihub.kojihub.context').start()
|
||||
# It seems MagicMock will not automatically handle attributes that
|
||||
|
|
@ -28,171 +47,208 @@ class TestPkglist(unittest.TestCase):
|
|||
self.context.session.assertLogin = mock.MagicMock()
|
||||
self.context.session.user_id = 112233
|
||||
self.context.session.user_data = {'name': 'username'}
|
||||
self.context_db = mock.patch('kojihub.db.context').start()
|
||||
self.context_db.event_id = 42
|
||||
self.context_db.session.user_id = 24
|
||||
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.run_callbacks = mock.patch('koji.plugin.run_callbacks').start()
|
||||
self.read_package_list = mock.patch('kojihub.kojihub.readPackageList').start()
|
||||
self.lookup_package = mock.patch('kojihub.kojihub.lookup_package').start()
|
||||
self._pkglist_add = mock.patch('kojihub.kojihub._pkglist_add').start()
|
||||
self._pkglist_owner_add = mock.patch('kojihub.kojihub._pkglist_owner_add').start()
|
||||
self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()
|
||||
self._pkglist_remove = mock.patch('kojihub.kojihub._pkglist_remove').start()
|
||||
self.assert_policy = mock.patch('kojihub.kojihub.assert_policy').start()
|
||||
self.get_user = mock.patch('kojihub.kojihub.get_user').start()
|
||||
self.tag = {'name': 'tag', 'id': 123}
|
||||
self.pkg = {'name': 'pkg', 'id': 321, 'owner_id': 3}
|
||||
self.users = [
|
||||
{'id': 3, 'name': 'user'},
|
||||
{'id': 112233, 'name': 'user'},
|
||||
]
|
||||
self.arches = 'arches'
|
||||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
|
||||
def test_pkglist_remove(self):
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.get_user.return_value = self.users[1]
|
||||
kojihub.pkglist_remove(self.tag['name'], self.pkg['name'])
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'remove', 'package': self.pkg['id'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='remove', tag=self.tag, package=self.pkg,
|
||||
user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='remove', tag=self.tag, package=self.pkg,
|
||||
user=self.users[1]),
|
||||
])
|
||||
self._pkglist_remove.assert_called_once_with(self.tag['id'], self.pkg['id'])
|
||||
|
||||
@mock.patch('kojihub.kojihub.pkglist_add')
|
||||
def test_pkglist_block(self, pkglist_add):
|
||||
force = mock.MagicMock()
|
||||
self.get_tag.return_value = {'name': 'tag', 'id': 123}
|
||||
self.lookup_package.return_value = {'name': 'pkg', 'id': 321}
|
||||
self.read_package_list.return_value = ['pkg']
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.read_package_list.return_value = [self.pkg['name']]
|
||||
|
||||
kojihub.pkglist_block('tag', 'pkg', force=force)
|
||||
kojihub.pkglist_block(self.tag['name'], self.pkg['name'], force=force)
|
||||
|
||||
self.get_tag.assert_called_once_with('tag', strict=True)
|
||||
self.lookup_package.assert_called_once_with('pkg', strict=True)
|
||||
pkglist_add.assert_called_once_with('tag', 'pkg', block=True, force=force)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=True)
|
||||
pkglist_add.assert_called_once_with(
|
||||
self.tag['name'], self.pkg['name'], block=True, force=force)
|
||||
|
||||
@mock.patch('kojihub.kojihub.pkglist_add')
|
||||
def test_pkglist_block_package_error(self, pkglist_add):
|
||||
pkg_name = 'pkg'
|
||||
tag_name = 'tag'
|
||||
force = mock.MagicMock()
|
||||
self.get_tag.return_value = {'name': tag_name, 'id': 123}
|
||||
self.lookup_package.return_value = {'name': pkg_name, 'id': 321}
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.read_package_list.return_value = []
|
||||
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub.pkglist_block('tag', 'pkg', force=force)
|
||||
self.assertEqual("Package %s is not in tag listing for %s" % (pkg_name, tag_name),
|
||||
str(ex.exception))
|
||||
kojihub.pkglist_block(self.tag['name'], self.pkg['name'], force=force)
|
||||
self.assertEqual(
|
||||
f"Package {self.pkg['name']} is not in tag listing for {self.tag['name']}",
|
||||
str(ex.exception))
|
||||
|
||||
self.get_tag.assert_called_once_with('tag', strict=True)
|
||||
self.lookup_package.assert_called_once_with('pkg', strict=True)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=True)
|
||||
pkglist_add.assert_not_called()
|
||||
|
||||
def test_pkglist_unblock(self,):
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'package', 'owner_id': 3}
|
||||
self.get_tag.return_value = tag
|
||||
self.lookup_package.return_value = pkg
|
||||
self.read_package_list.return_value = {pkg['id']: {
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'blocked': True,
|
||||
'tag_id': tag['id'],
|
||||
'owner_id': pkg['owner_id'],
|
||||
'tag_id': self.tag['id'],
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'extra_arches': ''}}
|
||||
users = [
|
||||
{'id': 3, 'name': 'user'},
|
||||
{'id': 112233, 'name': 'user'},
|
||||
]
|
||||
user = users[1]
|
||||
self.get_user.side_effect = get_user_factory(users)
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
|
||||
kojihub.pkglist_unblock('tag', 'pkg', force=False)
|
||||
kojihub.pkglist_unblock(self.tag['name'], self.pkg['name'], force=False)
|
||||
|
||||
self.get_tag.assert_called_once_with('tag', strict=True)
|
||||
self.lookup_package.assert_called_once_with('pkg', strict=True)
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag['id'], 'action': 'unblock', 'package': pkg['id'],
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'unblock', 'package': self.pkg['id'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.assertEqual(self.read_package_list.call_count, 2)
|
||||
self.read_package_list.assert_has_calls([
|
||||
mock.call(tag['id'], pkgID=pkg['id'], inherit=True),
|
||||
mock.call(tag['id'], pkgID=pkg['id'], inherit=True),
|
||||
mock.call(self.tag['id'], pkgID=self.pkg['id'], inherit=True),
|
||||
mock.call(self.tag['id'], pkgID=self.pkg['id'], inherit=True),
|
||||
])
|
||||
self._pkglist_add.assert_called_once_with(tag['id'], pkg['id'], pkg['owner_id'], False, '')
|
||||
self._pkglist_remove.assert_called_once_with(tag['id'], pkg['id'])
|
||||
self._pkglist_add.assert_called_once_with(
|
||||
self.tag['id'], self.pkg['id'], self.pkg['owner_id'], False, '')
|
||||
self._pkglist_remove.assert_called_once_with(self.tag['id'], self.pkg['id'])
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='unblock', tag=tag, package=pkg, user=user),
|
||||
mock.call('postPackageListChange', action='unblock', tag=tag, package=pkg, user=user),
|
||||
mock.call('prePackageListChange', action='unblock', tag=self.tag, package=self.pkg,
|
||||
user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='unblock', tag=self.tag, package=self.pkg,
|
||||
user=self.users[1]),
|
||||
])
|
||||
|
||||
def test_pkglist_unblock_inherited(self):
|
||||
tag_id, pkg_id, owner_id = 1, 2, 3
|
||||
self.get_tag.return_value = {'id': tag_id, 'name': 'tag'}
|
||||
self.lookup_package.return_value = {'id': pkg_id, 'name': 'pkg'}
|
||||
self.read_package_list.return_value = {pkg_id: {
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'blocked': True,
|
||||
'tag_id': 4,
|
||||
'owner_id': owner_id,
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'extra_arches': ''}}
|
||||
|
||||
kojihub.pkglist_unblock('tag', 'pkg', force=False)
|
||||
kojihub.pkglist_unblock(self.tag['name'], self.pkg['name'], force=False)
|
||||
|
||||
self.get_tag.assert_called_once_with('tag', strict=True)
|
||||
self.lookup_package.assert_called_once_with('pkg', strict=True)
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag_id, 'action': 'unblock', 'package': pkg_id,
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'unblock', 'package': self.pkg['id'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.read_package_list.assert_called_once_with(tag_id, pkgID=pkg_id, inherit=True)
|
||||
self._pkglist_add.assert_called_once_with(tag_id, pkg_id, owner_id, False, '')
|
||||
self.read_package_list.assert_called_once_with(
|
||||
self.tag['id'], pkgID=self.pkg['id'], inherit=True)
|
||||
self._pkglist_add.assert_called_once_with(
|
||||
self.tag['id'], self.pkg['id'], self.pkg['owner_id'], False, '')
|
||||
self._pkglist_remove.assert_not_called()
|
||||
|
||||
def test_pkglist_unblock_not_present(self):
|
||||
tag_id, pkg_id = 1, 2
|
||||
self.get_tag.return_value = {'id': tag_id, 'name': 'tag'}
|
||||
self.lookup_package.return_value = {'id': pkg_id, 'name': 'pkg'}
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
kojihub.pkglist_unblock('tag', 'pkg', force=False)
|
||||
kojihub.pkglist_unblock(self.tag['name'], self.pkg['name'], force=False)
|
||||
|
||||
self.get_tag.assert_called_once_with('tag', strict=True)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with('pkg', strict=True)
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag_id, 'action': 'unblock', 'package': pkg_id,
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'unblock', 'package': self.pkg['id'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.read_package_list.assert_called_once_with(tag_id, pkgID=pkg_id, inherit=True)
|
||||
self.read_package_list.assert_called_once_with(
|
||||
self.tag['id'], pkgID=self.pkg['id'], inherit=True)
|
||||
self._pkglist_add.assert_not_called()
|
||||
self._pkglist_remove.assert_not_called()
|
||||
|
||||
def test_pkglist_unblock_not_blocked(self):
|
||||
tag_id, pkg_id, owner_id = 1, 2, 3
|
||||
self.get_tag.return_value = {'id': tag_id, 'name': 'tag'}
|
||||
self.lookup_package.return_value = {'id': pkg_id, 'name': 'pkg'}
|
||||
self.read_package_list.return_value = {pkg_id: {
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'blocked': False,
|
||||
'tag_id': tag_id,
|
||||
'owner_id': owner_id,
|
||||
'tag_id': self.tag['id'],
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'extra_arches': ''}}
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
kojihub.pkglist_unblock('tag', 'pkg', force=False)
|
||||
kojihub.pkglist_unblock(self.tag['name'], self.pkg['name'], force=False)
|
||||
|
||||
self.get_tag.assert_called_once_with('tag', strict=True)
|
||||
self.lookup_package.assert_called_once_with('pkg', strict=True)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=True)
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag_id, 'action': 'unblock', 'package': pkg_id,
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'unblock', 'package': self.pkg['id'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.read_package_list.assert_called_once_with(tag_id, pkgID=pkg_id, inherit=True)
|
||||
self.read_package_list.assert_called_once_with(
|
||||
self.tag['id'], pkgID=self.pkg['id'], inherit=True)
|
||||
self._pkglist_add.assert_not_called()
|
||||
self._pkglist_remove.assert_not_called()
|
||||
|
||||
@mock.patch('kojihub.kojihub.pkglist_add')
|
||||
def test_pkglist_setowner(self, pkglist_add):
|
||||
force = mock.MagicMock()
|
||||
kojihub.pkglist_setowner('tag', 'pkg', 'owner', force=force)
|
||||
pkglist_add.assert_called_once_with('tag', 'pkg', owner='owner', force=force, update=True)
|
||||
kojihub.pkglist_setowner(
|
||||
self.tag['name'], self.pkg['name'], self.users[0]['name'], force=force)
|
||||
pkglist_add.assert_called_once_with(
|
||||
self.tag['name'], self.pkg['name'], owner=self.users[0]['name'], force=force,
|
||||
update=True)
|
||||
|
||||
@mock.patch('kojihub.kojihub.pkglist_add')
|
||||
def test_pkglist_setarches(self, pkglist_add):
|
||||
force = mock.MagicMock()
|
||||
kojihub.pkglist_setarches('tag', 'pkg', 'arches', force=force)
|
||||
pkglist_add.assert_called_once_with('tag', 'pkg', extra_arches='arches', force=force,
|
||||
update=True)
|
||||
kojihub.pkglist_setarches(self.tag['name'], self.pkg['name'], self.arches, force=force)
|
||||
pkglist_add.assert_called_once_with(self.tag['name'], self.pkg['name'],
|
||||
extra_arches=self.arches, force=force, update=True)
|
||||
|
||||
@mock.patch('kojihub.kojihub._direct_pkglist_add')
|
||||
def test_pkglist_add(self, _direct_pkglist_add):
|
||||
# just transition of params + policy=True
|
||||
kojihub.pkglist_add('tag', 'pkg', owner='owner', block='block',
|
||||
extra_arches='extra_arches', force='force', update='update')
|
||||
_direct_pkglist_add.assert_called_once_with('tag', 'pkg', 'owner', 'block', 'extra_arches',
|
||||
'force', 'update', policy=True)
|
||||
kojihub.pkglist_add(self.tag['name'], self.pkg['name'], owner=self.users[0]['name'],
|
||||
block='block', extra_arches=self.arches, force='force',
|
||||
update='update')
|
||||
_direct_pkglist_add.assert_called_once_with(
|
||||
self.tag['name'], self.pkg['name'], self.users[0]['name'], 'block', self.arches,
|
||||
'force', 'update', policy=True)
|
||||
|
||||
def test_direct_pkglist_add(self):
|
||||
block = False
|
||||
|
|
@ -200,43 +256,37 @@ class TestPkglist(unittest.TestCase):
|
|||
force = False
|
||||
update = False
|
||||
policy = True
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
users = [
|
||||
{'id': 3, 'name': 'user'},
|
||||
{'id': 112233, 'name': 'user'},
|
||||
]
|
||||
user = users[0]
|
||||
self.get_tag.return_value = tag
|
||||
self.lookup_package.return_value = pkg
|
||||
self.get_user.side_effect = get_user_factory(users)
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg['name'], user['name'], block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(pkg['name'], strict=False)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=False)
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(user['name'], strict=True),
|
||||
mock.call(112233),
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag['id'], 'action': 'add', 'package': pkg['name'],
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'add', 'package': self.pkg['name'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='add', tag=tag, package=pkg, owner=user['id'],
|
||||
mock.call('prePackageListChange', action='add', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[0]['id'],
|
||||
block=block, extra_arches=extra_arches, force=force, update=update,
|
||||
user=users[1]),
|
||||
mock.call('postPackageListChange', action='add', tag=tag, package=pkg,
|
||||
owner=user['id'], block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, user=users[1]),
|
||||
user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='add', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
])
|
||||
self._pkglist_add.assert_called_once_with(
|
||||
tag['id'], pkg['id'], user['id'], block, extra_arches)
|
||||
self.tag['id'], self.pkg['id'], self.users[0]['id'], block, extra_arches)
|
||||
|
||||
def test_direct_pkglist_add_no_package(self):
|
||||
block = False
|
||||
|
|
@ -244,104 +294,292 @@ class TestPkglist(unittest.TestCase):
|
|||
force = False
|
||||
update = False
|
||||
policy = True
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
user = {'id': 3, 'name': 'user'}
|
||||
self.get_tag.return_value = tag
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = None
|
||||
self.get_user.return_value = user
|
||||
self.get_user.return_value = self.users[0]
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
# package needs to be name, not dict
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg, user['name'], block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
self.assertEqual("No such package: %s" % pkg, str(ex.exception))
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['id'], self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
self.assertEqual(f"No such package: {self.pkg['id']}", str(ex.exception))
|
||||
|
||||
def test_direct_pkglist_add_not_previous_update_without_force(self):
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.get_user.return_value = self.users[0]
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
# package needs to be name, not dict
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=False, extra_arches='arch123', force=False,
|
||||
update=True, policy=False)
|
||||
self.assertEqual(
|
||||
f"cannot update: tag {self.tag['name']} has no data for package {self.pkg['name']}",
|
||||
str(ex.exception))
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 1)
|
||||
self.lookup_package.has_calls(mock.call(self.pkg['name'], strict=False))
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 0)
|
||||
self.run_callbacks.assert_not_called()
|
||||
self._pkglist_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_pkginfo_dict(self):
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
user = 'user'
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
expected = "Invalid type for id lookup: %s" % pkg
|
||||
expected = f"Invalid type for id lookup: {self.pkg['name']}"
|
||||
|
||||
self.get_tag.return_value = tag
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = koji.GenericError(expected)
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg, user, block=False, extra_arches='arch',
|
||||
force=False, update=True)
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=False, extra_arches='arch', force=False, update=True)
|
||||
self.assertEqual(expected, str(ex.exception))
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 1)
|
||||
self.lookup_package.has_calls(mock.call(self.pkg['name'], strict=False))
|
||||
self.get_user.assert_not_called()
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 0)
|
||||
self.run_callbacks.assert_not_called()
|
||||
self._pkglist_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_no_user(self):
|
||||
block = False
|
||||
extra_arches = 'arch123'
|
||||
force = False
|
||||
update = False
|
||||
policy = True
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
user = {'id': 3, 'name': 'user'}
|
||||
self.get_tag.return_value = tag
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = None
|
||||
self.get_user.side_effect = koji.GenericError
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg, user['name'], block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
|
||||
self.lookup_package.assert_called_once_with(pkg, strict=False)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=False)
|
||||
self.assertEqual(self.run_callbacks.call_count, 0)
|
||||
self._pkglist_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_previous_not_changed(self):
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = [None, self.pkg]
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'blocked': False,
|
||||
'extra_arches': 'x64_64'}
|
||||
}
|
||||
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=False, extra_arches='x64_64', force=False, update=False,
|
||||
policy=False)
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 2)
|
||||
self.lookup_package.has_calls(
|
||||
mock.call(self.pkg['name'], strict=False),
|
||||
mock.call(self.pkg['name'], create=True),
|
||||
)
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 0)
|
||||
self.run_callbacks.assert_not_called()
|
||||
self._pkglist_add.assert_not_called()
|
||||
self._pkglist_owner_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_not_previous_none_owner_without_force(self):
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = [None, self.pkg]
|
||||
self.get_user.return_value = self.users[1]
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], owner=None,
|
||||
block=False, extra_arches='x64_64', force=False,
|
||||
update=False,
|
||||
policy=False)
|
||||
self.assertEqual("owner not specified", str(ex.exception))
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 2)
|
||||
self.lookup_package.has_calls(
|
||||
mock.call(self.pkg['name'], strict=False),
|
||||
mock.call(self.pkg['name'], create=True),
|
||||
)
|
||||
self.get_user.assert_called_once_with(self.users[1]['id'])
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 0)
|
||||
self.run_callbacks.assert_not_called()
|
||||
self._pkglist_add.assert_not_called()
|
||||
self._pkglist_owner_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_not_previous_none_owner_with_force_act_block(self):
|
||||
block = True
|
||||
extra_arches = 'x86_64'
|
||||
force = True
|
||||
update = False
|
||||
policy = False
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = [None, self.pkg]
|
||||
self.get_user.return_value = self.users[1]
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], owner=None,
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update,
|
||||
policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 2)
|
||||
self.lookup_package.has_calls(
|
||||
mock.call(self.pkg['name'], strict=False),
|
||||
mock.call(self.pkg['name'], create=True),
|
||||
)
|
||||
self.get_user.assert_called_once_with(self.users[1]['id'])
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='block', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[1]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='block', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[1]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
])
|
||||
self._pkglist_add.assert_called_once_with(
|
||||
self.tag['id'], self.pkg['id'], self.users[1]['id'], block, extra_arches)
|
||||
self._pkglist_owner_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_previous_owner_block_extra_arches_none(self):
|
||||
block = None
|
||||
extra_arches = None
|
||||
force = False
|
||||
update = True
|
||||
policy = False
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = [None, self.pkg]
|
||||
self.get_user.return_value = self.users[1]
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'blocked': False,
|
||||
'extra_arches': 'x64_64'}
|
||||
}
|
||||
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], owner=None, block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 2)
|
||||
self.lookup_package.has_calls(
|
||||
mock.call(self.pkg['name'], strict=False),
|
||||
mock.call(self.pkg['name'], create=True),
|
||||
)
|
||||
self.get_user.assert_called_once_with(self.users[1]['id'])
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 0)
|
||||
self.run_callbacks.assert_not_called()
|
||||
self._pkglist_add.assert_not_called()
|
||||
self._pkglist_owner_add.assert_not_called()
|
||||
|
||||
def test_direct_pkglist_add_previous_change_owner_only(self):
|
||||
block = False
|
||||
extra_arches = 'x64_64'
|
||||
force = False
|
||||
update = True
|
||||
policy = False
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 2}
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = [None, pkg]
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
self.read_package_list.return_value = {pkg['id']: {
|
||||
'owner_id': pkg['owner_id'],
|
||||
'blocked': False,
|
||||
'extra_arches': 'x64_64'}
|
||||
}
|
||||
|
||||
kojihub._direct_pkglist_add(self.tag['name'], pkg['name'], owner=self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 2)
|
||||
self.lookup_package.has_calls(
|
||||
mock.call(pkg['name'], strict=False),
|
||||
mock.call(pkg['name'], create=True),
|
||||
)
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
self.assert_policy.assert_not_called()
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='update', tag=self.tag, package=pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='update', tag=self.tag, package=pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
])
|
||||
self._pkglist_add.assert_not_called()
|
||||
self._pkglist_owner_add.assert_called_once_with(self.tag['id'], pkg['id'], 3)
|
||||
|
||||
def test_direct_pkglist_add_new_package(self):
|
||||
block = False
|
||||
extra_arches = 'arch123'
|
||||
force = False
|
||||
update = False
|
||||
policy = True
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
users = [
|
||||
{'id': 3, 'name': 'user'},
|
||||
{'id': 112233, 'name': 'user1'},
|
||||
]
|
||||
user = users[0]
|
||||
self.get_tag.return_value = tag
|
||||
self.lookup_package.side_effect = [None, pkg]
|
||||
self.get_user.side_effect = get_user_factory(users)
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.side_effect = [None, self.pkg]
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
self.read_package_list.return_value = {}
|
||||
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg['name'], user['name'], block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(tag['name'], strict=True)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.assertEqual(self.lookup_package.call_count, 2)
|
||||
self.lookup_package.has_calls(
|
||||
mock.call(pkg['name'], strict=False),
|
||||
mock.call(pkg['name'], create=True),
|
||||
mock.call(self.pkg['name'], strict=False),
|
||||
mock.call(self.pkg['name'], create=True),
|
||||
)
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(user['name'], strict=True),
|
||||
mock.call(112233),
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag['id'], 'action': 'add', 'package': pkg['name'],
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'add', 'package': self.pkg['name'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='add', tag=tag, package=pkg, owner=user['id'],
|
||||
block=block, extra_arches=extra_arches, force=force, update=update,
|
||||
user=users[1]),
|
||||
mock.call('postPackageListChange', action='add', tag=tag, package=pkg,
|
||||
owner=user['id'], block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, user=users[1]),
|
||||
mock.call('prePackageListChange', action='add', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='add', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
])
|
||||
self._pkglist_add.assert_called_once_with(
|
||||
tag['id'], pkg['id'], user['id'], block, extra_arches)
|
||||
self.tag['id'], self.pkg['id'], self.users[0]['id'], block, extra_arches)
|
||||
|
||||
def test_direct_pkglist_add_blocked_previously(self):
|
||||
block = False
|
||||
|
|
@ -349,35 +587,29 @@ class TestPkglist(unittest.TestCase):
|
|||
force = False
|
||||
update = False
|
||||
policy = True
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
users = [
|
||||
{'id': 3, 'name': 'user', },
|
||||
{'id': 112233, 'name': 'user1'},
|
||||
]
|
||||
user = users[0]
|
||||
self.get_tag.return_value = tag
|
||||
self.lookup_package.return_value = pkg
|
||||
self.get_user.side_effect = get_user_factory(users)
|
||||
self.read_package_list.return_value = {pkg['id']: {
|
||||
'owner_id': pkg['owner_id'],
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'blocked': True,
|
||||
'extra_arches': ''}
|
||||
}
|
||||
|
||||
with self.assertRaises(koji.GenericError):
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg['name'], user['name'], block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(pkg['name'], strict=False)
|
||||
with self.assertRaises(koji.GenericError) as ex:
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
self.assertEqual(f"package {self.pkg['name']} is blocked in tag {self.tag['name']}",
|
||||
str(ex.exception))
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=False)
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(user['name'], strict=True),
|
||||
mock.call(112233),
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': tag['id'], 'action': 'add', 'package': pkg['name'],
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'add', 'package': self.pkg['name'],
|
||||
'force': False},
|
||||
force=False)
|
||||
self.run_callbacks.assert_not_called()
|
||||
|
|
@ -389,45 +621,99 @@ class TestPkglist(unittest.TestCase):
|
|||
force = True
|
||||
update = False
|
||||
policy = True
|
||||
tag = {'id': 1, 'name': 'tag'}
|
||||
pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}
|
||||
users = [
|
||||
{'id': 3, 'name': 'user', },
|
||||
{'id': 112233, 'name': 'user1'},
|
||||
]
|
||||
user = users[0]
|
||||
self.get_tag.return_value = tag
|
||||
self.lookup_package.return_value = pkg
|
||||
self.get_user.side_effect = get_user_factory(users)
|
||||
self.read_package_list.return_value = {pkg['id']: {
|
||||
'owner_id': pkg['owner_id'],
|
||||
self.get_tag.return_value = self.tag
|
||||
self.lookup_package.return_value = self.pkg
|
||||
self.get_user.side_effect = get_user_factory(self.users)
|
||||
self.read_package_list.return_value = {self.pkg['id']: {
|
||||
'owner_id': self.pkg['owner_id'],
|
||||
'blocked': True,
|
||||
'extra_arches': ''}
|
||||
}
|
||||
|
||||
kojihub._direct_pkglist_add(tag['name'], pkg['name'], user['name'], block=block,
|
||||
extra_arches=extra_arches, force=force, update=update,
|
||||
policy=policy)
|
||||
kojihub._direct_pkglist_add(self.tag['name'], self.pkg['name'], self.users[0]['name'],
|
||||
block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, policy=policy)
|
||||
|
||||
self.get_tag.assert_called_once_with(tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(pkg['name'], strict=False)
|
||||
self.get_tag.assert_called_once_with(self.tag['name'], strict=True)
|
||||
self.lookup_package.assert_called_once_with(self.pkg['name'], strict=False)
|
||||
self.get_user.assert_has_calls([
|
||||
mock.call(user['name'], strict=True),
|
||||
mock.call(112233),
|
||||
mock.call(self.users[0]['name'], strict=True),
|
||||
mock.call(self.users[1]['id']),
|
||||
])
|
||||
# force + admin
|
||||
self.assert_policy.assert_called_once_with(
|
||||
'package_list', {'tag': 1, 'action': 'add', 'package': 'pkg', 'force': True},
|
||||
'package_list', {'tag': self.tag['id'], 'action': 'add', 'package': 'pkg',
|
||||
'force': True},
|
||||
force=True)
|
||||
|
||||
self.assertEqual(self.run_callbacks.call_count, 2)
|
||||
self.run_callbacks.assert_has_calls([
|
||||
mock.call('prePackageListChange', action='add', tag=tag, package=pkg, owner=user['id'],
|
||||
block=block, extra_arches=extra_arches, force=force, update=update,
|
||||
user=users[1]),
|
||||
mock.call('postPackageListChange', action='add', tag=tag, package=pkg,
|
||||
owner=user['id'], block=block, extra_arches=extra_arches, force=force,
|
||||
update=update, user=users[1]),
|
||||
mock.call('prePackageListChange', action='add', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
mock.call('postPackageListChange', action='add', tag=self.tag, package=self.pkg,
|
||||
owner=self.users[0]['id'], block=block, extra_arches=extra_arches,
|
||||
force=force, update=update, user=self.users[1]),
|
||||
])
|
||||
self._pkglist_add.assert_called_once_with(
|
||||
tag['id'], pkg['id'], user['id'], block, extra_arches)
|
||||
self.tag['id'], self.pkg['id'], self.users[0]['id'], block, extra_arches)
|
||||
|
||||
def test_pkglist_remove_processor(self):
|
||||
kojihub._pkglist_remove(self.tag['id'], self.pkg['id'])
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'tag_packages')
|
||||
self.assertEqual(update.clauses, ['package_id=%(pkg_id)i', 'tag_id=%(tag_id)i'])
|
||||
self.assertEqual(update.data, {})
|
||||
self.assertEqual(update.rawdata, {})
|
||||
|
||||
def test_pkglist_owner_remove_processor(self):
|
||||
kojihub._pkglist_owner_remove(self.tag['id'], self.pkg['id'])
|
||||
|
||||
self.assertEqual(len(self.inserts), 0)
|
||||
self.assertEqual(len(self.updates), 1)
|
||||
update = self.updates[0]
|
||||
self.assertEqual(update.table, 'tag_package_owners')
|
||||
self.assertEqual(update.clauses, ['package_id=%(pkg_id)i', 'tag_id=%(tag_id)i'])
|
||||
self.assertEqual(update.data, {})
|
||||
self.assertEqual(update.rawdata, {})
|
||||
|
||||
@mock.patch('kojihub.kojihub._pkglist_owner_remove')
|
||||
def test_pkglist_owner_add_processor(self, pkglist_owner_remove):
|
||||
pkglist_owner_remove.return_value = None
|
||||
kojihub._pkglist_owner_add(self.tag['id'], self.pkg['id'], self.users[0]['name'])
|
||||
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'owner': self.users[0]['name'],
|
||||
'package_id': self.pkg['id'],
|
||||
'tag_id': self.tag['id']
|
||||
}
|
||||
self.assertEqual(insert.table, 'tag_package_owners')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
||||
@mock.patch('kojihub.kojihub._pkglist_remove')
|
||||
def test_pkglist_add_processor(self, pkglist_remove):
|
||||
pkglist_remove.return_value = None
|
||||
block = False
|
||||
extra_arches = 'test-arch'
|
||||
kojihub._pkglist_add(
|
||||
self.tag['id'], self.pkg['id'], self.users[0]['name'], block, extra_arches)
|
||||
|
||||
self.assertEqual(len(self.updates), 0)
|
||||
self.assertEqual(len(self.inserts), 1)
|
||||
insert = self.inserts[0]
|
||||
values = {
|
||||
'blocked': block,
|
||||
'extra_arches': 'test-arch',
|
||||
'package_id': self.pkg['id'],
|
||||
'tag_id': self.tag['id']
|
||||
}
|
||||
self.assertEqual(insert.table, 'tag_packages')
|
||||
self.assertEqual(insert.data, values)
|
||||
self.assertEqual(insert.rawdata, {})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue