refactory CLI unittests including:
- introduce new CliTestCase class - implement useful assert functions in CliTestCase
This commit is contained in:
Franz Chih-Ping Hsieh
Mike McLean
parent
f79cab8c0d
commit
d331fdfa11
6 changed files with 402 additions and 292 deletions
|
|
@ -1,41 +1,24 @@
|
|||
from __future__ import absolute_import
|
||||
import mock
|
||||
import os
|
||||
import six
|
||||
import sys
|
||||
import unittest
|
||||
import json
|
||||
|
||||
from koji_cli.commands import handle_call
|
||||
from . import utils
|
||||
|
||||
|
||||
class TestCall(unittest.TestCase):
|
||||
class TestCall(utils.CliTestCase):
|
||||
|
||||
# Show long diffs in error output...
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.progname = os.path.basename(sys.argv[0]) or 'koji'
|
||||
|
||||
def format_error_message(self, message):
|
||||
return """Usage: %s call [options] name [arg...]
|
||||
self.error_format = """Usage: %s call [options] name [arg...]
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
%s: error: %s
|
||||
""" % (self.progname, self.progname, message)
|
||||
|
||||
def assert_console_output(self, device, expected, wipe=True, regex=False):
|
||||
if not isinstance(device, six.StringIO):
|
||||
raise TypeError('Not a StringIO object')
|
||||
|
||||
output = device.getvalue()
|
||||
if not regex:
|
||||
self.assertMultiLineEqual(output, expected)
|
||||
else:
|
||||
six.assertRegex(self, output, expected)
|
||||
if wipe:
|
||||
device.truncate(0)
|
||||
device.seek(0)
|
||||
%s: error: {message}
|
||||
""" % (self.progname, self.progname)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
|
|
@ -52,7 +35,7 @@ class TestCall(unittest.TestCase):
|
|||
handle_call(options, session, arguments)
|
||||
activate_session_mock.assert_called_with(session, options)
|
||||
session.ssl_login.assert_called_with('debug', cert='/etc/pki/cert')
|
||||
self.assert_console_output(stdout, "'%s'\n" % response)
|
||||
self.assert_console_message(stdout, "'%s'\n" % response)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
|
|
@ -75,7 +58,7 @@ class TestCall(unittest.TestCase):
|
|||
handle_call(options, session, arguments)
|
||||
activate_session_mock.assert_called_with(session, options)
|
||||
session.ssl_login.assert_called_with(cert='/etc/pki/cert')
|
||||
self.assert_console_output(stdout, "'%s'\n" % response[1])
|
||||
self.assert_console_message(stdout, "'%s'\n" % response[1])
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
|
|
@ -102,7 +85,7 @@ class TestCall(unittest.TestCase):
|
|||
session.ssl_login.assert_called_with(cert='/etc/pki/cert')
|
||||
|
||||
expect = json.dumps(response, indent=2, separators=(',', ': '))
|
||||
self.assert_console_output(stdout, '%s\n' % expect)
|
||||
self.assert_console_message(stdout, '%s\n' % expect)
|
||||
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
|
|
@ -116,13 +99,16 @@ class TestCall(unittest.TestCase):
|
|||
|
||||
# Run it and check immediate output
|
||||
# argument is empty
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_call(options, session, arguments)
|
||||
expected = self.format_error_message(
|
||||
"Please specify the name of the XML-RPC method")
|
||||
self.assert_console_output(stderr, expected)
|
||||
self.assert_system_exit(
|
||||
handle_call,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr=expected,
|
||||
activate_session=None)
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
|
||||
arguments = ['ssl_login', '--python', '--json-output']
|
||||
|
||||
|
|
@ -136,26 +122,15 @@ class TestCall(unittest.TestCase):
|
|||
self.assertRaises(SystemExit) as cm:
|
||||
handle_call(options, session, arguments)
|
||||
expected = self.format_error_message(msg)
|
||||
self.assert_console_output(stderr, expected)
|
||||
self.assert_console_message(stderr, expected)
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_call_help(
|
||||
self, activate_session_mock, stderr, stdout):
|
||||
"""Test handle_call help message full output"""
|
||||
arguments = ['--help']
|
||||
options = mock.MagicMock()
|
||||
|
||||
# Mock out the xmlrpc server
|
||||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_call(options, session, arguments)
|
||||
expected = """Usage: %s call [options] name [arg...]
|
||||
def test_handle_call_help(self):
|
||||
"""Test handle_call help message"""
|
||||
self.assert_help(
|
||||
handle_call,
|
||||
"""Usage: %s call [options] name [arg...]
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
Options:
|
||||
|
|
@ -164,13 +139,8 @@ Options:
|
|||
--kwargs=KWARGS Specify keyword arguments as a dictionary (implies
|
||||
--python)
|
||||
--json-output Use JSON syntax for output
|
||||
""" % (self.progname)
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_output(stderr, '')
|
||||
""" % (self.progname))
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -1,36 +1,21 @@
|
|||
from __future__ import absolute_import
|
||||
import mock
|
||||
import os
|
||||
import six
|
||||
import sys
|
||||
import unittest
|
||||
import koji
|
||||
|
||||
from koji_cli.commands import handle_moshimoshi
|
||||
from . import utils
|
||||
|
||||
|
||||
class TestHello(unittest.TestCase):
|
||||
class TestHello(utils.CliTestCase):
|
||||
|
||||
# Show long diffs in error output...
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.progname = os.path.basename(sys.argv[0]) or 'koji'
|
||||
self.huburl = "https://%s.local/%shub" % (self.progname, self.progname)
|
||||
|
||||
def assert_console_output(self, device, expected, wipe=True, regex=False):
|
||||
if not isinstance(device, six.StringIO):
|
||||
raise TypeError('Not a StringIO object')
|
||||
|
||||
output = device.getvalue()
|
||||
if not regex:
|
||||
self.assertMultiLineEqual(output, expected)
|
||||
else:
|
||||
six.assertRegex(self, output, expected)
|
||||
if wipe:
|
||||
device.truncate(0)
|
||||
device.seek(0)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands._printable_unicode')
|
||||
|
|
@ -52,14 +37,18 @@ class TestHello(unittest.TestCase):
|
|||
session.getLoggedInUser.return_value = None
|
||||
print_unicode_mock.return_value = "Hello"
|
||||
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_moshimoshi(options, session, ['argument'])
|
||||
expect = """Usage: %s moshimoshi [options]
|
||||
|
||||
%s: error: This command takes no arguments
|
||||
""" % (self.progname, self.progname)
|
||||
self.assert_console_output(stderr, expect)
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
|
||||
self.assert_system_exit(
|
||||
handle_moshimoshi,
|
||||
options,
|
||||
session,
|
||||
['argument'],
|
||||
stderr=expect,
|
||||
activate_session=None)
|
||||
|
||||
auth_tests = {
|
||||
koji.AUTHTYPE_NORMAL: 'Authenticated via password',
|
||||
|
|
@ -73,7 +62,7 @@ class TestHello(unittest.TestCase):
|
|||
message = "Not authenticated\n" + "Hello, anonymous user!"
|
||||
hubinfo = "You are using the hub at %s" % self.huburl
|
||||
handle_moshimoshi(options, session, [])
|
||||
self.assert_console_output(
|
||||
self.assert_console_message(
|
||||
stdout, "{0}\n\n{1}\n".format(message, hubinfo))
|
||||
|
||||
session.getLoggedInUser.return_value = user
|
||||
|
|
@ -85,35 +74,18 @@ class TestHello(unittest.TestCase):
|
|||
print_unicode_mock.return_value = "Hello"
|
||||
handle_moshimoshi(options, session, [])
|
||||
print_unicode_mock.assert_called_once()
|
||||
self.assert_console_output(
|
||||
self.assert_console_message(
|
||||
stdout, "{0}\n\n{1}\n{2}\n".format(message, hubinfo, authinfo))
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_moshimoshi_help(
|
||||
self, activate_session_mock, stderr, stdout):
|
||||
"""Test handle_moshimoshi help message full output"""
|
||||
arguments = ['--help']
|
||||
options = mock.MagicMock()
|
||||
|
||||
# Mock out the xmlrpc server
|
||||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_moshimoshi(options, session, arguments)
|
||||
expected_stdout = """Usage: %s moshimoshi [options]
|
||||
def test_handle_moshimoshi_help(self):
|
||||
self.assert_help(
|
||||
handle_moshimoshi,
|
||||
"""Usage: %s moshimoshi [options]
|
||||
|
||||
Options:
|
||||
-h, --help show this help message and exit
|
||||
""" % (self.progname)
|
||||
self.assert_console_output(stdout, expected_stdout)
|
||||
self.assert_console_output(stderr, '')
|
||||
""" % self.progname)
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -1,43 +1,27 @@
|
|||
from __future__ import absolute_import
|
||||
import mock
|
||||
import os
|
||||
import six
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
from koji_cli.commands import handle_maven_chain
|
||||
from . import utils
|
||||
|
||||
|
||||
class TestMavenChain(unittest.TestCase):
|
||||
class TestMavenChain(utils.CliTestCase):
|
||||
|
||||
# Show long diffs in error output...
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.progname = os.path.basename(sys.argv[0]) or 'koji'
|
||||
self.target = 'target'
|
||||
self.config = 'config'
|
||||
self.task_id = 101
|
||||
|
||||
def format_error_message(self, message):
|
||||
return """Usage: %s maven-chain [options] target config...
|
||||
self.error_format = """Usage: %s maven-chain [options] target config...
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
%s: error: %s
|
||||
""" % (self.progname, self.progname, message)
|
||||
|
||||
def assert_console_output(self, device, expected, wipe=True, regex=False):
|
||||
if not isinstance(device, six.StringIO):
|
||||
raise TypeError('Not a StringIO object')
|
||||
|
||||
output = device.getvalue()
|
||||
if not regex:
|
||||
self.assertMultiLineEqual(output, expected)
|
||||
else:
|
||||
six.assertRegex(self, output, expected)
|
||||
if wipe:
|
||||
device.truncate(0)
|
||||
device.seek(0)
|
||||
%s: error: {message}
|
||||
""" % (self.progname, self.progname)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
|
|
@ -75,40 +59,48 @@ class TestMavenChain(unittest.TestCase):
|
|||
}
|
||||
|
||||
# Unknonw target test
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_maven_chain(options, session, arguments)
|
||||
expected = self.format_error_message("Unknown build target: %s" %
|
||||
self.target)
|
||||
self.assert_console_output(stderr, expected)
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
expected = self.format_error_message(
|
||||
"Unknown build target: %s" % self.target)
|
||||
self.assert_system_exit(
|
||||
handle_maven_chain,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr=expected)
|
||||
|
||||
# Unknow destination tag test
|
||||
session.getBuildTarget.return_value = target_info
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_maven_chain(options, session, arguments)
|
||||
expected = self.format_error_message("Unknown destination tag: %s" %
|
||||
target_info['dest_tag_name'])
|
||||
self.assert_console_output(stderr, expected)
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
expected = self.format_error_message(
|
||||
"Unknown destination tag: %s" % target_info['dest_tag_name'])
|
||||
self.assert_system_exit(
|
||||
handle_maven_chain,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr=expected)
|
||||
|
||||
# Distination is locked and --scratch is not specified
|
||||
session.getTag.return_value = tag_info
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_maven_chain(options, session, arguments)
|
||||
expected = self.format_error_message("Destination tag %s is locked" %
|
||||
tag_info['name'])
|
||||
self.assert_console_output(stderr, expected)
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
expected = self.format_error_message(
|
||||
"Destination tag %s is locked" % tag_info['name'])
|
||||
self.assert_system_exit(
|
||||
handle_maven_chain,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr=expected)
|
||||
|
||||
# Test ValueError exception asserted in parse_maven_chain
|
||||
arguments.extend(['--skip-tag', '--scratch',
|
||||
'--force', '--background'])
|
||||
parse_maven_chain_mock.side_effect = ValueError('fake-value-error')
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_maven_chain(options, session, arguments)
|
||||
expected = self.format_error_message("fake-value-error")
|
||||
self.assert_console_output(stderr, expected)
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
self.assert_system_exit(
|
||||
handle_maven_chain,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr=expected)
|
||||
|
||||
# Background or --nowait is true
|
||||
parse_maven_chain_mock.side_effect = None
|
||||
|
|
@ -117,7 +109,7 @@ class TestMavenChain(unittest.TestCase):
|
|||
expected = "Created task: %d\n" % self.task_id
|
||||
expected += "Task info: %s/taskinfo?taskID=%s\n" % \
|
||||
(options.weburl, self.task_id)
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
# reset mocks to run full test
|
||||
activate_session_mock.reset_mock()
|
||||
|
|
@ -131,7 +123,7 @@ class TestMavenChain(unittest.TestCase):
|
|||
expected = "Created task: %d\n" % self.task_id
|
||||
expected += "Task info: %s/taskinfo?taskID=%s\n" % \
|
||||
(options.weburl, self.task_id)
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_called_with(session, options)
|
||||
|
|
@ -149,12 +141,12 @@ class TestMavenChain(unittest.TestCase):
|
|||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_maven_chain_help_compat(
|
||||
def test_handle_maven_no_argument_error(
|
||||
self,
|
||||
activate_session_mock,
|
||||
stderr,
|
||||
stdout):
|
||||
"""Test handle_maven_chain help message compact output"""
|
||||
"""Test handle_maven_chain no argument error"""
|
||||
arguments = []
|
||||
options = mock.MagicMock()
|
||||
|
||||
|
|
@ -162,33 +154,25 @@ class TestMavenChain(unittest.TestCase):
|
|||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_maven_chain(options, session, arguments)
|
||||
expected_stderr = self.format_error_message(
|
||||
expected = self.format_error_message(
|
||||
"Two arguments (a build target and a config file) are required")
|
||||
self.assert_console_output(stdout, '')
|
||||
self.assert_console_output(stderr, expected_stderr)
|
||||
self.assert_system_exit(
|
||||
handle_maven_chain,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stdout='',
|
||||
stderr=expected,
|
||||
activate_session=None)
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_maven_chain_help_full(
|
||||
self, activate_session_mock, stderr, stdout):
|
||||
def test_handle_maven_chain_help(self):
|
||||
"""Test handle_maven_chain help message full output"""
|
||||
arguments = ['--help']
|
||||
options = mock.MagicMock()
|
||||
|
||||
# Mock out the xmlrpc server
|
||||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_maven_chain(options, session, arguments)
|
||||
expected_stdout = """Usage: %s maven-chain [options] target config...
|
||||
self.assert_help(
|
||||
handle_maven_chain,
|
||||
"""Usage: %s maven-chain [options] target config...
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
Options:
|
||||
|
|
@ -199,13 +183,8 @@ Options:
|
|||
--force Force rebuilds of all packages
|
||||
--nowait Don't wait on build
|
||||
--background Run the build at a lower priority
|
||||
""" % (self.progname)
|
||||
self.assert_console_output(stdout, expected_stdout)
|
||||
self.assert_console_output(stderr, '')
|
||||
""" % (self.progname))
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -1,20 +1,19 @@
|
|||
from __future__ import absolute_import
|
||||
from __future__ import print_function
|
||||
import mock
|
||||
import os
|
||||
import six
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
from koji_cli.commands import handle_resubmit
|
||||
from . import utils
|
||||
|
||||
|
||||
class TestResubmit(unittest.TestCase):
|
||||
class TestResubmit(utils.CliTestCase):
|
||||
|
||||
# Show long diffs in error output...
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.progname = os.path.basename(sys.argv[0]) or 'koji'
|
||||
self.task_id = 101
|
||||
self.taskinfo = """Task: 101
|
||||
Type: createrepo
|
||||
|
|
@ -29,28 +28,11 @@ Log Files:
|
|||
/mnt/koji/work/tasks/2/2/mergerepos.log
|
||||
"""
|
||||
|
||||
def format_error_message(self, message):
|
||||
return """Usage: %s resubmit [options] taskID
|
||||
self.error_format = """Usage: %s resubmit [options] taskID
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
%s: error: %s
|
||||
""" % (self.progname, self.progname, message)
|
||||
|
||||
def assert_console_output(self, device, expected, wipe=True, regex=False):
|
||||
if not isinstance(device, six.StringIO):
|
||||
raise TypeError('Not a StringIO object')
|
||||
|
||||
output = device.getvalue()
|
||||
if not regex:
|
||||
self.assertMultiLineEqual(output, expected)
|
||||
else:
|
||||
six.assertRegex(self, output, expected)
|
||||
if wipe:
|
||||
device.truncate(0)
|
||||
device.seek(0)
|
||||
|
||||
def print_taskinfo(self, *args, **kwargs):
|
||||
print(self.taskinfo)
|
||||
%s: error: {message}
|
||||
""" % (self.progname, self.progname)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
|
|
@ -62,7 +44,7 @@ Log Files:
|
|||
watch_tasks_mock,
|
||||
stderr,
|
||||
stdout):
|
||||
"""Test handle_maven_chain function"""
|
||||
"""Test handle_resubmit function"""
|
||||
arguments = [str(self.task_id)]
|
||||
options = mock.MagicMock(quiet=False)
|
||||
new_task_id = self.task_id + 100
|
||||
|
|
@ -75,14 +57,14 @@ Log Files:
|
|||
|
||||
# Generate task info and nowait tests
|
||||
with mock.patch('koji_cli.commands._printTaskInfo') as p_mock:
|
||||
p_mock.side_effect = self.print_taskinfo
|
||||
p_mock.side_effect = lambda *args, **kwargs: print(self.taskinfo)
|
||||
handle_resubmit(options, session, arguments + ['--nowait'])
|
||||
activate_session_mock.assert_called_with(session, options)
|
||||
expected = "Resubmitting the following task:" + "\n"
|
||||
expected += self.taskinfo + "\n"
|
||||
expected += "Resubmitted task %s as new task %s" % \
|
||||
(self.task_id, new_task_id) + "\n"
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
session.logout.reset_mock()
|
||||
session.resubmitTask.reset_mock()
|
||||
|
|
@ -102,14 +84,14 @@ Log Files:
|
|||
session.logout.assert_called_once()
|
||||
session.resubmitTask.assert_called_with(self.task_id)
|
||||
session.resubmitTask.assert_called_once()
|
||||
self.assert_console_output(stdout, '')
|
||||
self.assert_console_message(stdout, '')
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_resubmit_help_compat(
|
||||
def test_handle_resubmit_argument_error(
|
||||
self, activate_session_mock, stderr, stdout):
|
||||
"""Test handle_resubmit help message compact output"""
|
||||
"""Test handle_resubmit argument error"""
|
||||
arguments = []
|
||||
options = mock.MagicMock()
|
||||
|
||||
|
|
@ -117,33 +99,28 @@ Log Files:
|
|||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_resubmit(options, session, arguments)
|
||||
expected_stderr = self.format_error_message(
|
||||
expected = self.format_error_message(
|
||||
"Please specify a single task ID")
|
||||
self.assert_console_output(stdout, '')
|
||||
self.assert_console_output(stderr, expected_stderr)
|
||||
|
||||
self.assert_system_exit(
|
||||
handle_resubmit,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr=expected,
|
||||
activate_session=None)
|
||||
|
||||
# Check there is no message on stdout
|
||||
self.assert_console_message(stdout, '')
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_resubmit_help_full(
|
||||
self, activate_session_mock, stderr, stdout):
|
||||
"""Test handle_resubmit help message full output"""
|
||||
arguments = ['--help']
|
||||
options = mock.MagicMock()
|
||||
|
||||
# Mock out the xmlrpc server
|
||||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_resubmit(options, session, arguments)
|
||||
expected_stdout = """Usage: %s resubmit [options] taskID
|
||||
def test_handle_resubmit_help(self):
|
||||
"""Test handle_resubmit help message output"""
|
||||
self.assert_help(
|
||||
handle_resubmit,
|
||||
"""Usage: %s resubmit [options] taskID
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
Options:
|
||||
|
|
@ -151,13 +128,7 @@ Options:
|
|||
--nowait Don't wait on task
|
||||
--nowatch An alias for --nowait
|
||||
--quiet Do not print the task information
|
||||
""" % (self.progname)
|
||||
self.assert_console_output(stdout, expected_stdout)
|
||||
self.assert_console_output(stderr, '')
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 0)
|
||||
""" % self.progname)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -1,37 +1,28 @@
|
|||
from __future__ import absolute_import
|
||||
import mock
|
||||
import os
|
||||
import six
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
from koji_cli.commands import handle_wrapper_rpm
|
||||
from . import utils
|
||||
|
||||
|
||||
class TestWrapperRpm(unittest.TestCase):
|
||||
class TestWrapperRpm(utils.CliTestCase):
|
||||
|
||||
# Show long diffs in error output...
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.progname = os.path.basename(sys.argv[0]) or 'koji'
|
||||
self.target = 'target'
|
||||
self.build = '1'
|
||||
self.scm_url = 'git+https://github.com/project/test#12345'
|
||||
self.task_id = 1
|
||||
|
||||
def assert_console_output(self, device, expected, wipe=True, regex=False):
|
||||
if not isinstance(device, six.StringIO):
|
||||
raise TypeError('Not a StringIO object')
|
||||
self.error_format = """Usage: %s wrapper-rpm [options] target build-id|n-v-r URL
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
output = device.getvalue()
|
||||
if not regex:
|
||||
self.assertMultiLineEqual(output, expected)
|
||||
else:
|
||||
six.assertRegex(self, output, expected)
|
||||
if wipe:
|
||||
device.truncate(0)
|
||||
device.seek(0)
|
||||
%s: error: {message}
|
||||
""" % (self.progname, self.progname)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
|
|
@ -63,24 +54,24 @@ class TestWrapperRpm(unittest.TestCase):
|
|||
|
||||
arguments.extend(['--create-build', '--skip-tag', '--scratch'])
|
||||
self.assertEqual(None, handle_wrapper_rpm(options, session, arguments))
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
# Background off but --nowait is specified
|
||||
running_in_bg_mock.return_value = False
|
||||
|
||||
args = arguments + ['--nowait']
|
||||
self.assertEqual(None, handle_wrapper_rpm(options, session, args))
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
# proirity test
|
||||
args = arguments + ['--nowait', '--background']
|
||||
self.assertEqual(None, handle_wrapper_rpm(options, session, args))
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
# watch task case
|
||||
watch_tasks_mock.return_value = True
|
||||
self.assertTrue(handle_wrapper_rpm(options, session, arguments))
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_called_with(session, options)
|
||||
|
|
@ -134,47 +125,56 @@ class TestWrapperRpm(unittest.TestCase):
|
|||
self.scm_url,
|
||||
'--ini=/etc/koji.ini'
|
||||
]
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_wrapper_rpm(options, session, arguments)
|
||||
expected_stderr = """Usage: %s wrapper-rpm [options] target build-id|n-v-r URL
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
%s: error: Exactly one argument (a build target) is required
|
||||
""" % (self.progname, self.progname)
|
||||
self.assert_console_output(stdout, '')
|
||||
self.assert_console_output(stderr, expected_stderr)
|
||||
expected = self.format_error_message(
|
||||
"Exactly one argument (a build target) is required")
|
||||
self.assert_system_exit(
|
||||
handle_wrapper_rpm,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stdout='',
|
||||
stderr=expected,
|
||||
activate_session=None)
|
||||
|
||||
# If koji.util.parse_maven_param has troubles
|
||||
# ValueError exception
|
||||
arguments = [self.target, '--ini=/etc/koji.ini']
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
parse_maven_mock.side_effect = ValueError('fake-value-error')
|
||||
handle_wrapper_rpm(options, session, arguments)
|
||||
self.assert_console_output(
|
||||
stderr, '.*error: fake-value-error', regex=True)
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
parse_maven_mock.side_effect = ValueError('fake-value-error')
|
||||
self.assert_system_exit(
|
||||
handle_wrapper_rpm,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr={'message': '.*error: fake-value-error',
|
||||
'regex': True}
|
||||
)
|
||||
|
||||
parse_maven_mock.side_effect = None
|
||||
|
||||
# type != wrapper case
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
bad_param = {'pkg1': maven_param['pkg1'].copy()}
|
||||
bad_param['pkg1']['type'] = 'undefined'
|
||||
parse_maven_mock.return_value = bad_param
|
||||
handle_wrapper_rpm(options, session, arguments)
|
||||
self.assert_console_output(
|
||||
stderr,
|
||||
'Section .* does not contain a wrapper-rpm config',
|
||||
regex=True)
|
||||
bad_param = {'pkg1': maven_param['pkg1'].copy()}
|
||||
bad_param['pkg1']['type'] = 'undefined'
|
||||
parse_maven_mock.return_value = bad_param
|
||||
self.assert_system_exit(
|
||||
handle_wrapper_rpm,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr={'message': 'Section .* does not contain a wrapper-rpm config',
|
||||
'regex': True}
|
||||
)
|
||||
|
||||
# Lastest build does not exist case
|
||||
parse_maven_mock.return_value = maven_param
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_wrapper_rpm(options, session, arguments)
|
||||
self.assert_console_output(
|
||||
stderr,
|
||||
'.*error: No build of .* in %s' % target_info['dest_tag_name'],
|
||||
regex=True)
|
||||
self.assert_system_exit(
|
||||
handle_wrapper_rpm,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stderr={'message': '.*error: No build of .* in %s' %
|
||||
target_info['dest_tag_name'],
|
||||
'regex': True}
|
||||
)
|
||||
|
||||
# Check everything should work fine
|
||||
session.getLatestBuilds.return_value = [{'nvr': 'r1'}]
|
||||
|
|
@ -182,7 +182,7 @@ class TestWrapperRpm(unittest.TestCase):
|
|||
expected = "Created task: %d\n" % self.task_id
|
||||
expected += "Task info: %s/taskinfo?taskID=%s\n" % \
|
||||
(options.weburl, self.task_id)
|
||||
self.assert_console_output(stdout, expected)
|
||||
self.assert_console_message(stdout, expected)
|
||||
|
||||
activate_session_mock.assert_called_with(session, options)
|
||||
watch_tasks_mock.assert_called_with(
|
||||
|
|
@ -192,8 +192,8 @@ class TestWrapperRpm(unittest.TestCase):
|
|||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def test_handle_wrapper_rpm_help(self, activate_session_mock,
|
||||
stderr, stdout):
|
||||
def test_handle_wrapper_rpm_argument_error(
|
||||
self, activate_session_mock, stderr, stdout):
|
||||
"""Test handle_wrapper_rpm help message output"""
|
||||
arguments = []
|
||||
options = mock.MagicMock()
|
||||
|
|
@ -202,19 +202,39 @@ class TestWrapperRpm(unittest.TestCase):
|
|||
session = mock.MagicMock()
|
||||
|
||||
# Run it and check immediate output
|
||||
with self.assertRaises(SystemExit) as cm:
|
||||
handle_wrapper_rpm(options, session, arguments)
|
||||
expected_stderr = """Usage: %s wrapper-rpm [options] target build-id|n-v-r URL
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
%s: error: You must provide a build target, a build ID or NVR, and a SCM URL to a specfile fragment
|
||||
""" % (self.progname, self.progname)
|
||||
self.assert_console_output(stdout, '')
|
||||
self.assert_console_output(stderr, expected_stderr)
|
||||
expected = self.format_error_message(
|
||||
"You must provide a build target, a build ID or NVR, and a SCM URL to a specfile fragment")
|
||||
self.assert_system_exit(
|
||||
handle_wrapper_rpm,
|
||||
options,
|
||||
session,
|
||||
arguments,
|
||||
stdout='',
|
||||
stderr=expected,
|
||||
activate_session=None)
|
||||
|
||||
# Finally, assert that things were called as we expected.
|
||||
activate_session_mock.assert_not_called()
|
||||
self.assertEqual(cm.exception.code, 2)
|
||||
|
||||
def test_handle_wrapper_rpm_help(self):
|
||||
"""Test handle_wrapper_rpm help message output"""
|
||||
self.assert_help(
|
||||
handle_wrapper_rpm,
|
||||
"""Usage: %s wrapper-rpm [options] target build-id|n-v-r URL
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
||||
Options:
|
||||
-h, --help show this help message and exit
|
||||
--create-build Create a new build to contain wrapper rpms
|
||||
--ini=CONFIG Pass build parameters via a .ini file
|
||||
-s SECTION, --section=SECTION
|
||||
Get build parameters from this section of the .ini
|
||||
--skip-tag If creating a new build, don't tag it
|
||||
--scratch Perform a scratch build
|
||||
--nowait Don't wait on build
|
||||
--background Run the build at a lower priority
|
||||
""" % self.progname)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
|
|||
198
tests/test_cli/utils.py
Normal file
198
tests/test_cli/utils.py
Normal file
|
|
@ -0,0 +1,198 @@
|
|||
import os
|
||||
import sys
|
||||
import six
|
||||
import mock
|
||||
import unittest
|
||||
|
||||
|
||||
"""
|
||||
Classes
|
||||
"""
|
||||
|
||||
|
||||
#
|
||||
# Dummy Mock class
|
||||
#
|
||||
class _dummy_(object):
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *arg, **kwargs):
|
||||
pass
|
||||
|
||||
def assert_called_once(self, *arg, **kwargs):
|
||||
pass
|
||||
|
||||
def assert_called_once_with(self, *arg, **kwargs):
|
||||
pass
|
||||
|
||||
|
||||
#
|
||||
# CLI TestCase
|
||||
#
|
||||
class CliTestCase(unittest.TestCase):
|
||||
|
||||
# public attribute
|
||||
progname = os.path.basename(sys.argv[0]) or 'koji'
|
||||
error_format = None
|
||||
|
||||
#
|
||||
# private methods
|
||||
#
|
||||
def __assert_callable(self, obj):
|
||||
if not callable(obj):
|
||||
raise ValueError('%s is not callable' %
|
||||
getattr(obj, "__name__", 'function'))
|
||||
|
||||
#
|
||||
# public methods
|
||||
#
|
||||
def format_error_message(self, error_message, progname=None):
|
||||
return self.error_format.format(message=error_message) \
|
||||
if self.error_format else error_message
|
||||
|
||||
def assert_function_wrapper(self, callableObj, *args, **kwargs):
|
||||
"""wrapper func with anonymous funtion without argument"""
|
||||
self.__assert_callable(callableObj)
|
||||
return lambda: callableObj(*args, **kwargs)
|
||||
|
||||
def assert_console_message(
|
||||
self, device, message, wipe=True, regex=False):
|
||||
|
||||
# don't care condition
|
||||
if message is None:
|
||||
return
|
||||
|
||||
output = device.getvalue()
|
||||
if not regex:
|
||||
self.assertMultiLineEqual(output, message)
|
||||
else:
|
||||
six.assertRegex(self, output, message)
|
||||
|
||||
if wipe:
|
||||
device.seek(0)
|
||||
device.truncate(0)
|
||||
|
||||
def assert_system_exit(self, callableObj, *args, **kwargs):
|
||||
"""Test if SystemExit exception is issued
|
||||
|
||||
Arguments:
|
||||
callableObj: the test function
|
||||
*args: Vaiable length arguments that will be passed to callableObj
|
||||
**kwargs: keyword arguments (see below)
|
||||
|
||||
Keyword arguments: (reseverd for assert_system_exit)
|
||||
activate_session (string):
|
||||
Mock koji_cli.commands.activate_session and test if it is
|
||||
called.
|
||||
Default is on, use None to stop mocking.
|
||||
|
||||
stdout:
|
||||
stderr: Arguments for messages comparison on stdout/stderr. These
|
||||
arguments allows different data types.
|
||||
|
||||
``None`` type will skip message comparison.
|
||||
``string`` type will do multiple line comparison.
|
||||
``dict`` type is an advanced type to allow regular expression
|
||||
format. the format is::
|
||||
{
|
||||
'message': 'message string or regular expression',
|
||||
'wipe': 'wipe the output device, default is True',
|
||||
'regex': 'True if message format is regular expression'
|
||||
}
|
||||
|
||||
assert_func:
|
||||
Callable object with no arguments for customized tests.
|
||||
|
||||
exit_code:
|
||||
The exit code when SystemExit is raised
|
||||
|
||||
Important! all the other keyword arguments that are not listed
|
||||
above will be passed to callableObj
|
||||
"""
|
||||
|
||||
# check callableObj callable
|
||||
self.__assert_callable(callableObj)
|
||||
|
||||
# these arguments are reseverd and used in assert_system_exit
|
||||
reserved = [
|
||||
'activate_session', 'stdout', 'stderr',
|
||||
'assert_func', 'exit_code'
|
||||
]
|
||||
|
||||
activate = kwargs.get(
|
||||
'activate_session', 'koji_cli.commands.activate_session')
|
||||
|
||||
# stdout/stderr message comparison, None means don't care
|
||||
# message/error allows many different data types, None, string and dict
|
||||
message = {}
|
||||
for key in ['stdout', 'stderr']:
|
||||
data = kwargs.get(key, None)
|
||||
message[key] = {'message': None, 'wipe': True, 'regex': False}
|
||||
|
||||
if data is None or isinstance(data, six.string_types):
|
||||
message[key]['message'] = data
|
||||
|
||||
elif isinstance(data, dict):
|
||||
message[key] = {
|
||||
'message': data.get('message', None),
|
||||
'wipe': data.get('wipe', True),
|
||||
'regex': data.get('regex', False),
|
||||
}
|
||||
|
||||
else:
|
||||
raise ValueError('Invalid data type for %s' % key)
|
||||
|
||||
assert_function = kwargs.get(
|
||||
'assert_func', lambda *args, **kwargs: True)
|
||||
|
||||
exit_code = kwargs.get('exit_code', 2)
|
||||
|
||||
# args for testee
|
||||
test_args = args
|
||||
|
||||
# kwargs for testee, excludes those that are used in assert_system_exit
|
||||
test_kwargs = dict((k, v) for k, v in kwargs.items()
|
||||
if k not in reserved)
|
||||
|
||||
# check activate_session must be type of None or string
|
||||
if activate and not isinstance(activate, six.string_types):
|
||||
raise ValueError('activate_session is not a string')
|
||||
|
||||
session_patch = mock.patch(activate) if activate else _dummy_()
|
||||
stdout_patch = mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
stderr_patch = mock.patch('sys.stderr', new_callable=six.StringIO)
|
||||
|
||||
with session_patch as session, \
|
||||
stdout_patch as stdout, \
|
||||
stderr_patch as stderr, \
|
||||
self.assertRaises(SystemExit) as cm:
|
||||
callableObj(*test_args, **test_kwargs)
|
||||
session.assert_called_once()
|
||||
self.assert_console_message(stdout, **message['stdout'])
|
||||
self.assert_console_message(stderr, **message['stderr'])
|
||||
assert_function()
|
||||
self.assertEqual(cm.exception.code, exit_code)
|
||||
|
||||
@mock.patch('koji_cli.commands.activate_session')
|
||||
def assert_help(self, callableObj, message, activate_session_mock):
|
||||
self.assert_system_exit(
|
||||
callableObj,
|
||||
mock.MagicMock(),
|
||||
mock.MagicMock(),
|
||||
['--help'],
|
||||
stdout=message,
|
||||
stderr='',
|
||||
activate_session=None,
|
||||
exit_code=0)
|
||||
activate_session_mock.assert_not_called()
|
||||
|
||||
|
||||
#
|
||||
# Open /dev/stdout and write message directly.
|
||||
# This function is useful when sys.stdout is redirected
|
||||
#
|
||||
def debug_print(message):
|
||||
with open('/dev/stdout', 'w') as stdout:
|
||||
stdout.write(message)
|
||||
stdout.write('\n')
|
||||
Loading…
Add table
Add a link
Reference in a new issue