add unit tests for cli commands, coverage(46%)

tests/test_cli/test_block_group.py

@@ -2,7 +2,6 @@ from __future__ import absolute_import
 
 import mock
 import six
-import unittest
 
 from koji_cli.commands import handle_block_group
 from . import utils

tests/test_cli/test_hello.py

@@ -11,21 +11,19 @@ from . import utils
 
 class TestPrintUnicode(utils.CliTestCase):
 
-    greetings = ('hello',
-                 'bonjour',
-                u'céad míle fáilte',
-                u'hylô',
-                u'你好',
-                u'こんにちは',
-                u'안녕하세요')
+    greetings = (u'céad míle fáilte',
+                 u'hylô',
+                 u'你好',
+                 u'こんにちは',
+                 u'안녕하세요')
 
     def test_printable_unicode(self):
         for s in self.greetings:
-            result = _printable_unicode(s)
+            # make sure the type is unicode before convert in python2
             if six.PY2:
-                self.assertEqual(type(result), type(str()))
-            else:
-                self.assertEqual(type(result), type(str()))
+                self.assertEqual(type(s), type(unicode()))
+            result = _printable_unicode(s)
+            self.assertEqual(type(result), type(str()))
 
 
 class TestHello(utils.CliTestCase):

tests/test_cli/test_image_build_indirection.py

