#509 CLI block-group command
Merged 2 years ago by mikem. Opened 2 years ago by tkopecek.
tkopecek/koji issue462  into  master

file modified
+30

@@ -79,6 +79,36 @@ 

      session.groupListAdd(tag, group)

  

  

+ def handle_block_group(goptions, session, args):

+     "[admin] Block group in tag"

+     usage = _("usage: %prog block-group <tag> <group>")

+     usage += _("\n(Specify the --help global option for a list of other help options)")

+     parser = OptionParser(usage=usage)

+     (options, args) = parser.parse_args(args)

+     if len(args) != 2:

+         parser.error(_("Please specify a tag name and a group name"))

+         assert False  # pragma: no cover

+     tag = args[0]

+     group = args[1]

+ 

+     activate_session(session, goptions)

+     if not session.hasPerm('admin'):

+         print("This action requires admin privileges")

+         return 1

+ 

+     dsttag = session.getTag(tag)

+     if not dsttag:

+         print("Unknown tag: %s" % tag)

+         return 1

+ 

+     groups = dict([(p['name'], p['group_id']) for p in session.getTagGroups(tag, inherit=False)])

+     group_id = groups.get(group, None)

+     if group_id is None:

+         print("Group %s doesn't exist within tag %s" % (group, tag))

+         return 1

+ 

+     session.groupListBlock(tag, group)

+ 

  def handle_assign_task(goptions, session, args):

      "[admin] Assign a task to a host"

      usage = _('usage: %prog assign-task task_id hostname')

file modified
+3 -3

@@ -1580,7 +1580,7 @@ 

  

  def _grplist_add(taginfo, grpinfo, block, force, **opts):

      """grplist_add without permission check"""

-     tag = get_tag(taginfo)

+     tag = get_tag(taginfo, strict=True)

      group = lookup_group(grpinfo, create=True)

      block = bool(block)

      # check current group status (incl inheritance)

@@ -1638,8 +1638,8 @@ 

  

  

  def _grplist_remove(taginfo, grpinfo, force):

-     """grplist_remove without permssion check"""

-     tag = get_tag(taginfo)

+     """grplist_remove without permission check"""

+     tag = get_tag(taginfo, strict=True)

      group = lookup_group(grpinfo, strict=True)

      tag_id = tag['id']

      grp_id = group['id']

@@ -14,6 +14,7 @@ 

          add-user                  Add a user

          add-volume                Add a new storage volume

          assign-task               Assign a task to a host

+         block-group               Block group in tag

          block-group-pkg           Block a package from a group's package listing

          block-group-req           Block a group's requirement listing

          block-pkg                 Block a package in the listing for tag

@@ -14,6 +14,7 @@ 

          add-user                  Add a user

          add-volume                Add a new storage volume

          assign-task               Assign a task to a host

+         block-group               Block group in tag

          block-group-pkg           Block a package from a group's package listing

          block-group-req           Block a group's requirement listing

          block-pkg                 Block a package in the listing for tag

@@ -0,0 +1,95 @@ 

+ from __future__ import absolute_import

+ 

+ import mock

+ import six

+ import unittest

+ 

+ from koji_cli.commands import handle_block_group

+ 

+ class TestBlockGroup(unittest.TestCase):

+     # Show long diffs in error output...

+     maxDiff = None

+ 

+     @mock.patch('sys.stdout', new_callable=six.StringIO)

+     @mock.patch('koji_cli.commands.activate_session')

+     def test_handle_block_group_nonexistent_tag(self, activate_session_mock, stdout):

+         tag = 'nonexistent-tag'

+         group = 'group'

+         arguments = [tag, group]

+         options = mock.MagicMock()

+ 

+         # Mock out the xmlrpc server

+         session = mock.MagicMock()

+         session.hasPerm.return_value = True

+         session.getTag.return_value = None

+ 

+         # Run it and check immediate output

+         rv = handle_block_group(options, session, arguments)

+         actual = stdout.getvalue()

+         expected = 'Unknown tag: %s\n' % tag

+         self.assertMultiLineEqual(actual, expected)

+ 

+         # Finally, assert that things were called as we expected.

+         activate_session_mock.assert_called_once_with(session, options)

+         session.hasPerm.assert_called_once_with('admin')

+         session.getTag.assert_called_once_with(tag)

