241 lines
9.5 KiB
Python
241 lines
9.5 KiB
Python
from __future__ import absolute_import
|
|
from __future__ import print_function
|
|
import copy
|
|
import mock
|
|
import six
|
|
import unittest
|
|
|
|
from koji_cli.commands import anon_handle_wait_repo
|
|
from . import utils
|
|
|
|
|
|
class TestWaitRepo(utils.CliTestCase):
|
|
|
|
# Show long diffs in error output...
|
|
maxDiff = None
|
|
longMessage = True
|
|
|
|
TAG = {
|
|
'maven_support': False,
|
|
'locked': False,
|
|
'name': 'fedora26-build',
|
|
'extra': {},
|
|
'perm': None,
|
|
'id': 2,
|
|
'arches': 'x86_64',
|
|
'maven_include_all': False,
|
|
'perm_id': None
|
|
}
|
|
|
|
def setUp(self):
|
|
self.task_id = 1001
|
|
self.tag_name = self.TAG['name']
|
|
|
|
self.options = mock.MagicMock()
|
|
self.options.quiet = True
|
|
self.options.poll_interval = 1 # second
|
|
self.options.weburl = 'https://localhost.local'
|
|
|
|
self.session = mock.MagicMock()
|
|
self.session.getTag.return_value = copy.deepcopy(self.TAG)
|
|
self.session.newRepo.return_value = self.task_id
|
|
self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name}
|
|
|
|
self.error_format = """Usage: %s wait-repo [options] <tag>
|
|
(Specify the --help global option for a list of other help options)
|
|
|
|
%s: error: {message}
|
|
""" % (self.progname, self.progname)
|
|
|
|
self.setUpMocks()
|
|
|
|
def setUpMocks(self):
|
|
self.activate_session = mock.patch('koji_cli.commands.activate_session').start()
|
|
self.checkForBuilds = mock.patch('koji.util.checkForBuilds').start()
|
|
|
|
def tearDown(self):
|
|
mock.patch.stopall()
|
|
|
|
@mock.patch('time.time')
|
|
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
|
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
|
def __test_wait_repo(self, args, expected, stderr, stdout, time_mock, ret_code=0):
|
|
self.options.quiet = False
|
|
time_mock.side_effect = [0, 1, 2, 3]
|
|
if ret_code:
|
|
with self.assertRaises(SystemExit) as ex:
|
|
anon_handle_wait_repo(self.options, self.session, args)
|
|
self.assertExitCode(ex, ret_code)
|
|
self.assert_console_message(stderr, expected)
|
|
self.assert_console_message(stdout, '')
|
|
else:
|
|
rv = anon_handle_wait_repo(self.options, self.session, args)
|
|
self.assert_console_message(stdout, expected)
|
|
self.assert_console_message(stderr, '')
|
|
self.assertIn(rv, [0, None])
|
|
|
|
@mock.patch('time.time')
|
|
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
|
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
|
def __test_wait_repo_timeout(self, args, expected, stderr, stdout, time_mock, ret_code=0):
|
|
self.options.quiet = False
|
|
time_mock.side_effect = [0, 61, 62]
|
|
if ret_code:
|
|
with self.assertRaises(SystemExit) as ex:
|
|
anon_handle_wait_repo(self.options, self.session, args + ['--timeout', '1'])
|
|
self.assertExitCode(ex, ret_code)
|
|
self.assert_console_message(stderr, expected)
|
|
self.assert_console_message(stdout, '')
|
|
else:
|
|
rv = anon_handle_wait_repo(self.options, self.session, args + ['--timeout', '1'])
|
|
self.assert_console_message(stdout, expected)
|
|
self.assert_console_message(stderr, '')
|
|
self.assertIn(rv, [0, None])
|
|
|
|
def test_anon_handle_wait_repo(self):
|
|
"""Test anon_handle_wait_repo function"""
|
|
arguments = [self.tag_name]
|
|
|
|
self.options.quiet = False
|
|
self.session.getRepo.side_effect = [{}, {}, {'id': 1, 'name': 'DEFAULT'}]
|
|
expected = 'Successfully waited 0:03 for a new %s repo' % self.tag_name + '\n'
|
|
self.__test_wait_repo(arguments, expected)
|
|
|
|
def test_anon_handle_wait_repo_with_target_opt(self):
|
|
"""Test anon_handle_wait_repo function with --target option"""
|
|
arguments = [self.tag_name, '--target']
|
|
|
|
self.options.quiet = False
|
|
self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name, 'build_tag': 1}
|
|
self.session.getRepo.side_effect = [{}, {}, {'id': 1, 'name': 'DEFAULT'}]
|
|
expected = 'Successfully waited 0:03 for a new %s repo' % self.tag_name + '\n'
|
|
self.__test_wait_repo(arguments, expected)
|
|
|
|
def test_anon_handle_wait_repo_timeout(self):
|
|
"""Test anon_handle_wait_repo function on timeout case"""
|
|
arguments = [self.tag_name]
|
|
|
|
self.options.quiet = False
|
|
self.session.getRepo.return_value = {}
|
|
self.checkForBuilds.return_value = True
|
|
expected = 'Unsuccessfully waited 1:02 for a new %s repo' % self.tag_name + '\n'
|
|
self.__test_wait_repo_timeout(arguments, expected, ret_code=1)
|
|
|
|
def test_anon_handle_wait_repo_with_build(self):
|
|
"""Test anon_handle_wait_repo function with --build options"""
|
|
builds = ['bash-4.4.12-5.fc26', 'sed-4.4-1.fc26']
|
|
new_ver = 'bash-4.4.12-7.fc26'
|
|
arguments = [self.tag_name]
|
|
pkgs = ''
|
|
for b in builds:
|
|
arguments += ['--build', b]
|
|
pkgs += b + ':'
|
|
pkgs = pkgs[:-1].replace(':', ' and ')
|
|
|
|
self.options.quiet = False
|
|
self.session.getLatestBuilds.side_effect = [
|
|
[{'nvr': new_ver}], []
|
|
]
|
|
self.checkForBuilds.return_value = True
|
|
self.session.getRepo.side_effect = [
|
|
{}, {}, {'id': 1, 'name': 'DEFAULT', 'create_event': 1}
|
|
]
|
|
expected = 'Warning: nvr %s is not current in tag %s\n latest build in %s is %s' % \
|
|
(builds[0], self.tag_name, self.tag_name, new_ver) + "\n"
|
|
expected += 'Warning: package sed is not in tag %s' % self.tag_name + '\n'
|
|
expected += 'Successfully waited 0:03 for %s to appear in the %s repo' % (pkgs, self.tag_name) + '\n'
|
|
self.__test_wait_repo(arguments, expected)
|
|
|
|
def test_anon_handle_wait_repo_with_build_timeout(self):
|
|
"""Test anon_handle_wait_repo function with --build options on timeout cases"""
|
|
builds = ['bash-4.4.12-5.fc26', 'sed-4.4-1.fc26']
|
|
arguments = [self.tag_name]
|
|
pkgs = ''
|
|
for b in builds:
|
|
arguments += ['--build', b]
|
|
pkgs += b + ':'
|
|
pkgs = pkgs[:-1].replace(':', ' and ')
|
|
|
|
self.options.quiet = False
|
|
self.session.getLatestBuilds.side_effect = [
|
|
[{'nvr': builds[0]}],
|
|
[{'nvr': builds[1]}],
|
|
]
|
|
self.checkForBuilds.return_value = True
|
|
self.session.getRepo.return_value = {}
|
|
expected = 'Unsuccessfully waited 1:02 for %s to appear in the %s repo' % (pkgs, self.tag_name) + '\n'
|
|
self.__test_wait_repo_timeout(arguments, expected, ret_code=1)
|
|
|
|
def test_anon_handle_wait_repo_errors(self):
|
|
"""Test anon_handle_wait_repo function errors and exceptions"""
|
|
tests = [
|
|
# [ arguments, error_string ]
|
|
[[], "Please specify a tag name"],
|
|
[['tag1', 'tag2'], "Only one tag may be specified"],
|
|
[[self.tag_name], "Invalid tag: %s" % self.tag_name],
|
|
[[self.tag_name, '--target'], "Invalid build target: %s" % self.tag_name],
|
|
]
|
|
|
|
self.session.getBuildTarget.return_value = {}
|
|
self.session.getTag.return_value = {}
|
|
|
|
for test in tests:
|
|
self.assert_system_exit(
|
|
anon_handle_wait_repo,
|
|
self.options,
|
|
self.session,
|
|
test[0],
|
|
stderr=self.format_error_message(test[1]),
|
|
activate_session=None)
|
|
self.activate_session.assert_not_called()
|
|
|
|
@mock.patch('sys.stderr', new_callable=six.StringIO)
|
|
def test_anon_handle_wait_repo_target_not_found(self, stderr):
|
|
"""Test anon_handle_wait_repo function on target not found cases"""
|
|
|
|
# Case 1. both build and dest targets are not found
|
|
self.session.getTag.return_value = self.TAG.copy()
|
|
self.session.getBuildTargets.return_value = []
|
|
with self.assertRaises(SystemExit) as ex:
|
|
anon_handle_wait_repo(self.options, self.session, [self.tag_name])
|
|
self.assertExitCode(ex, 1)
|
|
expected = "%(name)s is not a build tag for any target" % self.TAG + "\n"
|
|
self.assert_console_message(stderr, expected)
|
|
|
|
# Cas 2. dest is matched, show suggestion
|
|
self.session.getBuildTargets.side_effect = [
|
|
[],
|
|
[
|
|
{'build_tag_name': 'build-tag-1'},
|
|
{'build_tag_name': 'build-tag-2'},
|
|
{'build_tag_name': 'build-tag-3'},
|
|
],
|
|
]
|
|
with self.assertRaises(SystemExit) as ex:
|
|
anon_handle_wait_repo(self.options, self.session, [self.tag_name])
|
|
self.assertExitCode(ex, 1)
|
|
expected = "%(name)s is not a build tag for any target" % self.TAG + "\n"
|
|
expected += "Suggested tags: build-tag-1, build-tag-2, build-tag-3\n"
|
|
self.assert_console_message(stderr, expected)
|
|
|
|
def test_anon_handle_wait_repo_help(self):
|
|
"""Test anon_handle_wait_repo help message"""
|
|
self.assert_help(
|
|
anon_handle_wait_repo,
|
|
"""Usage: %s wait-repo [options] <tag>
|
|
(Specify the --help global option for a list of other help options)
|
|
|
|
Options:
|
|
-h, --help show this help message and exit
|
|
--build=NVR Check that the given build is in the newly-generated repo
|
|
(may be used multiple times)
|
|
--target Interpret the argument as a build target name
|
|
--timeout=TIMEOUT Amount of time to wait (in minutes) before giving up
|
|
(default: 120)
|
|
--quiet Suppress output, success or failure will be indicated by
|
|
the return value only
|
|
""" % self.progname)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|