@@ -0,0 +1,262 @@
+from __future__ import absolute_import
+import mock
+import six
+import unittest
+import koji
+
+from koji_cli.commands import handle_image_build_indirection, _build_image_indirection
+from . import utils
+
+
+TASK_OPTIONS = {
+    "background": True,
+    "config": "build-image-config.conf",
+    "name": "image",
+    "version": "26",
+    "release": "1",
+    "arch": "x86_64 i386",
+    "target": "target",
+    "base_image_task": 2,
+    "base_image_build": None,
+    "utility_image_task": 4,
+    "utility_image_build": None,
+    "indirection_template": "template",
+    "indirection_template_url": "git://git.github.org/git/",
+    "results_loc": "results",
+    "scratch": None,
+    "wait": None,
+    "noprogress": None,
+    "skip_tag": False,
+}
+
+
+class Options(object):
+    def __init__(self, init_dict):
+        for k, v in init_dict.items():
+            setattr(self, k, v)
+
+
+class TestBuildImageIndirection(utils.CliTestCase):
+
+    # Show long diffs in error output...
+    maxDiff = None
+
+    def setUp(self):
+        self.task_id = 1001
+        self.weburl = 'https://web.url'
+        self.options = mock.MagicMock()
+        self.options.quiet = False
+        self.options.weburl = self.weburl
+        self.session = mock.MagicMock()
+        self.activate_session = mock.patch('koji_cli.commands.activate_session').start()
+        self.unique_path = mock.patch('koji_cli.commands._unique_path').start()
+
+        self.task_opts = Options(TASK_OPTIONS)
+
+        self.build_target = {
+            'id': 1,
+            'name': 'target',
+            'dest_tag': 1,
+            'build_tag': 2,
+            'build_tag_name': 'target-build',
+            'dest_tag_name': 'target'
+        }
+
+        self.dest_tag = {'id': 1, 'name': 'dest-tag', 'arch': 'x86_64'}
+
+        self.session.getBuildTarget.return_value = self.build_target
+        self.session.getTag.return_value = self.dest_tag
+        self.session.buildImageIndirection.return_value = self.task_id
+        self.unique_path.return_value = '/path/to/cli-image-indirection'
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def test_build_image_indirection(self):
+        """Test _build_image_indirection function"""
+        expected = "Created task: %d" % self.task_id + "\n"
+        expected += "Task info: %s/taskinfo?taskID=%s" % (self.weburl, self.task_id) + "\n"
+
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assert_console_message(stdout, expected)
+
+    def test_build_image_indirection_with_scratch(self):
+        """Test _build_image_indirection function with scratch option"""
+        expected = "\n"
+        expected += "Created task: %d" % self.task_id + "\n"
+        expected += "Task info: %s/taskinfo?taskID=%s" % (self.weburl, self.task_id) + "\n"
+
+        self.task_opts.indirection_template_url = None
+        self.task_opts.scratch = True
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assert_console_message(stdout, expected)
+
+    def test_build_image_indirection_with_noprogress(self):
+        """Test _build_image_indirection function with noprogress option"""
+        expected = "\n"
+        expected += "Created task: %d" % self.task_id + "\n"
+        expected += "Task info: %s/taskinfo?taskID=%s" % (self.weburl, self.task_id) + "\n"
+
+        self.task_opts.indirection_template_url = None
+        self.task_opts.scratch = True
+        self.task_opts.noprogress = True
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assert_console_message(stdout, expected)
+        args, kwargs = self.session.uploadWrapper.call_args
+        self.assertEqual(kwargs['callback'], None)
+
+    def test_build_image_indirection_expections(self):
+        """Test _build_image_indirection all exceptions"""
+
+        # Case 1. sanity check for utility image options
+        self.task_opts.utility_image_task = None
+        self.task_opts.utility_image_build = None
+        expected = "You must specify either a utility-image task or build ID/NVR"
+        with self.assertRaises(koji.GenericError) as cm:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assertEqual(str(cm.exception), expected)
+        self.activate_session.assert_not_called()
+
+        # Case 2. sanity check for base image options
+        self.task_opts.utility_image_build = 'image-utils-26.1'
+        self.task_opts.base_image_task = None
+        self.task_opts.base_image_build = None
+        expected = "You must specify either a base-image task or build ID/NVR"
+        with self.assertRaises(koji.GenericError) as cm:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assertEqual(str(cm.exception), expected)
+        self.activate_session.assert_not_called()
+
+        self.task_opts.base_image_build = 'image-base-26.1'
+
+        # Case 3. missing required options
+        required = ['name', 'version', 'arch', 'target',
+                    'indirection_template', 'results_loc']
+        for r in required:
+            orig = getattr(self.task_opts, r)
+            setattr(self.task_opts, r, None)
+            expected = "Missing the following required options: "
+            expected += "--" + r.replace('_', '-') + "\n"
+            with self.assertRaises(koji.GenericError) as cm, \
+                    mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+                _build_image_indirection(
+                    self.options, self.task_opts, self.session, [])
+            self.assert_console_message(stdout, expected)
+            self.assertEqual(
+                str(cm.exception), "Missing required options specified above")
+            self.activate_session.assert_not_called()
+            setattr(self.task_opts, r, orig)
+
+        # Case 4. target not found error
+        self.session.getBuildTarget.return_value = {}
+        expected = "Unknown build target: %s" % {}
+        with self.assertRaises(koji.GenericError) as cm:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assertEqual(str(cm.exception), expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+
+        # Case 5. tag not found error
+        self.session.getBuildTarget.return_value = self.build_target
+        self.session.getTag.return_value = {}
+        expected = "Unknown destination tag: %s" % self.build_target['dest_tag_name']
+        with self.assertRaises(koji.GenericError) as cm:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assertEqual(str(cm.exception), expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+
+        # Case 6. --indirection-template-url is not given and --scatch is not set
+        self.session.getTag.return_value = self.dest_tag
+        self.task_opts.indirection_template_url = None
+        self.task_opts.scratch = None
+        expected = "Non-scratch builds must provide a URL for the indirection template"
+        with self.assertRaises(koji.GenericError) as cm:
+            _build_image_indirection(
+                self.options, self.task_opts, self.session, [])
+        self.assertEqual(str(cm.exception), expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+
+
+class TestImageBuildIndirection(utils.CliTestCase):
+
+    # Show long diffs in error output...
+    maxDiff = None
+
+    def setUp(self):
+        self.options = mock.MagicMock()
+        self.session = mock.MagicMock()
+
+        self.error_format = """Usage: %s image-build-indirection [base_image] [utility_image] [indirection_build_template]
+       %s image-build --config FILE
+
+(Specify the --help global option for a list of other help options)
+
+%s: error: {message}
+""" % (self.progname, self.progname, self.progname)
+
+    @mock.patch('koji_cli.commands._build_image_indirection')
+    def test_handle_image_build_indirection(self, build_image_mock):
+        """Test handle_image_build_indirection function"""
+        handle_image_build_indirection(self.options, self.session, [])
+        args, kwargs = build_image_mock.call_args
+        empty_opts = dict((k, None) for k in TASK_OPTIONS)
+        self.assertDictEqual(empty_opts, args[1].__dict__)
+
+    def test_handle_image_build_indirection_help(self):
+        """Test handle_image_build_indirection help message"""
+        self.assert_help(
+            handle_image_build_indirection,
+            """Usage: %s image-build-indirection [base_image] [utility_image] [indirection_build_template]
+       %s image-build --config FILE
+
+(Specify the --help global option for a list of other help options)
+
+Options:
+  -h, --help            show this help message and exit
+  --config=CONFIG       Use a configuration file to define image-build options
+                        instead of command line options (they will be
+                        ignored).
+  --background          Run the image creation task at a lower priority
+  --name=NAME           Name of the output image
+  --version=VERSION     Version of the output image
+  --release=RELEASE     Release of the output image
+  --arch=ARCH           Architecture of the output image and input images
+  --target=TARGET       Build target to use for the indirection build
+  --skip-tag            Do not tag the resulting build
+  --base-image-task=BASE_IMAGE_TASK
+                        ID of the createImage task of the base image to be
+                        used
+  --base-image-build=BASE_IMAGE_BUILD
+                        NVR or build ID of the base image to be used
+  --utility-image-task=UTILITY_IMAGE_TASK
+                        ID of the createImage task of the utility image to be
+                        used
+  --utility-image-build=UTILITY_IMAGE_BUILD
+                        NVR or build ID of the utility image to be used
+  --indirection-template=INDIRECTION_TEMPLATE
+                        Name of the local file, or SCM file containing the
+                        template used to drive the indirection plugin
+  --indirection-template-url=INDIRECTION_TEMPLATE_URL
+                        SCM URL containing the template used to drive the
+                        indirection plugin
+  --results-loc=RESULTS_LOC
+                        Relative path inside the working space image where the
+                        results should be extracted from
+  --scratch             Create a scratch image
+  --wait                Wait on the image creation, even if running in the
+                        background
+  --noprogress          Do not display progress of the upload
+""" % (self.progname, self.progname))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_cli/test_move_build.py

@@ -0,0 +1,158 @@
+from __future__ import absolute_import
+import mock
+import six
+import unittest
+from koji_cli.commands import handle_move_build
+from . import utils
+
+
+class TestMoveBuild(utils.CliTestCase):
+
+    # Show long diffs in error output...
+    maxDiff = None
+
+    def setUp(self):
+        self.session = mock.MagicMock()
+        self.options = mock.MagicMock()
+
+        self.error_format = """Usage: %s move-build [options] <tag1> <tag2> <pkg> [<pkg>...]
+(Specify the --help global option for a list of other help options)
+
+%s: error: {message}
+""" % (self.progname, self.progname)
+
+        self.activate_session = mock.patch('koji_cli.commands.activate_session').start()
+        self.running_in_bg = mock.patch('koji_cli.commands._running_in_bg').start()
+        self.running_in_bg.return_value = False
+        self.watch_tasks = mock.patch('koji_cli.commands.watch_tasks').start()
+        self.watch_tasks.return_value = True
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def test_handle_move_build(self):
+        """Test handle_move_build function"""
+        pkgs = ['pkg_a-1.0-1fc26', 'pkg_b-2.0-1fc26', 'pkg_c-2.2-2fc26']
+        arguments = ['tag-a', 'tag-b'] + pkgs
+        tasks = [202, 303]
+
+        self.options.quiet = False
+        self.options.force = False
+        self.options.poll_interval = 100
+
+        self.session.getBuild.side_effect = [
+            {'id': 11, 'name': 'pkg_a', 'version': '1.0', 'release': '1fc26'},
+            {'id': 22, 'name': 'pkg_b', 'version': '2.0', 'release': '1fc26'},
+            {},             # assume pkg_c-2.2-2fc26 is invalid
+        ]
+        self.session.moveBuild.side_effect = tasks
+
+        expected = 'Invalid build %s, skipping.' % 'pkg_c-2.2-2fc26' + "\n"
+        for i, t in enumerate(tasks):
+            expected += "Created task %d, moving %s" % (t, pkgs[i]) + "\n"
+
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            rv = handle_move_build(self.options, self.session, arguments)
+
+        # sanity checks
+        self.assertEqual(rv, True)
+        self.assert_console_message(stdout, expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+        self.session.logout.assert_called_once()
+        self.watch_tasks.assert_called_with(
+            self.session,
+            tasks,
+            quiet=self.options.quiet,
+            poll_interval=self.options.poll_interval)
+
+    def test_handle_move_build_nowait(self):
+        """Test handle_move_build function with --nowait option"""
+        pkgs = ['pkg_a-1.0-1fc26']
+        arguments = ['tag-a', 'tag-b', '--nowait'] + pkgs
+        task_id = 999
+
+        self.session.getBuild.side_effect = [
+            {'id': 11, 'name': 'pkg_a', 'version': '1.0', 'release': '1fc26'},
+        ]
+        self.session.moveBuild.side_effect = [task_id]
+
+        expected = "Created task %d, moving %s" % (task_id, pkgs[0]) + "\n"
+
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            rv = handle_move_build(self.options, self.session, arguments)
+
+        # sanity checks
+        self.assertEqual(rv, None)
+        self.assert_console_message(stdout, expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+        self.session.logout.assert_not_called()
+        self.watch_tasks.assert_not_called()
+
+    def test_handle_move_build_with_all_option(self):
+        """Test handle_move_build function with --all option"""
+        pkgs = ['pkg_a-1.0-1fc26', 'pkg_b-2.0-1fc26', 'pkg_c-2.2-2fc26']
+        arguments = ['tag-a', 'tag-b', '--all', '--nowait'] + pkgs
+
+        self.session.getPackage.side_effect = [
+            {'id': 44, 'name': 'pkg_a', 'version': '1.0', 'release': '1fc26'},
+            {'id': 55, 'name': 'pkg_b', 'version': '2.0', 'release': '1fc26'},
+            {},             # assume pkg_c-2.2-2fc26 is invalid
+        ]
+        self.session.moveAllBuilds.side_effect = [
+            [500, 501, 502], [601, 602, 603]
+        ]
+
+        expected = 'Invalid package name %s, skipping.' % 'pkg_c-2.2-2fc26' + "\n"
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            rv = handle_move_build(self.options, self.session, arguments)
+
+        self.assertEqual(rv, None)
+        self.assert_console_message(stdout, expected)
+        self.session.moveAllBuilds.assert_has_calls(
+            [mock.call(arguments[0], arguments[1], p, None) for p in pkgs[:-1]]
+        )
+
+    def test_handle_move_build_argument_error(self):
+        """Test handle_move_build function with wrong argument"""
+
+        # Case 1. without --all option
+        expected = self.format_error_message(
+            "This command takes at least three arguments: two tags and one or more package n-v-r's")
+        for arg in [[], ['tag1'], ['tag1', 'tag2']]:
+            self.assert_system_exit(
+                handle_move_build,
+                self.options,
+                self.session,
+                arg,
+                stderr=expected,
+                activate_session=None)
+
+        # Case 2. with --all option
+        expected = self.format_error_message(
+            "This command, with --all, takes at least three arguments: two tags and one or more package names")
+        for arg in [['--all', 'tag1'], ['--all', 'tag1', 'tag2']]:
+            self.assert_system_exit(
+                handle_move_build,
+                self.options,
+                self.session,
+                arg,
+                stderr=expected,
+                activate_session=None)
+
+    def test_handle_move_build_help(self):
+        """Test handle_move_build help message"""
+        self.assert_help(
+            handle_move_build,
+            """Usage: %s move-build [options] <tag1> <tag2> <pkg> [<pkg>...]
+(Specify the --help global option for a list of other help options)
+
+Options:
+  -h, --help  show this help message and exit
+  --force     force operation
+  --nowait    do not wait on tasks
+  --all       move all instances of a package, <pkg>'s are package names
+""" % self.progname)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_cli/test_regen_repo.py

@@ -0,0 +1,177 @@
+from __future__ import absolute_import
+from __future__ import print_function
+import mock
+import six
+import unittest
+import copy
+
+from koji_cli.commands import handle_regen_repo
+from . import utils
+
+
+class TestRegenRepo(utils.CliTestCase):
+
+    # Show long diffs in error output...
+    maxDiff = None
+    longMessage = True
+
+    TAG = {
+        'maven_support': False,
+        'locked': False,
+        'name': 'fedora26-build',
+        'extra': {},
+        'perm': None,
+        'id': 2,
+        'arches': 'x86_64',
+        'maven_include_all': False,
+        'perm_id': None
+     }
+
+    def setUp(self):
+        self.task_id = 1001
+        self.tag_name = self.TAG['name']
+
+        self.options = mock.MagicMock()
+        self.options.quiet = True
+        self.options.poll_interval = 100
+        self.options.weburl = 'https://localhost.local'
+
+        self.session = mock.MagicMock()
+        self.session.getTag.return_value = copy.deepcopy(self.TAG)
+        self.session.newRepo.return_value = self.task_id
+        self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name}
+
+        self.error_format = """Usage: %s regen-repo [options] <tag>
+(Specify the --help global option for a list of other help options)
+
+%s: error: {message}
+""" % (self.progname, self.progname)
+
+        self.setUpMocks()
+
+    def setUpMocks(self):
+        self.activate_session = mock.patch('koji_cli.commands.activate_session').start()
+
+        self.running_in_bg = mock.patch('koji_cli.commands._running_in_bg').start()
+        self.running_in_bg.return_value = False     # assume run in foreground
+
+        self.watch_tasks = mock.patch('koji_cli.commands.watch_tasks').start()
+        self.watch_tasks.return_value = True
+
+        self.mocks_table = {}
+        for m in ('activate_session', 'running_in_bg', 'watch_tasks'):
+            self.mocks_table[m] = getattr(self, m)
+            self.addCleanup(self.mocks_table[m].stop)
+
+    def resetMocks(self):
+        for m in self.mocks_table.values():
+            m.reset()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def __run_test_handle_regen_repo(self, arguments, return_value=None, expected=''):
+        expected += "Regenerating repo for tag: %s" % self.tag_name + "\n"
+        expected += "Created task: %d" % self.task_id + "\n"
+        expected += "Task info: %s/taskinfo?taskID=%s" % (self.options.weburl, self.task_id) + "\n"
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            rv = handle_regen_repo(self.options, self.session, arguments)
+            self.assertEqual(rv, return_value)
+        self.assert_console_message(stdout, expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+
+    def test_handle_regen_repo(self):
+        """Test handle_regen_repo function"""
+
+        # show error if tag is not exist
+        self.session.getTag.return_value = {}
+        expected = self.format_error_message("No matching tag: " + self.tag_name)
+        self.assert_system_exit(
+            handle_regen_repo,
+            self.options,
+            self.session,
+            [self.tag_name],
+            stderr=expected)
+
+        self.resetMocks()
+
+        # show warning if tag is not a build tag
+        self.session.getTag.return_value = copy.copy(self.TAG)
+        self.session.getBuildTargets.return_value = []
+        expected = "Warning: %s is not a build tag" % self.tag_name + "\n"
+        self.__run_test_handle_regen_repo([self.tag_name], True, expected=expected)
+
+        self.resetMocks()
+        # show warning message if arch is empty
+        noarch_tag = copy.copy(self.TAG)
+        noarch_tag.update({'arches': ''})
+        self.session.getTag.return_value = noarch_tag
+        expected += "Warning: tag %s has an empty arch list" % noarch_tag['name'] + "\n"
+        self.__run_test_handle_regen_repo([self.tag_name], True, expected=expected)
+
+    def test_handle_regen_repo_with_target_opt(self):
+        """Test handle_regen_repo function with --target option"""
+        arguments = [self.tag_name, '--target']
+
+        # show error if target is not matched
+        self.session.getBuildTarget.return_value = {}
+        expected = self.format_error_message("No matching build target: " + self.tag_name)
+        self.assert_system_exit(
+            handle_regen_repo,
+            self.options,
+            self.session,
+            arguments,
+            stderr=expected)
+
+        self.resetMocks()
+
+        self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name}
+        self.__run_test_handle_regen_repo(arguments, True)
+
+    def test_handle_regen_repo_with_other_opts(self):
+        """Test handle_regen_repo function with options"""
+        # --nowait
+        self.__run_test_handle_regen_repo([self.tag_name, '--nowait'], None)
+        self.resetMocks()
+
+        # --source && --debuginfo
+        self.__run_test_handle_regen_repo([self.tag_name, '--source', '--debuginfo'], True)
+        self.session.newRepo.assert_called_with(self.tag_name, **{'debuginfo': True, 'src': True})
+
+    def test_handle_regen_repo_errors(self):
+        """Test handle_regen_repo function errors and exceptions"""
+        tests = [
+            # [ arguments, error_string ]
+            [[], self.format_error_message("A tag name must be specified")],
+            [['tag1', 'tag2'], self.format_error_message("Only a single tag name may be specified")],
+            [['tag1', 'tag2', '--target'], self.format_error_message("Only a single target may be specified")],
+        ]
+
+        for test in tests:
+            self.assert_system_exit(
+                handle_regen_repo,
+                self.options,
+                self.session,
+                test[0],
+                stderr=test[1],
+                activate_session=None)
+        self.activate_session.assert_not_called()
+
+    def test_handle_regen_repo_help(self):
+        """Test handle_regen_repo help message"""
+        self.assert_help(
+            handle_regen_repo,
+            """Usage: %s regen-repo [options] <tag>
+(Specify the --help global option for a list of other help options)
+
+Options:
+  -h, --help       show this help message and exit
+  --target         Interpret the argument as a build target name
+  --nowait         Don't wait on for regen to finish
+  --debuginfo      Include debuginfo rpms in repo
+  --source, --src  Include source rpms in the repo
+""" % self.progname)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_cli/test_tag_build.py

@@ -0,0 +1,99 @@
+from __future__ import absolute_import
+import mock
+import six
+import unittest
+from koji_cli.commands import handle_tag_build
+from . import utils
+
+
+class TestTagBuild(utils.CliTestCase):
+
+    # Show long diffs in error output...
+    maxDiff = None
+
+    def setUp(self):
+        self.session = mock.MagicMock()
+        self.options = mock.MagicMock()
+
+        self.error_format = """Usage: %s tag-build [options] <tag> <pkg> [<pkg>...]
+(Specify the --help global option for a list of other help options)
+
+%s: error: {message}
+""" % (self.progname, self.progname)
+
+        self.activate_session = mock.patch('koji_cli.commands.activate_session').start()
+        self.running_in_bg = mock.patch('koji_cli.commands._running_in_bg').start()
+        self.running_in_bg.return_value = False
+        self.watch_tasks = mock.patch('koji_cli.commands.watch_tasks').start()
+        self.watch_tasks.return_value = True
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def test_handle_tag_build(self):
+        """Test handle_tag_build function"""
+        pkgs = ['pkg_a-1.0-1fc26', 'pkg_b-2.0-1fc26', 'pkg_c-2.2-2fc26']
+        arguments = ['tag'] + pkgs
+        tasks = [1001, 2002, 3003]
+
+        self.options.quiet = False
+        self.options.poll_interval = 100
+
+        self.session.tagBuild.side_effect = tasks
+        expected = ''.join(["Created task %d" % t + "\n" for t in tasks])
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            rv = handle_tag_build(self.options, self.session, arguments)
+        self.assertEqual(rv, True)
+        self.assert_console_message(stdout, expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+
+        self.session.logout.assert_called_once()
+        self.watch_tasks.assert_called_with(
+            self.session,
+            tasks,
+            quiet=self.options.quiet,
+            poll_interval=self.options.poll_interval)
+
+    def test_handle_tag_build_quiet_mode(self):
+        """Test handle_tag_build function with --nowait option"""
+        pkgs = ['pkg_a-1.0-1fc26']
+        arguments = ['tag', '--nowait'] + pkgs
+        task_id = 4004
+
+        expected = "Created task %d" % task_id + "\n"
+        self.session.tagBuild.side_effect = [task_id]
+        with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
+            rv = handle_tag_build(self.options, self.session, arguments)
+        self.assertEqual(rv, None)
+        self.assert_console_message(stdout, expected)
+        self.activate_session.assert_called_with(self.session, self.options)
+        self.watch_tasks.assert_not_called()
+
+    def test_handle_tag_build_argument_error(self):
+        """Test handle_tag_build function with error argument"""
+        expected = self.format_error_message(
+            "This command takes at least two arguments: a tag name/ID and one or more package n-v-r's")
+        for arg in [[], ['tag']]:
+            self.assert_system_exit(
+                handle_tag_build,
+                self.options,
+                self.session,
+                arg,
+                stderr=expected,
+                activate_session=None)
+
+    def test_handle_tag_build_help(self):
+        self.assert_help(
+            handle_tag_build,
+            """Usage: %s tag-build [options] <tag> <pkg> [<pkg>...]
+(Specify the --help global option for a list of other help options)
+
+Options:
+  -h, --help  show this help message and exit
+  --force     force operation
+  --nowait    Do not wait on task
+""" % self.progname)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_cli/test_wait_repo.py

@@ -0,0 +1,219 @@
+from __future__ import absolute_import
+from __future__ import print_function
+import mock
+import six
+import unittest
+import copy
+
+from koji_cli.commands import anon_handle_wait_repo
+from . import utils
+
+
+class TestWaitRepo(utils.CliTestCase):
+
+    # Show long diffs in error output...
+    maxDiff = None
+    longMessage = True
+
+    TAG = {
+        'maven_support': False,
+        'locked': False,
+        'name': 'fedora26-build',
+        'extra': {},
+        'perm': None,
+        'id': 2,
+        'arches': 'x86_64',
+        'maven_include_all': False,
+        'perm_id': None
+     }
+
+    def setUp(self):
+        self.task_id = 1001
+        self.tag_name = self.TAG['name']
+
+        self.options = mock.MagicMock()
+        self.options.quiet = True
+        self.options.poll_interval = 1  # second
+        self.options.weburl = 'https://localhost.local'
+
+        self.session = mock.MagicMock()
+        self.session.getTag.return_value = copy.deepcopy(self.TAG)
+        self.session.newRepo.return_value = self.task_id
+        self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name}
+
+        self.error_format = """Usage: %s wait-repo [options] <tag>
+(Specify the --help global option for a list of other help options)
+
+%s: error: {message}
+""" % (self.progname, self.progname)
+
+        self.setUpMocks()
+
+    def setUpMocks(self):
+        self.activate_session = mock.patch('koji_cli.commands.activate_session').start()
+        self.checkForBuilds = mock.patch('koji.util.checkForBuilds').start()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    @mock.patch('time.time')
+    @mock.patch('sys.stdout', new_callable=six.StringIO)
+    def __test_wait_repo(self, args, expected, stdout, time_mock):
+        self.options.quiet = False
+        time_mock.side_effect = [0, 1, 2, 3]
+        anon_handle_wait_repo(self.options, self.session, args)
+        self.assert_console_message(stdout, expected)
+
+    @mock.patch('time.time')
+    @mock.patch('sys.stdout', new_callable=six.StringIO)
+    def __test_wait_repo_timeout(self, args, expected, stdout, time_mock):
+        self.options.quiet = False
+        time_mock.side_effect = [0, 61, 62]
+        anon_handle_wait_repo(self.options, self.session, args + ['--timeout', '1'])
+        self.assert_console_message(stdout, expected)
+
+    def test_anon_handle_wait_repo(self):
+        """Test anon_handle_wait_repo function"""
+        arguments = [self.tag_name]
+
+        self.options.quiet = False
+        self.session.getRepo.side_effect = [{}, {}, {'id': 1, 'name': 'DEFAULT'}]
+        expected = 'Successfully waited 0:03 for a new %s repo' % self.tag_name + '\n'
+        self.__test_wait_repo(arguments, expected)
+
+    def test_anon_handle_wait_repo_with_target_opt(self):
+        """Test anon_handle_wait_repo function with --target option"""
+        arguments = [self.tag_name, '--target']
+
+        self.options.quiet = False
+        self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name, 'build_tag': 1}
+        self.session.getRepo.side_effect = [{}, {}, {'id': 1, 'name': 'DEFAULT'}]
+        expected = 'Successfully waited 0:03 for a new %s repo' % self.tag_name + '\n'
+        self.__test_wait_repo(arguments, expected)
+
+    def test_anon_handle_wait_repo_timeout(self):
+        """Test anon_handle_wait_repo function on timeout case"""
+        arguments = [self.tag_name]
+
+        self.options.quiet = False
+        self.session.getRepo.return_value = {}
+        self.checkForBuilds.return_value = True
+        expected = 'Unsuccessfully waited 1:02 for a new %s repo' % self.tag_name + '\n'
+        self.__test_wait_repo_timeout(arguments, expected)
+
+    def test_anon_handle_wait_repo_with_build(self):
+        """Test anon_handle_wait_repo function with --build options"""
+        builds = ['bash-4.4.12-5.fc26', 'sed-4.4-1.fc26']
+        new_ver = 'bash-4.4.12-7.fc26'
+        arguments = [self.tag_name]
+        pkgs = ''
+        for b in builds:
+            arguments += ['--build', b]
+            pkgs += b + ':'
+        pkgs = pkgs[:-1].replace(':', ' and ')
+
+        self.options.quiet = False
+        self.session.getLatestBuilds.side_effect = [
+            [{'nvr': new_ver}], []
+        ]
+        self.checkForBuilds.return_value = True
+        self.session.getRepo.side_effect = [
+            {}, {}, {'id': 1, 'name': 'DEFAULT', 'create_event': 1}
+        ]
+        expected = 'Warning: nvr %s is not current in tag %s\n  latest build in %s is %s' % \
+                   (builds[0], self.tag_name, self.tag_name, new_ver) + "\n"
+        expected += 'Warning: package sed is not in tag %s' % self.tag_name + '\n'
+        expected += 'Successfully waited 0:03 for %s to appear in the %s repo' % (pkgs, self.tag_name) + '\n'
+        self.__test_wait_repo(arguments, expected)
+
+    def test_anon_handle_wait_repo_with_build_timeout(self):
+        """Test anon_handle_wait_repo function with --build options on timeout cases"""
+        builds = ['bash-4.4.12-5.fc26', 'sed-4.4-1.fc26']
+        arguments = [self.tag_name]
+        pkgs = ''
+        for b in builds:
+            arguments += ['--build', b]
+            pkgs += b + ':'
+        pkgs = pkgs[:-1].replace(':', ' and ')
+
+        self.options.quiet = False
+        self.session.getLatestBuilds.side_effect = [
+            [{'nvr': builds[0]}],
+            [{'nvr': builds[1]}],
+        ]
+        self.checkForBuilds.return_value = True
+        self.session.getRepo.return_value = {}
+        expected = 'Unsuccessfully waited 1:02 for %s to appear in the %s repo' % (pkgs, self.tag_name) + '\n'
+        self.__test_wait_repo_timeout(arguments, expected)
+
+    def test_anon_handle_wait_repo_errors(self):
+        """Test anon_handle_wait_repo function errors and exceptions"""
+        tests = [
+            # [ arguments, error_string ]
+            [[], "Please specify a tag name"],
+            [['tag1', 'tag2'], "Only one tag may be specified"],
+            [[self.tag_name], "Invalid tag: %s" % self.tag_name],
+            [[self.tag_name, '--target'], "Invalid build target: %s" % self.tag_name],
+        ]
+
+        self.session.getBuildTarget.return_value = {}
+        self.session.getTag.return_value = {}
+
+        for test in tests:
+            self.assert_system_exit(
+                anon_handle_wait_repo,
+                self.options,
+                self.session,
+                test[0],
+                stderr=self.format_error_message(test[1]),
+                activate_session=None)
+        self.activate_session.assert_not_called()
+
+    @mock.patch('sys.stdout', new_callable=six.StringIO)
+    def test_anon_handle_wait_repo_target_not_found(self, stdout):
+        """Test anon_handle_wait_repo function on target not found cases"""
+
+        # Case 1. both build and dest targets are not found
+        self.session.getTag.return_value = self.TAG.copy()
+        self.session.getBuildTargets.return_value = []
+        rv = anon_handle_wait_repo(self.options, self.session, [self.tag_name])
+        self.assertEqual(rv, 1)
+        expected = "%(name)s is not a build tag for any target" % self.TAG + "\n"
+        self.assert_console_message(stdout, expected)
+
+        # Cas 2. dest is matched, show suggestion
+        self.session.getBuildTargets.side_effect = [
+            [],
+            [
+                {'build_tag_name': 'build-tag-1'},
+                {'build_tag_name': 'build-tag-2'},
+                {'build_tag_name': 'build-tag-3'},
+            ],
+        ]
+        rv = anon_handle_wait_repo(self.options, self.session, [self.tag_name])
+        self.assertEqual(rv, 1)
+        expected = "%(name)s is not a build tag for any target" % self.TAG + "\n"
+        expected += "Suggested tags: build-tag-1, build-tag-2, build-tag-3\n"
+        self.assert_console_message(stdout, expected)
+
+    def test_anon_handle_wait_repo_help(self):
+        """Test anon_handle_wait_repo help message"""
+        self.assert_help(
+            anon_handle_wait_repo,
+            """Usage: %s wait-repo [options] <tag>
+(Specify the --help global option for a list of other help options)
+
+Options:
+  -h, --help         show this help message and exit
+  --build=NVR        Check that the given build is in the newly-generated repo
+                     (may be used multiple times)
+  --target           Interpret the argument as a build target name
+  --timeout=TIMEOUT  Amount of time to wait (in minutes) before giving up
+                     (default: 120)
+  --quiet            Suppress output, success or failure will be indicated by
+                     the return value only
+""" % self.progname)
+
+
+if __name__ == '__main__':
+    unittest.main()