update unit test
This commit is contained in:
Mike McLean
Tomas Kopecek
parent
e172a9ca08
commit
cee0b21061
1 changed files with 196 additions and 135 deletions
|
|
@ -1,35 +1,30 @@
|
|||
from __future__ import absolute_import
|
||||
import json
|
||||
import os
|
||||
import unittest
|
||||
import shutil
|
||||
import tempfile
|
||||
try:
|
||||
from unittest import mock
|
||||
from unittest.mock import call
|
||||
except ImportError:
|
||||
import mock
|
||||
from mock import call
|
||||
|
||||
import six
|
||||
|
||||
from koji_cli.commands import handle_import_cg
|
||||
import koji
|
||||
from koji_cli.commands import handle_import_cg, _progress_callback
|
||||
from . import utils
|
||||
|
||||
|
||||
class TestImportCG(utils.CliTestCase):
|
||||
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.maxDiff = None
|
||||
self.options = mock.MagicMock()
|
||||
self.session = mock.MagicMock()
|
||||
self.custom_os_path_exists = {}
|
||||
self.os_path_exists = os.path.exists
|
||||
self.workdir = tempfile.mkdtemp()
|
||||
self.outdir = self.workdir + '/output'
|
||||
self.unique_path_mock = mock.patch('koji_cli.commands.unique_path').start()
|
||||
self.running_in_bg = mock.patch('koji_cli.commands._running_in_bg').start()
|
||||
self.running_in_bg.return_value = False
|
||||
self.linked_upload_mock = mock.patch('koji_cli.commands.linked_upload').start()
|
||||
self.linked_upload = mock.patch('koji_cli.commands.linked_upload').start()
|
||||
self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start()
|
||||
self.error_format = """Usage: %s import-cg [options] <metadata_file> <files_dir>
|
||||
(Specify the --help global option for a list of other help options)
|
||||
|
|
@ -39,143 +34,210 @@ class TestImportCG(utils.CliTestCase):
|
|||
|
||||
def tearDown(self):
|
||||
mock.patch.stopall()
|
||||
shutil.rmtree(self.workdir)
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('koji_cli.commands._progress_callback')
|
||||
@mock.patch('koji.json')
|
||||
def test_handle_import_cg(self, json_mock, progress_callback_mock, stdout):
|
||||
def make_metadata(self, metadata=None, defaults=True, write_outputs=True):
|
||||
"""Fill in metadata and write to workdir"""
|
||||
if metadata is None:
|
||||
metadata = {}
|
||||
|
||||
if defaults:
|
||||
build = metadata.setdefault('build', {})
|
||||
build.setdefault('name', 'mypackage')
|
||||
build.setdefault('version', '1')
|
||||
build.setdefault('release', '2')
|
||||
|
||||
if 'output' not in metadata:
|
||||
metadata['output'] = [
|
||||
{'relpath': 'relative/path', 'filename': 'file-1'},
|
||||
{'relpath': 'relative/path', 'filename': 'file-2'},
|
||||
]
|
||||
|
||||
# write metdata
|
||||
fn = os.path.join(self.workdir, 'metadata.json')
|
||||
with open(fn, 'wt') as fd:
|
||||
json.dump(metadata, fd, indent=4)
|
||||
|
||||
if write_outputs:
|
||||
self.write_outputs(metadata)
|
||||
|
||||
return metadata, fn
|
||||
|
||||
def write_outputs(self, metadata):
|
||||
# write outputs
|
||||
for info in metadata.get('output', []):
|
||||
fdir = os.path.join(self.outdir, info.get('relpath', ''))
|
||||
koji.ensuredir(fdir)
|
||||
fn = os.path.join(fdir, info['filename'])
|
||||
with open(fn, 'wt') as fd:
|
||||
fd.write('fnord\n')
|
||||
|
||||
def test_handle_import_cg(self):
|
||||
"""Test handle_import_cg function"""
|
||||
arguments = ['fake-metafile', '/path/to/files/']
|
||||
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
|
||||
self.unique_path_mock.return_value = fake_srv_path
|
||||
metadata, fn = self.make_metadata()
|
||||
arguments = [fn, self.outdir]
|
||||
|
||||
# Case 1, running in fg, progress on
|
||||
with mock.patch(utils.get_builtin_open()):
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
calls, expected = gen_value("Uploading %s\n", progress_callback_mock)
|
||||
self.assert_console_message(stdout, expected)
|
||||
self.linked_upload_mock.assert_not_called()
|
||||
self.session.uploadWrapper.assert_has_calls(calls)
|
||||
self.session.CGImport.assert_called_with(metadata, fake_srv_path, None)
|
||||
self.assertEqual(len(self.session.uploadWrapper.mock_calls), len(metadata['output']))
|
||||
kwargs = self.session.uploadWrapper.call_args.kwargs
|
||||
self.assertEqual(kwargs['callback'], _progress_callback)
|
||||
self.session.CGImport.assert_called_once()
|
||||
args = self.session.CGImport.call_args.args
|
||||
self.assertEqual(args[0], metadata)
|
||||
self.linked_upload.assert_not_called()
|
||||
|
||||
# Case 2, running in fg, progress off
|
||||
with mock.patch(utils.get_builtin_open()):
|
||||
handle_import_cg(self.options, self.session, arguments + ['--noprogress'])
|
||||
def test_handle_import_cg_nodir(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
calls, expected = gen_value("Uploading %s", None)
|
||||
self.assert_console_message(stdout, expected)
|
||||
self.linked_upload_mock.assert_not_called()
|
||||
self.session.uploadWrapper.assert_has_calls(calls)
|
||||
self.session.CGImport.assert_called_with(metadata, fake_srv_path, None)
|
||||
metadata, fn = self.make_metadata(write_outputs=False)
|
||||
arguments = [fn]
|
||||
|
||||
# reset mocks
|
||||
self.linked_upload_mock.reset_mock()
|
||||
self.session.uploadWrapper.reset_mock()
|
||||
self.session.CGImport.reset_mock()
|
||||
|
||||
# Case 3, --test option
|
||||
with mock.patch(utils.get_builtin_open()):
|
||||
handle_import_cg(self.options, self.session, arguments + ['--test'])
|
||||
|
||||
self.linked_upload_mock.assert_not_called()
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.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(self.options, self.session, arguments + ['--link'])
|
||||
|
||||
self.linked_upload_mock.assert_has_calls(calls)
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.session.CGImport.assert_called_with(metadata, fake_srv_path, None)
|
||||
|
||||
# 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/']
|
||||
|
||||
# Case 1. empty argument
|
||||
self.assert_system_exit(
|
||||
handle_import_cg,
|
||||
self.options, self.session, [],
|
||||
stderr=self.format_error_message("Please specify metadata files directory"),
|
||||
activate_session=None)
|
||||
self.options, self.session, arguments,
|
||||
stderr=self.format_error_message('Please specify metadata files directory'),
|
||||
stdout='',
|
||||
activate_session=None
|
||||
)
|
||||
|
||||
# Case 2. JSON module does not exist
|
||||
# dropped - it is now part of stdlib
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.session.CGImport.assert_not_called()
|
||||
self.linked_upload.assert_not_called()
|
||||
|
||||
metadata = {
|
||||
'output': [
|
||||
{'metadata_only': True},
|
||||
{'relpath': '/real/path', 'filename': 'filename'}
|
||||
]
|
||||
}
|
||||
def test_handle_import_cg_output(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
#
|
||||
# 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
|
||||
metadata, fn = self.make_metadata({}, defaults=False)
|
||||
arguments = [fn, self.outdir]
|
||||
|
||||
with mock.patch(utils.get_builtin_open()):
|
||||
with mock.patch('os.path.exists', new=self.mock_os_path_exists):
|
||||
with mock.patch('koji.json') as json_mock:
|
||||
self.assert_system_exit(
|
||||
handle_import_cg,
|
||||
self.options, self.session, arguments,
|
||||
stderr='Metadata contains no output\n',
|
||||
stdout='',
|
||||
activate_session=None,
|
||||
exit_code=1
|
||||
)
|
||||
|
||||
# Case 3. metafile doesn't have output section
|
||||
json_mock.load.return_value = {}
|
||||
self.assert_system_exit(
|
||||
handle_import_cg,
|
||||
self.options, self.session, arguments,
|
||||
stderr="Metadata contains no output\n",
|
||||
exit_code=1)
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.session.CGImport.assert_not_called()
|
||||
self.linked_upload.assert_not_called()
|
||||
|
||||
# Case 4. path not exist
|
||||
file_path = '%(relpath)s/%(filename)s' % metadata['output'][1]
|
||||
json_mock.load.return_value = metadata
|
||||
self.assert_system_exit(
|
||||
handle_import_cg,
|
||||
self.options, self.session, arguments,
|
||||
stderr=self.format_error_message("No such file: %s" % file_path),
|
||||
exit_code=2)
|
||||
def test_handle_import_cg_nofiles(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata(write_outputs=False)
|
||||
arguments = [fn, self.outdir]
|
||||
|
||||
expect = "No such file: %s" % os.path.join(self.outdir, 'relative/path/file-1')
|
||||
self.assert_system_exit(
|
||||
handle_import_cg,
|
||||
self.options, self.session, arguments,
|
||||
stderr=self.format_error_message(expect),
|
||||
stdout='',
|
||||
activate_session=None,
|
||||
exit_code=2
|
||||
)
|
||||
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.linked_upload.assert_not_called()
|
||||
self.session.CGImport.assert_not_called()
|
||||
|
||||
def test_handle_import_cg_test(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata()
|
||||
arguments = [fn, self.outdir, '--test']
|
||||
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.linked_upload.assert_not_called()
|
||||
self.session.CGImport.assert_not_called()
|
||||
|
||||
def test_handle_import_cg_metaonly(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata(write_outputs=False)
|
||||
for info in metadata['output']:
|
||||
info['metadata_only'] = True
|
||||
metadata, fn = self.make_metadata(metadata, defaults=False, write_outputs=False)
|
||||
arguments = [fn, self.outdir]
|
||||
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.linked_upload.assert_not_called()
|
||||
self.session.CGImport.assert_called_once()
|
||||
args = self.session.CGImport.call_args.args
|
||||
self.assertEqual(args[0], metadata)
|
||||
|
||||
def test_handle_import_cg_draft(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata()
|
||||
arguments = [fn, self.outdir, '--draft']
|
||||
# metadata from the call should have draft flag added
|
||||
expect = metadata.copy()
|
||||
expect['build']['draft'] = True
|
||||
|
||||
# Case 1, running in fg, progress on
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
self.assertEqual(len(self.session.uploadWrapper.mock_calls), len(metadata['output']))
|
||||
self.session.CGImport.assert_called_once()
|
||||
args = self.session.CGImport.call_args.args
|
||||
self.assertEqual(args[0], metadata)
|
||||
|
||||
def test_handle_import_cg_reserve(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata()
|
||||
arguments = [fn, self.outdir, '--build-id', '12345']
|
||||
# metadata from the call should have draft flag added
|
||||
expect = metadata.copy()
|
||||
expect['build']['build_id'] = 12345
|
||||
|
||||
# Case 1, running in fg, progress on
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
self.assertEqual(len(self.session.uploadWrapper.mock_calls), len(metadata['output']))
|
||||
self.session.CGImport.assert_called_once()
|
||||
args = self.session.CGImport.call_args.args
|
||||
self.assertEqual(args[0], expect)
|
||||
|
||||
def test_handle_import_cg_linked(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata()
|
||||
arguments = [fn, self.outdir, '--link']
|
||||
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
self.session.uploadWrapper.assert_not_called()
|
||||
self.assertEqual(len(self.linked_upload.mock_calls), len(metadata['output']))
|
||||
self.session.CGImport.assert_called_once()
|
||||
args = self.session.CGImport.call_args.args
|
||||
self.assertEqual(args[0], metadata)
|
||||
|
||||
def test_handle_import_cg_noprogress(self):
|
||||
"""Test handle_import_cg function"""
|
||||
|
||||
metadata, fn = self.make_metadata()
|
||||
arguments = [fn, self.outdir, '--noprogress']
|
||||
|
||||
handle_import_cg(self.options, self.session, arguments)
|
||||
|
||||
self.assertEqual(len(self.session.uploadWrapper.mock_calls), len(metadata['output']))
|
||||
kwargs = self.session.uploadWrapper.call_args.kwargs
|
||||
self.assertEqual(kwargs['callback'], None)
|
||||
self.session.CGImport.assert_called_once()
|
||||
args = self.session.CGImport.call_args.args
|
||||
self.assertEqual(args[0], metadata)
|
||||
self.linked_upload.assert_not_called()
|
||||
|
||||
def test_handle_import_cg_help(self):
|
||||
"""Test handle_import_cg help message"""
|
||||
|
|
@ -195,5 +257,4 @@ Options:
|
|||
""" % self.progname)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
# the end
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue