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

Add unit tests for koji commands

Browse source
This commit is contained in:
Franz Chih-Ping Hsieh 2017-11-20 11:24:22 -05:00 • committed by Mike McLean
parent d331fdfa11
commit f16f26be63
17 changed files with 3502 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

92
tests/test_cli/test_add_tag.py Normal file
View file

@ -0,0 +1,92 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import handle_add_tag
from . import utils
class TestAddTag(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s add-tag [options] name
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_add_tag(
self,
activate_session_mock,
stdout):
"""Test handle_add_tag function"""
session = mock.MagicMock()
options = mock.MagicMock()
# Case 1. no argument error
expected = self.format_error_message(
"Please specify a name for the tag")
self.assert_system_exit(
handle_add_tag,
options,
session,
[],
stderr=expected,
activate_session=None)
# Case 2. not admin account
expected = "This action requires admin privileges\n"
session.hasPerm.return_value = None
handle_add_tag(options, session, ['test-tag'])
self.assert_console_message(stdout, expected)
# Case 3. options test
arguments = ['test-tag',
'--parent', 'parent',
'--arch', 'x86_64',
'--maven-support', '--include-all']
# extra fields,
arguments += ['--extra', 'mock.package_manager=dnf',
'--extra', 'mock.new_chroot=0']
opts = {
'parent': 'parent',
'arches': 'x86_64',
'maven_support': True,
'maven_include_all': True,
'extra':
{
'mock.package_manager': 'dnf',
'mock.new_chroot': 0,
}
}
session.hasPerm.return_value = True
handle_add_tag(options, session, arguments)
session.createTag.assert_called_with('test-tag', **opts)
def test_handle_add_tag_help(self):
self.assert_help(
handle_add_tag,
"""Usage: %s add-tag [options] name
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--parent=PARENT Specify parent
--arches=ARCHES Specify arches
--maven-support Enable creation of Maven repos for this tag
--include-all Include all packages in this tag when generating Maven
repos
-x key=value, --extra=key=value
Set tag extra option
""" % self.progname)
if __name__ == '__main__':
unittest.main()

118
tests/test_cli/test_disable_host.py Normal file
View file

@ -0,0 +1,118 @@
from __future__ import absolute_import
import mock
import six
import unittest
from mock import call
from koji_cli.commands import handle_disable_host
from . import utils
class TestDisableHost(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_disable_host(
self,
activate_session_mock,
stdout):
"""Test %s function""" % handle_disable_host.__name__
arguments = []
options = mock.MagicMock()
session = mock.MagicMock()
session.getHost.return_value = None
session.disableHost.return_value = True
session.editHost.return_value = True
#
# session.multiCall returns:
#
# [[{'comment': None,
# 'capacity': 2.0,
# 'name': 'kbuilder01',
# 'enabled': True,
# 'arches': 'x86_64',
# 'task_load': 0.0,
# 'ready': False,
# 'user_id': 4,
# 'id': 2, 'description': None}],
# [{'comment': None,
# 'capacity': 2.0,
# 'name': 'kbuilder02' ...}]
#
session.multiCall.return_value = [[None], [None]]
arguments = ['host1', 'host2']
self.assertEqual(1, handle_disable_host(options, session, arguments))
activate_session_mock.assert_called_once()
session.getHost.assert_has_calls([call('host1'), call('host2')])
session.multiCall.assert_called_once()
session.disableHost.assert_not_called()
session.editHost.assert_not_called()
expect = ''
for host in arguments:
expect += "Host %s does not exist\n" % host
expect += "No changes made. Please correct the command line.\n"
self.assert_console_message(stdout, expect)
# reset session mocks
activate_session_mock.reset_mock()
session.multiCall.reset_mock()
session.disableHost.reset_mock()
session.editHost.reset_mock()
session.multiCall.return_value = [
[{'id': 1, 'name': 'host1'}], [{'id': 2, 'name': 'host2'}]
]
arguments = ['host1', 'host2', '--comment', 'disable host test']
handle_disable_host(options, session, arguments)
activate_session_mock.assert_called_once()
session.getHost.assert_has_calls([call('host1'), call('host2')])
self.assertEqual(2, session.multiCall.call_count)
session.disableHost.assert_has_calls([call('host1'), call('host2')])
session.editHost.assert_has_calls(
[call('host1', comment='disable host test'),
call('host2', comment='disable host test')])
self.assert_console_message(stdout, '')
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_disable_host_no_argument(self, activate_session_mock, stdout):
"""Test %s function without arguments""" % handle_disable_host.__name__
options = mock.MagicMock()
session = mock.MagicMock()
session.getHost.return_value = None
session.multiCall.return_value = [[None]]
session.disableHost.return_value = True
session.editHost.return_value = True
handle_disable_host(options, session, [])
activate_session_mock.assert_called_once()
session.getHost.assert_not_called()
session.multiCall.assert_called()
session.disableHost.assert_not_called()
session.editHost.assert_not_called()
self.assert_console_message(stdout, '')
def test_handle_disable_host_help(self):
"""Test %s help message""" % handle_disable_host.__name__
self.assert_help(
handle_disable_host,
"""Usage: %s disable-host [options] hostname ...
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--comment=COMMENT Comment indicating why the host(s) are being disabled
""" % self.progname)
if __name__ == '__main__':
unittest.main()

118
tests/test_cli/test_enable_host.py Normal file
View file

@ -0,0 +1,118 @@
from __future__ import absolute_import
import mock
import six
import unittest
from mock import call
from koji_cli.commands import handle_enable_host
from . import utils
class TestEnableHost(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_enable_host(
self,
activate_session_mock,
stdout):
"""Test %s function""" % handle_enable_host.__name__
arguments = []
options = mock.MagicMock()
session = mock.MagicMock()
session.getHost.return_value = None
session.enableHost.return_value = True
session.editHost.return_value = True
#
# session.multiCall returns:
#
# [[{'comment': None,
# 'capacity': 2.0,
# 'name': 'kbuilder01',
# 'enabled': True,
# 'arches': 'x86_64',
# 'task_load': 0.0,
# 'ready': False,
# 'user_id': 4,
# 'id': 2, 'description': None}],
# [{'comment': None,
# 'capacity': 2.0,
# 'name': 'kbuilder02' ...}]
#
session.multiCall.return_value = [[None], [None]]
arguments = ['host1', 'host2']
self.assertEqual(1, handle_enable_host(options, session, arguments))
activate_session_mock.assert_called_once()
session.getHost.assert_has_calls([call('host1'), call('host2')])
session.multiCall.assert_called_once()
session.enableHost.assert_not_called()
session.editHost.assert_not_called()
expect = ''
for host in arguments:
expect += "Host %s does not exist\n" % host
expect += "No changes made. Please correct the command line.\n"
self.assert_console_message(stdout, expect)
# reset session mocks
activate_session_mock.reset_mock()
session.multiCall.reset_mock()
session.disableHost.reset_mock()
session.editHost.reset_mock()
session.multiCall.return_value = [
[{'id': 1, 'name': 'host1'}], [{'id': 2, 'name': 'host2'}]
]
arguments = ['host1', 'host2', '--comment', 'disable host test']
handle_enable_host(options, session, arguments)
activate_session_mock.assert_called_once()
session.getHost.assert_has_calls([call('host1'), call('host2')])
self.assertEqual(2, session.multiCall.call_count)
session.enableHost.assert_has_calls([call('host1'), call('host2')])
session.editHost.assert_has_calls(
[call('host1', comment='disable host test'),
call('host2', comment='disable host test')])
self.assert_console_message(stdout, '')
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_enable_host_no_argument(self, activate_session_mock, stdout):
"""Test %s function without arguments""" % handle_enable_host.__name__
options = mock.MagicMock()
session = mock.MagicMock()
session.getHost.return_value = None
session.multiCall.return_value = [[None]]
session.enableHost.return_value = True
session.editHost.return_value = True
handle_enable_host(options, session, [])
activate_session_mock.assert_called_once()
session.getHost.assert_not_called()
session.multiCall.assert_called()
session.enableHost.assert_not_called()
session.editHost.assert_not_called()
self.assert_console_message(stdout, '')
def test_handle_enable_host_help(self):
"""Test %s help message""" % handle_enable_host.__name__
self.assert_help(
handle_enable_host,
"""Usage: %s enable-host [options] hostname ...
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--comment=COMMENT Comment indicating why the host(s) are being enabled
""" % self.progname)
if __name__ == '__main__':
unittest.main()

690
tests/test_cli/test_import.py Normal file
View file

@ -0,0 +1,690 @@
from __future__ import absolute_import
import mock
import six
import unittest
import koji
import os
from koji_cli.commands import handle_import
from . import utils
class TestImport(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.huburl = "https://%s.local/%shub" % (self.progname, self.progname)
self.md5 = '00112233445566778899aabbccddeeff'
self.fake_srv_dir = '/path/to/server/import'
#
# RPM header example (bash-4.4.12-5.fc26.x86_64.rpm):
# {
# 'sourcepackage': None,
# 'name': 'bash',
# 'sigmd5': 'J\x15\xca\x94\xb1\xacY\xd5\xef\x9f\xc6\xd5\n\xd7?>',
# 'epoch': None,
# 'version': '4.4.12',
# 'release': '5.fc26',
# 'sourcerpm': 'bash-4.4.12-5.fc26.src.rpm',
# 'arch': 'x86_64'
# }
#
# SRPM header example (bash-4.4.12-5.fc26.src.rpm):
# {
# 'sourcepackage': 1,
# 'name': 'bash',
# 'sigmd5': '\x8a\x17\x05\xe8k\xef\x15\x16V[\x02\x9cs\xab\x7f\xdd',
# 'epoch': None,
# 'version': '4.4.12',
# 'release': '5.fc26',
# 'sourcerpm': None,
# 'arch': 'x86_64'
# }
self.srpm_header = {
'sourcepackage': 1,
'name': 'bash',
'sigmd5': bytearray.fromhex(self.md5),
'epoch': None,
'version': '4.4.12',
'release': '5.fc26',
'sourcerpm': None,
'arch': 'x86_64'
}
self.rpm_header = {
'sourcepackage': None,
'name': 'bash',
'sigmd5': bytearray.fromhex(self.md5),
'epoch': None,
'version': '4.4.12',
'release': '5.fc26',
'sourcerpm': 'bash-4.4.12-5.fc26.src.rpm',
'arch': 'x86_64'
}
# koji.BUILD_STATES
self.bstate = {
'BUILDING': 0,
'COMPLETE': 1,
'DELETED': 2,
'FAILED': 3,
'CANCELED': 4,
}
self.error_format = """Usage: %s import [options] package [package...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def __do_import_test(self, options, session, arguments, **kwargs):
expected = kwargs.get('expected', None)
rpm_header = kwargs.get('rpm_header', {})
fake_srv_path = kwargs.get('srv_path', '/path/to/server/import')
upload_rpm_mock = kwargs.get('upload_rpm_mock', session.uploadWrapper)
with mock.patch('koji.get_header_fields') as get_header_fields_mock, \
mock.patch('koji_cli.commands._unique_path') as unique_path_mock, \
mock.patch('koji_cli.commands.activate_session') as activate_session_mock, \
mock.patch('sys.stdout', new_callable=six.StringIO) as stdout, \
upload_rpm_mock:
get_header_fields_mock.return_value = rpm_header
unique_path_mock.return_value = fake_srv_path
handle_import(options, session, arguments)
# check output message
self.assert_console_message(stdout, expected)
# check mock calls
activate_session_mock.assert_called_with(session, options)
get_header_fields_mock.assert_called_with(
arguments[0],
('name', 'version', 'release', 'epoch',
'arch', 'sigmd5', 'sourcepackage', 'sourcerpm')
)
session.getRPM.assert_called_with(
dict((k, rpm_header.get(k, ''))
for k in ['release', 'version', 'arch', 'name'])
)
unique_path_mock.assert_called_with('cli-import')
upload_rpm_mock.assert_called_with(arguments[0], self.fake_srv_dir)
session.importRPM.assert_called_with(
self.fake_srv_dir, os.path.basename(arguments[0]))
# reset for next test
activate_session_mock.reset_mock()
get_header_fields_mock.reset_mock()
unique_path_mock.reset_mock()
upload_rpm_mock.reset_mock()
session.getRPM.reset_mock()
session.importRPM.reset_mock()
def __skip_import_test(self, options, session, arguments, **kwargs):
expected = kwargs.get('expected', None)
rpm_header = kwargs.get('rpm_header', {})
with mock.patch('koji.get_header_fields') as get_header_fields_mock, \
mock.patch('koji_cli.commands.activate_session') as activate_session_mock, \
mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
get_header_fields_mock.return_value = rpm_header
handle_import(options, session, arguments)
# check output message
self.assert_console_message(stdout, expected)
# check mock calls
activate_session_mock.assert_called_with(session, options)
get_header_fields_mock.assert_called_with(
arguments[0],
('name', 'version', 'release', 'epoch',
'arch', 'sigmd5', 'sourcepackage', 'sourcerpm')
)
session.getRPM.assert_called_with(
dict((k, rpm_header.get(k, ''))
for k in ['release', 'version', 'arch', 'name'])
)
session.uploadWrapper.assert_not_called()
session.importRPM.assert_not_called()
def test_handle_import_src_rpm_import_with_no_exist_build(self):
"""Test handle_import source RPM import.
No build is on the server case,
this tests are focusing on do_import() function coverage.
"""
arguments = ['/path/to/bash-4.4.12-5.fc26.src.rpm', '--src-epoch', 'None']
options = mock.MagicMock()
session = mock.MagicMock()
# No exist build test case
# import general case
session.getBuild.return_value = None
session.getRPM.return_value = None
expected = "uploading %s... done\n" % arguments[0]
expected += "importing %s... done\n" % arguments[0]
self.__do_import_test(
options, session, arguments,
rpm_header=self.srpm_header, expected=expected)
# import error case
session.importRPM.side_effect = koji.GenericError('fake-import-error')
expected = "uploading %s... done\n" % arguments[0]
expected += "importing %s... \n" % arguments[0]
expected += "Error importing: fake-import-error\n"
self.__do_import_test(
options, session, arguments,
rpm_header=self.srpm_header, expected=expected)
# import with --link (hardlink) option
session.importRPM.side_effect = None
expected = "importing %s... done\n" % arguments[0]
self.__do_import_test(
options, session, arguments + ['--link'],
upload_rpm_mock=mock.patch('koji_cli.commands.linked_upload').start(),
rpm_header=self.srpm_header, expected=expected)
session.uploadWrapper.assert_not_called()
def test_handle_import_src_rpm_import_with_exist_build(self):
"""Test handle_import source RPM import with exist build case."""
arguments = ['/path/to/bash-4.4.12-5.fc26.src.rpm']
options = mock.MagicMock()
session = mock.MagicMock()
false_md5 = 'ffeeddccbbaa99887766554433221100'
#
# getBuild return example (bash-4.4.12-5.fc26.src.rpm)
# {
# 'package_name': 'bash',
# 'extra': None,
# 'creation_time': '2017-11-08 18:56:31.359440',
# 'completion_time': '2017-11-08 18:56:31.359440',
# 'package_id': 1,
# 'id': 1,
# 'build_id': 1,
# 'state': 1,
# 'source': None,
# 'epoch': None,
# 'version': '4.4.12',
# 'completion_ts': 1510167391.35944,
# 'owner_id': 1,
# 'owner_name': 'kojiadmin',
# 'nvr': 'bash-4.4.12-5.fc26',
# 'start_time': '2017-11-08 18:56:31.359440',
# 'creation_event_id': 81,
# 'start_ts': 1510167391.35944,
# 'volume_id': 0,
# 'creation_ts': 1510167391.35944,
# 'name': 'bash',
# 'task_id': None,
# 'volume_name': 'DEFAULT',
# 'release': '5.fc26'
# }
#
#
# getRPM return example:
# {
# 'build_id': 1,
# 'name': 'bash',
# 'extra': None,
# 'external_repo_id': 0,
# 'buildtime': 1496119944,
# 'id': 1,
# 'epoch': None,
# 'version': '4.4.12',
# 'buildroot_id': None,
# 'metadata_only': False,
# 'release': '5.fc26',
# 'arch': 'src',
# 'payloadhash': '8a1705e86bef1516565b029c73ab7fdd',
# 'external_repo_name': 'INTERNAL',
# 'size': 9462614
# }
#
# Case 1: build exists and status is 'COMPLETE', md5 matched
# reseult: import skipped
session.getBuild.return_value = {'state': self.bstate['COMPLETE']}
session.getRPM.return_value = {
'external_repo_id': 0,
'external_repo_name': 'INTERNAL',
'payloadhash': self.md5
}
expected = "RPM already imported: %s\n" % arguments[0]
expected += "Skipping import\n"
self.__skip_import_test(
options, session, arguments,
rpm_header=self.srpm_header,
expected=expected)
# Case 2: build exists and status is 'COMPLETE', md5 mismatched
# reseult: import skipped
session.getRPM.return_value['payloadhash'] = false_md5
expected = "WARNING: md5sum mismatch for %s\n" % arguments[0]
expected += " A different rpm with the same name has already been imported\n"
expected += " Existing sigmd5 is %r, your import has %r\n" % (false_md5, self.md5)
expected += "Skipping import\n"
self.__skip_import_test(
options, session, arguments,
rpm_header=self.srpm_header,
expected=expected)
# Case 3: build exists and status is 'COMPLETE', has external_repo_id
# reseult: import will be performed
session.getRPM.return_value['external_repo_id'] = 1
expected = "uploading %s... done\n" % arguments[0]
expected += "importing %s... done\n" % arguments[0]
self.__do_import_test(
options, session, arguments,
rpm_header=self.srpm_header,
expected=expected)
nvr = '%(name)s-%(version)s-%(release)s' % self.srpm_header
# Case 4: build exists and the status is FAILED or CANCELED
# without --create-build option.
# result: import skipped
session.getBuild.return_value = {
'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['FAILED']}
session.getRPM.return_value = {
'external_repo_name': 'INTERNAL',
'payloadhash': self.md5
}
expected = "Build %s state is %s. Skipping import\n" % (nvr, 'FAILED')
with mock.patch('koji.get_header_fields') as get_header_fields_mock, \
mock.patch('koji_cli.commands.activate_session') as activate_session_mock, \
mock.patch('sys.stdout', new_callable=six.StringIO) as stdout:
get_header_fields_mock.return_value = self.srpm_header
handle_import(options, session, arguments)
activate_session_mock.assert_called_with(session, options)
self.assert_console_message(stdout, expected)
# Case 5: build exists and the status is FAILED or CANCELED
# with --create-build option.
# result: empty build will be created,
# import will be performed.
expected = "Creating empty build: %s\n" % nvr
expected += "uploading %s... done\n" % arguments[0]
expected += "importing %s... done\n" % arguments[0]
session.getBuild.side_effect = [
# simulate build fail case
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['FAILED']},
# after calling createEmptyBuild, getBuild should return:
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['COMPLETE']},
]
# no RPM info is returned
session.getRPM.return_value = {}
self.__do_import_test(
options, session, arguments + ['--create-build'],
rpm_header=self.srpm_header,
expected=expected)
kwarg = dict((k, self.rpm_header.get(k, None))
for k in ['name', 'version', 'release', 'epoch'])
session.createEmptyBuild.assert_called_with(**kwarg)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji.get_header_fields')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_import_binary_rpm_import_with_no_build_exist(
self,
activate_session_mock,
get_header_fields_mock,
stderr,
stdout):
"""Test handle_import binary RPM import with no exist build case."""
arguments = ['/path/to/bash-4.4.12-5.fc26.rpm']
options = mock.MagicMock()
session = mock.MagicMock()
get_header_fields_mock.return_value = self.rpm_header
nvr = '%(name)s-%(version)s-%(release)s' % get_header_fields_mock.return_value
# Case 1. without exist build on server
# result: abort
session.getBuild.return_value = None
handle_import(options, session, arguments)
expected = "Missing build or srpm: %s\n" % nvr
expected += "Aborting import\n"
self.assert_console_message(stdout, expected)
# Case 2. without exist build on server,
# with --create-build option
# result: import will be performed
session.getBuild.return_value = None
session.getRPM.return_value = None
expected = "Missing build or srpm: %s\n" % nvr
expected += "Creating empty build: %s\n" % nvr
expected += "uploading %s... done\n" % arguments[0]
expected += "importing %s... done\n" % arguments[0]
self.__do_import_test(
options, session, arguments + ['--create-build'],
rpm_header=self.rpm_header,
expected=expected)
kwarg = dict((k, self.rpm_header.get(k, None))
for k in ['name', 'version', 'release', 'epoch'])
session.createEmptyBuild.assert_called_with(**kwarg)
@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_import_binary_rpm_with_exist_build(
self,
activate_session_mock,
stderr,
stdout):
"""Test handle_import binary RPM import with exist build case."""
arguments = ['/path/to/bash-4.4.12-5.fc26.rpm']
options = mock.MagicMock()
session = mock.MagicMock()
nvr = '%(name)s-%(version)s-%(release)s' % self.rpm_header
# Case 1. have exist build on server, build status is COMPLETE case.
# result: skip import
session.getBuild.return_value = {'state': self.bstate['COMPLETE']}
session.getRPM.return_value = {
'external_repo_name': 'INTERNAL',
'payloadhash': self.md5,
}
expected = "RPM already imported: %s\n" % arguments[0]
expected += "Skipping import\n"
self.__skip_import_test(
options, session, arguments,
rpm_header=self.rpm_header,
expected=expected)
# Case 2. have exist build on server, build status is failed,
# without --create-build case.
# result: import will be skipped
session.getBuild.return_value = {'state': self.bstate['FAILED']}
expected = "Build %s state is %s. Skipping import\n" % (nvr, 'FAILED')
self.__skip_import_test(
options, session, arguments,
rpm_header=self.rpm_header,
expected=expected)
# Case 3. have exist build on server, build status is failed,
# with --create-build,
# result:
# empty build will be created,
# import will be performed.
expected = "Creating empty build: %s\n" % nvr
expected += "uploading %s... done\n" % arguments[0]
expected += "importing %s... done\n" % arguments[0]
session.getBuild.side_effect = [
# binary rpm case,
# getBuild is called to check build.
{'state': self.bstate['FAILED']},
# simulate build fail case
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['FAILED']},
# after calling createEmptyBuild, getBuild should return:
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['COMPLETE']},
]
# no RPM info is returned
session.getRPM.return_value = {}
self.__do_import_test(
options, session, arguments + ['--create-build'],
rpm_header=self.rpm_header,
expected=expected)
kwarg = dict((k, self.rpm_header.get(k, None))
for k in ['name', 'version', 'release', 'epoch'])
session.createEmptyBuild.assert_called_with(**kwarg)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.get_header_fields')
@mock.patch('koji_cli.commands._unique_path')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_import_with_test_option(
self,
activate_session_mock,
unique_path_mock,
get_header_fields_mock,
stdout):
"""Test handle_import RPM with --test option"""
options = mock.MagicMock()
session = mock.MagicMock()
nvr = lambda: '%(name)s-%(version)s-%(release)s' % get_header_fields_mock.return_value
# Case 1. SRPM case
# result: skip because of --test
get_header_fields_mock.return_value = self.srpm_header
session.getBuild.return_value = {}
session.getRPM.return_value = {}
arguments = ['/path/to/bash-4.4.12-5.fc26.src.rpm', '--test']
expected = "Test mode -- skipping import for %s\n" % arguments[0]
handle_import(options, session, arguments)
self.assert_console_message(stdout, expected)
unique_path_mock.assert_not_called()
session.importRPM.assert_not_called()
# Case 2. Binary RPM case (need --create-build option)
# result: skip because of --test
session.getBuild.side_effect = [
# binary rpm case,
# getBuild is called to check build.
{'state': self.bstate['FAILED']},
# simulate build fail case
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['FAILED']},
]
get_header_fields_mock.return_value = self.rpm_header
arguments = ['/path/to/bash-4.4.12-5.fc26.rpm', '--test', '--create-build']
expected = "Test mode -- would have created empty build: %s\n" % nvr()
expected += "Test mode -- skipping import for %s\n" % arguments[0]
with mock.patch('koji_cli.commands._unique_path') as unique_path_mock:
handle_import(options, session, arguments)
self.assert_console_message(stdout, expected)
unique_path_mock.assert_not_called()
session.createEmptyBuild.assert_not_called()
session.importRPM.assert_not_called()
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.get_header_fields')
@mock.patch('koji_cli.commands._unique_path')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_import_with_epoch_option(
self,
activate_session_mock,
unique_path_mock,
get_header_fields_mock,
stdout):
"""Test handle_import RPM with --src-epoch option"""
options = mock.MagicMock()
session = mock.MagicMock()
epoch = "1"
nvr = '%(name)s-%(version)s-%(release)s' % self.rpm_header
# Binary RPM case (need --create-build option)
# result: skip because of --test
arguments = ['/path/to/bash-4.4.12-5.fc26.rpm',
'--create-build',
'--src-epoch', epoch]
expected = "Creating empty build: %s\n" % nvr
expected += "uploading %s... done\n" % arguments[0]
expected += "importing %s... done\n" % arguments[0]
session.getBuild.side_effect = [
# binary rpm case,
# getBuild is called to check build.
{'state': self.bstate['FAILED']},
# simulate build fail case
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['FAILED']},
# after calling createEmptyBuild, getBuild should return:
{'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'state': self.bstate['COMPLETE']},
]
# no RPM info is returned
session.getRPM.return_value = {}
self.__do_import_test(
options, session, arguments + ['--create-build'],
rpm_header=self.rpm_header,
expected=expected)
kwarg = dict((k, self.rpm_header.get(k, None))
for k in ['name', 'version', 'release', 'epoch'])
kwarg['epoch'] = int(epoch)
session.createEmptyBuild.assert_called_with(**kwarg)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji.get_header_fields')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_import_has_build_states_test(
self,
activate_session_mock,
get_header_fields_mock,
stderr,
stdout):
"""Test handle_import RPM with existing build but status not COMPLETE case"""
arguments = ['/path/to/bash-4.4.12-5.fc26.src.rpm']
options = mock.MagicMock()
session = mock.MagicMock()
get_header_fields_mock.return_value = self.srpm_header
test_cases = [
{
'state': 'FAILED',
'msg': "Build %s state is %s. Skipping import"
},
{
'state': 'CANCELED',
'msg': "Build %s state is %s. Skipping import"
},
{
'state': 'DELETED',
'msg': 'Build %s exists with state=%s. Skipping import'
},
{
'state': 'BUILDING',
'msg': 'Build %s exists with state=%s. Skipping import'
}
]
nvr = '%(name)s-%(version)s-%(release)s' % \
get_header_fields_mock.return_value
# build failed cases
for case in test_cases:
session.getBuild.return_value = {
'state': self.bstate[case['state']]}
handle_import(options, session, arguments)
expected = case['msg'] % (nvr, case['state']) + "\n"
self.assert_console_message(stdout, expected)
get_header_fields_mock.assert_called_with(
arguments[0],
('name', 'version', 'release', 'epoch',
'arch', 'sigmd5', 'sourcepackage', 'sourcerpm')
)
activate_session_mock.assert_called_with(session, options)
session.getRPM.assert_not_called()
session.importRPM.assert_not_called()
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_import_argument_error(
self,
activate_session_mock,
stderr):
"""Test handle_import function with error arguments"""
options = mock.MagicMock()
session = mock.MagicMock()
epoch = 'zzz'
test_cases = [
# empty argument error
{
'arguments': [],
'error': "At least one package must be specified"
},
# invalid epoch number test (epoch must be integer)
{
'arguments': ['pkg', '--src-epoch', epoch],
'error': "Invalid value for epoch: %s" % epoch
}
]
for case in test_cases:
expect = self.format_error_message(case['error'])
self.assert_system_exit(
handle_import,
options,
session,
case['arguments'],
stderr=expect,
activate_session=None)
activate_session_mock.assert_not_called()
def test_handle_import_help(self):
"""Test handle_import function help message"""
self.assert_help(
handle_import,
"""Usage: %s import [options] package [package...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--link Attempt to hardlink instead of uploading
--test Don't actually import
--create-build Auto-create builds as needed
--src-epoch=SRC_EPOCH
When auto-creating builds, use this epoch
""" % self.progname)
if __name__ == '__main__':
unittest.main()

220
tests/test_cli/test_import_cg.py Normal file
View file

@ -0,0 +1,220 @@
from __future__ import absolute_import
import mock
import six
import unittest
from mock import call
from koji_cli.commands import handle_import_cg
from . import utils
import os
class TestImportCG(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def mock_os_path_exists(self, filepath):
if filepath in self.custom_os_path_exists:
return self.custom_os_path_exists[filepath]
return self.os_path_exists(filepath)
def setUp(self):
self.custom_os_path_exists = {}
self.os_path_exists = os.path.exists
self.error_format = """Usage: %s import-cg [options] metadata_file files_dir
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands._progress_callback')
@mock.patch('koji_cli.commands._unique_path')
@mock.patch('koji_cli.commands._running_in_bg', return_value=False)
@mock.patch('koji_cli.commands.linked_upload')
@mock.patch('koji_cli.commands.activate_session')
@mock.patch('koji_cli.commands.json')
def test_handle_import_cg(
self,
json_mock,
activate_session_mock,
linked_upload_mock,
running_in_bg_mock,
unique_path_mock,
progress_callback_mock,
stdout):
"""Test handle_import_cg function"""
arguments = ['fake-metafile', '/path/to/files/']
options = mock.MagicMock()
session = mock.MagicMock()
expected = ''
fake_srv_path = '/path/to/server/cli-import'
metadata = {
'output': [
{'relpath': '/real/path', 'filename': 'file-1'},
{'relpath': '/real/path', 'filename': 'file-2'}
]
}
#
# we just need to change original os.path.exists behavior, if the input
# is matched return the value we expected.
self.custom_os_path_exists = dict(('%(relpath)s/%(filename)s' % v, True)
for v in metadata['output'])
# setup and start os.path.exists patch
os_path_exists_patch = mock.patch('os.path.exists',
new=self.mock_os_path_exists)
os_path_exists_patch.start()
def gen_value(fmt, callback):
calls, expect = [], ''
for item in metadata['output']:
filepath = "%(relpath)s/%(filename)s" % item
calls.append(call(filepath,
item['relpath'],
callback=callback))
expect += fmt % filepath + "\n"
return calls, expect
json_mock.load.return_value = metadata
unique_path_mock.return_value = fake_srv_path
# Case 1, running in fg, progress on
with mock.patch(utils.get_builtin_open()):
handle_import_cg(options, session, arguments)
calls, expected = gen_value("Uploading %s\n", progress_callback_mock)
self.assert_console_message(stdout, expected)
linked_upload_mock.assert_not_called()
session.uploadWrapper.assert_has_calls(calls)
session.CGImport.assert_called_with(metadata, fake_srv_path)
# Case 2, running in fg, progress off
with mock.patch(utils.get_builtin_open()):
handle_import_cg(options, session, arguments + ['--noprogress'])
calls, expected = gen_value("Uploading %s", None)
self.assert_console_message(stdout, expected)
linked_upload_mock.assert_not_called()
session.uploadWrapper.assert_has_calls(calls)
session.CGImport.assert_called_with(metadata, fake_srv_path)
# reset mocks
linked_upload_mock.reset_mock()
session.uploadWrapper.reset_mock()
session.CGImport.reset_mock()
# Case 3, --test option
with mock.patch(utils.get_builtin_open()):
handle_import_cg(options, session, arguments + ['--test'])
linked_upload_mock.assert_not_called()
session.uploadWrapper.assert_not_called()
session.CGImport.assert_not_called()
calls = [call("%(relpath)s/%(filename)s" % item, item['relpath'])
for item in metadata['output']]
# Case 4, --link option
with mock.patch(utils.get_builtin_open()):
handle_import_cg(options, session, arguments + ['--link'])
linked_upload_mock.assert_has_calls(calls)
session.uploadWrapper.assert_not_called()
session.CGImport.assert_called_with(metadata, fake_srv_path)
# make sure there is no message on output
self.assert_console_message(stdout, '')
# stop os.path.exists patch
os_path_exists_patch.stop()
def test_handle_import_argument_test(self):
"""Test handle_import_cg function without arguments"""
arguments = ['fake-metafile', '/path/to/files/']
options = mock.MagicMock()
session = mock.MagicMock()
# Case 1. empty argument
expected = self.format_error_message(
"Please specify metadata files directory")
self.assert_system_exit(
handle_import_cg,
options,
session,
[],
stderr=expected,
activate_session=None)
# Case 2. JSON module does not exist
expected = self.format_error_message('Unable to find json module')
with mock.patch('koji_cli.commands.json', new=None):
self.assert_system_exit(
handle_import_cg,
options,
session,
arguments,
stderr=expected,
activate_session=None)
metadata = {
'output': [
{'metadata_only': True},
{'relpath': '/real/path', 'filename': 'filename'}
]
}
#
# we just need to change original os.path.exists behavior, if the input
# is matched return the value we expected.
self.custom_os_path_exists['/real/path/filename'] = False
with mock.patch(utils.get_builtin_open()), \
mock.patch('os.path.exists', new=self.mock_os_path_exists), \
mock.patch('koji_cli.commands.json') as json_mock:
# Case 3. metafile doesn't have output section
json_mock.load.return_value = {}
expected = "Metadata contains no output\n"
self.assert_system_exit(
handle_import_cg,
options,
session,
arguments,
stdout=expected,
exit_code=1)
# Case 4. path not exist
file_path = '%(relpath)s/%(filename)s' % metadata['output'][1]
json_mock.load.return_value = metadata
expected = self.format_error_message(
"No such file: %s" % file_path)
self.assert_system_exit(
handle_import_cg,
options,
session,
arguments,
stderr=expected)
def test_handle_import_cg_help(self):
"""Test handle_import_cg help message"""
self.assert_help(
handle_import_cg,
"""Usage: %s import-cg [options] metadata_file files_dir
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--noprogress Do not display progress of the upload
--link Attempt to hardlink instead of uploading
--test Don't actually import
""" % self.progname)
if __name__ == '__main__':
unittest.main()

284
tests/test_cli/test_import_sig.py Normal file
View file

@ -0,0 +1,284 @@
from __future__ import absolute_import
import mock
import six
import unittest
import random
import hashlib
import copy
import base64
from mock import call
from koji_cli.commands import handle_import_sig
from . import utils
import os
class TestImportSIG(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def md5sum(self, message):
md5 = hashlib.md5()
md5.update(message.encode('utf-8'))
return md5.hexdigest()
def mock_os_path_exists(self, filepath):
if filepath in self.custom_os_path_exists:
return self.custom_os_path_exists[filepath]
return self.os_path_exists(filepath)
def setUp(self):
self.custom_os_path_exists = {}
self.os_path_exists = os.path.exists
self.rpm_headers = [
{
'sourcepackage': 1,
'name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'arch': 'x86_64',
'siggpg': None,
'sigpgp': None,
},
{
'sourcepackage': 1,
'name': 'less',
'version': '487',
'release': '3.fc26',
'arch': 'x86_64',
'siggpg': None,
'sigpgp': None,
},
{
'sourcepackage': 1,
'name': 'sed',
'version': '4.4',
'release': '1.fc26',
'arch': 'x86_64',
'siggpg': None,
'sigpgp': None,
}
]
self.error_format = """Usage: %s import-sig [options] package [package...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.rip_rpm_sighdr')
@mock.patch('koji.get_sigpacket_key_id')
@mock.patch('koji.get_header_fields')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_import_sig(
self,
activate_session_mock,
get_header_fields_mock,
get_sigpacket_key_id_mock,
rip_rpm_sighdr_mock,
stdout):
"""Test handle_import_sig function"""
arguments = ['/path/to/bash', '/path/to/less', '/path/to/sed']
options = mock.MagicMock()
session = mock.MagicMock()
expected = ''
fake_sigkey = '00112233445566778899aAbBcCdDeEfF'
#
# we just need to change original os.path.exists behavior, if the input
# is matched return the value we expected.
self.custom_os_path_exists = dict((f, True) for f in arguments)
# setup and start os.path.exists patch
os_path_exists_patch = mock.patch('os.path.exists',
new=self.mock_os_path_exists)
os_path_exists_patch.start()
# Case 1, Unsigned pkg test (without ----with-unsigned)
# result: import skipped
for pkgfile in arguments:
expected += "Skipping unsigned package: %s" % pkgfile + "\n"
get_header_fields_mock.side_effect = copy.deepcopy(self.rpm_headers)
# Run
handle_import_sig(options, session, arguments)
self.assert_console_message(stdout, expected)
activate_session_mock.assert_called_once()
rip_rpm_sighdr_mock.assert_not_called()
session.getRPM.assert_not_called()
# Case 2, No RPM in system
# result: import skipped
expected = ''
for data in self.rpm_headers:
data['siggpg'] = fake_sigkey
data['sigpgp'] = fake_sigkey
tmp = data.copy()
tmp['arch'] = 'src' if tmp['sourcepackage'] else tmp['arch']
expected += "No such rpm in system: %(name)s-%(version)s-%(release)s.%(arch)s" % \
tmp + "\n"
get_header_fields_mock.side_effect = copy.deepcopy(self.rpm_headers)
get_sigpacket_key_id_mock.return_value = fake_sigkey
session.getRPM.return_value = {}
# Run
handle_import_sig(options, session, arguments)
self.assert_console_message(stdout, expected)
rip_rpm_sighdr_mock.assert_not_called()
session.queryRPMSigs.assert_not_called()
# Case 3, Find external repo RPM
# result: import skipped
ext_repos = ['ext-repo1.net', 'ext-repo2.net', 'ext-repo3.net']
expected = ''
rinfo = copy.deepcopy(self.rpm_headers)
for data in rinfo:
rid = random.randint(0, 999) % 3
data['external_repo_id'] = rid + 1
data['external_repo_name'] = ext_repos[rid]
data['arch'] = 'src' if data['sourcepackage'] else data['arch']
expected += "Skipping external rpm: %(name)s-%(version)s-%(release)s.%(arch)s@%(external_repo_name)s" % data + "\n"
get_header_fields_mock.side_effect = rinfo
session.getRPM.side_effect = rinfo
# Run
handle_import_sig(options, session, arguments)
self.assert_console_message(stdout, expected)
rip_rpm_sighdr_mock.assert_not_called()
session.queryRPMSigs.assert_not_called()
# Case 4, has previous RPM signature
# result: import skipped
# show match or mismatch message
expected = ''
# session.queryRPMSigs return example:
# [{'sigkey': '64dab85d', 'sighash': '7141c84f059d2f0722ff545051b2981d', 'rpm_id': 1}]
#
sighdr, sigRpm = [], []
for i, pkg in enumerate(arguments):
signature = 'sighdr-%s' % pkg
sighdr.append(signature.encode('utf-8'))
if i < 2:
sigRpm.append([{'sighash': self.md5sum(signature)}])
expected += "Signature already imported: %s" % pkg + "\n"
else:
sigRpm.append([{'sighash': self.md5sum('wrong-sig-case')}])
expected += "Warning: signature mismatch: %s" % pkg + "\n"
expected += " The system already has a signature for this rpm with key %s" % fake_sigkey + "\n"
expected += " The two signature headers are not the same" + "\n"
rinfo = copy.deepcopy(self.rpm_headers)
for i, data in enumerate(rinfo):
data['external_repo_id'] = 0
data['id'] = i + 1
get_header_fields_mock.side_effect = copy.deepcopy(rinfo)
session.getRPM.side_effect = rinfo
rip_rpm_sighdr_mock.side_effect = sighdr
session.queryRPMSigs.side_effect = sigRpm
# Run
handle_import_sig(options, session, arguments)
self.assert_console_message(stdout, expected)
# Case 5, --test options test
# result: everything works fine and addRPMSig/writeSignedRPM should
# not be called.
expected = ''
for pkg in arguments:
expected += "Importing signature [key %s] from %s..." % (fake_sigkey, pkg) + "\n"
expected += "Writing signed copy" + "\n"
get_header_fields_mock.side_effect = copy.deepcopy(rinfo)
session.getRPM.side_effect = rinfo
rip_rpm_sighdr_mock.side_effect = sighdr
session.queryRPMSigs.side_effect = None
session.queryRPMSigs.return_value = []
# Run
handle_import_sig(options, session, arguments + ['--test'])
self.assert_console_message(stdout, expected)
session.addRPMSig.assert_not_called()
session.writeSignedRPM.assert_not_called()
# Case 6, normal import
# result: addRPMSig/writeSignedRPM should be called
get_header_fields_mock.side_effect = copy.deepcopy(rinfo)
session.getRPM.side_effect = rinfo
rip_rpm_sighdr_mock.side_effect = sighdr
add_sig_calls, write_sig_calls = [], []
for i in range(0, 3):
add_sig_calls.append(call(rinfo[i]['id'], base64.encodestring(sighdr[i])))
write_sig_calls.append(call(rinfo[i]['id'], fake_sigkey))
# Run
handle_import_sig(options, session, arguments)
self.assert_console_message(stdout, expected)
session.addRPMSig.assert_has_calls(add_sig_calls)
session.writeSignedRPM.assert_has_calls(write_sig_calls)
# restore os.path.exists patch
os_path_exists_patch.stop()
def test_handle_import_sig_argument_test(self):
"""Test handle_import_sig function without arguments"""
options = mock.MagicMock()
session = mock.MagicMock()
# Case 1. empty argument
expected = self.format_error_message(
"At least one package must be specified")
self.assert_system_exit(
handle_import_sig,
options,
session,
[],
stderr=expected,
activate_session=None)
# Case 2. File not exists test
arguments = ['/bin/ls', '/tmp', '/path/to/file1', '/path/to/file2']
expected = self.format_error_message(
"No such file: %s" % arguments[2])
self.assert_system_exit(
handle_import_sig,
options,
session,
arguments,
stderr=expected,
activate_session=None)
def test_handle_import_sig_help(self):
"""Test handle_import_sig help message"""
self.assert_help(
handle_import_sig,
"""Usage: %s import-sig [options] package [package...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--with-unsigned Also import unsigned sig headers
--write Also write the signed copies
--test Test mode -- don't actually import
""" % self.progname)
if __name__ == '__main__':
unittest.main()

125
tests/test_cli/test_list_tasks.py
View file

@ -1,8 +1,12 @@
import mock
import six
import unittest
import koji
from koji_cli.lib import _list_tasks
from koji_cli.commands import handle_list_tasks
from . import utils
class TestListTasks(unittest.TestCase):
def setUp(self):
@ -29,7 +33,7 @@ class TestListTasks(unittest.TestCase):
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')],
'decode': True,
'owner': 1,
}, {'order' : 'priority,create_time'})
}, {'order': 'priority,create_time'})
# invalid me
session.getLoggedInUser.return_value = None
@ -50,7 +54,7 @@ class TestListTasks(unittest.TestCase):
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')],
'decode': True,
'owner': 2,
}, {'order' : 'priority,create_time'})
}, {'order': 'priority,create_time'})
# invalid user
session.getUser.return_value = None
@ -66,7 +70,7 @@ class TestListTasks(unittest.TestCase):
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')],
'decode': True,
'arch': ['x86_64', 'i386'],
}, {'order' : 'priority,create_time'})
}, {'order': 'priority,create_time'})
# only method
session.listTasks.reset_mock()
@ -77,7 +81,7 @@ class TestListTasks(unittest.TestCase):
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')],
'decode': True,
'method': 'method',
}, {'order' : 'priority,create_time'})
}, {'order': 'priority,create_time'})
# only channel
session.listTasks.reset_mock()
@ -89,7 +93,7 @@ class TestListTasks(unittest.TestCase):
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')],
'decode': True,
'channel_id': 123,
}, {'order' : 'priority,create_time'})
}, {'order': 'priority,create_time'})
session.getChannel.assert_called_once_with('channel')
# invalid channel
@ -107,7 +111,7 @@ class TestListTasks(unittest.TestCase):
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')],
'decode': True,
'host_id': 234,
}, {'order' : 'priority,create_time'})
}, {'order': 'priority,create_time'})
session.getHost.assert_called_once_with('host')
# invalid host
@ -127,7 +131,7 @@ class TestListTasks(unittest.TestCase):
{
'children': [
{
'children': [ {'id': 3, 'parent': 2, 'sub': True}],
'children': [{'id': 3, 'parent': 2, 'sub': True}],
'id': 2,
'parent': 1,
'sub': True
@ -148,3 +152,110 @@ class TestListTasks(unittest.TestCase):
'sub': True}
])
class TestCliListTasks(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s list-tasks [options]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands._list_tasks')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_list_tasks(
self,
activate_session_mock,
list_tasks_mock,
stdout):
"""Test handle_list_tasks function"""
session = mock.MagicMock()
options = mock.MagicMock(quiet=False)
tasks = [
{
'id': 5,
'sub': True,
'state': 0,
'owner_name': 'kojiadmin',
'method': 'createrepo',
'parent': 3,
'priority': 14,
'arch': 'noarch'
},
{
'children':
[
{
'id': 5,
'sub': True,
'state': 0,
'owner_name': 'kojiadmin',
'method': 'createrepo',
'parent': 3,
'priority': 14,
'arch': 'noarch'
},
],
'id': 3,
'state': 0,
'owner_name': 'kojiadmin',
'method': 'newRepo',
'priority': 15,
'arch': 'noarch',
}
]
header = \
"ID Pri Owner State Arch Name\n"
task_output = \
"3 15 kojiadmin FREE noarch newRepo (noarch)\n" + \
"5 14 kojiadmin FREE noarch +createrepo (noarch)\n"
expected = self.format_error_message(
"This command takes no arguments")
# Case 1, argument error test.
self.assert_system_exit(
handle_list_tasks,
options,
session,
['test'],
stderr=expected,
activate_session=None)
# Case 2, no tasks
list_tasks_mock.return_value = None
handle_list_tasks(options, session, [])
self.assert_console_message(stdout, '(no tasks)\n')
# Case 3, show tasks with header
list_tasks_mock.return_value = tasks
handle_list_tasks(options, session, [])
self.assert_console_message(stdout, header + task_output)
# Case 4. show task without header
handle_list_tasks(options, session, ['--quiet'])
self.assert_console_message(stdout, task_output)
def test_handle_list_tasks_help(self):
self.assert_help(
handle_list_tasks,
"""Usage: %s list-tasks [options]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--mine Just print your tasks
--user=USER Only tasks for this user
--arch=ARCH Only tasks for this architecture
--method=METHOD Only tasks of this method
--channel=CHANNEL Only tasks in this channel
--host=HOST Only tasks for this host
--quiet Do not display the column headers
""" % self.progname)

368
tests/test_cli/test_mock_config.py Normal file
View file

@ -0,0 +1,368 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import anon_handle_mock_config
from . import utils
class TestMockConfig(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.common_args = [
'--distribution', 'fedora',
'--topdir', '/top-dir',
'--topurl', '/top-url',
'--yum-proxy', '/yum-proxy'
]
self.common_opts = {
'distribution': 'fedora',
'mockdir': '/var/lib/mock',
'topdir': '/top-dir',
'topurl': '/top-url',
'yum_proxy': '/yum-proxy',
}
self.mock_output = """# Auto-generated by the Koji build system
config_opts['chroot_setup_cmd'] = 'groupinstall build'
config_opts['use_host_resolv'] = False
config_opts['root'] = 'fedora26-build-repo_1'
config_opts['yum.conf'] = '[main]\ncachedir=/var/cache/yum\ndebuglevel=1\nlogfile=/var/log/yum.log\nreposdir=/dev/null\nretries=20\nobsoletes=1\ngpgcheck=0\nassumeyes=1\nkeepcache=1\ninstall_weak_deps=0\nstrict=1\n\n# repos\n\n[build]\nname=build\nbaseurl=https://fedora.local/kojifiles/repos/fedora26-build/1/x86_64\n'
config_opts['rpmbuild_timeout'] = 86400
config_opts['chroothome'] = '/builddir'
config_opts['target_arch'] = 'x86_64'
config_opts['basedir'] = '/var/lib/mock'
config_opts['plugin_conf']['yum_cache_enable'] = False
config_opts['plugin_conf']['root_cache_enable'] = False
config_opts['plugin_conf']['ccache_enable'] = False
config_opts['macros']['%_rpmfilename'] = '%%{NAME}-%%{VERSION}-%%{RELEASE}.%%{ARCH}.rpm'
config_opts['macros']['%_topdir'] = '/builddir/build'
config_opts['macros']['%packager'] = 'Koji'
config_opts['macros']['%_host'] = 'x86_64-koji-linux-gnu'
config_opts['macros']['%_host_cpu'] = 'x86_64'
config_opts['macros']['%vendor'] = 'Koji'
config_opts['macros']['%distribution'] = 'Koji Testing'
"""
self.error_format = """Usage: %s mock-config [options]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.genMockConfig')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_mock_config_buildroot_option(
self, activate_session_mock, gen_config_mock, stdout, stderr):
"""Test anon_handle_mock_config buildroot options"""
arguments = []
options = mock.MagicMock()
buildroot_info = {
'repo_id': 101,
'tag_name': 'tag_name',
'arch': 'x86_64'
}
# Mock out the xmlrpc server
session = mock.MagicMock()
session.getBuildroot.return_value = buildroot_info
# Mock config
gen_config_mock.return_value = self.mock_output
# buildroot check
arguments = ['--buildroot', 'root', self.progname]
expected = self.format_error_message("Buildroot id must be an integer")
self.assert_system_exit(
anon_handle_mock_config,
options,
session,
arguments,
stderr=expected)
arguments = self.common_args + ['--buildroot', '1',
'--name', self.progname]
opts = self.common_opts.copy()
opts.update({
'repoid': buildroot_info['repo_id'],
'tag_name': buildroot_info['tag_name']
})
anon_handle_mock_config(options, session, arguments)
self.assert_console_message(
stdout, "%s\n" % gen_config_mock.return_value)
gen_config_mock.assert_called_with(
self.progname, buildroot_info['arch'], **opts)
arguments = self.common_args + ['--buildroot', '1',
'--name', self.progname,
'--latest']
opts['repoid'] = 'latest'
anon_handle_mock_config(options, session, arguments)
self.assert_console_message(
stdout, "%s\n" % gen_config_mock.return_value)
gen_config_mock.assert_called_with(
self.progname, buildroot_info['arch'], **opts)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.genMockConfig')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_mock_config_task_option(
self, activate_session_mock, gen_config_mock, stdout, stderr):
"""Test anon_handle_mock_config task options"""
arguments = []
task_id = 1001
options = mock.MagicMock()
session = mock.MagicMock()
session.listBuildroots.return_value = ''
# Mock config
gen_config_mock.return_value = ''
arguments = ['--task', 'task']
expected = self.format_error_message("Task id must be an integer")
self.assert_system_exit(
anon_handle_mock_config,
options,
session,
arguments,
stderr=expected)
arguments = ['--task', str(task_id)]
expected = "No buildroots for task %s (or no such task)\n" % str(task_id)
self.assertEqual(1, anon_handle_mock_config(options, session, arguments))
self.assert_console_message(stdout, expected)
multi_broots = [
{'id': 1101, 'repo_id': 101, 'tag_name': 'tag_101', 'arch': 'x86_64'},
{'id': 1111, 'repo_id': 111, 'tag_name': 'tag_111', 'arch': 'x86_64'},
{'id': 1121, 'repo_id': 121, 'tag_name': 'tag_121', 'arch': 'x86_64'}
]
session.listBuildroots.return_value = multi_broots
anon_handle_mock_config(options, session, arguments)
expected = "Multiple buildroots found: %s" % [br['id'] for br in multi_broots]
self.assert_console_message(stdout, "%s\n\n" % expected)
opts = self.common_opts.copy()
opts.update({
'repoid': 'latest',
'tag_name': multi_broots[0]['tag_name']
})
arguments = self.common_args + ['--task', str(task_id),
'--name', self.progname,
'--latest']
session.listBuildroots.return_value = [multi_broots[0]]
gen_config_mock.return_value = self.mock_output
anon_handle_mock_config(options, session, arguments)
self.assert_console_message(
stdout, "%s\n" % gen_config_mock.return_value)
gen_config_mock.assert_called_with(
self.progname, multi_broots[0]['arch'], **opts)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.genMockConfig')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_mock_config_tag_option(
self, activate_session_mock, gen_config_mock, stdout, stderr):
"""Test anon_handle_mock_config with tag option"""
arguments = []
tag = 'tag'
tag = {'id': 201, 'name': 'tag', 'arch': 'x86_64'}
options = mock.MagicMock()
# Mock out the xmlrpc server
session = mock.MagicMock()
session.getTag.return_value = None
session.getBuildConfig.return_value = None
session.getRepo.return_value = None
arguments = ['--tag', tag['name']]
expected = "Please specify an arch\n"
self.assertEqual(1, anon_handle_mock_config(options, session, arguments))
self.assert_console_message(stdout, expected)
arguments = ['--tag', tag['name'], '--arch', tag['arch']]
expected = self.format_error_message("Invalid tag: %s" % tag['name'])
self.assert_system_exit(
anon_handle_mock_config,
options,
session,
arguments,
stderr=expected)
# return tag info
session.getTag.return_value = tag
expected = "Could not get config info for tag: %(name)s\n" % tag
self.assertEqual(1, anon_handle_mock_config(options, session, arguments))
self.assert_console_message(stdout, expected)
# return build config
session.getBuildConfig.return_value = {'id': 301}
expected = "Could not get a repo for tag: %(name)s\n" % tag
self.assertEqual(1, anon_handle_mock_config(options, session, arguments))
self.assert_console_message(stdout, expected)
# return repo
session.getRepo.return_value = {'id': 101}
gen_config_mock.return_value = self.mock_output
anon_handle_mock_config(options, session, arguments)
self.assert_console_message(
stdout, "%s\n" % gen_config_mock.return_value)
arguments = self.common_args + ['--tag', tag['name'],
'--arch', tag['arch'],
'--name', self.progname,
'--latest']
opts = self.common_opts.copy()
opts.update({
'repoid': 'latest',
'tag_name': tag['name'],
})
anon_handle_mock_config(options, session, arguments)
self.assert_console_message(
stdout, "%s\n" % gen_config_mock.return_value)
gen_config_mock.assert_called_with(
self.progname, tag['arch'], **opts)
@mock.patch('sys.stderr', new_callable=six.StringIO)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji.genMockConfig')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_mock_config_target_option(
self, activate_session_mock, gen_config_mock, stdout, stderr):
"""Test anon_handle_mock_config with target option"""
arguments = []
arch = "x86_64"
target = {'id': 1,
'name': 'target',
'dest_tag': 1,
'build_tag': 2,
'build_tag_name': 'target-build',
'dest_tag_name': 'target'}
options = mock.MagicMock()
# Mock out the xmlrpc server
session = mock.MagicMock()
session.getBuildTarget.return_value = None
session.getRepo.return_value = None
arguments = ['--target', target['name']]
expected = "Please specify an arch\n"
self.assertEqual(1, anon_handle_mock_config(options, session, arguments))
self.assert_console_message(stdout, expected)
arguments = ['--target', target['name'],
'--arch', arch]
expected = self.format_error_message(
"Invalid target: %s" % target['name'])
self.assert_system_exit(
anon_handle_mock_config,
options,
session,
arguments,
stderr=expected)
session.getBuildTarget.return_value = target
expected = "Could not get a repo for tag: %s\n" % target['build_tag_name']
self.assertEqual(1, anon_handle_mock_config(options, session, arguments))
self.assert_console_message(stdout, expected)
arguments = self.common_args + ['--target', target['name'],
'--arch', arch,
'--name', self.progname]
opts = self.common_opts.copy()
opts.update({
'repoid': 101,
'tag_name': target['build_tag_name']
})
session.getRepo.return_value = {'id': 101}
gen_config_mock.return_value = self.mock_output
anon_handle_mock_config(options, session, arguments)
self.assert_console_message(
stdout, "%s\n" % gen_config_mock.return_value)
gen_config_mock.assert_called_with(
self.progname, arch, **opts)
# --latest and -o (output) test
opts['repoid'] = 'latest'
arguments.extend(['--latest', '-o', '/tmp/mock.out'])
with mock.patch('koji_cli.commands.open', create=True) as openf_mock:
anon_handle_mock_config(options, session, arguments)
openf_mock.assert_called_with('/tmp/mock.out', 'w')
handle = openf_mock()
handle.write.assert_called_once_with(self.mock_output)
gen_config_mock.assert_called_with(
self.progname, arch, **opts)
@mock.patch('sys.stderr', new_callable=six.StringIO)
def test_handle_mock_config_errors(self, stderr):
"""Test anon_handle_mock_config general error messages"""
arguments = []
options = mock.MagicMock()
# Mock out the xmlrpc server
session = mock.MagicMock()
# Run it and check immediate output
# argument is empty
expected = self.format_error_message(
"Please specify one of: --tag, --target, --task, --buildroot")
self.assert_system_exit(
anon_handle_mock_config,
options,
session,
arguments,
stderr=expected)
# name is specified twice case
arguments = [self.progname, '--name', 'name']
expected = self.format_error_message(
"Name already specified via option")
self.assert_system_exit(
anon_handle_mock_config,
options,
session,
arguments,
stderr=expected)
def test_handle_mock_config_help(self):
"""Test anon_handle_mock_config help message full output"""
self.assert_help(
anon_handle_mock_config,
"""Usage: %s mock-config [options]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
-a ARCH, --arch=ARCH Specify the arch
-n NAME, --name=NAME Specify the name for the buildroot
--tag=TAG Create a mock config for a tag
--target=TARGET Create a mock config for a build target
--task=TASK Duplicate the mock config of a previous task
--latest use the latest redirect url
--buildroot=BUILDROOT
Duplicate the mock config for the specified buildroot
id
--mockdir=DIR Specify mockdir
--topdir=DIR Specify topdir
--topurl=URL URL under which Koji files are accessible
--distribution=DISTRIBUTION
Change the distribution macro
--yum-proxy=YUM_PROXY
Specify a yum proxy
-o FILE Output to a file
""" % self.progname)
if __name__ == '__main__':
unittest.main()

198
tests/test_cli/test_restart_host.py Normal file
View file

@ -0,0 +1,198 @@
from __future__ import absolute_import
import mock
import six
import unittest
import koji
from koji_cli.commands import handle_restart_hosts
from . import utils
class TestRestartHosts(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.task_id = 101
@mock.patch('koji_cli.commands.watch_tasks')
@mock.patch('koji_cli.commands._running_in_bg')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_restart_hosts_force_options(
self, activate_session_mock, running_in_bg_mock, watch_tasks_mock):
"""Test %s function with --force option""" % handle_restart_hosts.__name__
arguments = ['--force']
options = mock.MagicMock(quiet=None, poll_interval=3)
session = mock.MagicMock()
# set running in foreground
running_in_bg_mock.return_value = False
session.getHost.return_value = None
session.restartHosts.return_value = self.task_id
session.logout.return_value = None
# has other restart tasks are running case
session.listTasks.return_value = [{'id': 1}, {'id': 2}, {'id': 3}]
handle_restart_hosts(options, session, arguments)
activate_session_mock.assert_called_once()
session.listTasks.assert_not_called()
session.restartHosts.assert_called_with()
session.logout.assert_called_once()
watch_tasks_mock.assert_called_with(
session, [self.task_id], quiet=None, poll_interval=3)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.watch_tasks')
@mock.patch('koji_cli.commands._running_in_bg')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_restart_hosts_has_other_tasks(
self,
activate_session_mock,
running_in_bg_mock,
watch_tasks_mock,
stdout):
"""Test %s function when there has other restart tasks exist""" % handle_restart_hosts.__name__
options = mock.MagicMock()
session = mock.MagicMock()
# set running in foreground
running_in_bg_mock.return_value = False
session.getHost.return_value = None
session.restartHosts.return_value = True
session.logout.return_value = None
#
# session.listTasks returns:
#
# [{'weight': 0.1,
# 'awaited': None,
# 'completion_time': None,
# 'create_time': '2017-11-02 18:30:08.933753',
# 'result': None,
# 'owner': 1,
# 'id': 11,
# 'state': 1,
# 'label': None,
# 'priority': 5,
# 'waiting': True,
# 'completion_ts': None,
# 'method': 'restartHosts',
# 'owner_name': 'kojiadmin',
# 'parent': None,
# 'start_time': '2017-11-02 18:30:28.028843',
# 'start_ts': 1509647428.02884,
# 'create_ts': 1509647408.93375,
# 'host_id': 1, 'arch': 'noarch',
# 'request': "<?xml version='1.0'?>\n<methodCall>\n<methodName>restartHosts</methodName>\n<params>\n</params>\n</methodCall>\n",
# 'channel_id': 1,
# 'owner_type': 0}]
#
# has other restart tasks are running case
session.listTasks.return_value = [{'id': 1}, {'id': 2}, {'id': 3}]
self.assertEqual(1, handle_restart_hosts(options, session, []))
activate_session_mock.assert_called_once()
query_opt = {
'method': 'restartHosts',
'state': [koji.TASK_STATES[s] for s in ('FREE', 'OPEN', 'ASSIGNED')]
}
session.listTasks.assert_called_with(query_opt)
session.restartHosts.assert_not_called()
session.logout.assert_not_called()
expect = "Found other restartHosts tasks running.\n"
expect += "Task ids: %r\n" % \
[t['id'] for t in session.listTasks.return_value]
expect += "Use --force to run anyway\n"
self.assert_console_message(stdout, expect)
@mock.patch('koji_cli.commands.watch_tasks')
@mock.patch('koji_cli.commands._running_in_bg')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_restart_hosts_wait_option(
self, activate_session_mock, running_in_bg_mock, watch_tasks_mock):
"""Test %s function with --force option""" % handle_restart_hosts.__name__
arguments = ['--wait']
options = mock.MagicMock(quiet=None, poll_interval=3)
session = mock.MagicMock()
# --wait is specified, running_in_bg() should not matter.
running_in_bg_mock.return_value = True
session.getHost.return_value = None
session.restartHosts.return_value = self.task_id
session.logout.return_value = None
# has other restart tasks are running case
session.listTasks.return_value = []
handle_restart_hosts(options, session, arguments)
activate_session_mock.assert_called_once()
session.listTasks.assert_called_once()
session.restartHosts.assert_called_with()
session.logout.assert_called_once()
watch_tasks_mock.assert_called_with(
session, [self.task_id], quiet=None, poll_interval=3)
@mock.patch('koji_cli.commands.watch_tasks')
@mock.patch('koji_cli.commands._running_in_bg')
@mock.patch('koji_cli.commands.activate_session')
def test_handle_restart_hosts_other_options(
self, activate_session_mock, running_in_bg_mock, watch_tasks_mock):
"""Test %s function with --force option""" % handle_restart_hosts.__name__
arguments = ['--nowait',
'--channel', 'createrepo',
'--arch', 'x86_64',
'--timeout', '10']
options = mock.MagicMock(quiet=None, poll_interval=3)
session = mock.MagicMock()
# --no-wait is specified, running_in_bg() should not matter.
running_in_bg_mock.return_value = True
session.getHost.return_value = None
session.restartHosts.return_value = 101
session.logout.return_value = None
# has other restart tasks are running case
session.listTasks.return_value = []
handle_restart_hosts(options, session, arguments)
activate_session_mock.assert_called_once()
session.listTasks.assert_called_once()
session.restartHosts.assert_called_with(
options={'arches': ['x86_64'], 'timeout': 10, 'channel': 'createrepo'})
session.logout.assert_not_called()
watch_tasks_mock.assert_not_called()
def test_handle_restart_hosts_help(self):
self.assert_help(
handle_restart_hosts,
"""Usage: %s restart-hosts [options]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--wait Wait on the task, even if running in the background
--nowait Don't wait on task
--quiet Do not print the task information
--force Ignore checks and force operation
--channel=CHANNEL Only hosts in this channel
-a ARCH, --arch=ARCH Limit to hosts of this architecture (can be given
multiple times)
--timeout=N Time out after N seconds
""" % self.progname)
if __name__ == '__main__':
unittest.main()

92
tests/test_cli/test_search.py Normal file
View file

@ -0,0 +1,92 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import anon_handle_search
from . import utils
class TestSearch(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s search [options] search_type pattern
Available search types: package, build, tag, target, user, host, rpm, maven, win
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
def test_anon_handle_search(
self,
stdout):
"""Test anon_handle_search function"""
session = mock.MagicMock()
options = mock.MagicMock()
s_type, s_pattern = 'build', 'fedora'
arguments = [s_type, s_pattern]
search_results = [
{'id': 166, 'name': 'f25'},
{'id': 177, 'name': 'f26'},
{'id': 202, 'name': 'f27'}
]
session.search.return_value = search_results
expected = ''.join('%s\n' % x['name'] for x in search_results)
# Case 1. normal search
anon_handle_search(options, session, arguments)
self.assert_console_message(stdout, expected)
session.search.assert_called_with(s_pattern, s_type, 'glob')
# Case 2. exact match
anon_handle_search(options, session, arguments + ['--exact'])
self.assert_console_message(stdout, expected)
session.search.assert_called_with(s_pattern, s_type, 'exact')
# Case 3. regex match
anon_handle_search(options, session, arguments + ['-r'])
self.assert_console_message(stdout, expected)
session.search.assert_called_with(s_pattern, s_type, 'regexp')
def test_anon_handle_search_argument_error(self):
"""Test anon_handle_search function with argument error"""
s_type, s_patt = 'unknown', 'unknown'
cases = [
{'argument': [], 'error': 'Please specify search type'},
{'argument': [s_type], 'error': 'Please specify search pattern'},
{'argument': [s_type, s_patt],
'error': 'Unknown search type: %s' % s_type}
]
for case in cases:
expected = self.format_error_message(case['error'])
self.assert_system_exit(
anon_handle_search,
mock.MagicMock(),
mock.MagicMock(),
case['argument'],
stderr=expected,
activate_session=None)
def test_anon_handle_search_help(self):
self.assert_help(
anon_handle_search,
"""Usage: %s search [options] search_type pattern
Available search types: package, build, tag, target, user, host, rpm, maven, win
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
-r, --regex treat pattern as regex
--exact exact matches only
""" % self.progname)
if __name__ == '__main__':
unittest.main()

66
tests/test_cli/test_set_pkg_arches.py Normal file
View file

@ -0,0 +1,66 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import handle_set_pkg_arches
from . import utils
class TestSetPkgArches(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s set-pkg-arches [options] arches tag package [package2 ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_set_pkg_arches(
self,
activate_session_mock,
stdout):
"""Test handle_set_pkg_arches function"""
session = mock.MagicMock()
options = mock.MagicMock()
arguments = ['x86_64', 'tag', '--force', 'bash', 'less', 'sed']
expected = self.format_error_message(
"Please specify an archlist, a tag, and at least one package")
# Case 1. argument error
self.assert_system_exit(
handle_set_pkg_arches,
options,
session,
[],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
# Case 2. run set arch to x86_64
calls = [mock.call('tag', pkg, 'x86_64', force=True) for pkg in arguments[3:]]
handle_set_pkg_arches(options, session, arguments)
activate_session_mock.assert_called_with(session, options)
session.packageListSetArches.assert_has_calls(calls)
self.assert_console_message(stdout, '')
def test_handle_set_pkg_arches_help(self):
self.assert_help(
handle_set_pkg_arches,
"""Usage: %s set-pkg-arches [options] arches tag package [package2 ...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--force Force operation
""" % self.progname)
if __name__ == '__main__':
unittest.main()

66
tests/test_cli/test_set_pkg_owner.py Normal file
View file

@ -0,0 +1,66 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import handle_set_pkg_owner
from . import utils
class TestSetPkgOwner(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s set-pkg-owner [options] owner tag package [package2 ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_set_pkg_owner(
self,
activate_session_mock,
stdout):
"""Test handle_set_pkg_owner function"""
session = mock.MagicMock()
options = mock.MagicMock()
arguments = ['owner', 'tag', '--force', 'bash', 'less', 'sed']
expected = self.format_error_message(
"Please specify an owner, a tag, and at least one package")
# Case 1. argument error
self.assert_system_exit(
handle_set_pkg_owner,
options,
session,
[],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
# Case 2. run set owner
calls = [mock.call('tag', pkg, 'owner', force=True) for pkg in arguments[3:]]
handle_set_pkg_owner(options, session, arguments)
activate_session_mock.assert_called_with(session, options)
session.packageListSetOwner.assert_has_calls(calls)
self.assert_console_message(stdout, '')
def test_handle_set_pkg_owner_help(self):
self.assert_help(
handle_set_pkg_owner,
"""Usage: %s set-pkg-owner [options] owner tag package [package2 ...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--force Force operation
""" % self.progname)
if __name__ == '__main__':
unittest.main()

93
tests/test_cli/test_set_task_priority.py Normal file
View file

@ -0,0 +1,93 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import handle_set_task_priority
from . import utils
class TestSetTaskPriority(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s set-task-priority [options] --priority=<priority> <task-id> [task-id]...
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_set_task_priority(
self,
activate_session_mock,
stdout):
"""Test handle_set_task_priority function"""
session = mock.MagicMock()
options = mock.MagicMock()
arguments = ['--priority', '10', '1', '11', '121', '1331']
# Case 1. no task id error
expected = self.format_error_message(
"You must specify at least one task id")
self.assert_system_exit(
handle_set_task_priority,
options,
session,
[],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
# Case 2. no --priority is specified
expected = self.format_error_message(
"You must specify --priority")
self.assert_system_exit(
handle_set_task_priority,
options,
session,
['1'],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
# Case 3 . Wrong task id (not integer format)
for case in ['1.0', '0.1', 'abc']:
expected = self.format_error_message(
"Task numbers must be integers")
self.assert_system_exit(
handle_set_task_priority,
options,
session,
[case, '--priority', '10'],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
calls = [mock.call(int(tid), 10, False) for tid in arguments[2:]]
handle_set_task_priority(options, session, arguments)
activate_session_mock.assert_called_with(session, options)
session.setTaskPriority.assert_has_calls(calls)
self.assert_console_message(stdout, '')
def test_handle_set_task_priority_help(self):
self.assert_help(
handle_set_task_priority,
"""Usage: %s set-task-priority [options] --priority=<priority> <task-id> [task-id]...
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--priority=PRIORITY New priority
--recurse Change priority of child tasks as well
""" % self.progname)
if __name__ == '__main__':
unittest.main()

648
tests/test_cli/test_taskinfo.py Normal file
View file

@ -0,0 +1,648 @@
from __future__ import absolute_import, print_function
import mock
import six
import unittest
import koji
import collections
import time
from koji_cli.commands import anon_handle_taskinfo, \
_printTaskInfo, _parseTaskParams
from . import utils
class TestParseTaskParams(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.session = mock.MagicMock()
self.build_templ = {
'package_name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'epoch': None,
'nvr': 'bash-4.4.12-5.fc26',
'build_id': 1,
}
def __run_parseTask_test(self, method, params, expect):
self.session.getTaskRequest.return_value = params
lines = _parseTaskParams(self.session, method, 1, '/mnt/koji')
self.assertEquals(lines, expect)
def test_error_with_param(self):
params = []
expect = ['Unable to parse task parameters']
with mock.patch('koji_cli.commands.logging.getLogger', create=True) as get_logger_mock:
h = get_logger_mock()
h.isEnabledFor.return_value = True
self.__run_parseTask_test('buildSRPMFromCVS', params, expect)
logger = get_logger_mock()
logger.isEnabledFor.assert_called_once()
logger.debug.assert_called_once()
def test_buildSRPMFrom(self):
# from CVS case
params = ['cvs.src.org']
expect = ["CVS URL: %s" % params[0]]
self.__run_parseTask_test('buildSRPMFromCVS', params, expect)
# from SCM case
params = ['git.github.com']
expect = ["SCM URL: %s" % params[0]]
self.__run_parseTask_test('buildSRPMFromSCM', params, expect)
def test_buildArch(self):
name = 'TEST'
topdir = '/mnt/koji'
self.session.getTag.return_value = {'name': name}
params = ['path/to/bash-4.4.12-5.fc26.src.rpm', 2, 'x86_64', True, {'repo_id': 1}]
expect = ["SRPM: %s/work/%s" % (topdir, params[0])]
expect.append("Build Tag: %s" % name)
expect.append("Build Arch: %s" % params[2])
expect.append("SRPM Kept: %r" % params[3])
expect.append("Options:")
expect.append(" repo_id: 1")
self.__run_parseTask_test('buildArch', params, expect)
def test_tagBuild(self):
params = [1, 1, False, None, True]
self.session.getTag.return_value = {'name': 'fedora'}
self.session.getBuild.return_value = self.build_templ
expect = ["Destination Tag: fedora"]
expect.append("Build: bash-4.4.12-5.fc26")
self.__run_parseTask_test('tagBuild', params, expect)
def test_buildNotification(self):
params = [['r1', 'r2'],
self.build_templ,
{'name': 'fedora'},
'fedoraproject.org']
expect = ["Recipients: %s" % (", ".join(params[0]))]
expect.append("Build: %s" % self.build_templ['nvr'])
expect.append("Build Target: %s" % params[2]['name'])
expect.append("Web URL: %s" % params[3])
self.__run_parseTask_test('buildNotification', params, expect)
def test_build(self):
params = ['path/to/bash-4.4.12-5.fc26.src.rpm',
'fedora26-build',
{'build-test': True}]
expect = ["Source: %s" % params[0]]
expect.append("Build Target: %s" % params[1])
expect.append("Options:")
expect.append(" build-test: True")
self.__run_parseTask_test('build', params, expect)
def test_maven(self):
params = ['scm.maven.org', 'maven-target', {'maven-test': True}]
expect = ["SCM URL: %s" % params[0]]
expect.append("Build Target: %s" % params[1])
expect.append("Options:")
expect.append(" maven-test: True")
self.__run_parseTask_test('maven', params, expect)
def test_buildMaven(self):
params = ['scm.maven.org', {'name': 'maven-tag'}, {'build-test': True}]
expect = ["SCM URL: %s" % params[0]]
expect.append("Build Tag: %s" % params[1]['name'])
expect.append("Options:")
expect.append(" build-test: True")
self.__run_parseTask_test('buildMaven', params, expect)
def test_wrapperRPM(self):
target = 'test-target'
params = ['http://path.to/pkg.spec', {'name': 'build-tag'},
self.build_templ,
{
'id': 1,
'method': 'wrapperRPM',
'arch': 'x86_64',
'request': [1, {'name': target}, self.build_templ]
},
{'wrapRPM-test': True}]
expect = ["Spec File URL: %s" % params[0]]
expect.append("Build Tag: %s" % params[1]['name'])
expect.append("Build: %s" % self.build_templ['nvr'])
task_info = "wrapperRPM (%s, %s)" % (target, self.build_templ['nvr'])
expect.append("Task: %s %s" % (params[3]['id'], task_info))
expect.append("Options:")
expect.append(" wrapRPM-test: True")
self.__run_parseTask_test('wrapperRPM', params, expect)
def test_chainmaven(self):
params = [{
'maven-pkg-1': {'build-opt': '--test'},
'maven-pkg-2': {'build-opt': '-O2'},
},
'build-target',
{'chainmaven-test': True}]
expect = ["Builds:"]
for pkg, opt in params[0].items():
expect.append(" %s" % pkg)
for k, v in opt.items():
expect.append(" %s: %s" % (k, v))
expect.append("Build Target: %s" % params[1])
expect.append("Options:")
expect.append(" chainmaven-test: True")
self.__run_parseTask_test('chainmaven', params, expect)
def test_winbuild(self):
params = ['vm-builder', 'github.com', 'target',
{'winver': 10}]
expect = ["VM: %s" % params[0]]
expect.append("SCM URL: %s" % params[1])
expect.append("Build Target: %s" % params[2])
expect.append("Options:")
expect.append(" winver: 10")
self.__run_parseTask_test('winbuild', params, expect)
def test_vmExec(self):
params = ['vm-name',
['x86_64', {'cpu': 'GenuineIntel'}],
{'optimize': '-O3'}]
expect = ["VM: %s" % params[0]]
expect.append("Exec Params:")
expect.append(" x86_64")
expect.append(" cpu: GenuineIntel")
expect.append("Options:")
expect.append(" optimize: -O3")
self.__run_parseTask_test('vmExec', params, expect)
def test_createXXX(self):
params = ['name', '1.0', '1', 'x86_64', 'target_info',
'build-tag', 'Repo', 'kickstart.ks', {}]
fields = ['Name', 'Version', 'Release', 'Arch', 'Target Info',
'Build Tag', 'Repo', 'Kickstart File']
template = list()
for n, v in zip(fields, params):
template.append("%s: %s" % (n, v))
for method in ('createLiveCD', 'createAppliance', 'createLiveMedia'):
params[-1] = {"extra": "test-%s" % method}
expect = list(template)
expect.append("Options:")
expect.append(" extra: test-%s" % method)
self.__run_parseTask_test(method, params, expect)
def test_appliance_livecd_livemedia(self):
params = ['name', '1.0', 'x86_64, ppc64', 'target_info',
'kickstart.ks', {}]
fields = ['Name', 'Version', 'Arches', 'Target', 'Kickstart']
template = list()
for n, v in zip(fields, params):
template.append("%s: %s" % (n, v))
for method in ('appliance', 'livecd', 'livemedia'):
params[-1] = {"extra": "test-%s" % method}
expect = list(template)
expect.append("Options:")
expect.append(" extra: test-%s" % method)
self.__run_parseTask_test(method, params, expect)
def test_newRepo(self):
params = [0]
self.session.getTag.return_value = {'name': 'f26'}
expect = ['Tag: f26']
self.__run_parseTask_test('newRepo', params, expect)
def test_prepRepo(self):
params = [{'name': 'f26'}]
expect = ['Tag: f26']
self.__run_parseTask_test('prepRepo', params, expect)
def test_createRepo(self):
params = [1, 'x86_64', {'id': 1, 'creation_time': '1970-1-1 0:0:0'},
[{'external_repo_name': 'fedoraproject.net'},
{'external_repo_name': 'centos.org'}]]
expect = ["Repo ID: %i" % params[0]]
expect.append("Arch: %s" % params[1])
expect.append("Old Repo ID: %i" % params[2]['id'])
expect.append("Old Repo Creation: Thu, 01 Jan 1970")
expect.append("External Repos: %s" % ', '.join(
[ext['external_repo_name'] for ext in params[3]]))
with mock.patch('koji.formatTimeLong',
return_value='Thu, 01 Jan 1970'):
self.__run_parseTask_test('createrepo', params, expect)
def test_tagNotification(self):
params = [['kojiadmin', 'user'], # recipients
True, # successful
1, # dest tag id
2, # src tag id
3, # build id
0, # user id
False, # Ignoresuccess
'no error'] # Failure message
destTag, srcTag = {'name': 'dest-tag-1'}, {'name': 'src-tag-2'}
user = {'name': 'kojiadmin'}
self.session.getTag.side_effect = [destTag, srcTag]
self.session.getBuild.return_value = self.build_templ
self.session.getUser.return_value = user
expect = ["Recipients: %s" % ", ".join(params[0])]
expect.append("Successful?: %s" % (params[1] and 'yes' or 'no'))
expect.append("Tagged Into: %s" % destTag['name'])
expect.append("Moved From: %s" % srcTag['name'])
expect.append("Build: %s" % self.build_templ['nvr'])
expect.append("Tagged By: %s" % user['name'])
expect.append("Ignore Success?: %s" % (params[6] and 'yes' or 'no'))
expect.append("Failure Message: %s" % params[7])
self.__run_parseTask_test('tagNotification', params, expect)
def test_dependantTask(self):
params = [
[1, 2, 3, 4], # dependant task ids
[
['buildSRPMFromSCM', ['param1', 'param2'], {'scm': 'github.com'}],
['build', ['param1', 'param2'], {'arch': 'x86_64'}],
['tagBuild', ['tagname', 'param2'], {}]
]
]
expect = ["Dependant Tasks: %s" % ", ".join([str(dep) for dep in params[0]])]
expect.append("Subtasks:")
for subtask in params[1]:
expect.append(" Method: %s" % subtask[0])
expect.append(" Parameters: %s" % ", ".join(
[str(subparam) for subparam in subtask[1]]))
if subtask[2]:
expect.append(' Options:')
for k, v in subtask[2].items():
expect.append(' %s: %s' % (k, v))
expect.append('')
self.__run_parseTask_test('dependantTask', params, expect)
def test_chainbuild(self):
params = [[
['base-grp', 'desktop-grp'],
['base-grp', 'devel-grp'],
],
'f26',
{'extra': 'f26-pre-release'}]
expect = ["Build Groups:"]
for i, grp in enumerate(params[0]):
expect.append(' %i: %s' % (i+1, ', '.join(grp)))
expect.append("Build Target: %s" % params[1])
expect.append("Options:")
expect.append(" extra: f26-pre-release")
self.__run_parseTask_test('chainbuild', params, expect)
def test_waitrepo(self):
params = ['build-taget', True, ['bash-4.4.12-5.fc26']]
expect = ["Build Target: %s" % params[0]]
expect.append("Newer Than: %s" % params[1])
expect.append("NVRs: %s" % ', '.join(params[2]))
self.__run_parseTask_test('waitrepo', params, expect)
class TestPrintTaskInfo(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.host_info = {
"comment": None,
"arches": "x86_64",
"task_load": 0.0,
"capacity": 2.0,
"name": "kojibuilder",
"ready": True,
"user_id": 3,
"enabled": True,
"id": 1,
"description": None
}
self.task_info_templ = {
'weight': 0.1,
'parent': None,
'create_ts': 1000,
'start_ts': 2000,
'completion_ts': 3000,
'state': 2,
'awaited': None,
'label': None,
'priority': 15,
'channel_id': 2,
'waiting': False,
'id': 1,
'owner': 1,
'host_id': 1,
'arch': 'noarch',
'method': 'newRepo'
}
self.tag_info = {
'maven_support': False,
'locked': False,
'name': 'fedora26-build',
'extra': {},
'perm': None,
'id': 2,
'arches': 'x86_64',
'maven_include_all': False,
'perm_id': None
}
self.user_info = {
'status': 0,
'usertype': 0,
'id': 1,
'name': 'kojiadmin',
'krb_principal': None
}
@mock.patch('koji_cli.commands.list_task_output_all_volumes')
def test_printTaskInfo_create_repo(self, list_task_output_mock):
session = mock.MagicMock()
parent = self.task_info_templ.copy()
parent.update({
'method': 'newRepo',
'host_id': None,
})
children = self.task_info_templ.copy()
children.update({
'id': 2,
'parent': 1,
'request': [1, 'x86_64', None],
'label': 'x86_64',
'channel_id': 2,
'host_id': None,
'method': 'createrepo'
})
session.getTaskInfo.side_effect = [parent, children]
session.listBuildroots.return_value = {}
session.listBuilds.return_value = {}
session.getTag.return_value = self.tag_info
session.getUser.return_value = self.user_info
session.getTaskRequest.return_value = [1, 'x86_64', None]
session.getTaskChildren.side_effect = [[children], []]
list_task_output_mock.side_effect = [
{},
{
'mergerepos.log': ['DEFAULT'],
'createrepo.log': ['DEFAULT']
}
]
expected = """\
Task: 1
Type: newRepo
Request Parameters:
Tag: fedora26-build
Owner: kojiadmin
State: closed
Created: Thu Jan 1 00:16:40 1970
Started: Thu Jan 1 00:33:20 1970
Finished: Thu Jan 1 00:50:00 1970
Task: 2
Type: createrepo
Request Parameters:
Repo ID: 1
Arch: x86_64
Owner: kojiadmin
State: closed
Created: Thu Jan 1 00:16:40 1970
Started: Thu Jan 1 00:33:20 1970
Finished: Thu Jan 1 00:50:00 1970
Log Files:
/mnt/koji/work/tasks/2/2/mergerepos.log
/mnt/koji/work/tasks/2/2/createrepo.log
"""
with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout, \
mock.patch('time.localtime', new=time.gmtime):
_printTaskInfo(session, 1, '/mnt/koji')
self.assert_console_message(stdout, expected)
@mock.patch('koji_cli.commands.list_task_output_all_volumes')
def test_printTaskInfo_build_srpm(self, list_task_output_mock):
session = mock.MagicMock()
parent = self.task_info_templ.copy()
parent.update({
'method': 'build',
})
child_build = parent.copy()
child_build.update({
'id': 2,
'parent': 1,
'request': ['path/to/bash.src.rpm', 2, 'x86_64', True, {'repo_id': 1}],
'method': 'buildArch',
})
child_tag = parent.copy()
child_tag.update({
'id': 3,
'parent': 1,
'request': [1, 1, False, None, True],
'arch': 'noarch',
'method': 'tagBuild'
})
buildroot_info = [{
'id': 1,
'repo_id': 1,
'tag_name': 'fedora26-build',
'host_name': 'kojibuilder',
}]
build_info = [{
'package_name': 'bash',
'version': '4.4.12',
'release': '5.fc26',
'epoch': None,
'nvr': 'bash-4.4.12-5.fc26',
'build_id': 1,
}]
files = {
'bash-debuginfo-4.4.12-5.fc26.x86_64.rpm': ['DEFAULT'],
'hw_info.log': ['DEFAULT'],
'build.log': ['DEFAULT'],
'bash-4.4.12-5.fc26.src.rpm': ['DEFAULT'],
'root.log': ['DEFAULT'],
'state.log': ['DEFAULT'],
'mock_output.log': ['DEFAULT'],
'bash-4.4.12-5.fc26.x86_64.rpm': ['DEFAULT'],
'installed_pkgs.log': ['DEFAULT'],
'bash-doc-4.4.12-5.fc26.x86_64.rpm': ['DEFAULT']
}
# need ordered dict to get same results
files = collections.OrderedDict(sorted(files.items(),
key=lambda t: t[0]))
list_task_output_mock.side_effect = [[], files, {}]
session.getTaskInfo.side_effect = [parent, child_build, child_tag]
session.listBuildroots.side_effect = [[], buildroot_info, []]
session.listBuilds.side_effect = [build_info, [], []]
session.getTag.return_value = self.tag_info
session.getUser.return_value = self.user_info
session.getBuild.return_value = build_info[0]
session.getHost.return_value = self.host_info
session.getTaskRequest.side_effect = [
['path/to/bash-4.4.12-5.fc26.src.rpm', 'fedora26-build', {}],
['path/to/bash-4.4.12-5.fc26.src.rpm', 2, 'x86_64', True, {'repo_id': 1}],
[1, 1, False, None, True]
]
session.getTaskChildren.side_effect = [[child_build, child_tag], [], []]
expected = """\
Task: 1
Type: build
Request Parameters:
Source: path/to/bash-4.4.12-5.fc26.src.rpm
Build Target: fedora26-build
Owner: kojiadmin
State: closed
Created: Thu Jan 1 00:16:40 1970
Started: Thu Jan 1 00:33:20 1970
Finished: Thu Jan 1 00:50:00 1970
Host: kojibuilder
Build: bash-4.4.12-5.fc26 (1)
Task: 2
Type: buildArch
Request Parameters:
SRPM: /mnt/koji/work/path/to/bash-4.4.12-5.fc26.src.rpm
Build Tag: fedora26-build
Build Arch: x86_64
SRPM Kept: True
Options:
repo_id: 1
Owner: kojiadmin
State: closed
Created: Thu Jan 1 00:16:40 1970
Started: Thu Jan 1 00:33:20 1970
Finished: Thu Jan 1 00:50:00 1970
Host: kojibuilder
Buildroots:
/var/lib/mock/fedora26-build-1-1/
Log Files:
/mnt/koji/work/tasks/2/2/build.log
/mnt/koji/work/tasks/2/2/hw_info.log
/mnt/koji/work/tasks/2/2/installed_pkgs.log
/mnt/koji/work/tasks/2/2/mock_output.log
/mnt/koji/work/tasks/2/2/root.log
/mnt/koji/work/tasks/2/2/state.log
Output:
/mnt/koji/work/tasks/2/2/bash-4.4.12-5.fc26.src.rpm
/mnt/koji/work/tasks/2/2/bash-4.4.12-5.fc26.x86_64.rpm
/mnt/koji/work/tasks/2/2/bash-debuginfo-4.4.12-5.fc26.x86_64.rpm
/mnt/koji/work/tasks/2/2/bash-doc-4.4.12-5.fc26.x86_64.rpm
Task: 3
Type: tagBuild
Request Parameters:
Destination Tag: fedora26-build
Build: bash-4.4.12-5.fc26
Owner: kojiadmin
State: closed
Created: Thu Jan 1 00:16:40 1970
Started: Thu Jan 1 00:33:20 1970
Finished: Thu Jan 1 00:50:00 1970
Host: kojibuilder
"""
with mock.patch('sys.stdout', new_callable=six.StringIO) as stdout, \
mock.patch('time.localtime', new=time.gmtime):
_printTaskInfo(session, 1, '/mnt/koji')
self.assert_console_message(stdout, expected)
def test_printTaskInfo_no_task(self):
"""Test _printTaskInfo with no task found"""
session = mock.MagicMock()
task_id = 1
session.getTaskInfo.return_value = None
with self.assertRaises(koji.GenericError) as cm:
_printTaskInfo(session, task_id, '/')
self.assertEquals(str(cm.exception), "No such task: %d" % task_id)
class TestTaskInfo(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s taskinfo [options] taskID [taskID...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_anon_handle_taskinfo(
self,
activate_session_mock,
stdout):
"""Test anon_handle_taskinfo function"""
session = mock.MagicMock()
options = mock.MagicMock()
# Case 1. no task id error
expected = self.format_error_message(
"You must specify at least one task ID")
self.assert_system_exit(
anon_handle_taskinfo,
options,
session,
[],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
# Case 2. show task info
task_output = """Task: 1
Type: newRepo
Owner: kojiadmin
State: closed
Created: Thu Nov 16 17:34:29 2017
Started: Thu Nov 16 17:51:07 2017
Finished: Thu Nov 16 17:54:55 2017
Host: kojibuilder
"""
def print_task(*args, **kwargs):
print(task_output, end='')
with mock.patch('koji_cli.commands._printTaskInfo', new=print_task):
anon_handle_taskinfo(options, session, ['1'])
self.assert_console_message(stdout, task_output)
def test_anon_handle_taskinfo_help(self):
self.assert_help(
anon_handle_taskinfo,
"""Usage: %s taskinfo [options] taskID [taskID...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
-r, --recurse Show children of this task as well
-v, --verbose Be verbose
""" % self.progname)
if __name__ == '__main__':
unittest.main()

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

@ -0,0 +1,65 @@
from __future__ import absolute_import
import mock
import six
import unittest
from koji_cli.commands import handle_unblock_pkg
from . import utils
class TestUnblockPkg(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def setUp(self):
self.error_format = """Usage: %s unblock-pkg [options] tag package [package2 ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_unblock_pkg(
self,
activate_session_mock,
stdout):
"""Test handle_unblock_pkg function"""
session = mock.MagicMock()
options = mock.MagicMock()
arguments = ['tag', 'bash', 'less', 'sed']
expected = self.format_error_message(
"Please specify a tag and at least one package")
# Case 1. argument error
self.assert_system_exit(
handle_unblock_pkg,
options,
session,
[],
stderr=expected,
activate_session=None)
activate_session_mock.assert_not_called()
# Case 2. run unlock
calls = [mock.call('tag', pkg) for pkg in arguments[1:]]
handle_unblock_pkg(options, session, arguments)
activate_session_mock.assert_called_with(session, options)
session.packageListUnblock.assert_has_calls(calls)
self.assert_console_message(stdout, '')
def test_handle_unblock_pkg_help(self):
self.assert_help(
handle_unblock_pkg,
"""Usage: %s unblock-pkg [options] tag package [package2 ...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
""" % self.progname)
if __name__ == '__main__':
unittest.main()

249
tests/test_cli/test_write_signed_rpm.py Normal file
View file

@ -0,0 +1,249 @@
from __future__ import absolute_import
import mock
import six
import unittest
import koji
import hashlib
from mock import call
from koji_cli.commands import handle_write_signed_rpm
from . import utils
import os
QUERY_RPM_RESULTS = [
{'sigkey': '64dab85d', 'sighash': '7141c84f059d2f0722ff545051b2981d', 'rpm_id': 1},
{'sigkey': '64dab85d', 'sighash': '65e5dc6e8690fc7a9a5453029d33f5b6', 'rpm_id': 2},
{'sigkey': '64dab85d', 'sighash': '522d23faed3fe55866caa9a7e72c2c94', 'rpm_id': 3},
{'sigkey': '64dab85d', 'sighash': 'e9e7c107379cff28a0732a6304a5741e', 'rpm_id': 4}
]
GET_RPM_RESULTS = [
{
'build_id': 1,
'name': 'bash',
'extra': None,
'external_repo_id': 0,
'buildtime': 1496119944,
'id': 1,
'epoch': None,
'version': '4.4.12',
'buildroot_id': None,
'metadata_only': False,
'release': '5.fc26',
'arch': 'src',
'payloadhash': '8a1705e86bef1516565b029c73ab7fdd',
'external_repo_name': 'INTERNAL',
'size': 9462614
}, {
'build_id': 2,
'name': 'less',
'extra': None,
'external_repo_id': 0,
'buildtime': 1494946131,
'id': 2, 'epoch': None,
'version': '487',
'buildroot_id': None,
'metadata_only': False,
'release': '3.fc26',
'arch': 'src',
'payloadhash': '860aa8152af63ab3868af3596764d46e',
'external_repo_name': 'INTERNAL',
'size': 358835
}, {
'build_id': 3,
'name': 'sed',
'extra': None,
'external_repo_id': 0,
'buildtime': 1486654285,
'id': 3,
'epoch': None,
'version': '4.4',
'buildroot_id': None,
'metadata_only': False,
'release': '1.fc26',
'arch': 'src',
'payloadhash': 'beb5ac98593fa1047941ad771eb88497',
'external_repo_name': 'INTERNAL',
'size': 1262814
}, {
'build_id': 1,
'name': 'bash',
'extra': None,
'external_repo_id': 0,
'buildtime': 1496120170,
'id': 4,
'epoch': None,
'version': '4.4.12',
'buildroot_id': None,
'metadata_only': False,
'release': '5.fc26',
'arch': 'x86_64',
'payloadhash': '4a15ca94b1ac59d5ef9fc6d50ad73f3e',
'external_repo_name': 'INTERNAL',
'size': 1611874
}
]
class TestWriteSignedRPM(utils.CliTestCase):
# Show long diffs in error output...
maxDiff = None
def md5sum(self, message):
md5 = hashlib.md5()
md5.update(message.encode('utf-8'))
return md5.hexdigest()
def mock_os_path_exists(self, filepath):
if filepath in self.custom_os_path_exists:
return self.custom_os_path_exists[filepath]
return self.os_path_exists(filepath)
def setUp(self):
self.custom_os_path_exists = {}
self.os_path_exists = os.path.exists
self.error_format = """Usage: %s write-signed-rpm [options] <signature-key> n-v-r [n-v-r...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
@mock.patch('sys.stdout', new_callable=six.StringIO)
@mock.patch('koji_cli.commands.activate_session')
def test_handle_write_signed_rpm(
self,
activate_session_mock,
stdout):
"""Test handle_write_signed_rpm function"""
fake_sigkey = '64dab85d'
arguments = [fake_sigkey]
options = mock.MagicMock()
session = mock.MagicMock()
def get_expect_data(rpm_data):
expected = ''
calls = []
for i, data in enumerate(rpm_data):
nvra = "%(name)s-%(version)s-%(release)s.%(arch)s" % data
expected += "[%d/%d] %s" % (i+1, len(rpm_data), nvra) + "\n"
calls.append(call(data['id'], fake_sigkey))
return expected, calls
# Case 1, specifies N-V-R-A or N-V-R format RPM
# result: write sigkey to specified RPMs
rpm_data = [GET_RPM_RESULTS[0], GET_RPM_RESULTS[3]]
session.getRPM.side_effect = [
rpm_data[0], # bash-4.4.12-5.fc26.src
koji.GenericError # bash-4.4.12-5.fc26
]
session.getBuild.return_value = {
'package_name': 'bash',
'id': 1,
'version': '4.4.12',
'nvr': 'bash-4.4.12-5.fc26',
'name': 'bash',
'release': '5.fc26'
}
session.listRPMs.return_value = [rpm_data[1]] # bash-4.4.12-5.fc26
args = arguments + ['bash-4.4.12-5.fc26.src', 'bash-4.4.12-5.fc26']
expect_msg, expect_calls = get_expect_data(rpm_data)
handle_write_signed_rpm(options, session, args)
self.assert_console_message(stdout, expect_msg)
session.writeSignedRPM.assert_has_calls(expect_calls)
session.queryRPMSigs.assert_not_called()
# Case 2, with --all option
# result: write sigkey to all RPMS
session.queryRPMSigs.return_value = QUERY_RPM_RESULTS
session.getRPM.side_effect = GET_RPM_RESULTS
expect_msg, expect_calls = get_expect_data(GET_RPM_RESULTS)
handle_write_signed_rpm(options, session, arguments + ['--all'])
self.assert_console_message(stdout, expect_msg)
session.writeSignedRPM.assert_has_calls(expect_calls)
session.queryRPMSigs.assert_called_with(sigkey=fake_sigkey)
session.queryRPMSigs.reset_mock()
# Case 3, with --buildid option
# result: write sigkey to specified build id RPM
rpm_data = [
GET_RPM_RESULTS[0], # build_id = 1
GET_RPM_RESULTS[3]] # build_id = 1
session.listRPMs.return_value = rpm_data
expect_msg, expect_calls = get_expect_data(rpm_data)
handle_write_signed_rpm(options, session, arguments + ['--buildid', '1'])
self.assert_console_message(stdout, expect_msg)
session.listRPMs.assert_called_with(1)
session.queryRPMSigs.assert_not_called()
session.writeSignedRPM.assert_has_calls(expect_calls)
session.listRPM.reset_mock()
session.writeSignedRPM.reset_mock()
# Case 4, RPM not exist
# result: raise koji.GenericError
session.getRPM.side_effect = koji.GenericError('fake-get-rpm-error')
session.getBuild.return_value = None
args = arguments + ['gawk-4.1.4-3.fc26.x86_64']
with self.assertRaises(koji.GenericError) as cm:
handle_write_signed_rpm(options, session, args)
self.assertEqual(
str(cm.exception),
'No such rpm or build: %s' % args[1])
session.listRPM.assert_not_called()
session.queryRPMSigs.assert_not_called()
session.writeSignedRPM.assert_not_called()
def test_handle_write_signed_rpm_argument_test(self):
"""Test handle_write_signed_rpm function without arguments"""
options = mock.MagicMock()
session = mock.MagicMock()
# Case 1. empty argument
expected = self.format_error_message(
"A signature key must be specified")
self.assert_system_exit(
handle_write_signed_rpm,
options,
session,
[],
stderr=expected,
activate_session=None)
# Case 2. no RPM package is specified
arguments = ['fake-signature-key']
expected = self.format_error_message(
"At least one RPM must be specified")
self.assert_system_exit(
handle_write_signed_rpm,
options,
session,
arguments,
stderr=expected,
activate_session=None)
def test_handle_write_signed_rpm_help(self):
"""Test handle_write_signed_rpm help message"""
self.assert_help(
handle_write_signed_rpm,
"""Usage: %s write-signed-rpm [options] <signature-key> n-v-r [n-v-r...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--all Write out all RPMs signed with this key
--buildid=BUILDID Specify a build id rather than an n-v-r
""" % self.progname)
if __name__ == '__main__':
unittest.main()

26
tests/test_cli/utils.py
View file

@ -1,3 +1,4 @@
from __future__ import print_function
import os
import sys
import six
@ -35,6 +36,8 @@ class CliTestCase(unittest.TestCase):
# public attribute
progname = os.path.basename(sys.argv[0]) or 'koji'
error_format = None
STDOUT = sys.stdout
STDERR = sys.stderr
#
# private methods
@ -51,8 +54,16 @@ class CliTestCase(unittest.TestCase):
return self.error_format.format(message=error_message) \
if self.error_format else error_message
def print_message(self, *args, **kwargs):
"""Print message on sys.stdout
This function will not be influenced when sys.stdout is mocked.
"""
kwargs['file'] = self.STDOUT
print(" ".join(map(str, args)), **kwargs)
def assert_function_wrapper(self, callableObj, *args, **kwargs):
"""wrapper func with anonymous funtion without argument"""
"""Wrapper func with anonymous funtion without argument"""
self.__assert_callable(callableObj)
return lambda: callableObj(*args, **kwargs)
@ -188,11 +199,8 @@ class CliTestCase(unittest.TestCase):
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')
def get_builtin_open():
if six.PY2:
return '__builtin__.open'
else:
return 'builtins.open'