particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add CLI related to channels + add comments to channels

Browse source 
Fixes: https://pagure.io/koji/issue/1711
Fixes: https://pagure.io/koji/issue/1849
This commit is contained in:
Jana Cupova 2021-06-09 13:16:17 +02:00 • committed by Tomas Kopecek
parent ed2b0ad19b
commit aec9fba121
18 changed files with 599 additions and 169 deletions
Download patch file Download diff file Expand all files Collapse all files

28
cli/koji_cli/commands.py
View file

@ -298,6 +298,19 @@ def handle_remove_host_from_channel(goptions, session, args):
session.removeHostFromChannel(host, channel)
def handle_add_channel(goptions, session, args):
"[admin] Add a channel"
usage = _("usage: %prog add-channel [options] <channel_name>")
parser = OptionParser(usage=get_usage_str(usage))
parser.add_option("--description", help=_("Description of channel"))
(options, args) = parser.parse_args(args)
if len(args) != 1:
parser.error(_("Please specify one channel name"))
activate_session(session, goptions)
channel_id = session.addChannel(args[0], description=options.description)
print("%s added: id %d" % (args[0], channel_id))
def handle_remove_channel(goptions, session, args):
"[admin] Remove a channel entirely"
usage = _("usage: %prog remove-channel [options] <channel>")
@ -318,6 +331,8 @@ def handle_rename_channel(goptions, session, args):
usage = _("usage: %prog rename-channel [options] <old-name> <new-name>")
parser = OptionParser(usage=get_usage_str(usage))
(options, args) = parser.parse_args(args)
print("rename-channel is deprecated and will be removed in 1.28, this call is replaced by "
"edit-channel")
if len(args) != 2:
parser.error(_("Incorrect number of arguments"))
activate_session(session, goptions)
@ -327,6 +342,19 @@ def handle_rename_channel(goptions, session, args):
session.renameChannel(args[0], args[1])
def handle_edit_channel(goptions, session, args):
"[admin] Edit a channel"
usage = _("usage: %prog edit-channel [options] <old-name>")
parser = OptionParser(usage=get_usage_str(usage))
parser.add_option("--name", help=_("New channel name"))
parser.add_option("--description", help=_("Description of channel"))
(options, args) = parser.parse_args(args)
if len(args) != 1:
parser.error(_("Incorrect number of arguments"))
activate_session(session, goptions)
session.editChannel(args[0], name=options.name, description=options.description)
def handle_add_pkg(goptions, session, args):
"[admin] Add a package to the listing for tag"
usage = _("usage: %prog add-pkg [options] --owner <owner> <tag> <package> [<package> ...]")

9
docs/schema-upgrade-1.25-1.26.sql Normal file
View file

@ -0,0 +1,9 @@
-- upgrade script to migrate the Koji database schema
-- from version 1.25 to 1.26
BEGIN;
ALTER TABLE channels ADD COLUMN description TEXT;
COMMIT;

3
docs/schema.sql
View file

@ -135,7 +135,8 @@ CREATE INDEX sessions_expired ON sessions(expired);
-- listening to.
CREATE TABLE channels (
id SERIAL NOT NULL PRIMARY KEY,
name VARCHAR(128) UNIQUE NOT NULL
name VARCHAR(128) UNIQUE NOT NULL,
description TEXT
) WITHOUT OIDS;
-- create default channel

64
hub/kojihub.py
View file

