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 of work needed to add new features or changes to the code most of these tests will be replaced with more abstract form of testing introduced in the following commits. --- tests/__init__.py | 44 ++-- tests/docker_source.py | 150 +++++++++++ tests/test_docker_source.py | 607 ------------------------------------------- tests/test_file_source.py | 171 ------------ tests/test_progress.py | 112 -------- tests/test_utils.py | 580 +---------------------------------------- tests/test_virt_bootstrap.py | 464 --------------------------------- 7 files changed, 180 insertions(+), 1948 deletions(-) create mode 100644 tests/docker_source.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_virt_bootstrap.py diff --git a/tests/__init__.py b/tests/__init__.py index e82c6d5..1b06616 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,22 +1,23 @@ -""" - Test suite for virt-bootstrap - - Authors: Radostin Stoyanov <rstoyanov1@xxxxxxxxx> - - Copyright (C) 2017 Radostin Stoyanov +# -*- coding: utf-8 -*- +# Authors: Radostin Stoyanov <rstoyanov1@xxxxxxxxx> +# +# 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/>. - 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/>. +""" +Test suite for virt-bootstrap """ import sys @@ -27,13 +28,12 @@ try: except ImportError: import unittest.mock as mock -sys.path += '../src' # noqa: E402 +sys.path.insert(0, '../src') # noqa: E402 -# pylint: disable=import-error +# pylint: disable=import-error, wrong-import-position 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'] +__all__ = ['virt_bootstrap', 'sources', 'progress', 'utils'] diff --git a/tests/docker_source.py b/tests/docker_source.py new file mode 100644 index 0000000..60404e6 --- /dev/null +++ b/tests/docker_source.py @@ -0,0 +1,150 @@ +# -*- coding: utf-8 -*- +# Authors: Radostin Stoyanov <rstoyanov1@xxxxxxxxx> +# +# 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/>. + +""" +Tests which aim is to exercise creation of root file system with DockerSource. +""" + +import unittest + +from . import mock +from . import sources + + +# pylint: disable=invalid-name +class TestDockerSource(unittest.TestCase): + """ + Unit tests for DockerSource + """ + ################################### + # 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 which aim to return an instance of the class + DockerSource with some util functions being mocked. + """ + 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] + ] + + with mock.patch('os.path.getsize') as m_getsize: + m_getsize.return_value = 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) 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@xxxxxxxxx> -# -# 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@xxxxxxxxx> -# -# 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@xxxxxxxxx> -# -# 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 index 0b6ccc0..c2f55b5 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- # Authors: Radostin Stoyanov <rstoyanov1@xxxxxxxxx> # # Copyright (C) 2017 Radostin Stoyanov @@ -19,126 +20,16 @@ """ 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 +import unittest +from . import utils # 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() ################################### @@ -172,241 +63,6 @@ class TestUtils(unittest.TestCase): 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): @@ -459,10 +115,11 @@ class TestUtils(unittest.TestCase): Ensures that make_async() sets O_NONBLOCK flag on PIPE. """ - pipe = utils.subprocess.Popen( + proc = utils.subprocess.Popen( ["echo"], stdout=utils.subprocess.PIPE - ).stdout + ) + pipe = proc.stdout fd = pipe.fileno() F_GETFL = utils.fcntl.F_GETFL @@ -471,36 +128,8 @@ class TestUtils(unittest.TestCase): 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) + proc.wait() + pipe.close() ################################### # Tests for: str2float() @@ -512,196 +141,3 @@ class TestUtils(unittest.TestCase): 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@xxxxxxxxx> -# -# 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) -- 2.13.5 _______________________________________________ virt-tools-list mailing list virt-tools-list@xxxxxxxxxx https://www.redhat.com/mailman/listinfo/virt-tools-list
