Add unit tests notifications, list-channels

This commit is contained in:
Jana Cupova 2021-09-10 21:20:24 +02:00 committed by Tomas Kopecek
parent d9b2f88b0e
commit 3b8b231a07
6 changed files with 416 additions and 10 deletions

View file

@ -1,8 +1,9 @@
from __future__ import absolute_import
import koji
import mock
from six.moves import StringIO
import koji
from koji_cli.commands import handle_block_notification
from . import utils
@ -39,3 +40,104 @@ class TestBlockNotification(utils.CliTestCase):
handle_block_notification(self.options, self.session, ['--package', pkg])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_block_notification_with_args(self, stderr, activate_session):
expected = "Usage: %s block-notification [options]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: This command takes no arguments\n" \
% (self.progname, self.progname)
with self.assertRaises(SystemExit) as ex:
handle_block_notification(self.options, self.session, ['1234'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_not_called()
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_block_notification_with_user_only(self, stderr, activate_session):
expected = "Usage: %s block-notification [options]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: One of --tag, --package or --all must be specified.\n" \
% (self.progname, self.progname)
with self.assertRaises(SystemExit) as ex:
handle_block_notification(self.options, self.session, ['--user', 'testuser'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_not_called()
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_block_notification_with_user_non_admin_tag(self, stderr, activate_session):
expected = "Usage: %s block-notification [options]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: --user requires admin permission\n" \
% (self.progname, self.progname)
self.session.hasPerm.return_value = False
with self.assertRaises(SystemExit) as ex:
handle_block_notification(self.options, self.session, ['--user', 'testuser',
'--tag', 'tagtest'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_called_once_with(self.session, self.options)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_handle_block_notification_with_user_pkg(self, stdout, activate_session):
expected = ""
self.session.hasPerm.return_value = True
self.session.getUser.return_value = {'id': 2, 'krb_principals': [],
'name': 'testuser', 'status': 0, 'usertype': 0}
self.session.getBuildNotificationBlocks.return_value = []
self.session.createNotificationBlock.return_value = None
self.session.getPackageID.return_value = 1
handle_block_notification(self.options, self.session, ['--user', 'testuser',
'--package', 'pkgtest'])
self.assert_console_message(stdout, expected)
activate_session.assert_called_once_with(self.session, self.options)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_block_notification_without_user_not_logged(self, stderr, activate_session):
expected = "Usage: %s block-notification [options]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: Please login with authentication or specify --user\n" \
% (self.progname, self.progname)
self.session.getLoggedInUser.return_value = None
with self.assertRaises(SystemExit) as ex:
handle_block_notification(self.options, self.session, ['--tag', 'tagtest'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_called_once_with(self.session, self.options)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_block_notification_existing_block(self, stderr, activate_session):
expected = "Usage: %s block-notification [options]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: Notification already exists.\n" \
% (self.progname, self.progname)
self.session.hasPerm.return_value = True
self.session.getUser.return_value = {'id': 2, 'krb_principals': [],
'name': 'testuser', 'status': 0, 'usertype': 0}
self.session.getBuildNotificationBlocks.return_value = [{'package_id': 1,
'tag_id': 2}]
self.session.createNotificationBlock.return_value = None
self.session.getPackageID.return_value = 1
self.session.getTagID.return_value = 2
with self.assertRaises(SystemExit) as ex:
handle_block_notification(self.options, self.session, ['--user', 'testuser',
'--package', 'pkgtest',
'--tag', 'testtag'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_called_once_with(self.session, self.options)

View file

@ -24,6 +24,13 @@ class TestListChannels(utils.CliTestCase):
{'id': 2, 'name': 'test', 'enabled': False, 'comment': 'test-comment-2',
'description': 'test-description-2'},
]
self.list_channels_without_enabled = [
{'id': 1, 'name': 'default', 'comment': 'test-comment-1',
'description': 'test-description-1'},
{'id': 2, 'name': 'test', 'comment': 'test-comment-2',
'description': 'test-description-2'},
]
self.list_hosts_mc = [
[[
{'enabled': True, 'ready': True, 'capacity': 2.0, 'task_load': 1.34},
@ -39,19 +46,20 @@ class TestListChannels(utils.CliTestCase):
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels(self, ensure_connection_mock, stdout):
def test_list_channels_not_quiet(self, ensure_connection_mock, stdout):
self.session.listChannels.return_value = self.list_channels
self.session.multiCall.return_value = self.list_hosts_mc
args = []
self.options.quiet = False
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = """\
default 3 1 0 1 6 22%
test [disabled] 2 2 1 1 6 28%
"""
expected = "Channel Enabled Ready Disbld Load Cap Perc \n" \
"default 3 1 0 1 6 22%\n" \
"test [disabled] 2 2 1 1 6 28%\n"
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@ -89,6 +97,148 @@ test [disabled] 2 2 1 1 6 28%
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_with_comment_desc_not_quiet(self, ensure_connection_mock, stdout):
self.session.listChannels.return_value = self.list_channels
self.session.multiCall.return_value = self.list_hosts_mc
args = ['--description', '--comment']
self.options.quiet = False
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = "Channel Enabled Ready Disbld Load Cap Perc " \
"Description " \
"Comment \n" \
"default 3 1 0 1 6 22% " \
"test-description-1 " \
"test-comment-1 \n" \
"test [disabled] 2 2 1 1 6 28% " \
"test-description-2 " \
"test-comment-2 \n"
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_with_enabled_hub_without_enabled(self, ensure_connection_mock, stdout):
self.session.listChannels.side_effect = [koji.ParameterError,
self.list_channels_without_enabled]
self.session.multiCall.return_value = self.list_hosts_mc
args = ['--enabled']
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = 'default 3 1 0 1 6 22%\n' \
'test 2 2 1 1 6 28%\n'
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_with_enabled(self, ensure_connection_mock, stdout):
list_channel_enabled = [
{'id': 1, 'name': 'default', 'enabled': True, 'comment': 'test-comment-1',
'description': 'test-description-1'},
]
list_hosts_mc = [
[[
{'enabled': True, 'ready': True, 'capacity': 2.0, 'task_load': 1.34},
{'enabled': True, 'ready': False, 'capacity': 2.0, 'task_load': 0.0},
{'enabled': True, 'ready': False, 'capacity': 2.0, 'task_load': 0.0},
]]
]
self.session.listChannels.return_value = list_channel_enabled
self.session.multiCall.return_value = list_hosts_mc
args = ['--enabled']
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = 'default 3 1 0 1 6 22%\n'
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_with_disabled(self, ensure_connection_mock, stdout):
list_channel_disabled = [
{'id': 2, 'name': 'test', 'enabled': False, 'comment': 'test-comment-2',
'description': 'test-description-2'},
]
list_hosts_mc = [
[[
{'enabled': True, 'ready': True, 'capacity': 2.0, 'task_load': 1.34},
{'enabled': False, 'ready': True, 'capacity': 2.0, 'task_load': 0.34},
{'enabled': True, 'ready': False, 'capacity': 2.0, 'task_load': 0.0},
]]
]
self.session.listChannels.return_value = list_channel_disabled
self.session.multiCall.return_value = list_hosts_mc
args = ['--disabled']
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = 'test [disabled] 2 2 1 1 6 28%\n'
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_with_empty_channels(self, ensure_connection_mock, stdout):
list_channels = []
list_hosts_mc = [[[]]]
self.session.listChannels.return_value = list_channels
self.session.multiCall.return_value = list_hosts_mc
args = []
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = ''
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_simple_without_quiet(self, ensure_connection_mock, stdout):
self.session.listChannels.return_value = self.list_channels
self.session.multiCall.return_value = self.list_hosts_mc
self.options.quiet = False
args = ['--simple']
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = """Channel
default
test [disabled]
"""
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
@mock.patch('koji_cli.commands.ensure_connection')
def test_list_channels_simple_with_quiet(self, ensure_connection_mock, stdout):
self.session.listChannels.return_value = self.list_channels
self.session.multiCall.return_value = self.list_hosts_mc
args = ['--simple', '--quiet']
anon_handle_list_channels(self.options, self.session, args)
actual = stdout.getvalue()
print(actual)
expected = """default
test [disabled]
"""
self.assertMultiLineEqual(actual, expected)
ensure_connection_mock.assert_called_once_with(self.session, self.options)
def test_list_channels_help(self):
self.assert_help(
anon_handle_list_channels,

View file

@ -22,6 +22,23 @@ class TestAddHost(utils.CliTestCase):
self.session.deleteNotification.assert_has_calls([mock.call(1), mock.call(3),
mock.call(5)])
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_handle_remove_notification_not_quiet(self, stdout, activate_session_mock):
self.options.quiet = False
expected = "Notification 1 successfully removed.\n" \
"Notification 3 successfully removed.\n" \
"Notification 5 successfully removed.\n"
handle_remove_notification(self.options, self.session, ['1', '3', '5'])
self.session.deleteNotification.assert_has_calls([mock.call(1), mock.call(3),
mock.call(5)])
activate_session_mock.assert_called_once_with(self.session, self.options)
actual = stdout.getvalue()
print(actual)
self.assertMultiLineEqual(actual, expected)
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_remove_notification_bogus(self, stderr):
expected = """Usage: %s remove-notification [options] <notification_id> [<notification_id> ...]

View file

@ -17,7 +17,7 @@ class TestRemoveSig(utils.CliTestCase):
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
def test_delete_sig_help(self):
def test_remove_sig_help(self):
self.assert_help(
handle_remove_sig,
"""Usage: %s remove-sig [options] <rpm-id/n-v-r.a/rpminfo>
@ -33,7 +33,7 @@ Options:
""" % self.progname)
@mock.patch('sys.stderr', new_callable=StringIO)
def test_delete_sig_without_option(self, stderr):
def test_remove_sig_without_option(self, stderr):
expected = "Usage: %s remove-sig [options] <rpm-id/n-v-r.a/rpminfo>\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: Please specify an RPM\n" % (self.progname, self.progname)
@ -43,7 +43,7 @@ Options:
self.assert_console_message(stderr, expected)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_delete_sig_non_exist_rpm(self, stdout):
def test_remove_sig_non_exist_rpm(self, stdout):
rpm = '1234'
expected = "No such rpm in system: %s\n" % rpm
self.session.deleteRPMSig.side_effect = koji.GenericError('No such rpm: DATA')
@ -58,8 +58,36 @@ Options:
activate_session=None)
self.session.deleteRPMSig.assert_called_with('1234', sigkey=None, all_sigs=True)
def test_delete_sig_valid(self):
def test_remove_sig_valid(self):
rpm = '1'
self.session.deleteRPMSig.return_value = None
handle_remove_sig(self.options, self.session, [rpm, '--sigkey', 'testkey'])
self.session.deleteRPMSig.assert_called_with('1', sigkey='testkey', all_sigs=False)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_sig_without_all_and_sigkey(self, stdout):
rpm = '1234'
expected = "Either --sigkey or --all options must be given\n"
self.assert_system_exit(
handle_remove_sig,
self.options,
self.session,
[rpm],
stderr=self.format_error_message(expected),
exit_code=1,
activate_session=None)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_sig_with_all_and_sigkey(self, stdout):
rpm = '1234'
expected = "Conflicting options specified\n"
self.assert_system_exit(
handle_remove_sig,
self.options,
self.session,
[rpm, '--all', '--sigkey', 'testkey'],
stderr=self.format_error_message(expected),
exit_code=1,
activate_session=None)

View file

@ -0,0 +1,71 @@
from __future__ import absolute_import
import mock
from six.moves import StringIO
import koji
from koji_cli.commands import handle_unblock_notification
from . import utils
class TestUnblockNotification(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.options.debug = False
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_unblock_notification_without_args(self, stderr, activate_session):
expected = "Usage: %s unblock-notification [options] <notification_id> " \
"[<notification_id> ...]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: At least one notification block id has to be specified\n" \
% (self.progname, self.progname)
with self.assertRaises(SystemExit) as ex:
handle_unblock_notification(self.options, self.session, [])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_called_once_with(self.session, self.options)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_unblock_notification_not_all_integers(self, stderr, activate_session):
expected = "Usage: %s unblock-notification [options] <notification_id> " \
"[<notification_id> ...]\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: All notification block ids has to be integers\n" \
% (self.progname, self.progname)
with self.assertRaises(SystemExit) as ex:
handle_unblock_notification(self.options, self.session, ['0', 'abcd'])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
activate_session.assert_called_once_with(self.session, self.options)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_handle_unblock_notification(self, stdout, activate_session):
self.options.quiet = False
expected = "Notification block 0 successfully removed.\n" \
"Notification block 1234 successfully removed.\n"
self.session.deleteNotificationBlock.side_effect = [None, None]
handle_unblock_notification(self.options, self.session, ['0', '1234'])
activate_session.assert_called_once_with(self.session, self.options)
actual = stdout.getvalue()
print(actual)
self.assertMultiLineEqual(actual, expected)
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_handle_unblock_notification_quiet(self, stdout, activate_session):
expected = ""
self.options.quiet = True
self.session.deleteNotificationBlock.side_effect = [None, None]
handle_unblock_notification(self.options, self.session, ['0', '1234'])
activate_session.assert_called_once_with(self.session, self.options)
actual = stdout.getvalue()
print(actual)
self.assertMultiLineEqual(actual, expected)

View file

@ -0,0 +1,38 @@
from __future__ import absolute_import
import mock
from six.moves import StringIO
import koji
from koji_cli.commands import handle_version
from . import utils
class TestVersion(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.options.debug = False
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
@mock.patch('sys.stdout', new_callable=StringIO)
def test_version_valid(self, stdout):
expected = """Client: 1.26.0
Hub: 1.26.0
"""
self.session.getKojiVersion.return_value = '1.26.0'
rv = handle_version(self.options, self.session, [])
self.assertEqual(rv, None)
self.assert_console_message(stdout, expected)
self.session.getKojiVersion.assert_called_once_with()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_version_invalid(self, stdout):
expected = """Client: 1.26.0
Hub: Can't determine (older than 1.23)
"""
self.session.getKojiVersion.side_effect = koji.GenericError()
rv = handle_version(self.options, self.session, [])
self.assertEqual(rv, None)
self.assert_console_message(stdout, expected)
self.session.getKojiVersion.assert_called_once_with()