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

Increase unit tests

Browse source
This commit is contained in:
Jana Cupova 2022-08-09 07:52:23 +02:00
parent d81033c71d
commit 4f5b69f8a7
8 changed files with 624 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

108
tests/test_cli/test_add_tag_inheritance.py
View file

@ -1,6 +1,7 @@
from __future__ import absolute_import
import mock
from six.moves import StringIO
import koji
from koji_cli.commands import handle_add_tag_inheritance
@ -19,6 +20,41 @@ class TestAddTagInheritance(utils.CliTestCase):
%s: error: {message}
""" % (self.progname, self.progname)
self.tag_inheritance = {'child_id': 1,
'intransitive': False,
'maxdepth': None,
'name': 'parent-test-tag',
'noconfig': False,
'parent_id': 2,
'pkg_filter': '',
'priority': 10}
self.tag_1 = {'arches': 'x86_64',
'extra': {},
'id': 1,
'locked': False,
'maven_include_all': False,
'maven_support': False,
'name': 'test-tag',
'perm': None,
'perm_id': None}
self.tag_2 = {'arches': 'x86_64',
'extra': {},
'id': 2,
'locked': False,
'maven_include_all': False,
'maven_support': False,
'name': 'parent-test-tag',
'perm': None,
'perm_id': None}
self.tag_3 = {'arches': 'x86_64',
'extra': {},
'id': 3,
'locked': False,
'maven_include_all': False,
'maven_support': False,
'name': 'parent-test-tag-3',
'perm': None,
'perm_id': None}
def test_add_tag_inheritance_without_option(self):
arguments = []
@ -50,20 +86,10 @@ class TestAddTagInheritance(utils.CliTestCase):
self.activate_session_mock.assert_called_once_with(self.session, self.options)
def test_add_tag_inheritance_non_exist_parent_tag(self):
side_effect_result = [{'arches': 'x86_64',
'extra': {},
'id': 1,
'locked': False,
'maven_include_all': False,
'maven_support': False,
'name': 'test-tag',
'perm': None,
'perm_id': None},
None]
tag = 'test-tag'
parent_tag = 'parent-test-tag'
arguments = [tag, parent_tag]
self.session.getTag.side_effect = side_effect_result
self.session.getTag.side_effect = [self.tag_1, None]
self.assert_system_exit(
handle_add_tag_inheritance,
self.options, self.session, arguments,
@ -72,3 +98,63 @@ class TestAddTagInheritance(utils.CliTestCase):
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
def test_add_tag_inheritance_same_parents_without_force(self):
tag = 'test-tag'
parent_tag = 'parent-test-tag'
arguments = ['--priority=10', tag, parent_tag]
self.session.getInheritanceData.return_value = [self.tag_inheritance, self.tag_inheritance]
self.session.getTag.side_effect = [self.tag_1, self.tag_2]
expected_error = "Error: You are attempting to add %s as %s's parent even though it " \
"already is %s's parent.\nPlease use --force if this is what you " \
"really want to do.\n" % (parent_tag, tag, tag)
self.assert_system_exit(
handle_add_tag_inheritance,
self.options, self.session, arguments,
stdout='',
stderr=expected_error,
exit_code=1,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
def test_add_tag_inheritance_same_priority_without_force(self):
tag = 'test-tag'
parent_tag = 'parent-test-tag-3'
arguments = ['--priority=10', tag, parent_tag]
self.session.getInheritanceData.return_value = [self.tag_inheritance, self.tag_inheritance]
self.session.getTag.side_effect = [self.tag_1, self.tag_3]
expected_error = "Error: There is already an active inheritance with that priority " \
"on %s, please specify a different priority with --priority.\n" % tag
self.assert_system_exit(
handle_add_tag_inheritance,
self.options, self.session, arguments,
stdout='',
stderr=expected_error,
exit_code=1,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stderr', new_callable=StringIO)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_add_tag_inheritance_valid(self, stdout, stderr):
tag = 'test-tag'
parent_tag = 'parent-test-tag-3'
self.session.getInheritanceData.return_value = [self.tag_inheritance, self.tag_inheritance]
self.session.getTag.side_effect = [self.tag_1, self.tag_3]
handle_add_tag_inheritance(self.options, self.session, ['--priority=5', tag, parent_tag])
self.assert_console_message(stdout, '')
self.assert_console_message(stderr, '')
self.activate_session_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stderr', new_callable=StringIO)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_add_tag_inheritance_valid_with_maxdepth(self, stdout, stderr):
tag = 'test-tag'
parent_tag = 'parent-test-tag-3'
self.session.getInheritanceData.return_value = [self.tag_inheritance, self.tag_inheritance]
self.session.getTag.side_effect = [self.tag_1, self.tag_3]
handle_add_tag_inheritance(self.options, self.session,
['--maxdepth=10', '--priority=5', tag, parent_tag])
self.assert_console_message(stdout, '')
self.assert_console_message(stderr, '')
self.activate_session_mock.assert_called_once_with(self.session, self.options)

19
tests/test_cli/test_buildinfo.py
View file

@ -21,6 +21,11 @@ class TestBuildinfo(utils.CliTestCase):
self.options.debug = False
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
self.error_format = """Usage: %s buildinfo [options] <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)
self.original_timezone = os.environ.get('TZ')
os.environ['TZ'] = 'UTC'
time.tzset()
@ -120,3 +125,17 @@ Tags:
stdout='',
activate_session=None,
exit_code=1)
def test_buildinfo_without_option(self):
arguments = []
self.assert_system_exit(
anon_handle_buildinfo,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("Please specify a build"),
exit_code=2,
activate_session=None)
self.session.listTags.assert_not_called()
self.session.getMavenBuild.assert_not_called()
self.session.getWinBuild.assert_not_called()
self.session.listRPMs.assert_not_called()

