From 4bde1481180e8440e04714b0987c5159e9fe0a2f Mon Sep 17 00:00:00 2001 From: Mike McLean Date: Oct 13 2016 04:49:57 +0000 Subject: PR#154: ut: cli - test_watch_tasks / test_add_group / test_add_host ... Merges #154 https://pagure.io/koji/pull-request/154 --- diff --git a/cli/koji b/cli/koji index cb4ede7..3dc78fe 100755 --- a/cli/koji +++ b/cli/koji @@ -817,6 +817,9 @@ def handle_block_pkg(options, session, args): tag = args[0] # check if list of packages exists for that tag already dsttag=session.getTag(tag) + if dsttag is None: + print "No such tag: %s" % tag + return 1 pkglist = dict([(p['package_name'], p['package_id']) for p in session.listPackages(tagID=dsttag['id'], inherited=True)]) ret = 0 for package in args[1:]: @@ -826,9 +829,10 @@ def handle_block_pkg(options, session, args): ret = 1 if ret: return ret + session.multicall = True for package in args[1:]: - #really should implement multicall... session.packageListBlock(tag,package) + session.multiCall(strict=True) def handle_remove_pkg(options, session, args): "[admin] Remove a package from the listing for tag" @@ -846,6 +850,9 @@ def handle_remove_pkg(options, session, args): opts['force'] = options.force # check if list of packages exists for that tag already dsttag=session.getTag(tag) + if dsttag is None: + print "No such tag: %s" % tag + return 1 pkglist = dict([(p['package_name'], p['package_id']) for p in session.listPackages(tagID=dsttag['id'])]) ret = 0 for package in args[1:]: @@ -855,9 +862,10 @@ def handle_remove_pkg(options, session, args): ret = 1 if ret: return ret + session.multicall = True for package in args[1:]: - #really should implement multicall... session.packageListRemove(tag, package, **opts) + session.multiCall(strict=True) def _unique_path(prefix): """Create a unique path fragment by appending a path component diff --git a/tests/test_cli/test_add_group.py b/tests/test_cli/test_add_group.py new file mode 100644 index 0000000..929d42a --- /dev/null +++ b/tests/test_cli/test_add_group.py @@ -0,0 +1,171 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +import loadcli + +cli = loadcli.cli + + +class TestAddGroup(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_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 = 'dsttag' + session.getTagGroups.return_value = [ + {'name': 'otherGroup', 'group_id': 'otherGroupId'}] + + # Run it and check immediate output + rv = cli.handle_add_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) + session.hasPerm.assert_called_once_with('admin') + session.getTag.assert_called_once_with(tag) + session.getTagGroups.assert_called_once_with(tag, inherit=False) + session.groupListAdd.assert_called_once_with(tag, group) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_group_dupl(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 = 'dsttag' + session.getTagGroups.return_value = [ + {'name': 'group', 'group_id': 'groupId'}] + + # Run it and check immediate output + rv = cli.handle_add_group(options, session, arguments) + actual = stdout.getvalue() + expected = 'Group group already exists for tag tag\n' + self.assertMultiLineEqual(actual, expected) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.hasPerm.assert_called_once_with('admin') + session.getTag.assert_called_once_with(tag) + session.getTagGroups.assert_called_once_with(tag, inherit=False) + session.groupListAdd.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_group_help( + self, + activate_session_mock, + stderr, + stdout): + arguments = [] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + rv = cli.handle_add_group(options, session, arguments) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + progname = os.path.basename(sys.argv[0]) or 'koji' + expected_stderr = """Usage: %s add-group +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a tag name and a group name +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.hasPerm.assert_not_called() + session.getTag.assert_not_called() + session.getTagGroups.assert_not_called() + session.groupListAdd.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_group_no_perm(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 = False + + # Run it and check immediate output + rv = cli.handle_add_group(options, session, arguments) + actual = stdout.getvalue() + expected = 'This action requires admin privileges\n' + self.assertMultiLineEqual(actual, expected) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.hasPerm.assert_called_once_with('admin') + session.getTag.assert_not_called() + session.getTagGroups.assert_not_called() + session.groupListAdd.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_group_no_tag(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 = None + + # Run it and check immediate output + rv = cli.handle_add_group(options, session, arguments) + actual = stdout.getvalue() + expected = 'Unknown tag: tag\n' + self.assertMultiLineEqual(actual, expected) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.hasPerm.assert_called_once_with('admin') + session.getTag.assert_called_once_with(tag) + session.getTagGroups.assert_not_called() + session.groupListAdd.assert_not_called() + self.assertEqual(rv, 1) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_add_host.py b/tests/test_cli/test_add_host.py new file mode 100644 index 0000000..552bcf3 --- /dev/null +++ b/tests/test_cli/test_add_host.py @@ -0,0 +1,166 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +import loadcli + +cli = loadcli.cli + + +class TestAddHost(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host(self, activate_session_mock, stdout): + host = 'host' + host_id = 1 + arches = ['arch1', 'arch2'] + krb_principal = '--krb-principal=krb' + arguments = [host] + arches + arguments.append(krb_principal) + kwargs = {'krb_principal': 'krb'} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getHost.return_value = None + session.addHost.return_value = host_id + # Run it and check immediate output + # args: host, arch1, arch2, --krb-principal=krb + # expected: success + rv = cli.handle_add_host(options, session, arguments) + actual = stdout.getvalue() + expected = 'host added: id 1\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.addHost.assert_called_once_with(host, arches, **kwargs) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_no_krb_principal( + self, activate_session_mock, stdout): + host = 'host' + host_id = 1 + arches = ['arch1', 'arch2'] + arguments = [host] + arches + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + session.getHost.return_value = None + session.addHost.return_value = host_id + # Run it and check immediate output + # args: host, arch1, arch2 + # expected: success + rv = cli.handle_add_host(options, session, arguments) + actual = stdout.getvalue() + expected = 'host added: id 1\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.addHost.assert_called_once_with(host, arches) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_dupl(self, activate_session_mock, stdout): + host = 'host' + host_id = 1 + arches = ['arch1', 'arch2'] + krb_principal = '--krb-principal=krb' + arguments = [host] + arches + arguments.append(krb_principal) + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + session.getHost.return_value = host_id + # Run it and check immediate output + # args: host, arch1, arch2, --krb-principal=krb + # expected: failed, host already exists + rv = cli.handle_add_host(options, session, arguments) + actual = stdout.getvalue() + expected = 'host is already in the database\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.addHost.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_help(self, activate_session_mock, stderr, stdout): + arguments = [] + options = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_add_host(options, session, arguments) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s add-host [options] hostname arch [arch2 ...] +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a hostname and at least one arch +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.hasHost.assert_not_called() + session.addHost.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_failed(self, activate_session_mock, stdout): + host = 'host' + arches = ['arch1', 'arch2'] + krb_principal = '--krb-principal=krb' + arguments = [host] + arches + arguments.append(krb_principal) + kwargs = {'krb_principal': 'krb'} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getHost.return_value = None + session.addHost.return_value = None + # Run it and check immediate output + # args: host, arch1, arch2, --krb-principal=krb + # expected: failed + cli.handle_add_host(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) + session.getHost.assert_called_once_with(host) + session.addHost.assert_called_once_with(host, arches, **kwargs) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_add_host_to_channel.py b/tests/test_cli/test_add_host_to_channel.py new file mode 100644 index 0000000..4324c06 --- /dev/null +++ b/tests/test_cli/test_add_host_to_channel.py @@ -0,0 +1,205 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +import loadcli + +cli = loadcli.cli + + +class TestAddHostToChannel(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_to_channel(self, activate_session_mock, stdout): + host = 'host' + host_info = mock.ANY + channel = 'channel' + channel_info = mock.ANY + args = [host, channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + session.getHost.return_value = host_info + # Run it and check immediate output + # args: host, channel + # expected: success + rv = cli.handle_add_host_to_channel(options, session, args) + 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) + session.getChannel.assert_called_once_with(channel) + session.getHost.assert_called_once_with(host) + session.addHostToChannel.assert_called_once_with(host, channel) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_to_channel_list( + self, activate_session_mock, stdout): + list_arg = '--list' + args = [list_arg] + channel_infos = [{'name': 'channel1'}, {'name': 'channel2'}] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.listChannels.return_value = channel_infos + # Run it and check immediate output + # args: --list + # expected: list all channel names + rv = cli.handle_add_host_to_channel(options, session, args) + actual = stdout.getvalue() + expected = 'channel1\nchannel2\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.listChannels.assert_called_once() + session.getChannel.assert_not_called() + session.getHost.assert_not_called() + session.addHostToChannel.assert_not_called() + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_to_channel_new( + self, activate_session_mock, stdout): + host = 'host' + host_info = mock.ANY + channel = 'channel' + new_arg = '--new' + args = [host, channel, new_arg] + kwargs = {'create': True} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getHost.return_value = host_info + # Run it and check immediate output + # args: host, channel, --new + # expected: success + rv = cli.handle_add_host_to_channel(options, session, args) + 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) + session.getChannel.assert_not_called() + session.getHost.assert_called_once_with(host) + session.addHostToChannel.assert_called_once_with( + host, channel, **kwargs) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_to_channel_no_channel( + self, activate_session_mock, stdout): + host = 'host' + channel = 'channel' + channel_info = None + args = [host, channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + # Run it and check immediate output + # args: host, channel + # expected: failed, channel not found + rv = cli.handle_add_host_to_channel(options, session, args) + actual = stdout.getvalue() + expected = 'No such channel: channel\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getChannel.assert_called_once_with(channel) + session.getHost.assert_not_called() + session.addHostToChannel.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_to_channel_no_host( + self, activate_session_mock, stdout): + host = 'host' + host_info = None + channel = 'channel' + channel_info = mock.ANY + args = [host, channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + session.getHost.return_value = host_info + # Run it and check immediate output + # args: host, channel + # expected: success + rv = cli.handle_add_host_to_channel(options, session, args) + actual = stdout.getvalue() + expected = 'No such host: host\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getChannel.assert_called_once_with(channel) + session.getHost.assert_called_once_with(host) + session.addHostToChannel.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_host_to_channel_help( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + # args: _empty_ + # expected: failed, help msg shows + with self.assertRaises(SystemExit) as cm: + cli.handle_add_host_to_channel(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s add-host-to-channel [options] hostname channel +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a hostname and a channel +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getHost.assert_not_called() + session.getChannel.assert_not_called() + session.listChannels.assert_not_called() + session.addHostToChannel.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_add_pkg.py b/tests/test_cli/test_add_pkg.py new file mode 100644 index 0000000..94483b0 --- /dev/null +++ b/tests/test_cli/test_add_pkg.py @@ -0,0 +1,256 @@ +import unittest + + +import StringIO as stringio + +import os + +import sys + +import mock + +from mock import call + +import loadcli + +cli = loadcli.cli + + +class TestAddPkg(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_pkg(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + package = 'package' + owner = 'owner' + owner_info = mock.ANY + extra_arches = 'arch1,arch2 arch3, arch4' + args = [ + '--owner=' + + owner, + '--extra-arches=' + + extra_arches, + tag, + package] + kwargs = {'force': None, + 'block': False, + 'extra_arches': 'arch1 arch2 arch3 arch4'} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getUser.return_value = owner_info + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'other_package', 'package_id': 2}] + # Run it and check immediate output + # args: --owner, --extra-arches='arch1,arch2 arch3, arch4', tag, package + # expected: success + rv = cli.handle_add_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'Adding 1 packages to tag tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getUser.assert_called_once_with(owner) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_called_once_with(tagID=dsttag['id']) + session.packageListAdd.assert_called_once_with( + tag, package, owner, **kwargs) + session.multiCall.assert_called_once_with(strict=True) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_pkg_multi_pkg(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + packages = ['package1', 'package2', 'package3'] + owner = 'owner' + owner_info = mock.ANY + extra_arches = 'arch1,arch2 arch3, arch4' + args = [ + '--owner=' + owner, + '--extra-arches=' + extra_arches, + tag] + packages + kwargs = {'force': None, + 'block': False, + 'extra_arches': 'arch1 arch2 arch3 arch4'} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getUser.return_value = owner_info + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'package2', 'package_id': 2}] + # Run it and check immediate output + # args: --owner, --extra-arches='arch1,arch2 arch3, arch4', + # tag, package1, package2, package3 + # expected: success + rv = cli.handle_add_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'Package package2 already exists in tag tag\nAdding 2 packages to tag tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + self.assertEqual(session.mock_calls, + [call.getUser(owner), + call.getTag(tag), + call.listPackages(tagID=dsttag['id']), + call.packageListAdd(tag, packages[0], owner, **kwargs), + call.packageListAdd(tag, packages[2], owner, **kwargs), + call.multiCall(strict=True)]) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_pkg_owner_no_exists( + self, activate_session_mock, stdout): + tag = 'tag' + packages = ['package1', 'package2', 'package3'] + owner = 'owner' + owner_info = None + extra_arches = 'arch1,arch2 arch3, arch4' + args = [ + '--owner=' + owner, + '--extra-arches=' + extra_arches, + tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getUser.return_value = owner_info + # Run it and check immediate output + # args: --owner, --extra-arches='arch1,arch2 arch3, arch4', + # tag, package1, package2, package3 + # expected: failed: owner does not exist + rv = cli.handle_add_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'User owner does not exist\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + self.assertEqual(session.mock_calls, + [call.getUser(owner)]) + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_pkg_tag_no_exists(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = None + packages = ['package1', 'package2', 'package3'] + owner = 'owner' + owner_info = mock.ANY + extra_arches = 'arch1,arch2 arch3, arch4' + args = [ + '--owner=' + owner, + '--extra-arches=' + extra_arches, + tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getUser.return_value = owner_info + session.getTag.return_value = dsttag + # Run it and check immediate output + # args: --owner, --extra-arches='arch1,arch2 arch3, arch4', + # tag, package1, package2, package3 + # expected: failed: tag does not exist + with self.assertRaises(SystemExit) as cm: + cli.handle_add_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'No such tag: tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + self.assertEqual(session.mock_calls, + [call.getUser(owner), + call.getTag(tag)]) + self.assertEqual(cm.exception.code, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_pkg_no_owner( + self, activate_session_mock, stderr, stdout): + tag = 'tag' + packages = ['package1', 'package2', 'package3'] + extra_arches = 'arch1,arch2 arch3, arch4' + args = ['--extra-arches=' + extra_arches, tag] + packages + options = mock.MagicMock() + + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_add_pkg(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s add-pkg [options] tag package [package2 ...] +(Specify the --help global option for a list of other help options) + +%s: error: Please specify an owner for the package(s) +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getUser.assert_not_called() + session.getTag.assert_not_called() + session.listPackages.assert_not_called() + session.packageListAdd.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_add_pkg_no_arg( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_add_pkg(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s add-pkg [options] tag package [package2 ...] +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a tag and at least one package +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getUser.assert_not_called() + session.getTag.assert_not_called() + session.listPackages.assert_not_called() + session.packageListAdd.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_block_pkg.py b/tests/test_cli/test_block_pkg.py new file mode 100644 index 0000000..f493c32 --- /dev/null +++ b/tests/test_cli/test_block_pkg.py @@ -0,0 +1,191 @@ +import unittest + + +import StringIO as stringio + +import os + +import sys + +import mock + +from mock import call + +import loadcli + +cli = loadcli.cli + + +class TestBlockPkg(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_block_pkg(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + package = 'package' + args = [tag, package] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': package, 'package_id': 1}] + # Run it and check immediate output + # args: tag, package + # expected: success + rv = cli.handle_block_pkg(options, session, args) + 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) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_called_once_with( + tagID=dsttag['id'], inherited=True) + session.packageListBlock.assert_called_once_with( + tag, package) + session.multiCall.assert_called_once_with(strict=True) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_block_pkg_multi_pkg(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + packages = ['package1', 'package2', 'package3'] + args = [tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'package1', 'package_id': 1}, + {'package_name': 'package2', 'package_id': 2}, + {'package_name': 'package3', 'package_id': 3}, + {'package_name': 'other_package', 'package_id': 4} + ] + # Run it and check immediate output + # args: tag, package1, package2, package3 + # expected: success + rv = cli.handle_block_pkg(options, session, args) + 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) + 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)]) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_block_pkg_no_package(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + packages = ['package1', 'package2', 'package3'] + args = [tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'package1', 'package_id': 1}, + {'package_name': 'package3', 'package_id': 3}, + {'package_name': 'other_package', 'package_id': 4}] + # Run it and check immediate output + # args: tag, package1, package2, package3 + # expected: failed: can not find package2 under tag + rv = cli.handle_block_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'Package package2 doesn\'t exist in tag tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_called_once_with( + tagID=dsttag['id'], inherited=True) + session.packageListBlock.assert_not_called() + session.multiCall.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_block_pkg_tag_no_exists( + self, activate_session_mock, stdout): + tag = 'tag' + dsttag = None + packages = ['package1', 'package2', 'package3'] + args = [tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + # Run it and check immediate output + # args: tag, package1, package2, package3 + # expected: failed: tag does not exist + rv = cli.handle_block_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'No such tag: tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_not_called() + session.packageListBlock.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_block_pkg_help( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_block_pkg(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s block-pkg [options] tag package [package2 ...] +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a tag and at least one package +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getTag.assert_not_called() + session.listPackages.assert_not_called() + session.packageListBlock.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_edit_host.py b/tests/test_cli/test_edit_host.py new file mode 100644 index 0000000..0680fe6 --- /dev/null +++ b/tests/test_cli/test_edit_host.py @@ -0,0 +1,221 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +from mock import call + +import loadcli + +cli = loadcli.cli + + +class TestEditHost(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_edit_host(self, activate_session_mock, stdout): + host = 'host' + host_info = mock.ANY + arches = 'arch1 arch2' + capacity = 0.22 + description = 'description' + comment = 'comment' + args = [host] + args.append('--arches=' + arches) + args.append('--capacity=' + str(capacity)) + args.append('--description=' + description) + args.append('--comment=' + comment) + kwargs = {'arches': arches, + 'capacity': capacity, + 'description': description, + 'comment': comment} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.multiCall.side_effect = [[[host_info]], [[True]]] + # Run it and check immediate output + # args: host, --arches='arch1 arch2', --capacity=0.22, + # --description=description, --comment=comment + # expected: success + rv = cli.handle_edit_host(options, session, args) + actual = stdout.getvalue() + expected = 'Edited host\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.editHost.assert_called_once_with(host, **kwargs) + self.assertEqual(session.multiCall.call_count, 2) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_edit_host_failed(self, activate_session_mock, stdout): + host = 'host' + host_info = mock.ANY + arches = 'arch1 arch2' + capacity = 0.22 + description = 'description' + comment = 'comment' + args = [host] + args.append('--arches=' + arches) + args.append('--capacity=' + str(capacity)) + args.append('--description=' + description) + args.append('--comment=' + comment) + kwargs = {'arches': arches, + 'capacity': capacity, + 'description': description, + 'comment': comment} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.multiCall.side_effect = [[[host_info]], [[False]]] + # Run it and check immediate output + # args: host, --arches='arch1 arch2', --capacity=0.22, + # --description=description, --comment=comment + # expected: failed - session.editHost == False + rv = cli.handle_edit_host(options, session, args) + actual = stdout.getvalue() + expected = 'No changes made to host\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.editHost.assert_called_once_with(host, **kwargs) + self.assertEqual(session.multiCall.call_count, 2) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_edit_multi_host(self, activate_session_mock, stdout): + hosts = ['host1', 'host2'] + host_infos = [mock.ANY, mock.ANY] + arches = 'arch1 arch2' + capacity = 0.22 + description = 'description' + comment = 'comment' + args = hosts + args.append('--arches=' + arches) + args.append('--capacity=' + str(capacity)) + args.append('--description=' + description) + args.append('--comment=' + comment) + kwargs = {'arches': arches, + 'capacity': capacity, + 'description': description, + 'comment': comment} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.multiCall.side_effect = [[[info] + for info in host_infos], [[True], [True]]] + # Run it and check immediate output + # args: host1, host2, --arches='arch1 arch2', --capacity=0.22, + # --description=description, --comment=comment + # expected: success + rv = cli.handle_edit_host(options, session, args) + actual = stdout.getvalue() + expected = 'Edited host1\nEdited host2\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + self.assertEqual(session.mock_calls, + [call.getHost(hosts[0]), + call.getHost(hosts[1]), + call.multiCall(strict=True), + call.editHost(hosts[0], + **kwargs), + call.editHost(hosts[1], + **kwargs), + call.multiCall(strict=True)]) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_edit_host_no_arg( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + # Mock out the xmlrpc server + session = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + # Run it and check immediate output + # args: _empty_ + # expected: failed - should specify host + with self.assertRaises(SystemExit) as cm: + cli.handle_edit_host(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s edit-host hostname ... [options] +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a hostname +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getHost.assert_not_called() + session.editHost.assert_not_called() + session.multiCall.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_edit_host_no_host(self, activate_session_mock, stdout): + host = 'host' + host_info = None + arches = 'arch1 arch2' + capacity = 0.22 + description = 'description' + comment = 'comment' + args = [host] + args.append('--arches=' + arches) + args.append('--capacity=' + str(capacity)) + args.append('--description=' + description) + args.append('--comment=' + comment) + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.multiCall.return_value = [[host_info]] + # Run it and check immediate output + # args: host, --arches='arch1 arch2', --capacity=0.22, + # --description=description, --comment=comment + # expected: failed -- getHost() == None + rv = cli.handle_edit_host(options, session, args) + actual = stdout.getvalue() + expected = """Host host does not exist +No changes made, please correct the command line +""" + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.editHost.assert_not_called() + self.assertEqual(session.multiCall.call_count, 1) + self.assertEqual(rv, 1) + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_list_commands.py b/tests/test_cli/test_list_commands.py index a4de0d9..afc9a14 100644 --- a/tests/test_cli/test_list_commands.py +++ b/tests/test_cli/test_list_commands.py @@ -6,22 +6,12 @@ import StringIO as stringio import mock - -# We have to do this craziness because 'import koji' is ambiguous. Is it the -# koji module, or the koji cli module. Jump through hoops accordingly. -# http://stackoverflow.com/questions/67631/how-to-import-a-module-given-the-full-path -CLI_FILENAME = os.path.dirname(__file__) + "/../../cli/koji" -if sys.version_info[0] >= 3: - import importlib.util - spec = importlib.util.spec_from_file_location("koji_cli", CLI_FILENAME) - cli = importlib.util.module_from_spec(spec) - spec.loader.exec_module(cli) -else: - import imp - cli = imp.load_source('koji_cli', CLI_FILENAME) +import loadcli +cli = loadcli.cli class TestListCommands(unittest.TestCase): + def setUp(self): self.options = mock.MagicMock() self.session = mock.MagicMock() diff --git a/tests/test_cli/test_remove_channel.py b/tests/test_cli/test_remove_channel.py new file mode 100644 index 0000000..b76691a --- /dev/null +++ b/tests/test_cli/test_remove_channel.py @@ -0,0 +1,132 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +import loadcli + +cli = loadcli.cli + + +class TestRemoveChannel(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_channel(self, activate_session_mock, stdout): + channel = 'channel' + channel_info = mock.ANY + args = [channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + # Run it and check immediate output + # args: channel + # expected: success + rv = cli.handle_remove_channel(options, session, args) + 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) + session.getChannel.assert_called_once_with(channel) + session.removeChannel.assert_called_once_with(channel, force=None) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_channel_force(self, activate_session_mock, stdout): + channel = 'channel' + channel_info = mock.ANY + force_arg = '--force' + args = [force_arg, channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + # Run it and check immediate output + # args: --force, channel + # expected: success + rv = cli.handle_remove_channel(options, session, args) + 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) + session.getChannel.assert_called_once_with(channel) + session.removeChannel.assert_called_once_with(channel, force=True) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_channel_no_channel( + self, activate_session_mock, stdout): + channel = 'channel' + channel_info = None + args = [channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + # Run it and check immediate output + # args: channel + # expected: failed: no such channel + rv = cli.handle_remove_channel(options, session, args) + actual = stdout.getvalue() + expected = 'No such channel: channel\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getChannel.assert_called_once_with(channel) + session.removeChannel.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_channel_help( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_remove_channel(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s remove-channel [options] channel +(Specify the --help global option for a list of other help options) + +%s: error: Incorrect number of arguments +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getChannel.assert_not_called() + session.removeChannel.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_remove_host_from_channel.py b/tests/test_cli/test_remove_host_from_channel.py new file mode 100644 index 0000000..490edf4 --- /dev/null +++ b/tests/test_cli/test_remove_host_from_channel.py @@ -0,0 +1,147 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +import loadcli + +cli = loadcli.cli + + +class TestRemoveHostFromChannel(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_host_from_channel( + self, activate_session_mock, stdout): + host = 'host' + host_info = {'id': 1} + channel = 'channel' + args = [host, channel] + channel_infos = [{'name': 'channel'}, {'name': 'other_channel'}] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getHost.return_value = host_info + session.listChannels.return_value = channel_infos + # Run it and check immediate output + # args: host, channel + # expected: success + rv = cli.handle_remove_host_from_channel(options, session, args) + 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) + session.getHost.assert_called_once_with(host) + session.listChannels.assert_called_once_with(host_info['id']) + session.removeHostFromChannel.assert_called_once_with(host, channel) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_host_from_channel_no_host( + self, activate_session_mock, stdout): + host = 'host' + host_info = None + channel = 'channel' + args = [host, channel] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getHost.return_value = host_info + # Run it and check immediate output + # args: host, channel + # expected: failed: no such host + rv = cli.handle_remove_host_from_channel(options, session, args) + actual = stdout.getvalue() + expected = 'No such host: host\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.listChannels.assert_not_called() + session.removeHostFromChannel.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_host_from_channel_not_a_member( + self, activate_session_mock, stdout): + host = 'host' + host_info = {'id': 1} + channel = 'channel' + args = [host, channel] + channel_infos = [{'name': 'other_channel1'}, + {'name': 'other_channel2'}] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getHost.return_value = host_info + session.listChannels.return_value = channel_infos + # Run it and check immediate output + # args: host, channel + # expected: success: host isn't belong to channel + rv = cli.handle_remove_host_from_channel(options, session, args) + actual = stdout.getvalue() + expected = 'Host host is not a member of channel channel\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getHost.assert_called_once_with(host) + session.listChannels.assert_called_once_with(host_info['id']) + session.removeHostFromChannel.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_host_from_channel_help( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + # args: _empty_ + # expected: failed, help msg shows + with self.assertRaises(SystemExit) as cm: + cli.handle_remove_host_from_channel(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s remove-host-from-channel [options] hostname channel +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a hostname and a channel +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getHost.assert_not_called() + session.listChannels.assert_not_called() + session.removeHostFromChannel.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_remove_pkg.py b/tests/test_cli/test_remove_pkg.py new file mode 100644 index 0000000..9926158 --- /dev/null +++ b/tests/test_cli/test_remove_pkg.py @@ -0,0 +1,232 @@ +import unittest + + +import StringIO as stringio + +import os + +import sys + +import mock + +from mock import call + +import loadcli + +cli = loadcli.cli + + +class TestRemovePkg(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_pkg(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + package = 'package' + args = [tag, package] + kwargs = {'force': None} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': package, 'package_id': 1}] + # Run it and check immediate output + # args: tag, package + # expected: success + rv = cli.handle_remove_pkg(options, session, args) + 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) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_called_once_with( + tagID=dsttag['id']) + session.packageListRemove.assert_called_once_with( + tag, package, **kwargs) + session.multiCall.assert_called_once_with(strict=True) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_pkg_multi_pkg(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + packages = ['package1', 'package2', 'package3'] + args = [tag] + packages + kwargs = {'force': None} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'package1', 'package_id': 1}, + {'package_name': 'package2', 'package_id': 2}, + {'package_name': 'package3', 'package_id': 3}, + {'package_name': 'other_package', 'package_id': 4} + ] + # Run it and check immediate output + # args: tag, package1, package2, package3 + # expected: success + rv = cli.handle_remove_pkg(options, session, args) + 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) + self.assertEqual( + session.mock_calls, [ + call.getTag(tag), call.listPackages( + tagID=dsttag['id']), call.packageListRemove( + tag, packages[0], **kwargs), call.packageListRemove( + tag, packages[1], **kwargs), call.packageListRemove( + tag, packages[2], **kwargs), call.multiCall( + strict=True)]) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_pkg_force(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + packages = ['package1', 'package2', 'package3'] + args = ['--force', tag] + packages + kwargs = {'force': True} + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'package1', 'package_id': 1}, + {'package_name': 'package2', 'package_id': 2}, + {'package_name': 'package3', 'package_id': 3}, + {'package_name': 'other_package', 'package_id': 4} + ] + # Run it and check immediate output + # args: --force, tag, package1, package2, package3 + # expected: success + rv = cli.handle_remove_pkg(options, session, args) + 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) + self.assertEqual( + session.mock_calls, [ + call.getTag(tag), call.listPackages( + tagID=dsttag['id']), call.packageListRemove( + tag, packages[0], **kwargs), call.packageListRemove( + tag, packages[1], **kwargs), call.packageListRemove( + tag, packages[2], **kwargs), call.multiCall( + strict=True)]) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_pkg_no_package(self, activate_session_mock, stdout): + tag = 'tag' + dsttag = {'name': tag, 'id': 1} + packages = ['package1', 'package2', 'package3'] + args = [tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + session.listPackages.return_value = [ + {'package_name': 'package1', 'package_id': 1}, + {'package_name': 'package3', 'package_id': 3}, + {'package_name': 'other_package', 'package_id': 4}] + # Run it and check immediate output + # args: tag, package1, package2, package3 + # expected: failed: can not find package2 under tag + rv = cli.handle_remove_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'Package package2 is not in tag tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_called_once_with( + tagID=dsttag['id']) + session.packageListRemove.assert_not_called() + session.multiCall.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_pkg_tag_no_exists( + self, activate_session_mock, stdout): + tag = 'tag' + dsttag = None + packages = ['package1', 'package2', 'package3'] + args = [tag] + packages + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getTag.return_value = dsttag + # Run it and check immediate output + # args: tag, package1, package2, package3 + # expected: failed: tag does not exist + rv = cli.handle_remove_pkg(options, session, args) + actual = stdout.getvalue() + expected = 'No such tag: tag\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getTag.assert_called_once_with(tag) + session.listPackages.assert_not_called() + session.packageListRemove.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_remove_pkg_help( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_remove_pkg(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s remove-pkg [options] tag package [package2 ...] +(Specify the --help global option for a list of other help options) + +%s: error: Please specify a tag and at least one package +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getTag.assert_not_called() + session.listPackages.assert_not_called() + session.packageListRemove.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_rename_channel.py b/tests/test_cli/test_rename_channel.py new file mode 100644 index 0000000..fb64c01 --- /dev/null +++ b/tests/test_cli/test_rename_channel.py @@ -0,0 +1,108 @@ +import unittest + +import StringIO as stringio + +import os + +import sys + +import mock + +import loadcli + +cli = loadcli.cli + + +class TestRenameChannel(unittest.TestCase): + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_rename_channel(self, activate_session_mock, stdout): + old_name = 'old_name' + new_name = 'new_name' + channel_info = mock.ANY + args = [old_name, new_name] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + # Run it and check immediate output + # args: old_name, new_name + # expected: success + rv = cli.handle_rename_channel(options, session, args) + 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) + session.getChannel.assert_called_once_with(old_name) + session.renameChannel.assert_called_once_with(old_name, new_name) + self.assertNotEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_rename_channel_no_channel( + self, activate_session_mock, stdout): + old_name = 'old_name' + new_name = 'new_name' + channel_info = None + args = [old_name, new_name] + options = mock.MagicMock() + + # Mock out the xmlrpc server + session = mock.MagicMock() + + session.getChannel.return_value = channel_info + # Run it and check immediate output + # args: old_name, new_name + # expected: failed: no such channel + rv = cli.handle_rename_channel(options, session, args) + actual = stdout.getvalue() + expected = 'No such channel: old_name\n' + self.assertMultiLineEqual(actual, expected) + # Finally, assert that things were called as we expected. + activate_session_mock.assert_called_once_with(session) + session.getChannel.assert_called_once_with(old_name) + session.renameChannel.assert_not_called() + self.assertEqual(rv, 1) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('sys.stderr', new_callable=stringio.StringIO) + @mock.patch('koji_cli.activate_session') + def test_handle_rename_channel_help( + self, activate_session_mock, stderr, stdout): + args = [] + options = mock.MagicMock() + progname = os.path.basename(sys.argv[0]) or 'koji' + + # Mock out the xmlrpc server + session = mock.MagicMock() + + # Run it and check immediate output + with self.assertRaises(SystemExit) as cm: + cli.handle_rename_channel(options, session, args) + actual_stdout = stdout.getvalue() + actual_stderr = stderr.getvalue() + expected_stdout = '' + expected_stderr = """Usage: %s rename-channel [options] old-name new-name +(Specify the --help global option for a list of other help options) + +%s: error: Incorrect number of arguments +""" % (progname, progname) + self.assertMultiLineEqual(actual_stdout, expected_stdout) + self.assertMultiLineEqual(actual_stderr, expected_stderr) + + # Finally, assert that things were called as we expected. + activate_session_mock.assert_not_called() + session.getChannel.assert_not_called() + session.renameChannel.assert_not_called() + self.assertEqual(cm.exception.code, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_cli/test_runroot.py b/tests/test_cli/test_runroot.py index 7fd2777..4545ef2 100644 --- a/tests/test_cli/test_runroot.py +++ b/tests/test_cli/test_runroot.py @@ -6,22 +6,12 @@ import StringIO as stringio import mock - -# We have to do this craziness because 'import koji' is ambiguous. Is it the -# koji module, or the koji cli module. Jump through hoops accordingly. -# http://stackoverflow.com/questions/67631/how-to-import-a-module-given-the-full-path -CLI_FILENAME = os.path.dirname(__file__) + "/../../cli/koji" -if sys.version_info[0] >= 3: - import importlib.util - spec = importlib.util.spec_from_file_location("koji_cli", CLI_FILENAME) - cli = importlib.util.module_from_spec(spec) - spec.loader.exec_module(cli) -else: - import imp - cli = imp.load_source('koji_cli', CLI_FILENAME) +import loadcli +cli = loadcli.cli class TestListCommands(unittest.TestCase): + def setUp(self): self.options = mock.MagicMock() self.session = mock.MagicMock() @@ -29,7 +19,7 @@ class TestListCommands(unittest.TestCase): self.original_parser = cli.OptionParser cli.OptionParser = mock.MagicMock() self.parser = cli.OptionParser.return_value - cli.options = self.options # globals!!! + cli.options = self.options # globals!!! def tearDown(self): cli.OptionParser = self.original_parser @@ -63,7 +53,8 @@ class TestListCommands(unittest.TestCase): # Finally, assert that things were called as we expected. self.session.getTaskInfo.assert_called_once_with(1) self.session.listTaskOutput.assert_called_once_with(1) - self.session.downloadTaskOutput.assert_called_once_with(1, 'runroot.log') + self.session.downloadTaskOutput.assert_called_once_with( + 1, 'runroot.log') self.session.runroot.assert_called_once_with( tag, arch, command, repo_id=mock.ANY, weight=mock.ANY, mounts=mock.ANY, packages=mock.ANY, skip_setarch=mock.ANY, diff --git a/tests/test_cli/test_watch_tasks.py b/tests/test_cli/test_watch_tasks.py new file mode 100644 index 0000000..417cacd --- /dev/null +++ b/tests/test_cli/test_watch_tasks.py @@ -0,0 +1,229 @@ +import unittest + +import os +import sys +import StringIO as stringio + +import mock + +from mock import call + +import loadcli + +cli = loadcli.cli + + +class TestWatchTasks(unittest.TestCase): + + def setUp(self): + self.options = mock.MagicMock() + cli.options = self.options + self.session = mock.MagicMock(name='sessionMock') + self.args = mock.MagicMock() + self.original_parser = cli.OptionParser + cli.OptionParser = mock.MagicMock() + self.parser = cli.OptionParser.return_value + + def tearDown(self): + cli.OptionParser = self.original_parser + + # Show long diffs in error output... + maxDiff = None + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + def test_watch_tasks_no_tasklist(self, stdout): + returned = cli.watch_tasks(self.session, []) + actual = stdout.getvalue() + expected = "" + self.assertIsNone(returned) + self.assertEqual(actual, expected) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.TaskWatcher') + @mock.patch('koji_cli.display_tasklist_status') + @mock.patch('koji_cli.display_task_results') + def test_watch_tasks(self, dtrMock, dtsMock, twClzMock, stdout): + self.options.poll_interval = 0 + manager = mock.MagicMock() + manager.attach_mock(twClzMock, 'TaskWatcherMock') + manager.attach_mock(dtrMock, 'display_task_results_mock') + manager.attach_mock(dtsMock, 'display_tasklist_status_mock') + tw1 = manager.tw1 + tw1.level = 0 + tw1.is_done.side_effect = [False, True, False, True, True] + tw1.update.side_effect = [False, False, True, True, True] + tw1.is_success.return_value = False + tw2 = manager.tw2 + tw2.level = 0 + tw2.is_done.side_effect = [False, False, False, False, True] + tw2.update.side_effect = [True, False, False, True, True] + tw2.is_success.return_value = False + self.session.getTaskChildren.side_effect = lambda p: [ + {'id': 11}, {'id': 12}] if (0 == p) else [] + manager.attach_mock(self.session, 'sessionMock') + + def side_effect(*args, **kwargs): + rt = None + if args[0] not in range(2): + rt = mock.MagicMock() + rt.level = args[2] + rt.is_done.return_value = True + rt.update.return_value = True + rt.is_success.return_value = True + manager.attach_mock(rt, 'tw' + str(args[0])) + else: + rt = {0: tw1, 1: tw2}.get(args[0]) + return rt + + twClzMock.side_effect = side_effect + rv = cli.watch_tasks(self.session, range(2), quiet=False) + actual = stdout.getvalue() + self.assertMultiLineEqual( + actual, "Watching tasks (this may be safely interrupted)...\n\n") + self.assertEqual(rv, 1) + self.assertEqual(manager.mock_calls, + [call.TaskWatcherMock(0, self.session, quiet=False), + call.TaskWatcherMock(1, self.session, quiet=False), + call.tw1.update(), + call.tw1.is_done(), + call.sessionMock.getTaskChildren(0), + call.TaskWatcherMock(11, self.session, 1, quiet=False), + call.tw11.update(), + call.TaskWatcherMock(12, self.session, 1, quiet=False), + call.tw12.update(), + call.tw2.update(), + call.tw2.is_done(), + call.sessionMock.getTaskChildren(1), + call.tw1.update(), + call.tw1.is_done(), + call.tw1.is_success(), + call.sessionMock.getTaskChildren(0), + call.tw2.update(), + call.tw2.is_done(), + call.sessionMock.getTaskChildren(1), + call.tw11.update(), + call.tw11.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw11.is_success(), + call.sessionMock.getTaskChildren(11), + call.tw12.update(), + call.tw12.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw12.is_success(), + call.sessionMock.getTaskChildren(12), + call.tw1.update(), + call.tw1.is_done(), + call.sessionMock.getTaskChildren(0), + call.tw2.update(), + call.tw2.is_done(), + call.sessionMock.getTaskChildren(1), + call.tw11.update(), + call.tw11.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw11.is_success(), + call.sessionMock.getTaskChildren(11), + call.tw12.update(), + call.tw12.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw12.is_success(), + call.sessionMock.getTaskChildren(12), + call.tw1.update(), + call.tw1.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw1.is_success(), + call.sessionMock.getTaskChildren(0), + call.tw2.update(), + call.tw2.is_done(), + call.sessionMock.getTaskChildren(1), + call.tw11.update(), + call.tw11.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw11.is_success(), + call.sessionMock.getTaskChildren(11), + call.tw12.update(), + call.tw12.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw12.is_success(), + call.sessionMock.getTaskChildren(12), + call.tw1.update(), + call.tw1.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw1.is_success(), + call.sessionMock.getTaskChildren(0), + call.tw2.update(), + call.tw2.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw2.is_success(), + call.sessionMock.getTaskChildren(1), + call.tw11.update(), + call.tw11.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw11.is_success(), + call.sessionMock.getTaskChildren(11), + call.tw12.update(), + call.tw12.is_done(), + call.display_tasklist_status_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}), + call.tw12.is_success(), + call.sessionMock.getTaskChildren(12), + call.display_task_results_mock({0: tw1, 1: tw2, 11: manager.tw11, 12: manager.tw12}) + ]) + + @mock.patch('sys.stdout', new_callable=stringio.StringIO) + @mock.patch('koji_cli.TaskWatcher') + @mock.patch('koji_cli.display_tasklist_status') + @mock.patch('koji_cli.display_task_results') + def test_watch_tasks_with_keyboardinterrupt( + self, dtrMock, dtsMock, twClzMock, stdout): + """Raise KeyboardInterrupt inner watch_tasks. + Raising it by SIGNAL might be better""" + self.options.poll_interval = 0 + manager = mock.MagicMock() + manager.attach_mock(twClzMock, 'TaskWatcherMock') + manager.attach_mock(dtrMock, 'display_task_results_mock') + manager.attach_mock(dtsMock, 'display_tasklist_status_mock') + tw1 = manager.tw1 + tw1.level = 0 + tw1.is_done.side_effect = [False, KeyboardInterrupt, False] + tw1.update.side_effect = [False, False] + tw1.is_success.return_value = False + tw1.str.return_value = 'tw1' + tw1.display_state.return_value = 'tw1.display_state' + tw2 = manager.tw2 + tw2.level = 0 + tw2.is_done.side_effect = [False, False, False, False, True] + tw2.update.side_effect = [True, False, False, True, True] + tw2.is_success.return_value = False + tw2.str.return_value = 'tw2' + tw2.display_state.return_value = 'tw2.display_state' + self.session.getTaskChildren.side_effect = lambda p: [ + {'id': 11}, {'id': 12}] if (0 == p) else [] + manager.attach_mock(self.session, 'sessionMock') + + def side_effect(*args, **kwargs): + rt = None + if args[0] not in range(2): + rt = mock.MagicMock() + rt.level = args[2] + rt.is_done.return_value = True + rt.update.return_value = True + rt.is_success.return_value = True + manager.attach_mock(rt, 'tw' + str(args[0])) + else: + rt = {0: tw1, 1: tw2}.get(args[0]) + return rt + + twClzMock.side_effect = side_effect + + cli.watch_tasks(self.session, range(2), quiet=False) + + actual = stdout.getvalue() + self.assertMultiLineEqual( + actual, """Watching tasks (this may be safely interrupted)... +Tasks still running. You can continue to watch with the '%s watch-task' command. +Running Tasks: +tw1: tw1.display_state +tw2: tw2.display_state +""" % (os.path.basename(sys.argv[0]) or 'koji')) + +if __name__ == '__main__': + unittest.main()