@ -2293,6 +2293,7 @@ def remove_host_from_channel(hostname, channel_name):
def rename_channel(old, new):
"""Rename a channel"""
logger.warning("renameChannel call is deprecated and will be removed in 1.28")
context.session.assertPerm('admin')
if not isinstance(new, str):
raise koji.GenericError("new channel name must be a string")
@ -2305,8 +2306,40 @@ def rename_channel(old, new):
update.execute()
def edit_channel(channelInfo, name=None, description=None):
"""Edit information for an existing channel.
:param str/int channelInfo: channel name or ID
:param str name: new channel name
:param str description: description of channel
"""
context.session.assertPerm('admin')
channel = get_channel(channelInfo, strict=True)
if name:
if not isinstance(name, str):
raise koji.GenericError("new channel name must be a string")
dup_check = get_channel(name, strict=False)
if dup_check:
raise koji.GenericError("channel %(name)s already exists (id=%(id)i)" % dup_check)
update = UpdateProcessor('channels',
values={'channelID': channel['id']},
clauses=['id = %(channelID)i'])
if name:
update.set(name=name)
if description:
update.set(description=description)
update.execute()
return None
def remove_channel(channel_name, force=False):
"""Remove a channel
"""Remove a channel.
:param str channel_name: channel name
:param bool force: remove channel which has hosts
Channel must have no hosts, unless force is set to True
If a channel has associated tasks, it cannot be removed
@ -2334,6 +2367,26 @@ def remove_channel(channel_name, force=False):
_dml(delete, locals())
def add_channel(channel_name, description=None):
"""Add a channel.
:param str channel_name: channel name
:param str description: description of channel
"""
context.session.assertPerm('admin')
if not isinstance(channel_name, str):
raise koji.GenericError("Channel name must be a string")
dup_check = get_channel(channel_name, strict=False)
if dup_check:
raise koji.GenericError("channel %(name)s already exists (id=%(id)i)" % dup_check)
table = 'channels'
channel_id = _singleValue("SELECT nextval('%s_id_seq')" % table, strict=True)
insert = InsertProcessor(table)
insert.set(id=channel_id, name=channel_name, description=description)
insert.execute()
return channel_id
def get_ready_hosts():
"""Return information about hosts that are ready to build.
@ -5318,7 +5371,7 @@ def get_channel(channelInfo, strict=False):
:returns: dict of the channel ID and name, or None.
For example, {'id': 20, 'name': 'container'}
"""
fields = ('id', 'name')
fields = ('id', 'name', 'description')
query = """SELECT %s FROM channels
WHERE """ % ', '.join(fields)
if isinstance(channelInfo, int):
@ -5473,9 +5526,10 @@ def list_channels(hostID=None, event=None):
settings. You must specify a hostID parameter with this
option.
:returns: list of dicts, one per channel. For example,
[{'id': 20, 'name': 'container'}]
[{'id': 20, 'name': 'container', 'description': 'container channel'}]
"""
fields = {'channels.id': 'id', 'channels.name': 'name'}
fields = {'channels.id': 'id', 'channels.name': 'name',
'channels.description': 'description'}
columns, aliases = zip(*fields.items())
if hostID:
tables = ['host_channels']
@ -12589,7 +12643,9 @@ class RootExports(object):
addHostToChannel = staticmethod(add_host_to_channel)
removeHostFromChannel = staticmethod(remove_host_from_channel)
renameChannel = staticmethod(rename_channel)
editChannel = staticmethod(edit_channel)
removeChannel = staticmethod(remove_channel)
addChannel = staticmethod(add_channel)
def listHosts(self, arches=None, channelID=None, ready=None, enabled=None, userID=None,
queryOpts=None):

2
tests/test_cli/data/list-commands-admin.txt
View file

@ -1,6 +1,7 @@
Available commands:
admin commands:
add-channel Add a channel
add-external-repo Create an external repo and/or add one to a tag
add-group Add a group to a tag
add-group-pkg Add a package to a group's package listing
@ -21,6 +22,7 @@ admin commands:
clone-tag Duplicate the contents of one tag onto another tag
disable-host Mark one or more hosts as disabled
disable-user Disable logins by a user
edit-channel Edit a channel
edit-external-repo Edit data for an external repo
edit-host Edit a host
edit-tag Alter tag information

2
tests/test_cli/data/list-commands.txt
View file

@ -1,6 +1,7 @@
Available commands:
admin commands:
add-channel Add a channel
add-external-repo Create an external repo and/or add one to a tag
add-group Add a group to a tag
add-group-pkg Add a package to a group's package listing
@ -21,6 +22,7 @@ admin commands:
clone-tag Duplicate the contents of one tag onto another tag
disable-host Mark one or more hosts as disabled
disable-user Disable logins by a user
edit-channel Edit a channel
edit-external-repo Edit data for an external repo
edit-host Edit a host
edit-tag Alter tag information

96
tests/test_cli/test_add_channel.py Normal file
View file

@ -0,0 +1,96 @@
from __future__ import absolute_import
import unittest
import mock
import six
import koji
from koji_cli.commands import handle_add_channel
from . import utils
class TestAddChannel(utils.CliTestCase):
def setUp(self):
self.maxDiff = None
self.channel_name = 'test-channel'
self.description = 'test-description'
self.channel_id = 1
self.options = mock.MagicMock()
self.session = mock.MagicMock()
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_add_channel(self, activate_session_mock, stdout):
self.session.addChannel.return_value = self.channel_id
rv = handle_add_channel(self.options, self.session,
['--description', self.description, self.channel_name])
actual = stdout.getvalue()
expected = '%s added: id %s\n' % (self.channel_name, self.channel_id)
self.assertMultiLineEqual(actual, expected)
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.addChannel.assert_called_once_with(self.channel_name,
description=self.description)
self.assertNotEqual(rv, 1)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_add_channel_exist(self, activate_session_mock, stderr):
expected = 'channel %(name)s already exists (id=%(id)i)'
self.session.addChannel.side_effect = koji.GenericError(expected)
with self.assertRaises(koji.GenericError) as ex:
handle_add_channel(self.options, self.session,
['--description', self.description, self.channel_name])
self.assertEqual(str(ex.exception), expected)
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.addChannel.assert_called_once_with(self.channel_name,
description=self.description)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_add_channel_without_args(self, activate_session_mock, stderr):
with self.assertRaises(SystemExit) as ex:
handle_add_channel(self.options, self.session, [])
self.assertExitCode(ex, 2)
actual = stderr.getvalue()
expected_stderr = """Usage: %s add-channel [options] <channel_name>
(Specify the --help global option for a list of other help options)
%s: error: Please specify one channel name
""" % (self.progname, self.progname)
self.assertMultiLineEqual(actual, expected_stderr)
activate_session_mock.assert_not_called()
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_add_channel_more_args(self, activate_session_mock, stderr):
channel_2 = 'channel-2'
with self.assertRaises(SystemExit) as ex:
handle_add_channel(self.options, self.session, [self.channel_name, channel_2])
self.assertExitCode(ex, 2)
actual = stderr.getvalue()
expected_stderr = """Usage: %s add-channel [options] <channel_name>
(Specify the --help global option for a list of other help options)
%s: error: Please specify one channel name
""" % (self.progname, self.progname)
self.assertMultiLineEqual(actual, expected_stderr)
activate_session_mock.assert_not_called()
def test_handle_add_host_help(self):
self.assert_help(
handle_add_channel,
"""Usage: %s add-channel [options] <channel_name>
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--description=DESCRIPTION
Description of channel
""" % self.progname)
if __name__ == '__main__':
unittest.main()

25
tests/test_cli/test_add_host.py
View file

@ -1,13 +1,16 @@
from __future__ import absolute_import
import mock
import os
import six
import sys
import mock
import six
import koji
from koji_cli.commands import handle_add_host
from . import utils
class TestAddHost(utils.CliTestCase):
# Show long diffs in error output...
@ -101,7 +104,7 @@ class TestAddHost(utils.CliTestCase):
@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_add_host_help(self, activate_session_mock, stderr, stdout):
def test_handle_add_host_without_args(self, activate_session_mock, stderr, stdout):
arguments = []
options = mock.MagicMock()
progname = os.path.basename(sys.argv[0]) or 'koji'
@ -129,6 +132,20 @@ class TestAddHost(utils.CliTestCase):
session.hasHost.assert_not_called()
session.addHost.assert_not_called()
def test_handle_add_host_help(self):
self.assert_help(
handle_add_host,
"""Usage: %s add-host [options] <hostname> <arch> [<arch> ...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--krb-principal=KRB_PRINCIPAL
set a non-default kerberos principal for the host
--force if existing used is a regular user, convert it to a
host
""" % self.progname)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_add_host_failed(self, activate_session_mock, stderr):
@ -148,7 +165,7 @@ class TestAddHost(utils.CliTestCase):
# Run it and check immediate output
# args: host, arch1, arch2, --krb-principal=krb
# expected: failed
with self.assertRaises(koji.GenericError) as ex:
with self.assertRaises(koji.GenericError):
handle_add_host(options, session, arguments)
actual = stderr.getvalue()
expected = ''

65
tests/test_cli/test_edit_channel.py Normal file
View file

@ -0,0 +1,65 @@
# coding=utf-8
from __future__ import absolute_import
import unittest
import mock
import six
from koji_cli.commands import handle_edit_channel
from . import utils
class TestEditChannel(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.session = mock.MagicMock()
self.channel_old = 'test-channel'
self.channel_new = 'test-channel-new'
self.description = 'description'
self.maxDiff = None
def tearDown(self):
mock.patch.stopall()
def test_handle_edit_channel_help(self):
self.assert_help(
handle_edit_channel,
"""Usage: %s edit-channel [options] <old-name>
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--name=NAME New channel name
--description=DESCRIPTION
Description of channel
""" % self.progname)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_edit_channel_without_args(self, activate_session_mock, stderr):
with self.assertRaises(SystemExit) as ex:
handle_edit_channel(self.options, self.session, [])
self.assertExitCode(ex, 2)
actual = stderr.getvalue()
expected_stderr = """Usage: %s edit-channel [options] <old-name>
(Specify the --help global option for a list of other help options)
%s: error: Incorrect number of arguments
""" % (self.progname, self.progname)
self.assertMultiLineEqual(actual, expected_stderr)
activate_session_mock.assert_not_called()
@mock.patch('koji_cli.commands.activate_session')
def test_handle_edit_channel(self, activate_session_mock):
handle_edit_channel(self.options, self.session,
[self.channel_old, '--name', self.channel_new,
'--description', self.description])
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.editChannel.assert_called_once_with(self.channel_old, name=self.channel_new,
description=self.description)
if __name__ == '__main__':
unittest.main()

106
tests/test_cli/test_remove_channel.py
View file

@ -1,112 +1,78 @@
from __future__ import absolute_import
import mock
import os
import six
import sys
import unittest
import six
import mock
from koji_cli.commands import handle_remove_channel
from . import utils
class TestRemoveChannel(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.options = mock.MagicMock()
self.session = mock.MagicMock()
self.channel_name = 'test-channel'
self.description = 'description'
self.channel_info = {
'id': 123,
'name': self.channel_name,
'description': self.description,
}
self.maxDiff = None
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.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 = handle_remove_channel(options, session, args)
self.session.getChannel.return_value = self.channel_info
rv = handle_remove_channel(self.options, self.session, [self.channel_name])
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, options)
session.getChannel.assert_called_once_with(channel)
session.removeChannel.assert_called_once_with(channel, force=None)
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getChannel.assert_called_once_with(self.channel_name)
self.session.removeChannel.assert_called_once_with(self.channel_name, force=None)
self.assertNotEqual(rv, 1)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.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 = handle_remove_channel(options, session, args)
self.session.getChannel.return_value = self.channel_info
rv = handle_remove_channel(self.options, self.session, ['--force', self.channel_name])
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, options)
session.getChannel.assert_called_once_with(channel)
session.removeChannel.assert_called_once_with(channel, force=True)
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getChannel.assert_called_once_with(self.channel_name)
self.session.removeChannel.assert_called_once_with(self.channel_name, force=True)
self.assertNotEqual(rv, 1)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_remove_channel_no_channel(
self, activate_session_mock, stderr):
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
self.session.getChannel.return_value = channel_info
with self.assertRaises(SystemExit) as ex:
handle_remove_channel(options, session, args)
handle_remove_channel(self.options, self.session, [self.channel_name])
self.assertExitCode(ex, 1)
actual = stderr.getvalue()
expected = 'No such channel: channel\n'
expected = 'No such channel: %s\n' % self.channel_name
self.assertMultiLineEqual(actual, expected)
# Finally, assert that things were called as we expected.
activate_session_mock.assert_called_once_with(session, options)
session.getChannel.assert_called_once_with(channel)
session.removeChannel.assert_not_called()
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getChannel.assert_called_once_with(self.channel_name)
self.session.removeChannel.assert_not_called()
@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_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 ex:
handle_remove_channel(options, session, args)
handle_remove_channel(self.options, self.session, [])
self.assertExitCode(ex, 2)
actual_stdout = stdout.getvalue()
actual_stderr = stderr.getvalue()
@ -115,14 +81,12 @@ class TestRemoveChannel(utils.CliTestCase):
(Specify the --help global option for a list of other help options)
%s: error: Incorrect number of arguments
""" % (progname, progname)
""" % (self.progname, self.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.session.getChannel.assert_not_called()
self.session.removeChannel.assert_not_called()
if __name__ == '__main__':

124
tests/test_cli/test_rename_channel.py
View file

@ -1,104 +1,78 @@
from __future__ import absolute_import
import mock
import os
import six
import sys
import unittest
import mock
import six
from koji_cli.commands import handle_rename_channel
from . import utils
class TestRenameChannel(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.options = mock.MagicMock()
self.session = mock.MagicMock()
self.channel_name_old = 'old-channel'
self.channel_name_new = 'new-channel'
self.description = 'description'
self.channel_info = {
'id': 123,
'name': self.channel_name_old,
'description': self.description,
}
self.maxDiff = None
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.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
args = [self.channel_name_old, self.channel_name_new]
self.session.getChannel.return_value = self.channel_info
# Run it and check immediate output
# args: old_name, new_name
# expected: success
rv = handle_rename_channel(options, session, args)
actual = stdout.getvalue()
expected = ''
self.assertMultiLineEqual(actual, expected)
rv = handle_rename_channel(self.options, self.session, args)
depr_warn = 'rename-channel is deprecated and will be removed in 1.28'
self.assert_console_message(stdout, depr_warn, regex=True)
# Finally, assert that things were called as we expected.
activate_session_mock.assert_called_once_with(session, options)
session.getChannel.assert_called_once_with(old_name)
session.renameChannel.assert_called_once_with(old_name, new_name)
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getChannel.assert_called_once_with(self.channel_name_old)
self.session.renameChannel.assert_called_once_with(self.channel_name_old,
self.channel_name_new)
self.assertNotEqual(rv, 1)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_rename_channel_no_channel(
self, activate_session_mock, stderr):
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
with self.assertRaises(SystemExit) as ex:
handle_rename_channel(options, session, args)
self.assertExitCode(ex, 1)
actual = stderr.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, options)
session.getChannel.assert_called_once_with(old_name)
session.renameChannel.assert_not_called()
@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_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()
def test_handle_rename_channel_no_channel(self, activate_session_mock, stderr, stdout):
channel_info = None
args = [self.channel_name_old, self.channel_name_new]
self.session.getChannel.return_value = channel_info
# Run it and check immediate output
# args: old_name, new_name
# expected: failed: no such channel
with self.assertRaises(SystemExit) as ex:
handle_rename_channel(options, session, args)
self.assertExitCode(ex, 2)
actual_stdout = stdout.getvalue()
actual_stderr = stderr.getvalue()
expected_stdout = ''
expected_stderr = """Usage: %s rename-channel [options] <old-name> <new-name>
handle_rename_channel(self.options, self.session, args)
self.assertExitCode(ex, 1)
expected = 'No such channel: %s' % self.channel_name_old
depr_warn = 'rename-channel is deprecated and will be removed in 1.28'
self.assert_console_message(stderr, expected, wipe=False, regex=True)
self.assert_console_message(stdout, depr_warn, wipe=False, regex=True)
# Finally, assert that things were called as we expected.
activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getChannel.assert_called_once_with(self.channel_name_old)
self.session.renameChannel.assert_not_called()
def test_handle_rename_channel_help(self):
self.assert_help(
handle_rename_channel,
"""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()
Options:
-h, --help show this help message and exit
""" % self.progname)
if __name__ == '__main__':

65
tests/test_hub/test_add_channel.py Normal file
View file

@ -0,0 +1,65 @@
import unittest
import mock
import koji
import kojihub
UP = kojihub.UpdateProcessor
IP = kojihub.InsertProcessor
class TestAddChannel(unittest.TestCase):
def setUp(self):
self.context = mock.patch('kojihub.context').start()
self.context.session.assertPerm = mock.MagicMock()
self.exports = kojihub.RootExports()
self.channel_name = 'test-channel'
self.description = 'test-description'
self.InsertProcessor = mock.patch('kojihub.InsertProcessor',
side_effect=self.getInsert).start()
self.inserts = []
self.insert_execute = mock.MagicMock()
def tearDown(self):
mock.patch.stopall()
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = self.insert_execute
self.inserts.append(insert)
return insert
@mock.patch('kojihub.get_channel')
@mock.patch('kojihub._singleValue')
def test_add_channel_exists(self, _singleValue, get_channel):
get_channel.return_value = {'id': 123, 'name': self.channel_name}
with self.assertRaises(koji.GenericError):
self.exports.addChannel(self.channel_name)
get_channel.assert_called_once_with(self.channel_name, strict=False)
_singleValue.assert_not_called()
self.assertEqual(len(self.inserts), 0)
@mock.patch('kojihub.get_channel')
@mock.patch('kojihub._singleValue')
def test_add_channel_valid(self, _singleValue, get_channel):
get_channel.return_value = {}
_singleValue.side_effect = [12]
r = self.exports.addChannel(self.channel_name, description=self.description)
self.assertEqual(r, 12)
self.assertEqual(len(self.inserts), 1)
insert = self.inserts[0]
self.assertEqual(insert.data['name'], self.channel_name)
self.assertEqual(insert.data['id'], 12)
self.assertEqual(insert.data['description'], self.description)
self.assertEqual(insert.table, 'channels')
self.context.session.assertPerm.assert_called_once_with('admin')
get_channel.assert_called_once_with(self.channel_name, strict=False)
self.assertEqual(_singleValue.call_count, 1)
_singleValue.assert_has_calls([
mock.call("SELECT nextval('channels_id_seq')", strict=True)
])

99
tests/test_hub/test_edit_channel.py Normal file
View file

@ -0,0 +1,99 @@
import unittest
import mock
import koji
import kojihub
UP = kojihub.UpdateProcessor
IP = kojihub.InsertProcessor
class TestEditChannel(unittest.TestCase):
def getInsert(self, *args, **kwargs):
insert = IP(*args, **kwargs)
insert.execute = mock.MagicMock()
self.inserts.append(insert)
return insert
def getUpdate(self, *args, **kwargs):
update = UP(*args, **kwargs)
update.execute = mock.MagicMock()
self.updates.append(update)
return update
def setUp(self):
self.InsertProcessor = mock.patch('kojihub.InsertProcessor',
side_effect=self.getInsert).start()
self.inserts = []
self.UpdateProcessor = mock.patch('kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
self.context = mock.patch('kojihub.context').start()
self.context.session.assertPerm = mock.MagicMock()
self.exports = kojihub.RootExports()
self.channel_name = 'test-channel'
self.channel_name_new = 'test-channel-2'
def tearDown(self):
mock.patch.stopall()
@mock.patch('kojihub.get_channel')
def test_edit_channel_missing(self, get_channel):
expected = 'Invalid type for channelInfo: %s' % self.channel_name
get_channel.side_effect = koji.GenericError(expected)
with self.assertRaises(koji.GenericError) as ex:
self.exports.editChannel(self.channel_name, name=self.channel_name_new)
get_channel.assert_called_once_with(self.channel_name, strict=True)
self.assertEqual(self.inserts, [])
self.assertEqual(self.updates, [])
self.assertEqual(expected, str(ex.exception))
@mock.patch('kojihub.get_channel')
def test_edit_channel_already_exists(self, get_channel):
get_channel.side_effect = [
{
'id': 123,
'name': self.channel_name,
'description': 'description',
},
{
'id': 124,
'name': self.channel_name_new,
'description': 'description',
}
]
with self.assertRaises(koji.GenericError) as ex:
self.exports.editChannel(self.channel_name, name=self.channel_name_new)
expected_calls = [mock.call(self.channel_name, strict=True),
mock.call(self.channel_name_new, strict=False)]
get_channel.assert_has_calls(expected_calls)
self.assertEqual(self.inserts, [])
self.assertEqual(self.updates, [])
self.assertEqual('channel %s already exists (id=124)' % self.channel_name_new,
str(ex.exception))
@mock.patch('kojihub.get_channel')
def test_edit_channel_valid(self, get_channel):
kojihub.get_channel.side_effect = [{
'id': 123,
'name': self.channel_name,
'description': 'description',
},
{}]
r = self.exports.editChannel(self.channel_name, name=self.channel_name_new,
description='description_new')
self.assertIsNone(r)
expected_calls = [mock.call(self.channel_name, strict=True),
mock.call(self.channel_name_new, strict=False)]
get_channel.assert_has_calls(expected_calls)
self.assertEqual(len(self.updates), 1)
values = {'channelID': 123}
clauses = ['id = %(channelID)i']
update = self.updates[0]
self.assertEqual(update.table, 'channels')
self.assertEqual(update.values, values)
self.assertEqual(update.clauses, clauses)

15
tests/test_hub/test_get_channel.py
View file

@ -1,22 +1,31 @@
import unittest
import mock
import koji
import kojihub
class TestGetChannel(unittest.TestCase):
def setUp(self):
self.context = mock.patch('kojihub.context').start()
self.exports = kojihub.RootExports()
def tearDown(self):
mock.patch.stopall()
def test_wrong_type_channelInfo(self):
# dict
channel_info = {'channel': 'val'}
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_channel(channel_info)
self.exports.getChannel(channel_info)
self.assertEqual('Invalid type for channelInfo: %s' % type(channel_info),
str(cm.exception))
#list
# list
channel_info = ['channel']
with self.assertRaises(koji.GenericError) as cm:
kojihub.get_channel(channel_info)
self.exports.getChannel(channel_info)
self.assertEqual('Invalid type for channelInfo: %s' % type(channel_info),
str(cm.exception))

21
tests/test_hub/test_list_channels.py
View file

@ -1,6 +1,7 @@
import mock
import unittest
import mock
import koji
import kojihub
@ -17,7 +18,7 @@ class TestListChannels(unittest.TestCase):
def setUp(self):
self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
side_effect=self.getQuery).start()
side_effect=self.getQuery).start()
self.queries = []
self.context = mock.patch('kojihub.context').start()
# It seems MagicMock will not automatically handle attributes that
@ -27,16 +28,15 @@ class TestListChannels(unittest.TestCase):
def tearDown(self):
mock.patch.stopall()
def test_all(self):
kojihub.list_channels()
self.assertEqual(len(self.queries), 1)
query = self.queries[0]
self.assertEqual(query.tables, ['channels'])
self.assertEqual(query.aliases, ['id', 'name'])
self.assertEqual(query.aliases, ['description', 'id', 'name'])
self.assertEqual(query.joins, None)
self.assertEqual(query.values, {})
self.assertEqual(query.columns, ['channels.id', 'channels.name'])
self.assertEqual(query.columns, ['channels.description', 'channels.id', 'channels.name'])
self.assertEqual(query.clauses, None)
def test_host(self):
@ -50,10 +50,10 @@ class TestListChannels(unittest.TestCase):
'host_channels.host_id = %(host_id)s'
]
self.assertEqual(query.tables, ['host_channels'])
self.assertEqual(query.aliases, ['id', 'name'])
self.assertEqual(query.aliases, ['description', 'id', 'name'])
self.assertEqual(query.joins, joins)
self.assertEqual(query.values, {'host_id': 1234})
self.assertEqual(query.columns, ['channels.id', 'channels.name'])
self.assertEqual(query.columns, ['channels.description', 'channels.id', 'channels.name'])
self.assertEqual(query.clauses, clauses)
def test_host_and_event(self):
@ -63,14 +63,15 @@ class TestListChannels(unittest.TestCase):
query = self.queries[0]
joins = ['channels ON channels.id = host_channels.channel_id']
clauses = [
'(host_channels.create_event <= 2345 AND ( host_channels.revoke_event IS NULL OR 2345 < host_channels.revoke_event ))',
'(host_channels.create_event <= 2345 AND ( host_channels.revoke_event '
'IS NULL OR 2345 < host_channels.revoke_event ))',
'host_channels.host_id = %(host_id)s',
]
self.assertEqual(query.tables, ['host_channels'])
self.assertEqual(query.aliases, ['id', 'name'])
self.assertEqual(query.aliases, ['description', 'id', 'name'])
self.assertEqual(query.joins, joins)
self.assertEqual(query.values, {'host_id': 1234})
self.assertEqual(query.columns, ['channels.id', 'channels.name'])
self.assertEqual(query.columns, ['channels.description', 'channels.id', 'channels.name'])
self.assertEqual(query.clauses, clauses)
def test_event_only(self):

