250 lines
9.7 KiB
Python
250 lines
9.7 KiB
Python
from __future__ import absolute_import
|
|
from __future__ import absolute_import
|
|
from __future__ import print_function
|
|
|
|
import unittest
|
|
import copy
|
|
|
|
try:
|
|
from unittest import mock
|
|
except ImportError:
|
|
import mock
|
|
import pytest
|
|
import six
|
|
|
|
import koji
|
|
from koji_cli.commands import anon_handle_wait_repo
|
|
from . import utils
|
|
|
|
class TestWaitRepo(utils.CliTestCase):
|
|
|
|
"""
|
|
These older tests cover the non-request code path for the cli handler
|
|
"""
|
|
|
|
# Show long diffs in error output...
|
|
maxDiff = None
|
|
longMessage = True
|
|
|
|
TAG = {
|
|
'maven_support': False,
|
|
'locked': False,
|
|
'name': 'fedora26-build',
|
|
'extra': {'repo.auto': True},
|
|
'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 = 0.0001 # keep it fast
|
|
self.options.weburl = 'https://localhost.local'
|
|
|
|
self.session = mock.MagicMock()
|
|
self.session.hub_version = (1, 35, 0)
|
|
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.ensure_connection = mock.patch('koji_cli.commands.ensure_connection').start()
|
|
self.watcher = mock.MagicMock()
|
|
self.RepoWatcher = mock.patch('koji.util.RepoWatcher', return_value=self.watcher).start()
|
|
self.wait_logger = mock.MagicMock()
|
|
self.getLogger = mock.patch('logging.getLogger', return_value=self.wait_logger).start()
|
|
|
|
def tearDown(self):
|
|
mock.patch.stopall()
|
|
|
|
@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, ret_code=0,
|
|
expected_warn=''):
|
|
self.options.quiet = False
|
|
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, expected_warn)
|
|
self.assertIn(rv, [0, None])
|
|
|
|
def test_anon_handle_wait_repo(self):
|
|
"""Test anon_handle_wait_repo function"""
|
|
arguments = [self.tag_name, '--no-request']
|
|
|
|
self.options.quiet = False
|
|
self.watcher.waitrepo.return_value = {'id': 1, 'name': 'DEFAULT'}
|
|
expected = (
|
|
'Got repo 1\n'
|
|
'Repo info: https://localhost.local/repoinfo?repoID=1\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', '--no-request']
|
|
|
|
self.options.quiet = False
|
|
self.session.getBuildTarget.return_value = {'build_tag_name': self.tag_name,
|
|
'build_tag': 1}
|
|
self.watcher.waitrepo.return_value = {'id': 1, 'name': 'DEFAULT'}
|
|
expected = (
|
|
'Got repo 1\n'
|
|
'Repo info: https://localhost.local/repoinfo?repoID=1\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, '--no-request']
|
|
|
|
self.options.quiet = False
|
|
self.watcher.waitrepo.side_effect = koji.GenericError('timeout')
|
|
expected = 'Failed to get repo -- timeout\n'
|
|
self.__test_wait_repo(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, '--no-request']
|
|
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.watcher.waitrepo.return_value = {'id': 1, 'name': 'DEFAULT', 'create_event': 1}
|
|
|
|
expected_warn = 'nvr %s is not current in tag %s\n latest build is %s' % \
|
|
(builds[0], self.tag_name, new_ver) + "\n"
|
|
expected_warn += 'No sed builds in tag %s' % self.tag_name + '\n'
|
|
expected = (
|
|
'Got repo 1\n'
|
|
'Repo info: https://localhost.local/repoinfo?repoID=1\n'
|
|
)
|
|
self.__test_wait_repo(arguments, expected, expected_warn=expected_warn)
|
|
self.RepoWatcher.assert_called_with(self.session, self.TAG['id'], nvrs=builds, min_event=None, logger=self.wait_logger)
|
|
|
|
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, '--no-request']
|
|
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.watcher.waitrepo.side_effect = koji.GenericError('timeout')
|
|
expected = 'Failed to get repo -- timeout\n'
|
|
self.__test_wait_repo(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 ]
|
|
[['--no-request'], "Please specify a tag name"],
|
|
[['tag1', 'tag2', '--no-request'], "Only one tag may be specified"],
|
|
[[self.tag_name, '--no-request'], "No such tag: %s" % self.tag_name],
|
|
[[self.tag_name, '--target', '--no-request'], "No such 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"""
|
|
# Should warn, but continue to watch
|
|
|
|
# Case 1. both build and dest targets are not found
|
|
self.session.getTag.return_value = self.TAG.copy()
|
|
self.session.getBuildTargets.return_value = []
|
|
|
|
anon_handle_wait_repo(self.options, self.session, [self.tag_name, '--no-request'])
|
|
|
|
expected = "%(name)s is not a build tag for any target" % self.TAG + "\n"
|
|
self.assert_console_message(stderr, expected)
|
|
self.RepoWatcher.assert_called_with(self.session, self.TAG['id'], nvrs=[], min_event="last", logger=self.wait_logger)
|
|
|
|
# Cas 2. dest is matched, show suggestion
|
|
self.RepoWatcher.reset_mock()
|
|
self.session.getBuildTargets.side_effect = [
|
|
[],
|
|
[
|
|
{'build_tag_name': 'build-tag-1'},
|
|
{'build_tag_name': 'build-tag-2'},
|
|
{'build_tag_name': 'build-tag-3'},
|
|
],
|
|
]
|
|
anon_handle_wait_repo(self.options, self.session, [self.tag_name, '--no-request'])
|
|
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)
|
|
self.RepoWatcher.assert_called_with(self.session, self.TAG['id'], nvrs=[], min_event="last", logger=self.wait_logger)
|
|
|
|
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
|
|
--request Create a repo request (requires auth)
|
|
--no-request Do not create a repo request (the default)
|
|
--timeout=TIMEOUT Amount of time to wait (in minutes) before giving up
|
|
(default: 120)
|
|
-v, --verbose Be verbose
|
|
--quiet Suppress output, success or failure will be indicated by
|
|
the return value only
|
|
""" % self.progname)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|