#868 allow force for pkglist_add
Merged 5 years ago by mikem. Opened 6 years ago by tkopecek.
tkopecek/koji issue867  into  master

file modified
+2 -1
@@ -362,6 +362,7 @@ 

      usage = _("usage: %prog block-pkg [options] tag package [package2 ...]")

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

      parser = OptionParser(usage=usage)

+     parser.add_option("--force", action='store_true', default=False, help=_("Override blocks and owner if necessary"))

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

      if len(args) < 2:

          parser.error(_("Please specify a tag and at least one package"))
@@ -384,7 +385,7 @@ 

          return ret

      session.multicall = True

      for package in args[1:]:

-         session.packageListBlock(tag, package)

+         session.packageListBlock(tag, package, force=options.force)

      session.multiCall(strict=True)

  

  

file modified
+4 -7
@@ -855,7 +855,7 @@ 

      tag_id = tag['id']

      pkg = lookup_package(pkginfo, strict=False)

      if not pkg:

-         if not isinstance(pkginfo, basestring):

+         if not isinstance(pkginfo, six.string_types):

              raise koji.GenericError("Invalid package: %s" % pkginfo)

      if owner is not None:

          owner = get_user(owner, strict=True)['id']
@@ -880,10 +880,7 @@ 

      pkglist = readPackageList(tag_id, pkgID=pkg['id'], inherit=True)

      previous = pkglist.get(pkg['id'], None)

      if previous is None:

-         if block is None:

-             block = False

-         else:

-             block = bool(block)