+         session.getTagGroups.assert_not_called()

+         session.groupListBlock.assert_not_called()

+         self.assertEqual(rv, 1)

+ 

+     @mock.patch('sys.stdout', new_callable=six.StringIO)

+     @mock.patch('koji_cli.commands.activate_session')

+     def test_handle_block_group_nonexistent_group(self, activate_session_mock, stdout):

+         tag = 'tag'

+         group = 'group'

+         arguments = [tag, group]

+         options = mock.MagicMock()

+ 

+         # Mock out the xmlrpc server

+         session = mock.MagicMock()

+         session.hasPerm.return_value = True

+         session.getTag.return_value = tag

+         session.getTagGroups.return_value = []

+ 

+         # Run it and check immediate output

+         rv = handle_block_group(options, session, arguments)

+         actual = stdout.getvalue()

+         expected = "Group %s doesn't exist within tag %s\n" % (group, tag)

+         self.assertMultiLineEqual(actual, expected)

+ 

+         # Finally, assert that things were called as we expected.

+         activate_session_mock.assert_called_once_with(session, options)

+         session.hasPerm.assert_called_once_with('admin')

+         session.getTag.assert_called_once_with(tag)

+         session.getTagGroups.assert_called_once_with(tag, inherit=False)

+         session.groupListBlock.assert_not_called()

+         self.assertEqual(rv, 1)

+ 

+     @mock.patch('sys.stdout', new_callable=six.StringIO)

+     @mock.patch('koji_cli.commands.activate_session')

+     def test_handle_block_group_nonexistent_group(self, activate_session_mock, stdout):

+         tag = 'tag'

+         group = 'group'

+         arguments = [tag, group]

+         options = mock.MagicMock()

+ 

+         # Mock out the xmlrpc server

+         session = mock.MagicMock()

+         session.hasPerm.return_value = True

+         session.getTag.return_value = tag

+         session.getTagGroups.return_value = [

+             {'name': 'group', 'group_id': 'groupId'}]

+ 

+         # Run it and check immediate output

+         rv = handle_block_group(options, session, arguments)

+         actual = stdout.getvalue()

+         expected = ''

+         self.assertMultiLineEqual(actual, expected)

+ 

+         # Finally, assert that things were called as we expected.

+         activate_session_mock.assert_called_once_with(session, options)

+         session.hasPerm.assert_called_once_with('admin')

+         session.getTag.assert_called_once_with(tag)

+         session.getTagGroups.assert_called_once_with(tag, inherit=False)

+         session.groupListBlock.assert_called_once_with(tag, group)

+         self.assertEqual(rv, None)

@@ -0,0 +1,203 @@ 

+ import mock

+ import unittest

+ 

+ import koji

+ import kojihub

+ 

+ QP = kojihub.QueryProcessor

+ IP = kojihub.InsertProcessor

+ UP = kojihub.UpdateProcessor

+ 

+ class TestGrouplist(unittest.TestCase):

+     def getQuery(self, *args, **kwargs):

+         query = QP(*args, **kwargs)

+         query.execute = mock.MagicMock()

+         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.context = mock.patch('kojihub.context').start()

+         self.get_tag = mock.patch('kojihub.get_tag').start()

+         self.lookup_tag = mock.patch('kojihub.lookup_tag').start()

+         self.lookup_group = mock.patch('kojihub.lookup_group').start()

+         self.get_tag_groups = mock.patch('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.session.assertLogin = mock.MagicMock()

+ 

+         self.QueryProcessor = mock.patch('kojihub.QueryProcessor',

+                 side_effect=self.getQuery).start()

+         self.queries = []

+         self.InsertProcessor = mock.patch('kojihub.InsertProcessor',

+                 side_effect=self.getInsert).start()

+         self.inserts = []

+         self.UpdateProcessor = mock.patch('kojihub.UpdateProcessor',

+                 side_effect=self.getUpdate).start()

+         self.updates = []

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     def test_grplist_add(self):

+         tag = 'tag'

+         group = 'group'

+         self.get_tag.return_value = {'name': 'tag', 'id': 'tag_id'}

+         self.lookup_group.return_value = {'name': 'group', 'id': 'group_id'}

+         self.get_tag_groups.return_value = {}

+         self.context.event_id = 42

+         self.context.session.user_id = 24

+ 

+         kojihub.grplist_add(tag, group)

+ 

+         # what was called

+         self.context.session.assertPerm.assert_called_once_with('admin')

+         self.get_tag.assert_called_once_with(tag, strict=True)

+         self.lookup_group.assert_called_once_with(group, create=True)

+         self.get_tag_groups.assert_called_with('tag_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': 'group',

+             'biarchonly': False,

+             'exported': True,

+             'uservisible': True,

+             'create_event': 42,

+             'creator_id': 24,

+             'tag_id': 'tag_id',

+             'group_id': 'group_id',

+             'blocked': False,

+         }

+         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):