38
tests/test_hub/test_remove_channel.py Normal file
View file

@ -0,0 +1,38 @@
import unittest
import mock
import koji
import kojihub
UP = kojihub.UpdateProcessor
class TestRemoveChannel(unittest.TestCase):
def getUpdate(self, *args, **kwargs):
update = UP(*args, **kwargs)
update.execute = mock.MagicMock()
self.updates.append(update)
return update
def setUp(self):
self.UpdateProcessor = mock.patch('kojihub.UpdateProcessor',
side_effect=self.getUpdate).start()
self.updates = []
self.context = mock.patch('kojihub.context').start()
self.context.session.assertPerm = mock.MagicMock()
self.exports = kojihub.RootExports()
self.channel_name = 'test-channel'
def tearDown(self):
mock.patch.stopall()
@mock.patch('kojihub.get_channel_id')
def test_non_exist_channel(self, get_channel_id):
get_channel_id.side_effect = koji.GenericError('No such channel: %s' % self.channel_name)
with self.assertRaises(koji.GenericError):
kojihub.remove_channel(self.channel_name)
get_channel_id.assert_called_once_with(self.channel_name, strict=True)
self.assertEqual(len(self.updates), 0)

3
tests/test_hub/test_remove_host_from_channel.py
View file

@ -1,6 +1,7 @@
import mock
import unittest
import mock
import koji
import kojihub

3
www/kojiweb/channelinfo.chtml
View file

@ -11,6 +11,9 @@
<tr>
<th>ID</th><td>$channel.id</td>
</tr>
<tr>
<th>Description</th><td>$util.escapeHTML($channel.description)</td>
</tr>
<tr>
<th>Active Tasks</th><td><a href="tasks?view=flat&channelID=$channel.id">$taskCount</a></td>
</tr>