+         block = bool(block)

          if update and not force:

              #if update flag is true, require that there be a previous entry

              raise koji.GenericError("cannot update: tag %s has no data for package %s" \
@@ -946,9 +943,9 @@ 

      koji.plugin.run_callbacks('postPackageListChange', action='remove', tag=tag, package=pkg)

  

  

- def pkglist_block(taginfo, pkginfo):

+ def pkglist_block(taginfo, pkginfo, force=False):

      """Block the package in tag"""

-     pkglist_add(taginfo, pkginfo, block=True)

+     pkglist_add(taginfo, pkginfo, block=True, force=force)

  

  def pkglist_unblock(taginfo, pkginfo, force=False):

      """Unblock the package in tag

@@ -21,7 +21,7 @@ 

          tag = 'tag'

          dsttag = {'name': tag, 'id': 1}

          package = 'package'

-         args = [tag, package]

+         args = [tag, package, '--force']

          options = mock.MagicMock()

  

          # Mock out the xmlrpc server
@@ -43,7 +43,7 @@ 

          session.listPackages.assert_called_once_with(

              tagID=dsttag['id'], inherited=True)

          session.packageListBlock.assert_called_once_with(

-             tag, package)

+             tag, package, force=True)

          session.multiCall.assert_called_once_with(strict=True)

          self.assertNotEqual(rv, 1)

  
@@ -77,12 +77,12 @@ 

          activate_session_mock.assert_called_once_with(session, options)

          self.assertEqual(

              session.mock_calls, [

-                 call.getTag(tag), call.listPackages(

-                     tagID=dsttag['id'], inherited=True), call.packageListBlock(

-                     tag, packages[0]), call.packageListBlock(

-                     tag, packages[1]), call.packageListBlock(

-                         tag, packages[2]), call.multiCall(

-                             strict=True)])

+                 call.getTag(tag),

+                 call.listPackages(tagID=dsttag['id'], inherited=True),

+                 call.packageListBlock(tag, packages[0], force=False),

+                 call.packageListBlock(tag, packages[1], force=False),

+                 call.packageListBlock(tag, packages[2], force=False),

+                 call.multiCall(strict=True)])

          self.assertNotEqual(rv, 1)

  

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

@@ -0,0 +1,404 @@ 

+ import mock

+ try:

+     import unittest2 as unittest

+ except ImportError:

+     import unittest

+ 

+ import koji

+ import kojihub

+ 

+ class TestPkglistBlock(unittest2.TestCase):

+     def setUp(self):

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

+         # It seems MagicMock will not automatically handle attributes that

+         # start with "assert"

+         self.context.session.assertLogin = mock.MagicMock()

+         self.run_callbacks = mock.patch('koji.plugin.run_callbacks').start()

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     @mock.patch('kojihub.pkglist_add')

+     def test_pkglist_block(self, pkglist_add):

+         force = mock.MagicMock()

+         kojihub.pkglist_block('tag', 'pkg', force=force)

+ 

+         pkglist_add.assert_called_once_with('tag', 'pkg', block=True, force=force)

+ 

+     @mock.patch('kojihub._pkglist_remove')

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_pkglist_unblock(self, lookup_package, get_tag, assert_policy,

+             readPackageList, _pkglist_add, _pkglist_remove):

+         tag = {'id': 1, 'name': 'tag'}

+         pkg = {'id': 2, 'name': 'package', 'owner_id': 3}

+         get_tag.return_value = tag

+         lookup_package.return_value = pkg

+         readPackageList.return_value = {pkg['id']: {

+             'blocked': True,

+             'tag_id': tag['id'],

+             'owner_id': pkg['owner_id'],

+             'extra_arches': ''}}

+ 

+         kojihub.pkglist_unblock('tag', 'pkg', force=False)

+ 

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

+         lookup_package.assert_called_once_with('pkg', strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag['id'],

+             'action': 'unblock', 'package': pkg['id'], 'force': False})

+         self.assertEqual(readPackageList.call_count, 2)

+         readPackageList.assert_has_calls([

+             mock.call(tag['id'], pkgID=pkg['id'], inherit=True),

+             mock.call(tag['id'], pkgID=pkg['id'], inherit=True),

+         ])

+         _pkglist_add.assert_called_once_with(tag['id'], pkg['id'], pkg['owner_id'], False, '')

+         _pkglist_remove.assert_called_once_with(tag['id'], 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),

+             mock.call('postPackageListChange', action='unblock', tag=tag, package=pkg),

+         ])

+ 

+     @mock.patch('kojihub._pkglist_remove')

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_pkglist_unblock_inherited(self, lookup_package, get_tag, assert_policy,

+             readPackageList, _pkglist_add, _pkglist_remove):

+         tag_id, pkg_id, owner_id = 1, 2, 3

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

+         lookup_package.return_value = {'id': pkg_id, 'name': 'pkg'}

+         readPackageList.return_value = {pkg_id: {

+             'blocked': True,

+             'tag_id': 4,

+             'owner_id': owner_id,

+             'extra_arches': ''}}

+ 

+         kojihub.pkglist_unblock('tag', 'pkg', force=False)

+ 

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

+         lookup_package.assert_called_once_with('pkg', strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag_id,

+             'action': 'unblock', 'package': pkg_id, 'force': False})

+         readPackageList.assert_called_once_with(tag_id, pkgID=pkg_id, inherit=True)

+         _pkglist_add.assert_called_once_with(tag_id, pkg_id, owner_id, False, '')

+         _pkglist_remove.assert_not_called()

+ 

+     @mock.patch('kojihub._pkglist_remove')

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_pkglist_unblock_not_present(self, lookup_package, get_tag, assert_policy,

+             readPackageList, _pkglist_add, _pkglist_remove):

+         tag_id, pkg_id = 1, 2

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

+         lookup_package.return_value = {'id': pkg_id, 'name': 'pkg'}

+         readPackageList.return_value = {}

+ 

+         with self.assertRaises(koji.GenericError):

+             kojihub.pkglist_unblock('tag', 'pkg', force=False)

+ 

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

+         lookup_package.assert_called_once_with('pkg', strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag_id,

+             'action': 'unblock', 'package': pkg_id, 'force': False})

+         readPackageList.assert_called_once_with(tag_id, pkgID=pkg_id, inherit=True)

+         _pkglist_add.assert_not_called()

+         _pkglist_remove.assert_not_called()

+ 

+     @mock.patch('kojihub._pkglist_remove')

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_pkglist_unblock_not_blocked(self, lookup_package, get_tag, assert_policy,

+             readPackageList, _pkglist_add, _pkglist_remove):

+         tag_id, pkg_id, owner_id = 1, 2, 3

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

+         lookup_package.return_value = {'id': pkg_id, 'name': 'pkg'}

+         readPackageList.return_value = {pkg_id: {

+             'blocked': False,

+             'tag_id': tag_id,

+             'owner_id': owner_id,

+             'extra_arches': ''}}

+ 

+ 

+         with self.assertRaises(koji.GenericError):

+             kojihub.pkglist_unblock('tag', 'pkg', force=False)

+ 

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

+         lookup_package.assert_called_once_with('pkg', strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag_id,

+             'action': 'unblock', 'package': pkg_id, 'force': False})

+         readPackageList.assert_called_once_with(tag_id, pkgID=pkg_id, inherit=True)

+         _pkglist_add.assert_not_called()

+         _pkglist_remove.assert_not_called()

+ 

+     @mock.patch('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)

+ 

+     @mock.patch('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)

+ 

+     @mock.patch('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)

+ 

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_user')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_direct_pkglist_add(self, lookup_package, get_tag, get_user,

+             assert_policy, readPackageList, _pkglist_add):

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

+         get_tag.return_value = tag

+         lookup_package.return_value = pkg

+         get_user.return_value = user

+         readPackageList.return_value = {}

+ 

+ 

+         kojihub._direct_pkglist_add(tag['name'], pkg['name'],

+             user['name'], block=block, extra_arches=extra_arches,

+             force=force, update=update, policy=policy)

+ 

+         get_tag.assert_called_once_with(tag['name'], strict=True)

+         lookup_package.assert_called_once_with(pkg['name'], strict=False)

+         get_user.assert_called_once_with(user['name'], strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag['id'],

+             'action': 'add', 'package': pkg['name'], '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),

+             mock.call('postPackageListChange', action='add', tag=tag,

+                 package=pkg, owner=user['id'], block=block,

+                 extra_arches=extra_arches, force=force, update=update),

+         ])

+         _pkglist_add.assert_called_once_with(tag['id'], pkg['id'],

+             user['id'], block, extra_arches)

+ 

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_user')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_direct_pkglist_add_no_package(self, lookup_package,

+             get_tag, get_user, assert_policy, readPackageList, _pkglist_add):

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

+         get_tag.return_value = tag

+         lookup_package.return_value = None

+         get_user.return_value = user

+         readPackageList.return_value = {}

+ 

+         # package needs to be name, not dict

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

+ 

+         lookup_package.assert_called_once_with(pkg, strict=False)

+         self.assertEqual(self.run_callbacks.call_count, 0)

+         _pkglist_add.assert_not_called()

+ 

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_user')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_direct_pkglist_add_no_user(self, lookup_package,

+             get_tag, get_user, assert_policy, readPackageList, _pkglist_add):

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

+         get_tag.return_value = tag

+         lookup_package.return_value = None

+         get_user.side_effect = koji.GenericError

+         readPackageList.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)

+ 

+         lookup_package.assert_called_once_with(pkg, strict=False)

+         self.assertEqual(self.run_callbacks.call_count, 0)

+         _pkglist_add.assert_not_called()

+ 

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_user')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_direct_pkglist_add_new_package(self, lookup_package, get_tag, get_user,

+             assert_policy, readPackageList, _pkglist_add):

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

+         get_tag.return_value = tag

+         lookup_package.side_effect = [None, pkg]

+         get_user.return_value = user

+         readPackageList.return_value = {}

+ 

+ 

+         kojihub._direct_pkglist_add(tag['name'], pkg['name'],

+             user['name'], block=block, extra_arches=extra_arches,

+             force=force, update=update, policy=policy)

+ 

+         get_tag.assert_called_once_with(tag['name'], strict=True)

+         self.assertEqual(lookup_package.call_count, 2)

+         lookup_package.has_calls(

+             mock.call(pkg['name'], strict=False),

+             mock.call(pkg['name'], create=True),

+         )

+         get_user.assert_called_once_with(user['name'], strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag['id'],

+             'action': 'add', 'package': pkg['name'], '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),

+             mock.call('postPackageListChange', action='add', tag=tag,

+                 package=pkg, owner=user['id'], block=block,

+                 extra_arches=extra_arches, force=force, update=update),

+         ])

+         _pkglist_add.assert_called_once_with(tag['id'], pkg['id'],

+             user['id'], block, extra_arches)

+ 

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_user')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_direct_pkglist_add_blocked_previously(self,

+             lookup_package, get_tag, get_user,

+             assert_policy, readPackageList, _pkglist_add):

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

+         get_tag.return_value = tag

+         lookup_package.return_value = pkg

+         get_user.return_value = user

+         readPackageList.return_value = {pkg['id']: {

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

+ 

+         get_tag.assert_called_once_with(tag['name'], strict=True)

+         lookup_package.assert_called_once_with(pkg['name'], strict=False)

+         get_user.assert_called_once_with(user['name'], strict=True)

+         assert_policy.assert_called_once_with('package_list', {'tag': tag['id'],

+             'action': 'add', 'package': pkg['name'], 'force': False})

+         self.run_callbacks.assert_called_once_with(

+                 'prePackageListChange', action='add', tag=tag,

+                 package=pkg, owner=user['id'], block=block,

+                 extra_arches=extra_arches, force=force, update=update)

+         _pkglist_add.assert_not_called()

+ 

+     @mock.patch('kojihub._pkglist_add')

+     @mock.patch('kojihub.readPackageList')

+     @mock.patch('kojihub.assert_policy')

+     @mock.patch('kojihub.get_user')

+     @mock.patch('kojihub.get_tag')

+     @mock.patch('kojihub.lookup_package')

+     def test_direct_pkglist_add_blocked_previously_force(self,

+             lookup_package, get_tag, get_user,

+             assert_policy, readPackageList, _pkglist_add):

+         block = False

+         extra_arches = 'arch123'

+         force=True

+         update=False

+         policy=True

+         tag = {'id': 1, 'name': 'tag'}

+         pkg = {'id': 2, 'name': 'pkg', 'owner_id': 3}

+         user = {'id': 3, 'name': 'user'}

+         get_tag.return_value = tag

+         lookup_package.return_value = pkg

+         get_user.return_value = user

+         readPackageList.return_value = {pkg['id']: {

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

+ 

+         get_tag.assert_called_once_with(tag['name'], strict=True)

+         lookup_package.assert_called_once_with(pkg['name'], strict=False)

+         get_user.assert_called_once_with(user['name'], strict=True)

+         # force + admin

+         assert_policy.assert_not_called()

+ 

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

+             mock.call('postPackageListChange', action='add', tag=tag,

+                 package=pkg, owner=user['id'], block=block,

+                 extra_arches=extra_arches, force=force, update=update),

+         ])

+         _pkglist_add.assert_called_once_with(tag['id'], pkg['id'],

+             user['id'], block, extra_arches)

Using force option will allow 'pre-blocking' of packages which are not
in tag inheritance yet, but could cause a problem if they'll get to
inheritance chain somehow. In such case we don't have an owner for
package and using 'force' allows admin to introduce blocked package with
him as an owner.

Fixes: https://pagure.io/koji/issue/867

I don't think the help string is correct:

+    parser.add_option("--force", action='store_true', default=False, help=_("Override owner if necessary"))

The force option does more than this, even in the block case.

rebased onto 38cc637

6 years ago

New unit test has unqualified import unittest2

1 new commit added

  • use unittest2 only if available
5 years ago

Commit f37e203 fixes this pull-request

Pull-Request has been merged by mikem

5 years ago