[virt-tools-list] [virt-bootstrap] [PATCH v6 01/26] Drop unit tests
Cedric Bosdonnat
cbosdonnat at suse.com
Sat Aug 19 05:27:07 UTC 2017
ACK, but you may have forgotten to remove the mock dependency in setup.py.
--
Cedric
On Thu, 2017-08-17 at 10:39 +0100, Radostin Stoyanov wrote:
> Unit tests were used to ensure that functions and methods work as
> expected. However, these tests are closely related to the
> implementation and will result in major changes after
> some refactoring. To reduce the amount needed to add new features or
> changes to the code these tests will be replaced with more abstract
> form of testing introduced in the following commits.
> ---
> tests/__init__.py | 39 ---
> tests/test_docker_source.py | 607 -------------------------------------
> tests/test_file_source.py | 171 -----------
> tests/test_progress.py | 112 -------
> tests/test_utils.py | 707 -------------------------------------------
> tests/test_virt_bootstrap.py | 464 ----------------------------
> 6 files changed, 2100 deletions(-)
> delete mode 100644 tests/__init__.py
> delete mode 100644 tests/test_docker_source.py
> delete mode 100644 tests/test_file_source.py
> delete mode 100644 tests/test_progress.py
> delete mode 100644 tests/test_utils.py
> delete mode 100644 tests/test_virt_bootstrap.py
>
> diff --git a/tests/__init__.py b/tests/__init__.py
> deleted file mode 100644
> index e82c6d5..0000000
> --- a/tests/__init__.py
> +++ /dev/null
> @@ -1,39 +0,0 @@
> -"""
> - Test suite for virt-bootstrap
> -
> - Authors: Radostin Stoyanov <rstoyanov1 at gmail.com>
> -
> - Copyright (C) 2017 Radostin Stoyanov
> -
> - This program is free software: you can redistribute it and/or modify
> - it under the terms of the GNU General Public License as published by
> - the Free Software Foundation, either version 3 of the License, or
> - (at your option) any later version.
> -
> - This program is distributed in the hope that it will be useful,
> - but WITHOUT ANY WARRANTY; without even the implied warranty of
> - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
> - GNU General Public License for more details.
> -
> - You should have received a copy of the GNU General Public License
> - along with this program. If not, see <http://www.gnu.org/licenses/>.
> -"""
> -
> -import sys
> -import unittest
> -
> -try:
> - import mock
> -except ImportError:
> - import unittest.mock as mock
> -
> -sys.path += '../src' # noqa: E402
> -
> -# pylint: disable=import-error
> -from virtBootstrap import virt_bootstrap
> -from virtBootstrap import sources
> -from virtBootstrap import progress
> -from virtBootstrap import utils
> -
> -__all__ = ['unittest', 'mock',
> - 'virt_bootstrap', 'sources', 'progress', 'utils']
> diff --git a/tests/test_docker_source.py b/tests/test_docker_source.py
> deleted file mode 100644
> index 4859e1b..0000000
> --- a/tests/test_docker_source.py
> +++ /dev/null
> @@ -1,607 +0,0 @@
> -# Authors: Radostin Stoyanov <rstoyanov1 at gmail.com>
> -#
> -# Copyright (C) 2017 Radostin Stoyanov
> -#
> -# This program is free software: you can redistribute it and/or modify
> -# it under the terms of the GNU General Public License as published by
> -# the Free Software Foundation, either version 3 of the License, or
> -# (at your option) any later version.
> -
> -# This program is distributed in the hope that it will be useful,
> -# but WITHOUT ANY WARRANTY; without even the implied warranty of
> -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
> -# GNU General Public License for more details.
> -
> -# You should have received a copy of the GNU General Public License
> -# along with this program. If not, see <http://www.gnu.org/licenses/>.
> -
> -
> -"""
> -Unit tests for methods defined in virtBootstrap.sources.DockerSource
> -"""
> -
> -from tests import unittest
> -from tests import mock
> -from tests import sources
> -
> -try:
> - from urlparse import urlparse
> -except ImportError:
> - from urllib.parse import urlparse
> -
> -
> -# pylint: disable=invalid-name
> -# pylint: disable=too-many-public-methods
> -class TestDockerSource(unittest.TestCase):
> - """
> - Test cases for DockerSource
> - """
> - def _mock_docker_source(self):
> - """
> - This method returns an instance of Mock object
> - that acts as the specification for the DockerSource.
> - """
> - m_self = mock.Mock(spec=sources.DockerSource)
> - m_self.progress = mock.Mock()
> - m_self.no_cache = False
> - m_self.url = "docker://test"
> - m_self.images_dir = "/images_path"
> - m_self.insecure = True
> - m_self.username = 'user'
> - m_self.password = 'password'
> - m_self.layers = [
> - ['sha256', '75c416ea', '/images_path/75c416ea.tar', ''],
> - ['sha256', 'a7050fc1', '/images_path/a7050fc1.tar', '']
> - ]
> - return m_self
> -
> - ###################################
> - # Tests for: __init__()
> - ###################################
> - def test_argument_assignment(self):
> - """
> - Ensures that __init__() assigns the arguments' values to instance
> - variables.
> - """
> - kwargs = {'uri': '',
> - 'fmt': 'dir',
> - 'not_secure': False,
> - 'no_cache': False,
> - 'progress': mock.Mock(),
> - 'username': 'username',
> - 'password': 'password'}
> -
> - with mock.patch('virtBootstrap.utils'
> - '.get_image_dir') as m_get_image_dir:
> - with mock.patch.multiple('virtBootstrap.sources.DockerSource',
> - retrieve_layers_info=mock.DEFAULT,
> - gen_valid_uri=mock.DEFAULT) as mocked:
> - src_instance = sources.DockerSource(**kwargs)
> -
> - test_values = {
> - src_instance.url: mocked['gen_valid_uri'].return_value,
> - src_instance.progress: kwargs['progress'].update_progress,
> - src_instance.username: kwargs['username'],
> - src_instance.password: kwargs['password'],
> - src_instance.output_format: kwargs['fmt'],
> - src_instance.no_cache: kwargs['no_cache'],
> - src_instance.insecure: kwargs['not_secure'],
> - src_instance.images_dir: m_get_image_dir()
> - }
> - for value in test_values:
> - self.assertIs(value, test_values[value])
> -
> - def test_source_password_is_required_if_username_specifed(self):
> - """
> - Ensures that __init__() calls getpass() to request password
> - when username is specified and password is not.
> - """
> - test_password = 'secret'
> -
> - kwargs = {arg: '' for arg
> - in ['uri', 'fmt', 'not_secure', 'password', 'no_cache']}
> - kwargs['progress'] = mock.Mock()
> - kwargs['username'] = 'test'
> -
> - with mock.patch('virtBootstrap.utils.get_image_dir'):
> - with mock.patch('getpass.getpass') as m_getpass:
> - m_getpass.return_value = test_password
> - with mock.patch.multiple('virtBootstrap.sources.DockerSource',
> - retrieve_layers_info=mock.DEFAULT,
> - gen_valid_uri=mock.DEFAULT):
> - src_instance = sources.DockerSource(**kwargs)
> -
> - m_getpass.assert_called_once()
> - self.assertIs(test_password, src_instance.password)
> -
> - ###################################
> - # Tests for: retrieve_layers_info()
> - ###################################
> - def _mock_retrieve_layers_info(self, manifest, kwargs):
> - """
> - This method is gather common test pattern used in the following
> - two test cases.
> - """
> - with mock.patch.multiple('virtBootstrap.utils',
> - get_image_details=mock.DEFAULT,
> - get_image_dir=mock.DEFAULT) as m_utils:
> -
> - m_utils['get_image_details'].return_value = manifest
> - m_utils['get_image_dir'].return_value = '/images_path'
> -
> - patch_method = 'virtBootstrap.sources.DockerSource.gen_valid_uri'
> - with mock.patch(patch_method) as m_uri:
> - src_instance = sources.DockerSource(**kwargs)
> - return (src_instance, m_uri, m_utils)
> -
> - def test_retrieve_layers_info_pass_arguments_to_get_image_details(self):
> - """
> - Ensures that retrieve_layers_info() calls get_image_details()
> - with all passed arguments.
> - """
> - src_kwargs = {
> - 'uri': '',
> - 'progress': mock.Mock()
> - }
> -
> - manifest = {'schemaVersion': 2, 'layers': []}
> - (src_instance,
> - m_uri, m_utils) = self._mock_retrieve_layers_info(manifest,
> - src_kwargs)
> -
> - kwargs = {
> - 'insecure': src_instance.insecure,
> - 'username': src_instance.username,
> - 'password': src_instance.password,
> - 'raw': True
> - }
> - m_utils['get_image_details'].assert_called_once_with(m_uri(), **kwargs)
> -
> - def test_retrieve_layers_info_schema_version_1(self):
> - """
> - Ensures that retrieve_layers_info() extracts the layers' information
> - from manifest with schema version 1 a list with format:
> - ["digest", "sum_type", "file_path", "size"].
> - """
> - kwargs = {
> - 'uri': '',
> - 'progress': mock.Mock()
> - }
> -
> - manifest = {
> - 'schemaVersion': 1,
> - 'fsLayers': [
> - {'blobSum': 'sha256:75c416ea'},
> - {'blobSum': 'sha256:c6ff40b6'},
> - {'blobSum': 'sha256:a7050fc1'}
> - ]
> - }
> -
> - expected_result = [
> - ['sha256', 'a7050fc1', '/images_path/a7050fc1.tar', None],
> - ['sha256', 'c6ff40b6', '/images_path/c6ff40b6.tar', None],
> - ['sha256', '75c416ea', '/images_path/75c416ea.tar', None]
> - ]
> -
> - src_instance = self._mock_retrieve_layers_info(manifest, kwargs)[0]
> - self.assertEqual(src_instance.layers, expected_result)
> -
> - def test_retrieve_layers_info_schema_version_2(self):
> - """
> - Ensures that retrieve_layers_info() extracts the layers' information
> - from manifest with schema version 2 a list with format:
> - ["digest", "sum_type", "file_path", "size"].
> - """
> - kwargs = {
> - 'uri': '',
> - 'progress': mock.Mock()
> - }
> -
> - manifest = {
> - 'schemaVersion': 2,
> - "layers": [
> - {"size": 47103294, "digest": "sha256:75c416ea"},
> - {"size": 814, "digest": "sha256:c6ff40b6"},
> - {"size": 513, "digest": "sha256:a7050fc1"}
> - ]
> - }
> -
> - expected_result = [
> - ['sha256', '75c416ea', '/images_path/75c416ea.tar', 47103294],
> - ['sha256', 'c6ff40b6', '/images_path/c6ff40b6.tar', 814],
> - ['sha256', 'a7050fc1', '/images_path/a7050fc1.tar', 513]
> - ]
> -
> - src_instance = self._mock_retrieve_layers_info(manifest, kwargs)[0]
> - self.assertEqual(src_instance.layers, expected_result)
> -
> - def test_retrieve_layers_info_raise_error_on_invalid_schema_version(self):
> - """
> - Ensures that retrieve_layers_info() calls get_image_details()
> - with all passed arguments.
> - """
> - kwargs = {
> - 'uri': '',
> - 'progress': mock.Mock()
> - }
> -
> - manifest = {'schemaVersion': 3}
> - with self.assertRaises(ValueError):
> - self._mock_retrieve_layers_info(manifest, kwargs)
> -
> - ###################################
> - # Tests for: gen_valid_uri()
> - ###################################
> - def test_gen_valid_uri(self):
> - """
> - Validates the output of gen_valid_uri() for some test cases.
> - """
> - m_self = self._mock_docker_source()
> - test_values = {
> - 'docker:///repo': 'docker://repo',
> - 'docker:/repo': 'docker://repo',
> - 'docker://repo/': 'docker://repo',
> - 'docker://repo/image/': 'docker://repo/image',
> - 'docker:///repo/image/': 'docker://repo/image',
> - }
> - for uri in test_values:
> - uri_obj = urlparse(uri)
> - result = sources.DockerSource.gen_valid_uri(m_self, uri_obj)
> - expected = test_values[uri]
> - self.assertEqual(result, expected)
> -
> - ###################################
> - # Tests for: download_image()
> - ###################################
> - def test_download_image(self):
> - """
> - Ensures that download_image() calls read_skopeo_progress() with
> - expected skopeo copy command and removes tha leftover manifest file.
> - """
> - m_self = self._mock_docker_source()
> - m_self.read_skopeo_progress = mock.Mock()
> - manifest_path = "%s/manifest.json" % m_self.images_dir
> - with mock.patch('os.remove') as m_remove:
> - sources.DockerSource.download_image(m_self)
> -
> - expected_call = ["skopeo", "copy", m_self.url,
> - "dir:" + m_self.images_dir,
> - '--src-tls-verify=false',
> - '--src-creds={}:{}'.format(m_self.username,
> - m_self.password)]
> - m_self.read_skopeo_progress.assert_called_once_with(expected_call)
> - m_remove.assert_called_once_with(manifest_path)
> -
> - ###################################
> - # Tests for: parse_output()
> - ###################################
> - def test_parse_output_return_false_on_fail(self):
> - """
> - Ensures that parse_output() returns False when process call
> - exits with non-zero code.
> - """
> - m_self = mock.Mock(spec=sources.DockerSource)
> - m_self.layers = []
> - m_proc = mock.Mock()
> - m_proc.returncode = 1
> - self.assertFalse(sources.DockerSource.parse_output(m_self, m_proc))
> -
> - def test_parse_output(self):
> - """
> - Ensures that parse_output() recognises processing of different
> - layers from the skopeo's output.
> - """
> - m_self = self._mock_docker_source()
> - m_proc = mock.Mock()
> - m_proc.poll.return_value = None
> - m_proc.returncode = 0
> - test_values = '\n'.join([
> - 'Skipping fetch of repeat blob sha256:c6ff40',
> - 'Copying blob sha256:75c416ea735c4',
> - '40.00 MB / 44.92 MB [======================>------]',
> - 'Copying config sha256:d355ed35',
> - '40.00 MB / 44.92 MB [======================>------]'
> - ])
> -
> - expected_progress_calls = [
> - mock.call("Downloading layer (1/2)"),
> - mock.call("Downloading layer (2/2)"),
> - ]
> -
> - with mock.patch('select.select') as m_select:
> - m_select.return_value = [[test_values], [], []]
> - with mock.patch('virtBootstrap.utils.read_async') as m_read_async:
> - m_read_async.return_value = test_values
> - self.assertTrue(sources.DockerSource.parse_output(m_self,
> - m_proc))
> - m_select.assert_called_once_with([m_proc.stdout], [], [])
> - m_read_async.assert_called_once_with(test_values)
> - m_self.progress.assert_has_calls(expected_progress_calls)
> - m_self.update_progress_from_output.assert_called_once()
> - m_proc.wait.assert_called_once()
> -
> - ###################################
> - # Tests for: update_progress_from_output()
> - ###################################
> - def _mock_update_progress_from_output(self, test_values):
> - """
> - This method is gather common test pattern used in the following
> - two test cases.
> - """
> - m_self = self._mock_docker_source()
> - test_method = sources.DockerSource.update_progress_from_output
> - for line in test_values:
> - test_method(m_self, line.split(), 1, len(test_values))
> -
> - return m_self.progress.call_args_list
> -
> - def test_update_progress_from_output(self):
> - """
> - Ensures that update_progress_from_output() recognises the current
> - downloaded size, the total layer's size and calculates correct
> - percentage value.
> - """
> - test_values = [
> - '500.00 KB / 4.00 MB [======>------]',
> - '25.00 MB / 24.10 MB [======>------]',
> - '40.00 MB / 50.00 MB [======>------]',
> - ]
> - expected_values = [2, 17.33, 13.33]
> -
> - calls = self._mock_update_progress_from_output(test_values)
> - for call, expected in zip(calls, expected_values):
> - self.assertAlmostEqual(call[1]['value'], expected, places=1)
> -
> - def test_update_progress_from_output_ignore_failures(self):
> - """
> - Ensures that update_progress_from_output() ignores invalid lines
> - from skopeo's output.
> - """
> - test_values = [
> - 'a ',
> - '1 ' * 5,
> - '500.00 MB / 0.00 MB [======>------]',
> - '00.00 MB / 00.00 MB [======>------]',
> - ]
> - self._mock_update_progress_from_output(test_values)
> -
> - ###################################
> - # Tests for: read_skopeo_progress()
> - ###################################
> - def _mock_read_skopeo_progress(self, test_cmd, parse_output_return):
> - """
> - This method is gather common test pattern used in the following
> - two test cases.
> - """
> - m_self = mock.Mock(spec=sources.DockerSource)
> - m_self.parse_output.return_value = parse_output_return
> - with mock.patch.multiple('virtBootstrap.sources.'
> - 'docker_source.subprocess',
> - Popen=mock.DEFAULT,
> - PIPE=mock.DEFAULT) as mocked:
> - with mock.patch('virtBootstrap.utils.make_async') as m_make_async:
> - sources.DockerSource.read_skopeo_progress(m_self, test_cmd)
> -
> - return (mocked, m_make_async)
> -
> - def test_read_skopeo_progress(self):
> - """
> - Ensures that read_skopeo_progress() calls make_async() with
> - the stdout pipe of skopeo's process.
> - """
> - test_cmd = 'test'
> - mocked, m_make_async = self._mock_read_skopeo_progress(test_cmd, True)
> -
> - mocked['Popen'].assert_called_once_with(test_cmd,
> - stdout=mocked['PIPE'],
> - stderr=mocked['PIPE'],
> - universal_newlines=True)
> - m_make_async.assert_called_once_with(mocked['Popen']().stdout)
> -
> - def test_read_skopeo_progress_raise_error(self):
> - """
> - Ensures that read_skopeo_progress() raise CalledProcessError
> - when parse_output() returns false.
> - """
> - with self.assertRaises(sources.docker_source
> - .subprocess.CalledProcessError):
> - self._mock_read_skopeo_progress('test', False)
> -
> - ###################################
> - # Tests for: validate_image_layers()
> - ###################################
> - def _mock_validate_image_layers(self,
> - checksum_return,
> - path_exists_return,
> - expected_result,
> - check_calls=False):
> - """
> - This method is gather common test pattern used in the following
> - three test cases.
> - """
> - m_self = self._mock_docker_source()
> -
> - with mock.patch('os.path.exists') as m_path_exists:
> - with mock.patch('virtBootstrap.utils.checksum') as m_checksum:
> - m_checksum.return_value = checksum_return
> - m_path_exists.return_value = path_exists_return
> - result = sources.DockerSource.validate_image_layers(m_self)
> - self.assertEqual(result, expected_result)
> -
> - if check_calls:
> - path_exists_expected_calls = []
> - checksum_expected_calls = []
> - # Generate expected calls
> - for sum_type, hash_sum, path, _ignore in m_self.layers:
> - path_exists_expected_calls.append(mock.call(path))
> - checksum_expected_calls.append(
> - mock.call(path, sum_type, hash_sum))
> -
> - m_path_exists.assert_has_calls(path_exists_expected_calls)
> - m_checksum.assert_has_calls(checksum_expected_calls)
> -
> - def test_validate_image_layers_should_return_true(self):
> - """
> - Ensures that validate_image_layers() returns True when:
> - - checksum() returns True for all layers
> - - the file path of all layers exist
> - - all layers are validated
> - """
> - self._mock_validate_image_layers(True, True, True, True)
> -
> - def test_validate_image_layers_return_false_if_path_not_exist(self):
> - """
> - Ensures that validate_image_layers() returns False when
> - checksum() returns False.
> - """
> - self._mock_validate_image_layers(False, True, False)
> -
> - def test_validate_image_layers_return_false_if_checksum_fail(self):
> - """
> - Ensures that validate_image_layers() returns False when
> - the file path of layer does not exist.
> - """
> - self._mock_validate_image_layers(True, False, False)
> -
> - ###################################
> - # Tests for: fetch_layers()
> - ###################################
> - def _mock_fetch_layers(self, validate_return):
> - """
> - This method is gather common test pattern used in the following
> - two test cases.
> - """
> - m_self = mock.Mock(spec=sources.DockerSource)
> - m_self.validate_image_layers.return_value = validate_return
> - sources.DockerSource.fetch_layers(m_self)
> - return m_self
> -
> - def test_fetch_layers_should_call_download_image(self):
> - """
> - Ensures that fetch_layers() calls download_image()
> - when validate_image_layers() returns False.
> - """
> - m_self = self._mock_fetch_layers(False)
> - m_self.download_image.assert_called_once()
> -
> - def test_fetch_layers_should_not_call_download_image(self):
> - """
> - Ensures that fetch_layers() does not call download_image()
> - when validate_image_layers() returns True.
> - """
> - m_self = self._mock_fetch_layers(True)
> - m_self.download_image.assert_not_called()
> -
> - ###################################
> - # Tests for: unpack()
> - ###################################
> - def _unpack_test_fmt(self, output_format, patch_method=None,
> - side_effect=None, m_self=None):
> - """
> - This method is gather common test pattern used in the following
> - two test cases.
> - """
> - m_self = m_self if m_self else self._mock_docker_source()
> - m_self.output_format = output_format
> - dest = 'foo'
> -
> - if patch_method:
> - with mock.patch(patch_method) as mocked:
> - if side_effect:
> - mocked.side_effect = side_effect
> - sources.DockerSource.unpack(m_self, dest)
> -
> - mocked.assert_called_once_with(m_self.layers, dest,
> - m_self.progress)
> - else:
> - sources.DockerSource.unpack(m_self, dest)
> -
> - m_self.fetch_layers.assert_called_once()
> -
> - def test_unpack_dir_format(self):
> - """
> - Ensures that unpack() calls untar_layers() when the output format
> - is set to 'dir'.
> - """
> - self._unpack_test_fmt('dir', 'virtBootstrap.utils.untar_layers')
> -
> - def test_unpack_qcow2_format(self):
> - """
> - Ensures that unpack() calls extract_layers_in_qcow2() when the
> - output format is set to 'qcow2'.
> - """
> - self._unpack_test_fmt('qcow2',
> - 'virtBootstrap.utils.extract_layers_in_qcow2')
> -
> - def unpack_raise_error_test(self,
> - output_format,
> - patch_method,
> - side_effect=None,
> - msg=None):
> - """
> - This method is gather common test pattern used in the following
> - four test cases.
> - """
> - with self.assertRaises(Exception) as err:
> - self._unpack_test_fmt(output_format, patch_method,
> - side_effect)
> - if msg:
> - self.assertEqual(msg, str(err.exception))
> -
> - def test_unpack_raise_error_for_unknown_format(self):
> - """
> - Ensures that unpack() throws an Exception when called with
> - invalid output format.
> - """
> - msg = 'Unknown format:foo'
> - self.unpack_raise_error_test('foo', None, None, msg)
> -
> - def test_unpack_raise_error_if_untar_fail(self):
> - """
> - Ensures that unpack() throws an Exception when untar_layers()
> - fails.
> - """
> - msg = 'Caught untar failure'
> - side_effect = Exception(msg)
> - patch_method = 'virtBootstrap.utils.untar_layers'
> - self.unpack_raise_error_test('dir', patch_method, side_effect, msg)
> -
> - def test_unpack_raise_error_if_extract_in_qcow2_fail(self):
> - """
> - Ensures that unpack() throws an Exception when
> - extract_layers_in_qcow2() fails.
> - """
> - msg = 'Caught extract_layers_in_qcow2 failure'
> - side_effect = Exception(msg)
> - patch_method = 'virtBootstrap.utils.extract_layers_in_qcow2'
> - self.unpack_raise_error_test('qcow2', patch_method, side_effect, msg)
> -
> - def test_unpack_no_cache_clean_up(self):
> - """
> - Ensures that unpack() removes the folder which stores tar archives
> - of image layers when no_cache is set to True.
> - """
> - output_formats = ['dir', 'qcow2']
> - patch_methods = [
> - 'virtBootstrap.utils.untar_layers',
> - 'virtBootstrap.utils.extract_layers_in_qcow2'
> - ]
> - for fmt, patch_mthd in zip(output_formats, patch_methods):
> - m_self = self._mock_docker_source()
> - m_self.no_cache = True
> - with mock.patch('shutil.rmtree') as m_shutil:
> - self._unpack_test_fmt(fmt, patch_mthd, m_self=m_self)
> - m_shutil.assert_called_once_with(m_self.images_dir)
> -
> - def test_unpack_no_cache_clean_up_on_failure(self):
> - """
> - Ensures that unpack() removes the folder which stores tar archives
> - of image layers when no_cache is set to True and exception was
> - raised.
> - """
> - m_self = self._mock_docker_source()
> - m_self.no_cache = True
> - with self.assertRaises(Exception):
> - with mock.patch('shutil.rmtree') as m_rmtree:
> - self._unpack_test_fmt('foo', None, m_self=m_self)
> - m_rmtree.assert_called_once_with(m_self.images_dir)
> diff --git a/tests/test_file_source.py b/tests/test_file_source.py
> deleted file mode 100644
> index 6e89aa2..0000000
> --- a/tests/test_file_source.py
> +++ /dev/null
> @@ -1,171 +0,0 @@
> -# Authors: Radostin Stoyanov <rstoyanov1 at gmail.com>
> -#
> -# Copyright (C) 2017 Radostin Stoyanov
> -#
> -# This program is free software: you can redistribute it and/or modify
> -# it under the terms of the GNU General Public License as published by
> -# the Free Software Foundation, either version 3 of the License, or
> -# (at your option) any later version.
> -
> -# This program is distributed in the hope that it will be useful,
> -# but WITHOUT ANY WARRANTY; without even the implied warranty of
> -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
> -# GNU General Public License for more details.
> -
> -# You should have received a copy of the GNU General Public License
> -# along with this program. If not, see <http://www.gnu.org/licenses/>.
> -
> -
> -"""
> -Unit tests for methods defined in virtBootstrap.sources.FileSource
> -"""
> -
> -from tests import unittest
> -from tests import mock
> -from tests import sources
> -
> -
> -# pylint: disable=invalid-name
> -class TestFileSource(unittest.TestCase):
> - """
> - Test cases for FileSource
> - """
> -
> - ###################################
> - # Tests for: __init__()
> - ###################################
> - def test_argument_assignment(self):
> - """
> - Ensures that __init__() assigns the arguments' values to instance
> - variables.
> - """
> - kwargs = {'uri': mock.Mock(),
> - 'fmt': 'dir',
> - 'progress': mock.Mock()}
> -
> - src_instance = sources.FileSource(**kwargs)
> -
> - test_values = {
> - src_instance.path: kwargs['uri'].path,
> - src_instance.output_format: kwargs['fmt'],
> - src_instance.progress: kwargs['progress'].update_progress
> - }
> - for value in test_values:
> - self.assertIs(value, test_values[value])
> -
> - ###################################
> - # Tests for: unpack()
> - ###################################
> - def test_unpack_invalid_source_raise_exception(self):
> - """
> - Ensures that unpack() throws an Exception when called with
> - invalid file source.
> - """
> - m_self = mock.Mock(spec=sources.FileSource)
> - m_self.path = 'foo'
> - with mock.patch('os.path.isfile') as m_isfile:
> - m_isfile.return_value = False
> - with self.assertRaises(Exception) as err:
> - sources.FileSource.unpack(m_self, 'bar')
> - self.assertIn('Invalid file source', str(err.exception))
> -
> - def test_unpack_to_dir(self):
> - """
> - Ensures that unpack() calls safe_untar() when the output format
> - is set to 'dir'.
> - """
> - m_self = mock.Mock(spec=sources.FileSource)
> - m_self.progress = mock.Mock()
> - m_self.path = 'foo'
> - m_self.output_format = 'dir'
> - dest = 'bar'
> -
> - with mock.patch('os.path.isfile') as m_isfile:
> - m_isfile.return_value = True
> - with mock.patch('virtBootstrap.utils.safe_untar') as m_untar:
> - sources.FileSource.unpack(m_self, dest)
> -
> - m_untar.assert_called_once_with(m_self.path, dest)
> -
> - def test_unpack_to_qcow2(self):
> - """
> - Ensures that unpack() calls create_qcow2() when the output
> - format is set to 'qcow2'.
> - """
> - m_self = mock.Mock(spec=sources.FileSource)
> - m_self.progress = mock.Mock()
> - m_self.path = 'foo'
> - m_self.output_format = 'qcow2'
> - dest = 'bar'
> - qcow2_file_path = 'foobar'
> -
> - with mock.patch.multiple('os.path',
> - isfile=mock.DEFAULT,
> - realpath=mock.DEFAULT) as mocked:
> -
> - mocked['isfile'].return_value = True
> - mocked['realpath'].return_value = qcow2_file_path
> - with mock.patch('virtBootstrap.utils.create_qcow2') as m_qcow2:
> - sources.FileSource.unpack(m_self, dest)
> -
> - m_qcow2.assert_called_once_with(m_self.path, qcow2_file_path)
> -
> - def _unpack_raise_error_test(self,
> - output_format,
> - side_effect=None,
> - patch_method=None,
> - msg=None):
> - """
> - This method is gather common test pattern used in the following
> - three test cases.
> - """
> - m_self = mock.Mock(spec=sources.FileSource)
> - m_self.progress = mock.Mock()
> - m_self.path = 'foo'
> - m_self.output_format = output_format
> - dest = 'bar'
> -
> - with mock.patch.multiple('os.path',
> - isfile=mock.DEFAULT,
> - realpath=mock.DEFAULT) as m_path:
> - m_path['isfile'].return_value = True
> - with self.assertRaises(Exception) as err:
> - if patch_method:
> - with mock.patch(patch_method) as mocked_method:
> - mocked_method.side_effect = side_effect
> - sources.FileSource.unpack(m_self, dest)
> - else:
> - sources.FileSource.unpack(m_self, dest)
> - if msg:
> - self.assertEqual(msg, str(err.exception))
> -
> - def test_unpack_invalid_format_raise_exception(self):
> - """
> - Ensures that unpack() throws an Exception when called with
> - invalid output format.
> - """
> - self._unpack_raise_error_test('foo', msg='Unknown format:foo')
> -
> - def test_unpack_raise_error_if_untar_fail(self):
> - """
> - Ensures that unpack() throws an Exception when safe_untar()
> - fails.
> - """
> - msg = 'Caught untar failure'
> - patch_method = 'virtBootstrap.utils.safe_untar'
> - self._unpack_raise_error_test(output_format='dir',
> - side_effect=Exception(msg),
> - patch_method=patch_method,
> - msg=msg)
> -
> - def test_unpack_raise_error_if_extract_in_qcow2_fail(self):
> - """
> - Ensures that unpack() throws an Exception when create_qcow2()
> - fails.
> - """
> - msg = 'Caught extract_layers_in_qcow2 failure'
> - patch_method = 'virtBootstrap.utils.create_qcow2'
> - self._unpack_raise_error_test(output_format='qcow2',
> - side_effect=Exception(msg),
> - patch_method=patch_method,
> - msg=msg)
> diff --git a/tests/test_progress.py b/tests/test_progress.py
> deleted file mode 100644
> index 1f609d5..0000000
> --- a/tests/test_progress.py
> +++ /dev/null
> @@ -1,112 +0,0 @@
> -# Authors: Radostin Stoyanov <rstoyanov1 at gmail.com>
> -#
> -# Copyright (C) 2017 Radostin Stoyanov
> -#
> -# This program is free software: you can redistribute it and/or modify
> -# it under the terms of the GNU General Public License as published by
> -# the Free Software Foundation, either version 3 of the License, or
> -# (at your option) any later version.
> -
> -# This program is distributed in the hope that it will be useful,
> -# but WITHOUT ANY WARRANTY; without even the implied warranty of
> -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
> -# GNU General Public License for more details.
> -
> -# You should have received a copy of the GNU General Public License
> -# along with this program. If not, see <http://www.gnu.org/licenses/>.
> -
> -
> -"""
> -Unit tests for methods defined in virtBootstrap.progress
> -"""
> -
> -from tests import unittest
> -from tests import mock
> -from tests import progress
> -
> -
> -# pylint: disable=invalid-name
> -class TestFileSource(unittest.TestCase):
> - """
> - Test cases for Progress module
> - """
> -
> - ###################################
> - # Tests for: __init__()
> - ###################################
> - def test_progress_init(self):
> - """
> - Ensures that __init__() assigns the collback value to instance
> - variable and creates dictionary with 'status', 'value' keys.
> - """
> - callback = mock.Mock()
> - test_instance = progress.Progress(callback)
> - for key in ['status', 'value']:
> - self.assertIn(key, test_instance.progress)
> - self.assertIs(callback, test_instance.callback)
> -
> - ###################################
> - # Tests for: get_progress()
> - ###################################
> - def test_get_progress(self):
> - """
> - Ensures that get_progress() returns copy of the progress dictionary
> - which has the same keys and values.
> - """
> - test_instance = progress.Progress()
> - test_result = test_instance.get_progress()
> - self.assertIsNot(test_instance.progress, test_result)
> - self.assertDictEqual(test_instance.progress, test_result)
> -
> - ###################################
> - # Tests for: update_progress()
> - ###################################
> - def test_update_progress_creates_log_record(self):
> - """
> - Ensures that update_progress() creates log record with info level
> - and pass the status value as message.
> - """
> - test_instance = progress.Progress()
> - logger = mock.Mock()
> - status = "Test"
> - test_instance.update_progress(status=status, logger=logger)
> - logger.info.assert_called_once_with(status)
> -
> - def test_update_progress_update_status_and_value(self):
> - """
> - Ensures that update_progress() creates log record with info level
> - and pass the status value as message.
> - """
> - test_instance = progress.Progress()
> - test_instance.progress = {'status': '', 'value': 0}
> - new_status = 'Test'
> - new_value = 100
> - new_progress = {'status': new_status, 'value': new_value}
> - test_instance.update_progress(status=new_status, value=new_value)
> - self.assertDictEqual(test_instance.progress, new_progress)
> -
> - def test_update_progress_update_raise_logger_error(self):
> - """
> - Ensures that update_progress() raise ValueError when creating
> - log record has failed.
> - """
> - msg = 'test'
> - test_instance = progress.Progress()
> - logger = mock.Mock()
> - logger.info.side_effect = Exception(msg)
> - with self.assertRaises(ValueError) as err:
> - test_instance.update_progress(logger=logger)
> - self.assertIn(msg, str(err.exception))
> -
> - def test_update_progress_update_raise_callback_error(self):
> - """
> - Ensures that update_progress() raise ValueError when calling
> - callback failed.
> - """
> - msg = 'test'
> - callback = mock.Mock()
> - callback.side_effect = Exception(msg)
> - test_instance = progress.Progress(callback)
> - with self.assertRaises(ValueError) as err:
> - test_instance.update_progress('foo', 'bar')
> - self.assertIn(msg, str(err.exception))
> diff --git a/tests/test_utils.py b/tests/test_utils.py
> deleted file mode 100644
> index 0b6ccc0..0000000
> --- a/tests/test_utils.py
> +++ /dev/null
> @@ -1,707 +0,0 @@
> -# Authors: Radostin Stoyanov <rstoyanov1 at gmail.com>
> -#
> -# Copyright (C) 2017 Radostin Stoyanov
> -#
> -# This program is free software: you can redistribute it and/or modify
> -# it under the terms of the GNU General Public License as published by
> -# the Free Software Foundation, either version 3 of the License, or
> -# (at your option) any later version.
> -
> -# This program is distributed in the hope that it will be useful,
> -# but WITHOUT ANY WARRANTY; without even the implied warranty of
> -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
> -# GNU General Public License for more details.
> -
> -# You should have received a copy of the GNU General Public License
> -# along with this program. If not, see <http://www.gnu.org/licenses/>.
> -
> -
> -"""
> -Unit tests for functions defined in virtBootstrap.utils
> -"""
> -
> -from tests import unittest
> -from tests import mock
> -from tests import utils
> -try:
> - # pylint: disable=redefined-builtin
> - from importlib import reload
> -except ImportError:
> - pass
> -
> -
> -# pylint: disable=invalid-name
> -# pylint: disable=too-many-public-methods
> -class TestUtils(unittest.TestCase):
> - """
> - Ensures that functions defined in the utils module of virtBootstrap
> - work as expected.
> - """
> -
> - ###################################
> - # Tests for: checksum()
> - ###################################
> - def test_utils_checksum_return_false_on_invalid_hash(self):
> - """
> - Ensures that checksum() returns False if the actual and expected
> - hash sum of file are not equal.
> - """
> - with mock.patch.multiple(utils,
> - open=mock.DEFAULT,
> - logger=mock.DEFAULT,
> - hashlib=mock.DEFAULT) as mocked:
> - path, sum_type, sum_expected = '/foo', 'sha256', 'bar'
> - mocked['hashlib'].sha256.hexdigest.return_value = False
> - self.assertFalse(utils.checksum(path, sum_type, sum_expected))
> -
> - def test_utils_checksum_return_false_if_file_could_not_be_opened(self):
> - """
> - Ensures that checksum() returns False if the file to be checked
> - cannot be open for read.
> - """
> - with mock.patch.multiple(utils,
> - open=mock.DEFAULT,
> - logger=mock.DEFAULT,
> - hashlib=mock.DEFAULT) as mocked:
> - mocked['open'].side_effect = IOError()
> - self.assertFalse(utils.checksum('foo', 'sha256', 'bar'))
> -
> - def test_utils_checksum_return_true_on_valid_hash(self):
> - """
> - Ensures that checksum() returns True when the actual and expected
> - hash sum of file are equal.
> - """
> - with mock.patch.multiple(utils,
> - open=mock.DEFAULT,
> - logger=mock.DEFAULT,
> - hashlib=mock.DEFAULT) as mocked:
> - path, sum_type, sum_expected = '/foo', 'sha256', 'bar'
> - mocked['hashlib'].sha256.return_value.hexdigest.return_value \
> - = sum_expected
> - self.assertTrue(utils.checksum(path, sum_type, sum_expected))
> -
> - ###################################
> - # Tests for: execute()
> - ###################################
> - def test_utils_execute_logging_on_successful_proc_call(self):
> - """
> - Ensures that execute() creates log record of cmd, stdout and stderr
> - when the exit code of process is 0.
> - """
> - with mock.patch.multiple(utils,
> - logger=mock.DEFAULT,
> - subprocess=mock.DEFAULT) as mocked:
> - cmd = ['foo']
> - output, err = 'test_out', 'test_err'
> -
> - mocked['subprocess'].Popen.return_value.returncode = 0
> - (mocked['subprocess'].Popen.return_value
> - .communicate.return_value) = (output.encode(), err.encode())
> -
> - utils.execute(cmd)
> - mocked['logger'].debug.assert_any_call("Call command:\n%s", cmd[0])
> - mocked['logger'].debug.assert_any_call("Stdout:\n%s", output)
> - mocked['logger'].debug.assert_any_call("Stderr:\n%s", err)
> -
> - def test_utils_execute_raise_error_on_unsuccessful_proc_call(self):
> - """
> - Ensures that execute() raise CalledProcessError exception when the
> - exit code of process is not 0.
> - """
> - with mock.patch('virtBootstrap.utils.subprocess.Popen') as m_popen:
> - m_popen.return_value.returncode = 1
> - m_popen.return_value.communicate.return_value = (b'output', b'err')
> - with self.assertRaises(utils.subprocess.CalledProcessError):
> - utils.execute(['foo'])
> -
> - ###################################
> - # Tests for: safe_untar()
> - ###################################
> - def test_utils_safe_untar_calls_execute(self):
> - """
> - Ensures that safe_untar() calls execute with virt-sandbox
> - command to extract source files to destination folder.
> - Test for users with EUID 0 and 1000.
> - """
> - with mock.patch('virtBootstrap.utils.os.geteuid') as m_geteuid:
> - for uid in [0, 1000]:
> - m_geteuid.return_value = uid
> - reload(utils)
> - with mock.patch('virtBootstrap.utils.execute') as m_execute:
> - src, dest = 'foo', 'bar'
> - utils.safe_untar('foo', 'bar')
> - cmd = ['virt-sandbox',
> - '-c', utils.LIBVIRT_CONN,
> - '-m', 'host-bind:/mnt=' + dest,
> - '--',
> - '/bin/tar', 'xf', src,
> - '-C', '/mnt',
> - '--exclude', 'dev/*']
> - m_execute.assert_called_once_with(cmd)
> -
> - ###################################
> - # Tests for: bytes_to_size()
> - ###################################
> - def test_utils_bytes_to_size(self):
> - """
> - Validates the output of bytes_to_size() for some test cases.
> - """
> - test_values = {
> - 0: '0', 1: '1', 512: '512', 1000: '0.98 KiB', 1024: '1 KiB',
> - 4096: '4 KiB', 5120: '5 KiB', 10 ** 10: '9.31 GiB'
> - }
> - for value in test_values:
> - self.assertEqual(utils.bytes_to_size(value), test_values[value])
> -
> - ###################################
> - # Tests for: size_to_bytes()
> - ###################################
> - def test_utils_size_to_bytes(self):
> - """
> - Validates the output of size_to_bytes() for some test cases.
> - """
> - test_values = [1, '0']
> - test_formats = ['TB', 'GB', 'MB', 'KB', 'B']
> - expected_output = [1099511627776, 1073741824, 1048576, 1024, 1,
> - 0, 0, 0, 0, 0]
> - i = 0
> - for value in test_values:
> - for fmt in test_formats:
> - self.assertEqual(utils.size_to_bytes(value, fmt),
> - expected_output[i])
> - i += 1
> -
> - ###################################
> - # Tests for: log_layer_extract()
> - ###################################
> - def test_utils_log_layer_extract(self):
> - """
> - Ensures that log_layer_extract() updates the progress and creates
> - log record with debug level.
> - """
> - m_progress = mock.Mock()
> - layer = ['sum_type', 'sum_value', 'layer_file', 'layer_size']
> - with mock.patch.multiple(utils, logger=mock.DEFAULT,
> - bytes_to_size=mock.DEFAULT) as mocked:
> - utils.log_layer_extract(layer, 'foo', 'bar', m_progress)
> - mocked['bytes_to_size'].assert_called_once_with('layer_size')
> - mocked['logger'].debug.assert_called_once()
> - m_progress.assert_called_once()
> -
> - ###################################
> - # Tests for: get_mime_type()
> - ###################################
> - @mock.patch('virtBootstrap.utils.subprocess.Popen')
> - def test_utils_get_mime_type(self, m_popen):
> - """
> - Ensures that get_mime_type() returns the detected MIME type
> - of /usr/bin/file.
> - """
> - path = "foo"
> - mime = "application/x-gzip"
> - stdout = ('%s: %s' % (path, mime)).encode()
> - m_popen.return_value.stdout.read.return_value = stdout
> - self.assertEqual(utils.get_mime_type(path), mime)
> - m_popen.assert_called_once_with(
> - ["/usr/bin/file", "--mime-type", path],
> - stdout=utils.subprocess.PIPE
> - )
> -
> - ###################################
> - # Tests for: untar_layers()
> - ###################################
> - def test_utils_untar_all_layers_in_order(self):
> - """
> - Ensures that untar_layers() iterates through all passed layers
> - in order.
> - """
> - layers = ['l1', 'l2', 'l3']
> - layers_list = [['', '', layer] for layer in layers]
> - dest_dir = '/foo'
> - expected_calls = [mock.call(layer, dest_dir) for layer in layers]
> - with mock.patch.multiple(utils,
> - safe_untar=mock.DEFAULT,
> - log_layer_extract=mock.DEFAULT) as mocked:
> - utils.untar_layers(layers_list, dest_dir, mock.Mock())
> - mocked['safe_untar'].assert_has_calls(expected_calls)
> -
> - ###################################
> - # Tests for: create_qcow2()
> - ###################################
> - def _apply_test_to_create_qcow2(self, expected_calls, *args):
> - """
> - This method contains common test pattern used in the next two
> - test cases.
> - """
> - with mock.patch.multiple(utils,
> - execute=mock.DEFAULT,
> - logger=mock.DEFAULT,
> - get_mime_type=mock.DEFAULT) as mocked:
> - mocked['get_mime_type'].return_value = 'application/x-gzip'
> - utils.create_qcow2(*args)
> - mocked['execute'].assert_has_calls(expected_calls)
> -
> - def test_utils_create_qcow2_base_layer(self):
> - """
> - Ensures that create_qcow2() creates base layer when
> - backing_file = None.
> - """
> - tar_file = 'foo'
> - layer_file = 'bar'
> - size = '5G'
> - backing_file = None
> -
> - expected_calls = [
> - mock.call(["qemu-img", "create", "-f", "qcow2", layer_file, size]),
> -
> - mock.call(['virt-format',
> - '--format=qcow2',
> - '--partition=none',
> - '--filesystem=ext3',
> - '-a', layer_file]),
> -
> - mock.call(['guestfish',
> - '-a', layer_file,
> - '-m', '/dev/sda',
> - 'tar-in', tar_file, '/', 'compress:gzip'])
> - ]
> -
> - self._apply_test_to_create_qcow2(expected_calls, tar_file, layer_file,
> - backing_file, size)
> -
> - def test_utils_create_qcow2_layer_with_backing_chain(self):
> - """
> - Ensures that create_qcow2() creates new layer with backing chains
> - when backing_file is specified.
> - """
> - tar_file = 'foo'
> - layer_file = 'bar'
> - backing_file = 'base'
> - size = '5G'
> -
> - expected_calls = [
> - mock.call(['qemu-img', 'create',
> - '-b', backing_file,
> - '-f', 'qcow2',
> - layer_file, size]),
> -
> - mock.call(['guestfish',
> - '-a', layer_file,
> - '-m', '/dev/sda',
> - 'tar-in', tar_file, '/', 'compress:gzip'])
> - ]
> -
> - self._apply_test_to_create_qcow2(expected_calls, tar_file, layer_file,
> - backing_file, size)
> -
> - ###################################
> - # Tests for: extract_layers_in_qcow2()
> - ###################################
> - def test_utils_if_all_layers_extracted_in_order_in_qcow2(self):
> - """
> - Ensures that extract_layers_in_qcow2() iterates through all
> - layers in order.
> - """
> - layers = ['l1', 'l2', 'l3']
> - layers_list = [['', '', layer] for layer in layers]
> - dest_dir = '/foo'
> -
> - # Generate expected calls
> - expected_calls = []
> - qcow2_backing_file = None
> - for index, layer in enumerate(layers):
> - qcow2_layer_file = dest_dir + "/layer-%s.qcow2" % index
> - expected_calls.append(
> - mock.call(layer, qcow2_layer_file, qcow2_backing_file))
> - qcow2_backing_file = qcow2_layer_file
> -
> - # Mocking out and execute
> - with mock.patch.multiple(utils,
> - create_qcow2=mock.DEFAULT,
> - log_layer_extract=mock.DEFAULT) as mocked:
> - utils.extract_layers_in_qcow2(layers_list, dest_dir, mock.Mock())
> -
> - # Check actual calls
> - mocked['create_qcow2'].assert_has_calls(expected_calls)
> -
> - ###################################
> - # Tests for: get_image_dir()
> - ###################################
> - def test_utils_getimage_dir(self):
> - """
> - Ensures that get_image_dir() returns path to DEFAULT_IMG_DIR
> - if the no_cache argument is set to False and create it if
> - does not exist.
> - """
> - # Perform this test for UID 0 and 1000
> - for uid in [0, 1000]:
> - with mock.patch('os.geteuid') as m_geteuid:
> - m_geteuid.return_value = uid
> - reload(utils)
> - with mock.patch('os.makedirs') as m_makedirs:
> - with mock.patch('os.path.exists') as m_path_exists:
> - m_path_exists.return_value = False
> - self.assertEqual(utils.get_image_dir(False),
> - utils.DEFAULT_IMG_DIR)
> - m_makedirs.assert_called_once_with(utils.DEFAULT_IMG_DIR)
> -
> - @mock.patch('tempfile.mkdtemp')
> - def test_utils_getimage_dir_no_cache(self, m_mkdtemp):
> - """
> - Ensures that get_image_dir() returns temporary file path created
> - by tempfile.mkdtemp.
> - """
> - m_mkdtemp.return_value = 'foo'
> - self.assertEqual(utils.get_image_dir(True), 'foo')
> - m_mkdtemp.assert_called_once()
> -
> - ###################################
> - # Tests for: get_image_details()
> - ###################################
> - @mock.patch('virtBootstrap.utils.subprocess.Popen')
> - def test_utils_get_image_details_raise_error_on_fail(self, m_popen):
> - """
> - Ensures that get_image_details() throws ValueError exception
> - when stderr from skopeo is provided.
> - """
> - src = 'docker://foo'
> - m_popen.return_value.communicate.return_value = [b'', b'Error']
> - with self.assertRaises(ValueError):
> - utils.get_image_details(src)
> -
> - @mock.patch('virtBootstrap.utils.subprocess.Popen')
> - def test_utils_get_image_details_return_json_obj_on_success(self, m_popen):
> - """
> - Ensures that get_image_details() returns python dictionary which
> - represents the data provided from stdout of skopeo when stderr
> - is not present.
> - """
> - src = 'docker://foo'
> - json_dict = {'foo': 'bar'}
> - stdout = utils.json.dumps(json_dict).encode()
> - m_popen.return_value.communicate.return_value = [stdout, '']
> - self.assertDictEqual(utils.get_image_details(src), json_dict)
> -
> - def test_utils_get_image_details_all_argument_passed(self):
> - """
> - Ensures that get_image_details() pass all argument values to
> - skopeo inspect.
> - """
> - src = 'docker://foo'
> - raw, insecure = True, True
> - username, password = 'user', 'password'
> - cmd = ['skopeo', 'inspect', src,
> - '--raw',
> - '--tls-verify=false',
> - "--creds=%s:%s" % (username, password)]
> -
> - with mock.patch.multiple(utils.subprocess,
> - Popen=mock.DEFAULT,
> - PIPE=mock.DEFAULT) as mocked:
> - mocked['Popen'].return_value.communicate.return_value = [b'{}',
> - b'']
> - utils.get_image_details(src, raw, insecure, username, password)
> -
> - mocked['Popen'].assert_called_once_with(cmd,
> - stdout=mocked['PIPE'],
> - stderr=mocked['PIPE'])
> -
> - ###################################
> - # Tests for: is_new_layer_message()
> - ###################################
> - def test_utils_is_new_layer_message(self):
> - """
> - Ensures that is_new_layer_message() returns True when message
> - from the skopeo's stdout indicates processing of new layer
> - and False otherwise.
> - """
> -
> - valid_msgs = [
> - "Copying blob sha256:be232718519c940b04bc57",
> - "Skipping fetch of repeat blob sha256:75c416ea735c42a4a0b2"
> - ]
> -
> - invalid_msgs = [
> - 'Copying config sha256', 'test', ''
> - ]
> -
> - for msg in valid_msgs:
> - self.assertTrue(utils.is_new_layer_message(msg))
> - for msg in invalid_msgs:
> - self.assertFalse(utils.is_new_layer_message(msg))
> -
> - ###################################
> - # Tests for: is_layer_config_message()
> - ###################################
> - def test_utils_is_layer_config_message(self):
> - """
> - Ensures that is_layer_config_message() returns True when message
> - from the skopeo's stdout indicates processing of manifest file
> - of container image and False otherwise.
> - """
> - invalid_msgs = [
> - "Copying blob sha256:be232718519c940b04bc57",
> - "Skipping fetch of repeat blob sha256:75c416ea735c42a4a0b2",
> - ''
> - ]
> -
> - valid_msg = 'Copying config sha256:d355ed3537e94e76389fd78b7724'
> -
> - self.assertTrue(utils.is_layer_config_message(valid_msg))
> - for msg in invalid_msgs:
> - self.assertFalse(utils.is_layer_config_message(msg))
> -
> - ###################################
> - # Tests for: make_async()
> - ###################################
> - def test_utils_make_async(self):
> - """
> - Ensures that make_async() sets O_NONBLOCK flag on PIPE.
> - """
> -
> - pipe = utils.subprocess.Popen(
> - ["echo"],
> - stdout=utils.subprocess.PIPE
> - ).stdout
> -
> - fd = pipe.fileno()
> - F_GETFL = utils.fcntl.F_GETFL
> - O_NONBLOCK = utils.os.O_NONBLOCK
> -
> - self.assertFalse(utils.fcntl.fcntl(fd, F_GETFL) & O_NONBLOCK)
> - utils.make_async(fd)
> - self.assertTrue(utils.fcntl.fcntl(fd, F_GETFL) & O_NONBLOCK)
> -
> - ###################################
> - # Tests for: read_async()
> - ###################################
> - def test_utils_read_async_successful_read(self):
> - """
> - Ensures that read_async() calls read() of passed file descriptor.
> - """
> - m_fd = mock.MagicMock()
> - utils.read_async(m_fd)
> - m_fd.read.assert_called_once()
> -
> - def test_utils_read_async_return_empty_str_on_EAGAIN_error(self):
> - """
> - Ensures that read_async() ignores EAGAIN errors and returns
> - empty string.
> - """
> - m_fd = mock.MagicMock()
> - m_fd.read.side_effect = IOError(utils.errno.EAGAIN, '')
> - self.assertEqual(utils.read_async(m_fd), '')
> -
> - def test_utils_read_async_raise_errors(self):
> - """
> - Ensures that read_async() does not ignore IOError which is different
> - than EAGAIN and throws an exception.
> - """
> - m_fd = mock.MagicMock()
> - m_fd.read.side_effect = IOError()
> - with self.assertRaises(IOError):
> - utils.read_async(m_fd)
> -
> - ###################################
> - # Tests for: str2float()
> - ###################################
> - def test_utils_str2float(self):
> - """
> - Validates the output of str2float() for some test cases.
> - """
> - test_values = {'1': 1.0, 'test': None, '0': 0.0, '1.25': 1.25}
> - for test in test_values:
> - self.assertEqual(utils.str2float(test), test_values[test])
> -
> - ###################################
> - # Tests for: set_root_password_in_rootfs()
> - ###################################
> - def test_utils_set_root_password_in_rootfs_restore_permissions(self):
> - """
> - Ensures that set_root_password_in_rootfs() restore shadow
> - file permissions after edit.
> - """
> - permissions = 700
> - rootfs_path = '/foo'
> - shadow_file = '%s/etc/shadow' % rootfs_path
> -
> - m_open = mock.mock_open(read_data='')
> - with mock.patch('virtBootstrap.utils.open', m_open, create=True):
> - with mock.patch('virtBootstrap.utils.os') as m_os:
> - m_os.stat.return_value = [permissions]
> - m_os.path.join.return_value = shadow_file
> - utils.set_root_password_in_rootfs(rootfs_path, 'password')
> -
> - expected_calls = [
> - mock.call.path.join(rootfs_path, 'etc/shadow'),
> - mock.call.stat(shadow_file),
> - mock.call.chmod(shadow_file, 438),
> - mock.call.chmod(shadow_file, permissions)
> - ]
> - m_os.assert_has_calls(expected_calls)
> -
> - def test_utils_set_root_password_in_rootfs_restore_permissions_fail(self):
> - """
> - Ensures that set_root_password_in_rootfs() restore shadow file
> - permissions in case of failure.
> - """
> - permissions = 700
> - rootfs_path = '/foo'
> - shadow_file = '%s/etc/shadow' % rootfs_path
> -
> - m_open = mock.mock_open(read_data='')
> - with mock.patch('virtBootstrap.utils.open', m_open, create=True):
> - with mock.patch('virtBootstrap.utils.os') as m_os:
> - m_os.stat.return_value = [permissions]
> - m_os.path.join.return_value = shadow_file
> -
> - with self.assertRaises(Exception):
> - m_open.side_effect = Exception
> - utils.set_root_password_in_rootfs(rootfs_path, 'password')
> -
> - expected_calls = [
> - mock.call.path.join(rootfs_path, 'etc/shadow'),
> - mock.call.stat(shadow_file),
> - mock.call.chmod(shadow_file, 438),
> - mock.call.chmod(shadow_file, permissions)
> - ]
> - m_os.assert_has_calls(expected_calls)
> -
> - def test_utils_set_root_password_in_rootfs_store_hash(self):
> - """
> - Ensures that set_root_password_in_rootfs() stores the hashed
> - root password in shadow file.
> - """
> - rootfs_path = '/foo'
> - password = 'secret'
> - initial_value = '!locked'
> - hashed_password = 'hashed_password'
> - shadow_content = '\n'.join([
> - "root:%s::0:99999:7:::",
> - "bin:*:17004:0:99999:7:::"
> - "daemon:*:17004:0:99999:7:::",
> - "adm:*:17004:0:99999:7:::"
> - ])
> -
> - m_open = mock.mock_open(read_data=shadow_content % initial_value)
> - with mock.patch('virtBootstrap.utils.open', m_open, create=True):
> - with mock.patch('virtBootstrap.utils.os'):
> - with mock.patch('passlib.hosts.linux_context.hash') as m_hash:
> - m_hash.return_value = hashed_password
> - utils.set_root_password_in_rootfs(rootfs_path, password)
> -
> - m_hash.assert_called_once_with(password)
> - m_open().write.assert_called_once_with(shadow_content
> - % hashed_password)
> -
> - ###################################
> - # Tests for: set_root_password_in_image()
> - ###################################
> - @mock.patch('virtBootstrap.utils.execute')
> - def test_utils_set_root_password_in_image(self, m_execute):
> - """
> - Ensures that set_root_password_in_image() calls virt-edit
> - with correct arguments.
> - """
> - image, password = 'foo', 'password'
> - password_hash = ('$6$rounds=656000$PaQ/H4c/k8Ix9YOM$'
> - 'cyD47r9PtAE2LhnkpdbVzsiQbM0/h2S/1Bv'
> - 'u/sXqUtCg.3Ijp7TQy/8tEVstxMy5k5v4mh'
> - 'CGFqnVv7S6wd.Ah/')
> -
> - expected_call = [
> - 'virt-edit',
> - '-a', image, '/etc/shadow',
> - '-e', 's,^root:.*?:,root:%s:,' % utils.re.escape(password_hash)]
> -
> - hash_function = 'virtBootstrap.utils.passlib.hosts.linux_context.hash'
> - with mock.patch(hash_function) as m_hash:
> - m_hash.return_value = password_hash
> - utils.set_root_password_in_image(image, password)
> -
> - m_execute.assert_called_once_with(expected_call)
> -
> - ###################################
> - # Tests for: set_root_password()
> - ###################################
> - @mock.patch('virtBootstrap.utils.set_root_password_in_rootfs')
> - def test_utils_set_root_password_dir(self, m_set_root_password_in_rootfs):
> - """
> - Ensures that set_root_password() calls set_root_password_in_rootfs()
> - when the format is set to "dir".
> - """
> - fmt, dest, root_password = 'dir', 'dest', 'root_password'
> - utils.set_root_password(fmt, dest, root_password)
> -
> - m_set_root_password_in_rootfs.assert_called_once_with(
> - dest, root_password
> - )
> -
> - @mock.patch('virtBootstrap.utils.set_root_password_in_image')
> - def test_utils_set_root_password_qcow2(self, m_set_root_password_in_image):
> - """
> - Ensures that set_root_password() calls set_root_password_in_image()
> - when the format is set to "qcow2" with the path to the last
> - extracted layer.
> - """
> - fmt, dest, root_password = 'qcow2', 'dest', 'root_password'
> - layers = ['layer-0.qcow2', 'layer-1.qcow2']
> -
> - with mock.patch('os.listdir') as m_listdir:
> - m_listdir.return_value = layers
> - utils.set_root_password(fmt, dest, root_password)
> -
> - m_set_root_password_in_image.assert_called_once_with(
> - utils.os.path.join(dest, max(layers)),
> - root_password
> - )
> -
> - ###################################
> - # Tests for: write_progress()
> - ###################################
> - def test_utils_write_progress_fill_terminal_width(self):
> - """
> - Ensures that write_progress() outputs a message with length
> - equal to terminal width and last symbol '\r'.
> - """
> - terminal_width = 120
> - prog = {'status': 'status', 'value': 0}
> - with mock.patch.multiple(utils,
> - subprocess=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> -
> - (mocked['subprocess'].Popen.return_value.stdout
> - .read.return_value) = ("20 %s" % terminal_width).encode()
> -
> - utils.write_progress(prog)
> -
> - mocked['subprocess'].Popen.assert_called_once_with(
> - ["stty", "size"],
> - stdout=mocked['subprocess'].PIPE
> - )
> - output_message = mocked['sys'].stdout.write.call_args[0][0]
> - mocked['sys'].stdout.write.assert_called_once()
> - self.assertEqual(len(output_message), terminal_width + 1)
> - self.assertEqual(output_message[-1], '\r')
> -
> - def test_utils_write_progress_use_default_term_width_on_failure(self):
> - """
> - Ensures that write_progress() outputs a message with length equal
> - to default terminal width (80) when the detecting terminal width
> - has failed.
> - """
> - default_terminal_width = 80
> - prog = {'status': 'status', 'value': 0}
> - with mock.patch.multiple(utils,
> - subprocess=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['subprocess'].Popen.side_effect = Exception()
> - utils.write_progress(prog)
> -
> - self.assertEqual(len(mocked['sys'].stdout.write.call_args[0][0]),
> - default_terminal_width + 1)
> - mocked['sys'].stdout.write.assert_called_once()
> -
> -
> -if __name__ == '__main__':
> - unittest.main(exit=False)
> diff --git a/tests/test_virt_bootstrap.py b/tests/test_virt_bootstrap.py
> deleted file mode 100644
> index ff744f7..0000000
> --- a/tests/test_virt_bootstrap.py
> +++ /dev/null
> @@ -1,464 +0,0 @@
> -# Authors: Radostin Stoyanov <rstoyanov1 at gmail.com>
> -#
> -# Copyright (C) 2017 Radostin Stoyanov
> -#
> -# This program is free software: you can redistribute it and/or modify
> -# it under the terms of the GNU General Public License as published by
> -# the Free Software Foundation, either version 3 of the License, or
> -# (at your option) any later version.
> -
> -# This program is distributed in the hope that it will be useful,
> -# but WITHOUT ANY WARRANTY; without even the implied warranty of
> -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
> -# GNU General Public License for more details.
> -
> -# You should have received a copy of the GNU General Public License
> -# along with this program. If not, see <http://www.gnu.org/licenses/>.
> -
> -
> -"""
> -Unit tests for functions defined in virtBootstrap.virt-bootstrap
> -"""
> -
> -from tests import unittest
> -from tests import mock
> -from tests import virt_bootstrap
> -from tests import sources
> -
> -
> -# pylint: disable=invalid-name
> -class TestVirtBootstrap(unittest.TestCase):
> - """
> - Test cases for virt_bootstrap module
> - """
> -
> - ###################################
> - # Tests for: get_source(source_type)
> - ###################################
> - def test_get_invaid_source_type_should_fail(self):
> - """
> - Ensures that get_source() throws an Exception when invalid source
> - name was specified.
> - """
> - with self.assertRaises(Exception) as source:
> - virt_bootstrap.get_source('invalid')
> - self.assertIn('invalid', str(source.exception))
> -
> - def test_get_docker_source(self):
> - """
> - Ensures that get_source() returns DockerSource when source name
> - "docker" is requested.
> - """
> - self.assertIs(virt_bootstrap.get_source('docker'),
> - sources.DockerSource)
> -
> - def test_get_file_source(self):
> - """
> - Ensures that get_source() returns FileSource when source name
> - "file" is requested.
> - """
> - self.assertIs(virt_bootstrap.get_source('file'),
> - sources.FileSource)
> -
> - ###################################
> - # Tests for: mapping_uid_gid()
> - ###################################
> - def test_mapping_uid_gid(self):
> - """
> - Ensures that mapping_uid_gid() calls map_id() with
> - correct parameters.
> - """
> - dest = '/path'
> - calls = [
> - { # Call 1
> - 'dest': dest,
> - 'uid': [[0, 1000, 10]],
> - 'gid': [[0, 1000, 10]]
> - },
> - { # Call 2
> - 'dest': dest,
> - 'uid': [],
> - 'gid': [[0, 1000, 10]]
> - },
> - { # Call 3
> - 'dest': dest,
> - 'uid': [[0, 1000, 10]],
> - 'gid': []
> - },
> - { # Call 4
> - 'dest': dest,
> - 'uid': [[0, 1000, 10], [500, 500, 10]],
> - 'gid': [[0, 1000, 10]]
> - }
> - ]
> -
> - expected_calls = [
> - # Expected from call 1
> - mock.call(dest, [0, 1000, 10], [0, 1000, 10]),
> - # Expected from call 2
> - mock.call(dest, None, [0, 1000, 10]),
> - # Expected from call 3
> - mock.call(dest, [0, 1000, 10], None),
> - # Expected from call 4
> - mock.call(dest, [0, 1000, 10], [0, 1000, 10]),
> - mock.call(dest, [500, 500, 10], None)
> -
> - ]
> - with mock.patch('virtBootstrap.virt_bootstrap.map_id') as m_map_id:
> - for args in calls:
> - virt_bootstrap.mapping_uid_gid(args['dest'],
> - args['uid'],
> - args['gid'])
> -
> - m_map_id.assert_has_calls(expected_calls)
> -
> - ###################################
> - # Tests for: map_id()
> - ###################################
> - @mock.patch('os.path.realpath')
> - def test_map_id(self, m_realpath):
> - """
> - Ensures that the UID/GID mapping applies to all files
> - and directories in root file system.
> - """
> - root_path = '/root'
> - files = ['foo1', 'foo2']
> - m_realpath.return_value = root_path
> -
> - map_uid = [0, 1000, 10]
> - map_gid = [0, 1000, 10]
> - new_id = 'new_id'
> -
> - expected_calls = [
> - mock.call('/root', new_id, new_id),
> - mock.call('/root/foo1', new_id, new_id),
> - mock.call('/root/foo2', new_id, new_id)
> - ]
> -
> - with mock.patch.multiple('os',
> - lchown=mock.DEFAULT,
> - lstat=mock.DEFAULT,
> - walk=mock.DEFAULT) as mocked:
> -
> - mocked['walk'].return_value = [(root_path, [], files)]
> - mocked['lstat']().st_uid = map_uid[0]
> - mocked['lstat']().st_gid = map_gid[0]
> -
> - get_map_id = 'virtBootstrap.virt_bootstrap.get_map_id'
> - with mock.patch(get_map_id) as m_get_map_id:
> - m_get_map_id.return_value = new_id
> - virt_bootstrap.map_id(root_path, map_uid, map_gid)
> -
> - mocked['lchown'].assert_has_calls(expected_calls)
> -
> - ###################################
> - # Tests for: get_mapping_opts()
> - ###################################
> - def test_get_mapping_opts(self):
> - """
> - Ensures that get_mapping_opts() returns correct options for
> - mapping value.
> - """
> - test_values = [
> - {
> - 'mapping': [0, 1000, 10],
> - 'expected_result': {'first': 0, 'last': 10, 'offset': 1000},
> - },
> - {
> - 'mapping': [0, 1000, 10],
> - 'expected_result': {'first': 0, 'last': 10, 'offset': 1000},
> - },
> - {
> - 'mapping': [500, 1500, 1],
> - 'expected_result': {'first': 500, 'last': 501, 'offset': 1000},
> - },
> - {
> - 'mapping': [-1, -1, -1],
> - 'expected_result': {'first': 0, 'last': 1, 'offset': 0},
> - }
> - ]
> -
> - for test in test_values:
> - res = virt_bootstrap.get_mapping_opts(test['mapping'])
> - self.assertEqual(test['expected_result'], res)
> -
> - ###################################
> - # Tests for: get_map_id()
> - ###################################
> - def test_get_map_id(self):
> - """
> - Ensures that get_map_id() returns correct UID/GID mapping value.
> - """
> - test_values = [
> - {
> - 'old_id': 0,
> - 'mapping': [0, 1000, 10],
> - 'expected_result': 1000
> - },
> - {
> - 'old_id': 5,
> - 'mapping': [0, 500, 10],
> - 'expected_result': 505
> - },
> - {
> - 'old_id': 10,
> - 'mapping': [0, 100, 10],
> - 'expected_result': -1
> - },
> - ]
> - for test in test_values:
> - opts = virt_bootstrap.get_mapping_opts(test['mapping'])
> - res = virt_bootstrap.get_map_id(test['old_id'], opts)
> - self.assertEqual(test['expected_result'], res)
> -
> - ###################################
> - # Tests for: parse_idmap()
> - ###################################
> - def test_parse_idmap(self):
> - """
> - Ensures that parse_idmap() returns correct UID/GID mapping value.
> - """
> - test_values = [
> - {
> - 'mapping': ['0:1000:10', '0:100:10'],
> - 'expected_result': [[0, 1000, 10], [0, 100, 10]],
> - },
> - {
> - 'mapping': ['0:1000:10'],
> - 'expected_result': [[0, 1000, 10]],
> - },
> - {
> - 'mapping': ['500:1500:1'],
> - 'expected_result': [[500, 1500, 1]],
> - },
> - {
> - 'mapping': ['-1:-1:-1'],
> - 'expected_result': [[-1, -1, -1]],
> - },
> - {
> - 'mapping': [],
> - 'expected_result': None,
> - }
> - ]
> - for test in test_values:
> - res = virt_bootstrap.parse_idmap(test['mapping'])
> - self.assertEqual(test['expected_result'], res)
> -
> - def test_parse_idmap_raise_exception_on_invalid_mapping_value(self):
> - """
> - Ensures that parse_idmap() raise ValueError on mapping value.
> - """
> - with self.assertRaises(ValueError):
> - virt_bootstrap.parse_idmap(['invalid'])
> -
> - ###################################
> - # Tests for: bootstrap()
> - ###################################
> - def test_bootsrap_creates_directory_if_does_not_exist(self):
> - """
> - Ensures that bootstrap() creates destination directory if
> - it does not exists.
> - """
> - src, dest = 'foo', 'bar'
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = False
> - virt_bootstrap.bootstrap(src, dest)
> - mocked['os'].path.exists.assert_called_once_with(dest)
> - mocked['os'].makedirs.assert_called_once_with(dest)
> -
> - def test_bootstrap_exit_if_dest_is_invalid(self):
> - """
> - Ensures that bootstrap() exits with code 1 when the destination
> - path exists but it is not directory.
> - """
> - src, dest = 'foo', 'bar'
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - logger=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = True
> - mocked['os'].path.isdir.return_value = False
> - virt_bootstrap.bootstrap(src, dest)
> - mocked['os'].path.isdir.assert_called_once_with(dest)
> - mocked['sys'].exit.assert_called_once_with(1)
> -
> - def test_bootsrap_exit_if_no_write_access_on_dest(self):
> - """
> - Ensures that bootstrap() exits with code 1 when the current user
> - has not write permissions on the destination folder.
> - """
> - src, dest = 'foo', 'bar'
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - logger=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = True
> - mocked['os'].path.isdir.return_value = True
> - mocked['os'].access.return_value = False
> - virt_bootstrap.bootstrap(src, dest)
> - mocked['os'].access.assert_called_once_with(dest,
> - mocked['os'].W_OK)
> - mocked['sys'].exit.assert_called_once_with(1)
> -
> - def test_bootstrap_use_file_source_if_none_was_specified(self):
> - """
> - Ensures that bootstrap() calls get_source() with argument
> - 'file' when source format is not specified.
> - """
> - src, dest = 'foo', 'bar'
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - virt_bootstrap.bootstrap(src, dest)
> - mocked['get_source'].assert_called_once_with('file')
> -
> - def test_bootstrap_successful_call(self):
> - """
> - Ensures that bootstrap() creates source instance and calls the
> - unpack method with destination path as argument.
> - """
> - src, dest = 'foo', 'bar'
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = True
> - mocked['os'].path.isdir.return_value = True
> - mocked['os'].access.return_value = True
> - mocked_source = mock.Mock()
> - mocked_unpack = mock.Mock()
> - mocked_source.return_value.unpack = mocked_unpack
> - mocked['get_source'].return_value = mocked_source
> - virt_bootstrap.bootstrap(src, dest)
> - # sys.exit should not be called
> - mocked['sys'].exit.assert_not_called()
> - mocked_source.assert_called_once()
> - mocked_unpack.assert_called_once_with(dest)
> -
> - def test_bootstrap_all_params_are_passed_to_source_instance(self):
> - """
> - Ensures that bootstrap() is passing all arguments to newly created
> - source instance.
> - """
> - params_list = ['dest', 'fmt', 'username', 'password', 'root_password',
> - 'not_secure', 'no_cache', 'progress_cb']
> - params = {param: param for param in params_list}
> -
> - for kw_param in params_list:
> - params[kw_param] = kw_param
> -
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - urlparse=mock.DEFAULT,
> - progress=mock.DEFAULT,
> - utils=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = True
> - mocked['os'].path.isdir.return_value = True
> - mocked['os'].access.return_value = True
> -
> - mocked['progress'].Progress.return_value = params['progress_cb']
> -
> - mocked_source = mock.Mock()
> - mocked_unpack = mock.Mock()
> - mocked_source.return_value.unpack = mocked_unpack
> - mocked['get_source'].return_value = mocked_source
> -
> - mocked_uri = mock.Mock()
> - mocked['urlparse'].return_value = mocked_uri
> - params['uri'] = mocked_uri
> -
> - virt_bootstrap.bootstrap(**params)
> - # sys.exit should not be called
> - mocked['sys'].exit.assert_not_called()
> -
> - mocked_source.assert_called_once()
> - mocked_unpack.assert_called_once_with(params['dest'])
> -
> - called_with_args, called_with_kwargs = mocked_source.call_args
> - self.assertEqual(called_with_args, ())
> -
> - del params['dest']
> - del params['root_password']
> - params['progress'] = params.pop('progress_cb')
> - for kwarg in params:
> - self.assertEqual(called_with_kwargs[kwarg], params[kwarg])
> -
> - def test_if_bootstrap_calls_set_root_password(self):
> - """
> - Ensures that bootstrap() calls set_root_password() when the argument
> - root_password is specified.
> - """
> - src, fmt, dest, root_password = 'foo', 'fmt', 'bar', 'root_password'
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - utils=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = True
> - mocked['os'].path.isdir.return_value = True
> - mocked['os'].access.return_value = True
> -
> - virt_bootstrap.bootstrap(src, dest,
> - fmt=fmt,
> - root_password=root_password)
> -
> - mocked['utils'].set_root_password.assert_called_once_with(
> - fmt, dest, root_password)
> -
> - def test_if_bootstrap_calls_set_mapping_uid_gid(self):
> - """
> - Ensures that bootstrap() calls mapping_uid_gid() when the argument
> - uid_map or gid_map is specified.
> - """
> - src, dest, uid_map, gid_map = 'foo', 'bar', 'id', 'id'
> - expected_calls = [
> - mock.call('bar', None, 'id'),
> - mock.call('bar', 'id', None),
> - mock.call('bar', 'id', 'id')
> - ]
> -
> - with mock.patch.multiple(virt_bootstrap,
> - get_source=mock.DEFAULT,
> - os=mock.DEFAULT,
> - mapping_uid_gid=mock.DEFAULT,
> - utils=mock.DEFAULT,
> - sys=mock.DEFAULT) as mocked:
> - mocked['os'].path.exists.return_value = True
> - mocked['os'].path.isdir.return_value = True
> - mocked['os'].access.return_value = True
> -
> - virt_bootstrap.bootstrap(src, dest, gid_map=gid_map)
> - virt_bootstrap.bootstrap(src, dest, uid_map=uid_map)
> - virt_bootstrap.bootstrap(src, dest,
> - uid_map=uid_map, gid_map=gid_map)
> - mocked['mapping_uid_gid'].assert_has_calls(expected_calls)
> -
> - ###################################
> - # Tests for: set_logging_conf()
> - ###################################
> - def test_if_logging_level_format_handler_are_set(self):
> - """
> - Ensures that set_logging_conf() sets log level and adds new stream
> - handler with formatting.
> - """
> - with mock.patch('virtBootstrap.virt_bootstrap.logging') as m_logging:
> - mocked_stream_hdlr = mock.Mock()
> - m_logger = mock.Mock()
> - m_logging.getLogger.return_value = m_logger
> - m_logging.StreamHandler.return_value = mocked_stream_hdlr
> - virt_bootstrap.set_logging_conf()
> - m_logging.getLogger.assert_called_once_with('virtBootstrap')
> - mocked_stream_hdlr.setFormatter.assert_called_once()
> - m_logger.addHandler.assert_called_once_with(mocked_stream_hdlr)
> - m_logger.setLevel.assert_called_once()
> -
> -
> -if __name__ == '__main__':
> - unittest.main(exit=False)
More information about the virt-tools-list
mailing list