+             kojihub.grplist_add('tag', 'group')

+         self.context.session.assertPerm.assert_called_once_with('admin')

+         self.assertEqual(len(self.inserts), 0)

+         self.assertEqual(len(self.updates), 0)

+ 

+     def test_grplist_add_no_tag(self):

+         self.get_tag.side_effect = koji.GenericError

+         with self.assertRaises(koji.GenericError):

+             kojihub.grplist_add('tag', 'group')

+         self.context.session.assertPerm.assert_called_once_with('admin')

+         self.assertEqual(len(self.inserts), 0)

+         self.assertEqual(len(self.updates), 0)

+ 

+     def test_grplist_block(self):

+         # identical with test_grplist_add except blocked=True

+         tag = 'tag'

+         group = 'group'

+         self.get_tag.return_value = {'name': 'tag', 'id': 'tag_id'}

+         self.lookup_group.return_value = {'name': 'group', 'id': 'group_id'}

+         self.get_tag_groups.return_value = {}

+         self.context.event_id = 42

+         self.context.session.user_id = 24

+ 

+         kojihub.grplist_block(tag, group)

+ 

+         # what was called

+         self.context.session.assertPerm.assert_called_once_with('admin')

+         self.get_tag.assert_called_once_with(tag, strict=True)

+         self.lookup_group.assert_called_once_with(group, create=True)

+         self.get_tag_groups.assert_called_with('tag_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': 'group',

+             'biarchonly': False,

+             'exported': True,

+             'uservisible': True,

+             'create_event': 42,

+             'creator_id': 24,

+             'tag_id': 'tag_id',

+             'group_id': 'group_id',

+             'blocked': True,

+         }

+         self.assertEqual(insert.table, 'group_config')

+         self.assertEqual(insert.data, values)

+         self.assertEqual(insert.rawdata, {})

+ 

+     def test_grplist_remove(self):

+         tag = 'tag'

+         group = 'group'

+         self.get_tag.return_value = {'name': 'tag', 'id': 'tag_id'}

+         self.lookup_group.return_value = {'name': 'group', 'id': 'group_id'}

+         self.context.event_id = 42

+         self.context.session.user_id = 24

+ 

+         kojihub.grplist_remove(tag, group)

+ 

+         # what was called

+         self.context.session.assertPerm.assert_called_once_with('admin')

+         self.get_tag.assert_called_once_with(tag, strict=True)

+         self.lookup_group.assert_called_once_with(group, strict=True)

+ 

+         # db

+         self.assertEqual(len(self.inserts), 0)

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

+ 

+     def test_grplist_unblock(self):

+         # identical with test_grplist_add except blocked=True

+         tag = 'tag'

+         group = 'group'

+         self.lookup_tag.return_value = {'name': 'tag', 'id': 'tag_id'}

+         self.lookup_group.return_value = {'name': 'group', 'id': 'group_id'}

+         #self.context.event_id = 42

+         #self.context.session.user_id = 24

+ 

+         # will fail for non-blocked group

+         with self.assertRaises(koji.GenericError):

+             kojihub.grplist_unblock(tag, group)

+ 

+         # what was called

+         self.context.session.assertPerm.assert_called_once_with('admin')

+         self.lookup_tag.assert_called_once_with(tag, strict=True)

+         self.lookup_group.assert_called_once_with(group, strict=True)

+ 

+         # db

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         q = "SELECT blocked FROM group_config WHERE (active = TRUE) AND (group_id=%(grp_id)s) AND (tag_id=%(tag_id)s) FOR UPDATE"

+         actual = ' '.join(str(query).split())

+         self.assertEqual(actual, q)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.inserts), 0)