130
tests/test_cli/test_list_signed.py Normal file
View file

@ -0,0 +1,130 @@
from __future__ import absolute_import
import mock
from six.moves import StringIO
from koji_cli.commands import handle_list_signed
from . import utils
class TestListSigned(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.options.debug = True
self.session = mock.MagicMock()
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
self.error_format = """Usage: %s list-signed [options]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def test_list_signed_help(self):
self.assert_help(
handle_list_signed,
"""Usage: %s list-signed [options]
(Specify the --help global option for a list of other help options)
You must have local access to Koji's topdir filesystem.
Options:
-h, --help show this help message and exit
--key=KEY Only list RPMs signed with this key
--build=BUILD Only list RPMs from this build
--rpm=RPM Only list signed copies for this RPM
--tag=TAG Only list RPMs within this tag
""" % self.progname)
def test_list_signed_without_arg(self):
arguments = []
self.assert_system_exit(
handle_list_signed,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message(
"At least one from --build, --rpm, --tag needs to be specified."),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
self.session.getBuild.assert_not_called()
self.session.listRPMs.assert_not_called()
self.session.queryRPMSigs.assert_not_called()
self.session.getRPM.assert_not_called()
def test_list_signed_rpm_external_rpm_error(self):
arguments = ['--rpm=test-rpm']
rinfo = {'id': 123, 'name': 'test', 'version': '1.3', 'release': 1, 'arch': 'test-arch',
'external_repo_name': 'ext-repo', 'external_repo_id': 456}
err_msg = "External rpm: %(name)s-%(version)s-%(release)s.%(arch)s@" \
"%(external_repo_name)s" % rinfo
self.session.getRPM.return_value = rinfo
self.assert_system_exit(
handle_list_signed,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message(err_msg),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getBuild.assert_not_called()
self.session.listRPMs.assert_not_called()
self.session.queryRPMSigs.assert_not_called()
self.session.getRPM.assert_called_once_with('test-rpm', strict=True)
@mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_list_signed_rpm_external_build_non_exist_path(self, stdout, pb):
binfo = {'build_id': 1, 'id': 1, 'name': 'test-build', 'release': '1', 'task_id': 8,
'version': '1', 'state': 1, 'completion_ts': 1614869140.368759,
'owner_name': 'kojiadmin', 'volume_name': 'DEFAULT',
'package_name': 'test-package'}
list_rpms = [{'id': 123, 'name': 'test', 'version': '1.3', 'release': 1,
'arch': 'test-arch', 'external_repo_name': 'ext-repo',
'external_repo_id': 456, 'build_id': 1}]
sigRpm = [{'rpm_id': 123, 'sigkey': 'qwertyuiop'}]
self.session.getBuild.return_value = binfo
self.session.listRPMs.return_value = list_rpms
self.session.queryRPMSigs.return_value = sigRpm
handle_list_signed(self.options, self.session, ['--build=test-build'])
path = 'fakebuildpath/data/signed/%s/%s/%s-%s-%s.%s.rpm' \
% (sigRpm[0]['sigkey'], list_rpms[0]['arch'], list_rpms[0]['name'],
list_rpms[0]['version'], list_rpms[0]['release'], list_rpms[0]['arch'])
self.assert_console_message(stdout, 'No copy: %s\n' % path)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getBuild.assert_called_once_with('test-build', strict=True)
self.session.listRPMs.assert_called_once_with(buildID=binfo['id'])
self.session.queryRPMSigs.assert_called_once_with(rpm_id=list_rpms[0]['id'])
self.session.getRPM.assert_not_called()
@mock.patch('os.path.exists', return_value=True)
@mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
@mock.patch('sys.stdout', new_callable=StringIO)
def test_list_signed_rpm_external_build_valid(self, stdout, pb, os_path_exists):
binfo = {'build_id': 1,
'id': 1,
'name': 'test-build',
'release': '1',
'task_id': 8,
'version': '1',
'state': 1,
'completion_ts': 1614869140.368759,
'owner_name': 'kojiadmin',
'volume_name': 'DEFAULT',
'package_name': 'test-package'}
list_rpms = [
{'id': 123, 'name': 'test', 'version': '1.3', 'release': 1, 'arch': 'test-arch',
'external_repo_name': 'ext-repo', 'external_repo_id': 456, 'build_id': 1}]
sigRpm = [{'rpm_id': 123, 'sigkey': 'qwertyuiop'}]
self.session.getBuild.return_value = binfo
self.session.listRPMs.return_value = list_rpms
self.session.queryRPMSigs.return_value = sigRpm
handle_list_signed(self.options, self.session, ['--build=test-build'])
path = 'fakebuildpath/data/signed/%s/%s/%s-%s-%s.%s.rpm' \
% (sigRpm[0]['sigkey'], list_rpms[0]['arch'], list_rpms[0]['name'],
list_rpms[0]['version'], list_rpms[0]['release'], list_rpms[0]['arch'])
self.assert_console_message(stdout, '%s\n' % path)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.session.getBuild.assert_called_once_with('test-build', strict=True)
self.session.listRPMs.assert_called_once_with(buildID=binfo['id'])
self.session.queryRPMSigs.assert_called_once_with(rpm_id=list_rpms[0]['id'])
self.session.getRPM.assert_not_called()

133
tests/test_cli/test_remove_external_repo.py Normal file
View file

@ -0,0 +1,133 @@
from __future__ import absolute_import
import mock
from six.moves import StringIO
from koji_cli.commands import handle_remove_external_repo
from . import utils
class TestRemoveExternalRepo(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.session = mock.MagicMock()
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
self.error_format = """Usage: %s remove-external-repo <repo> [<tag> ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def test_remove_external_repo_help(self):
self.assert_help(
handle_remove_external_repo,
"""Usage: %s remove-external-repo <repo> [<tag> ...]
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--alltags Remove from all tags
--force Force action
""" % self.progname)
def test_remove_external_repo_without_arg(self):
arguments = []
self.assert_system_exit(
handle_remove_external_repo,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("Incorrect number of arguments"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
def test_remove_external_repo_delete_without_force(self):
self.session.getTagExternalRepos.return_value = [{'external_repo_id': 11,
'tag_id': 1,
'tag_name': 'test-tag',
'priority': 5,
'merge_mode': 'simple',
'arches': 'x86_64 i686'}]
repo = 'test-repo'
tag = 'test-tag'
exp_error = "Error: external repo %s used by tag(s): %s\nUse --force to remove anyway\n" \
% (repo, tag)
arguments = [repo]
self.assert_system_exit(
handle_remove_external_repo,
self.options, self.session, arguments,
stdout='',
stderr=exp_error,
exit_code=1,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_external_repo_delete_valid(self, stdout):
self.session.getTagExternalRepos.return_value = [{'external_repo_id': 11,
'tag_id': 1,
'tag_name': 'test-tag',
'priority': 5,
'merge_mode': 'simple',
'arches': 'x86_64 i686'}]
repo = 'test-repo'
handle_remove_external_repo(self.options, self.session, ['--force', repo])
self.assert_console_message(stdout, '')
self.activate_session_mock.assert_called_once_with(self.session, self.options)
def test_remove_external_repo_alltags_with_tag(self):
self.session.getTagExternalRepos.return_value = [{'external_repo_id': 11,
'tag_id': 1,
'tag_name': 'test-tag',
'priority': 5,
'merge_mode': 'simple',
'arches': 'x86_64 i686'}]
repo = 'test-repo'
tag = 'test-tag'
arguments = ['--alltags', repo, tag]
self.assert_system_exit(
handle_remove_external_repo,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("Do not specify tags when using --alltags"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stderr', new_callable=StringIO)
def test_remove_external_repo_alltags_not_associated_without_force(self, stderr):
self.session.getTagExternalRepos.return_value = []
repo = 'test-repo'
handle_remove_external_repo(self.options, self.session, ['--alltags', repo])
self.assert_console_message(stderr,
'External repo %s not associated with any tags\n' % repo)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_external_repo_remove_not_associated_tag(self, stdout):
self.session.getTagExternalRepos.return_value = [{'external_repo_id': 11,
'tag_id': None,
'tag_name': None,
'priority': 5,
'merge_mode': 'simple',
'arches': 'x86_64 i686'}]
repo = 'test-repo'
tag = 'test-tag'
handle_remove_external_repo(self.options, self.session, [repo, tag])
self.assert_console_message(
stdout, 'External repo %s not associated with tag %s\n' % (repo, tag))
self.activate_session_mock.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_external_repo_remove_valid(self, stdout):
self.session.getTagExternalRepos.return_value = [{'external_repo_id': 11,
'tag_id': 1,
'tag_name': 'test-tag',
'priority': 5,
'merge_mode': 'simple',
'arches': 'x86_64 i686'}]
repo = 'test-repo'
tag = 'test-tag'
handle_remove_external_repo(self.options, self.session, [repo, tag])
self.assert_console_message(stdout, '')
self.activate_session_mock.assert_called_once_with(self.session, self.options)

57
tests/test_cli/test_remove_sig.py
View file

@ -1,7 +1,6 @@
from __future__ import absolute_import
import mock
from six.moves import StringIO
import koji
from koji_cli.commands import handle_remove_sig
@ -16,6 +15,11 @@ class TestRemoveSig(utils.CliTestCase):
self.options.debug = False
self.session = mock.MagicMock()
self.session.getAPIVersion.return_value = koji.API_VERSION
self.error_format = """Usage: %s remove-sig [options] <rpm-id/n-v-r.a/rpminfo>
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def test_remove_sig_help(self):
self.assert_help(
@ -32,18 +36,18 @@ Options:
--all Remove all signed copies for specified RPM
""" % self.progname)
@mock.patch('sys.stderr', new_callable=StringIO)
def test_remove_sig_without_option(self, stderr):
expected = "Usage: %s remove-sig [options] <rpm-id/n-v-r.a/rpminfo>\n" \
"(Specify the --help global option for a list of other help options)\n\n" \
"%s: error: Please specify an RPM\n" % (self.progname, self.progname)
with self.assertRaises(SystemExit) as ex:
handle_remove_sig(self.options, self.session, [])
self.assertExitCode(ex, 2)
self.assert_console_message(stderr, expected)
def test_remove_sig_without_option(self):
arguments = []
self.assert_system_exit(
handle_remove_sig,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("Please specify an RPM"),
exit_code=2,
activate_session=None)
self.session.deleteRPMSig.assert_not_called()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_sig_non_exist_rpm(self, stdout):
def test_remove_sig_non_exist_rpm(self):
rpm = '1234'
expected = "No such rpm in system: %s\n" % rpm
self.session.deleteRPMSig.side_effect = koji.GenericError('No such rpm: DATA')
@ -53,7 +57,7 @@ Options:
self.options,
self.session,
[rpm, '--all'],
stderr=self.format_error_message(expected),
stderr=expected,
exit_code=1,
activate_session=None)
self.session.deleteRPMSig.assert_called_with('1234', sigkey=None, all_sigs=True)
@ -64,8 +68,7 @@ Options:
handle_remove_sig(self.options, self.session, [rpm, '--sigkey', 'testkey'])
self.session.deleteRPMSig.assert_called_with('1', sigkey='testkey', all_sigs=False)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_sig_without_all_and_sigkey(self, stdout):
def test_remove_sig_without_all_and_sigkey(self):
rpm = '1234'
expected = "Either --sigkey or --all options must be given\n"
@ -74,12 +77,11 @@ Options:
self.options,
self.session,
[rpm],
stderr=self.format_error_message(expected),
stderr=expected,
exit_code=1,
activate_session=None)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_remove_sig_with_all_and_sigkey(self, stdout):
def test_remove_sig_with_all_and_sigkey(self):
rpm = '1234'
expected = "Conflicting options specified\n"
@ -88,6 +90,23 @@ Options:
self.options,
self.session,
[rpm, '--all', '--sigkey', 'testkey'],
stderr=self.format_error_message(expected),
stderr=expected,
exit_code=1,
activate_session=None)
def test_remove_sig_signature_removal_failed(self):
rpm = '1234'
nvra = 'test-1.23-1.arch'
error_msg = "%s has no matching signatures to delete" % nvra
expected = "Signature removal failed: %s\n" % error_msg
self.session.deleteRPMSig.side_effect = koji.GenericError(error_msg)
self.assert_system_exit(
handle_remove_sig,
self.options,
self.session,
[rpm, '--all'],
stderr=expected,
exit_code=1,
activate_session=None)
self.session.deleteRPMSig.assert_called_with('1234', sigkey=None, all_sigs=True)

51
tests/test_cli/test_rpminfo.py
View file

@ -53,6 +53,11 @@ class TestRpminfo(utils.CliTestCase):
'version': '1.1',
'payloadhash': 'b2b95550390e5f213fc25f33822425f7',
'size': 7030}
self.error_format = """Usage: %s rpminfo [options] <n-v-r.a> [<n-v-r.a> ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def tearDown(self):
locale.resetlocale()
@ -94,19 +99,20 @@ Used in 1 buildroots:
self.session.getBuild.assert_called_once_with(self.getrpminfo['build_id'])
self.session.getRPM.assert_called_once_with(rpm_nvra)
@mock.patch('sys.stderr', new_callable=StringIO)
def test_handle_rpminfo_non_exist_nvra(self, stderr):
def test_handle_rpminfo_non_exist_nvra(self):
rpm_nvra = 'test-rpm-nvra.arch'
self.session.getRPM.return_value = None
arguments = ['--buildroot', rpm_nvra]
expected = "No such rpm: %s\n" % rpm_nvra + "\n"
with self.assertRaises(SystemExit) as ex:
anon_handle_rpminfo(self.options, self.session, ['--buildroot', rpm_nvra])
self.assertExitCode(ex, 1)
self.assert_console_message(stderr, expected)
self.assert_system_exit(
anon_handle_rpminfo,
self.options, self.session, arguments,
stdout='',
stderr=expected,
exit_code=1,
activate_session=None)
@mock.patch('sys.stderr', new_callable=StringIO)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_handle_rpminfo_more_nvra_non_exist_nvra(self, stdout, stderr):
def test_handle_rpminfo_more_nvra_non_exist_nvra(self):
rpm_nvra = 'test-rpm-1.1-11.noarch'
non_exist_rpm_nvra = 'not-test-rpm-11-112.arch'
self.session.getBuildroot.return_value = self.buildroot_info
@ -130,15 +136,30 @@ Used in 1 buildroots:
3 test-tag x86_64 kojibuilder
"""
arguments = ['--buildroot', non_exist_rpm_nvra, rpm_nvra]
expected_error = "No such rpm: %s\n" % non_exist_rpm_nvra + "\n"
with self.assertRaises(SystemExit) as ex:
anon_handle_rpminfo(self.options, self.session,
['--buildroot', non_exist_rpm_nvra, rpm_nvra])
self.assertExitCode(ex, 1)
self.assert_console_message(stdout, expected_output)
self.assert_console_message(stderr, expected_error)
self.assert_system_exit(
anon_handle_rpminfo,
self.options, self.session, arguments,
stdout=expected_output,
stderr=expected_error,
exit_code=1,
activate_session=None)
self.session.getBuildroot.assert_called_once_with(self.getrpminfo['buildroot_id'])
self.session.listBuildroots.assert_called_once_with(queryOpts={'order': 'buildroot.id'},
rpmID=self.getrpminfo['id'])
self.session.getBuild.assert_called_once_with(self.getrpminfo['build_id'])
self.assertEqual(self.session.getRPM.call_count, 2)
def test_rpminfo_without_option(self):
arguments = []
self.assert_system_exit(
anon_handle_rpminfo,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("Please specify an RPM"),
exit_code=2,
activate_session=None)
self.session.getBuildroot.assert_not_called()
self.session.listBuildroots.assert_not_called()
self.session.getBuild.assert_not_called()

86
tests/test_cli/test_watch_logs.py Normal file
View file

@ -0,0 +1,86 @@
from __future__ import absolute_import
import mock
import unittest
from six.moves import StringIO
from koji_cli.commands import anon_handle_watch_logs
from . import utils
class TestWatchLogs(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.session = mock.MagicMock()
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
self.ensure_connection = mock.patch('koji_cli.commands.ensure_connection').start()
self.list_tasks = mock.patch('koji_cli.commands._list_tasks').start()
self.error_format = """Usage: %s watch-logs [options] <task id> [<task id> ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def test_handle_watch_logs_help(self):
self.assert_help(
anon_handle_watch_logs,
"""Usage: %s watch-logs [options] <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
--log=LOG Watch only a specific log
--mine Watch logs for all your tasks, task_id arguments are forbidden
in this case.
--follow Follow spawned child tasks
""" % self.progname)
def test_watch_task_mine_and_task_id(self):
arguments = ['--mine', '1']
self.assert_system_exit(
anon_handle_watch_logs,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message(
"Selection options cannot be combined with a task list"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.ensure_connection.assert_not_called()
def test_watch_task_task_id_not_int(self):
arguments = ['task-id']
self.assert_system_exit(
anon_handle_watch_logs,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("task id must be an integer"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
self.ensure_connection.assert_called_once_with(self.session, self.options)
def test_watch_task_without_task(self):
arguments = []
self.assert_system_exit(
anon_handle_watch_logs,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("at least one task id must be specified"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
self.ensure_connection.assert_called_once_with(self.session, self.options)
@mock.patch('sys.stdout', new_callable=StringIO)
def test_watch_task_mine_without_tasks(self, stdout):
expected_output = "You've no active tasks.\n"
self.list_tasks.return_value = []
anon_handle_watch_logs(self.options, self.session, ['--mine'])
self.assert_console_message(stdout, expected_output)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.ensure_connection.assert_not_called()
if __name__ == '__main__':
unittest.main()

91
tests/test_cli/test_watch_tasks.py
View file

@ -6,15 +6,16 @@ import six
import sys
import unittest
from mock import call
from six.moves import range
from six.moves import StringIO
import koji
from koji_cli.lib import watch_tasks
from koji_cli.commands import anon_handle_watch_task
from .fakeclient import FakeClientSession, RecordingClientSession
from . import utils
class TestWatchTasks(unittest.TestCase):
class TestWatchTasksCliLib(unittest.TestCase):
def setUp(self):
self.options = mock.MagicMock()
@ -59,8 +60,7 @@ class TestWatchTasks(unittest.TestCase):
rv = watch_tasks(self.session, [1188], quiet=False, poll_interval=0,
topurl=self.options.topurl)
self.assertEqual(rv, 0)
expected = (
'''Watching tasks (this may be safely interrupted)...
expected = ('''Watching tasks (this may be safely interrupted)...
1188 build (f24, /users/mikem/fake.git:adaf62586b4b4a23b24394da5586abd7cd9f679e): closed
1189 buildSRPMFromSCM (/users/mikem/fake.git:adaf62586b4b4a23b24394da5586abd7cd9f679e): closed
1190 buildArch (fake-1.1-21.src.rpm, noarch): closed
@ -100,7 +100,7 @@ class TestWatchTasks(unittest.TestCase):
cfile = os.path.dirname(__file__) + '/data/calls/watchtasks2.json'
cdata = koji.load_json(cfile)
self.session.load_calls(cdata)
sleep.side_effect = [None] * 10 + [KeyboardInterrupt]
sleep.side_effect = [None] * 10 + [KeyboardInterrupt]
with self.assertRaises(KeyboardInterrupt):
# watch_tasks catches and re-raises it to display a message
watch_tasks(self.session, [1208], quiet=False, poll_interval=5,
@ -143,5 +143,84 @@ some output
self.assertMultiLineEqual(stdout.getvalue(), expected)
class TestWatchLogsCLI(utils.CliTestCase):
def setUp(self):
self.options = mock.MagicMock()
self.session = mock.MagicMock()
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
self.ensure_connection = mock.patch('koji_cli.commands.ensure_connection').start()
self.list_tasks = mock.patch('koji_cli.commands._list_tasks').start()
self.error_format = """Usage: %s watch-task [options] <task id> [<task id> ...]
(Specify the --help global option for a list of other help options)
%s: error: {message}
""" % (self.progname, self.progname)
def test_handle_watch_task_help(self):
self.assert_help(
anon_handle_watch_task,
"""Usage: %s watch-task [options] <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
--quiet Do not print the task information
--mine Just watch 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
""" % self.progname)
def test_watch_task_selection_and_task_id(self):
for arg in ['--mine', '--user=kojiadmin', '--arch=test-arcg', '--method=build',
'--channel=default', '--host=test-host']:
arguments = [arg, '1']
self.assert_system_exit(
anon_handle_watch_task,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("Selection options cannot be combined with a task list"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
self.ensure_connection.assert_not_called()
@mock.patch('sys.stdout', new_callable=StringIO)
def test_watch_task_mine_without_task(self, stdout):
expected_output = "(no tasks)\n"
self.list_tasks.return_value = []
anon_handle_watch_task(self.options, self.session, ['--mine'])
self.assert_console_message(stdout, expected_output)
self.activate_session_mock.assert_called_once_with(self.session, self.options)
self.ensure_connection.assert_not_called()
def test_watch_task_task_id_not_int(self):
arguments = ['task-id']
self.assert_system_exit(
anon_handle_watch_task,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("task id must be an integer"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
self.ensure_connection.assert_called_once_with(self.session, self.options)
def test_watch_task_without_task(self):
arguments = []
self.assert_system_exit(
anon_handle_watch_task,
self.options, self.session, arguments,
stdout='',
stderr=self.format_error_message("at least one task id must be specified"),
exit_code=2,
activate_session=None)
self.activate_session_mock.assert_not_called()
self.ensure_connection.assert_called_once_with(self.session, self.options)
if __name__ == '__main__':
unittest.main()