# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import builtins import datetime from http import client as http_client import io import os import shutil from unittest import mock from oslo_config import cfg from oslo_utils import uuidutils import requests from ironic.common import exception from ironic.common.glance_service import image_service as glance_v2_service from ironic.common import image_service from ironic.tests import base class HttpImageServiceTestCase(base.TestCase): def setUp(self): super(HttpImageServiceTestCase, self).setUp() self.service = image_service.HttpImageService() self.href = 'https://127.0.0.1:12345/fedora.qcow2' @mock.patch.object(os.path, 'exists', autospec=True) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_http_scheme(self, head_mock, path_mock): self.href = 'http://127.0.0.1:12345/fedora.qcow2' response = head_mock.return_value response.status_code = http_client.OK self.service.validate_href(self.href) path_mock.assert_not_called() head_mock.assert_called_once_with(self.href, verify=True, timeout=60) response.status_code = http_client.NO_CONTENT self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) response.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_false(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', 'False') response = head_mock.return_value response.status_code = http_client.OK self.service.validate_href(self.href) head_mock.assert_called_once_with(self.href, verify=False, timeout=60) response.status_code = http_client.NO_CONTENT self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) response.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_false_error(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', 'False') head_mock.side_effect = requests.ConnectionError() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) head_mock.assert_called_once_with(self.href, verify=False, timeout=60) head_mock.side_effect = requests.RequestException() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_true(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', 'True') response = head_mock.return_value response.status_code = http_client.OK self.service.validate_href(self.href) head_mock.assert_called_once_with(self.href, verify=True, timeout=60) response.status_code = http_client.NO_CONTENT self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) response.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_true_error(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', 'True') head_mock.side_effect = requests.ConnectionError() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) head_mock.assert_called_once_with(self.href, verify=True, timeout=60) head_mock.side_effect = requests.RequestException() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_valid_path(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') response = head_mock.return_value response.status_code = http_client.OK self.service.validate_href(self.href) head_mock.assert_called_once_with(self.href, verify='/some/path', timeout=60) response.status_code = http_client.NO_CONTENT self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) response.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_custom_timeout(self, head_mock): cfg.CONF.set_override('webserver_connection_timeout', 15) response = head_mock.return_value response.status_code = http_client.OK self.service.validate_href(self.href) head_mock.assert_called_once_with(self.href, verify=True, timeout=15) response.status_code = http_client.NO_CONTENT self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) response.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_connect_error(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') response = mock.Mock() response.status_code = http_client.OK head_mock.side_effect = requests.ConnectionError() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) head_mock.assert_called_once_with(self.href, verify='/some/path', timeout=60) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_error(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') head_mock.side_effect = requests.RequestException() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) head_mock.assert_called_once_with(self.href, verify='/some/path', timeout=60) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_verify_os_error(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') head_mock.side_effect = OSError() self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) head_mock.assert_called_once_with(self.href, verify='/some/path', timeout=60) @mock.patch.object(requests, 'head', autospec=True) def test_validate_href_error_with_secret_parameter(self, head_mock): cfg.CONF.set_override('webserver_verify_ca', 'False') head_mock.return_value.status_code = 204 e = self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href, True) self.assertIn('secreturl', str(e)) self.assertNotIn(self.href, str(e)) head_mock.assert_called_once_with(self.href, verify=False, timeout=60) @mock.patch.object(requests, 'head', autospec=True) def _test_show(self, head_mock, mtime, mtime_date): head_mock.return_value.status_code = http_client.OK head_mock.return_value.headers = { 'Content-Length': 100, 'Last-Modified': mtime } result = self.service.show(self.href) head_mock.assert_called_once_with(self.href, verify=True, timeout=60) self.assertEqual({'size': 100, 'updated_at': mtime_date, 'properties': {}}, result) def test_show_rfc_822(self): self._test_show(mtime='Tue, 15 Nov 2014 08:12:31 GMT', mtime_date=datetime.datetime(2014, 11, 15, 8, 12, 31)) def test_show_rfc_850(self): self._test_show(mtime='Tuesday, 15-Nov-14 08:12:31 GMT', mtime_date=datetime.datetime(2014, 11, 15, 8, 12, 31)) def test_show_ansi_c(self): self._test_show(mtime='Tue Nov 15 08:12:31 2014', mtime_date=datetime.datetime(2014, 11, 15, 8, 12, 31)) @mock.patch.object(requests, 'head', autospec=True) def test_show_no_content_length(self, head_mock): head_mock.return_value.status_code = http_client.OK head_mock.return_value.headers = {} self.assertRaises(exception.ImageRefValidationFailed, self.service.show, self.href) head_mock.assert_called_with(self.href, verify=True, timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_success_http_scheme(self, req_get_mock, shutil_mock): self.href = 'http://127.0.0.1:12345/fedora.qcow2' response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) self.service.download(self.href, file_mock) shutil_mock.assert_called_once_with( response_mock.raw.__enter__(), file_mock, image_service.IMAGE_CHUNK_SIZE ) req_get_mock.assert_called_once_with(self.href, stream=True, verify=True, timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_success_verify_false( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', 'False') response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) self.service.download(self.href, file_mock) shutil_mock.assert_called_once_with( response_mock.raw.__enter__(), file_mock, image_service.IMAGE_CHUNK_SIZE ) req_get_mock.assert_called_once_with(self.href, stream=True, verify=False, timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_success_verify_true( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', 'True') response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) self.service.download(self.href, file_mock) shutil_mock.assert_called_once_with( response_mock.raw.__enter__(), file_mock, image_service.IMAGE_CHUNK_SIZE ) req_get_mock.assert_called_once_with(self.href, stream=True, verify=True, timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_success_verify_path( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) self.service.download(self.href, file_mock) shutil_mock.assert_called_once_with( response_mock.raw.__enter__(), file_mock, image_service.IMAGE_CHUNK_SIZE ) req_get_mock.assert_called_once_with(self.href, stream=True, verify='/some/path', timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_fail_verify_false_connerror( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', False) req_get_mock.side_effect = requests.ConnectionError() file_mock = mock.Mock(spec=io.BytesIO) self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_fail_verify_false_ioerror( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', False) response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) shutil_mock.side_effect = IOError self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) req_get_mock.assert_called_once_with(self.href, stream=True, verify=False, timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_success_verify_true_connerror( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') response_mock = mock.Mock() response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) req_get_mock.side_effect = requests.ConnectionError file_mock = mock.Mock(spec=io.BytesIO) self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) req_get_mock.assert_called_once_with(self.href, stream=True, verify='/some/path', timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_fail_verify_true_ioerror( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) shutil_mock.side_effect = IOError self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) req_get_mock.assert_called_once_with(self.href, stream=True, verify='/some/path', timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_fail_verify_true_oserror( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_verify_ca', '/some/path') response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) shutil_mock.side_effect = OSError() self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) req_get_mock.assert_called_once_with(self.href, stream=True, verify='/some/path', timeout=60) @mock.patch.object(shutil, 'copyfileobj', autospec=True) @mock.patch.object(requests, 'get', autospec=True) def test_download_success_custom_timeout( self, req_get_mock, shutil_mock): cfg.CONF.set_override('webserver_connection_timeout', 15) response_mock = req_get_mock.return_value response_mock.status_code = http_client.OK response_mock.raw = mock.MagicMock(spec=io.BytesIO) file_mock = mock.Mock(spec=io.BytesIO) self.service.download(self.href, file_mock) shutil_mock.assert_called_once_with( response_mock.raw.__enter__(), file_mock, image_service.IMAGE_CHUNK_SIZE ) req_get_mock.assert_called_once_with(self.href, stream=True, verify=True, timeout=15) class FileImageServiceTestCase(base.TestCase): def setUp(self): super(FileImageServiceTestCase, self).setUp() self.service = image_service.FileImageService() self.href = 'file:///home/user/image.qcow2' self.href_path = '/home/user/image.qcow2' @mock.patch.object(os.path, 'isfile', return_value=True, autospec=True) def test_validate_href(self, path_exists_mock): self.service.validate_href(self.href) path_exists_mock.assert_called_once_with(self.href_path) @mock.patch.object(os.path, 'isfile', return_value=False, autospec=True) def test_validate_href_path_not_found_or_not_file(self, path_exists_mock): self.assertRaises(exception.ImageRefValidationFailed, self.service.validate_href, self.href) path_exists_mock.assert_called_once_with(self.href_path) @mock.patch.object(os.path, 'getmtime', return_value=1431087909.1641912, autospec=True) @mock.patch.object(os.path, 'getsize', return_value=42, autospec=True) @mock.patch.object(image_service.FileImageService, 'validate_href', autospec=True) def test_show(self, _validate_mock, getsize_mock, getmtime_mock): _validate_mock.return_value = self.href_path result = self.service.show(self.href) getsize_mock.assert_called_once_with(self.href_path) getmtime_mock.assert_called_once_with(self.href_path) _validate_mock.assert_called_once_with(mock.ANY, self.href) self.assertEqual({'size': 42, 'updated_at': datetime.datetime(2015, 5, 8, 12, 25, 9, 164191), 'properties': {}}, result) @mock.patch.object(os, 'link', autospec=True) @mock.patch.object(os, 'remove', autospec=True) @mock.patch.object(os, 'access', return_value=True, autospec=True) @mock.patch.object(os, 'stat', autospec=True) @mock.patch.object(image_service.FileImageService, 'validate_href', autospec=True) def test_download_hard_link(self, _validate_mock, stat_mock, access_mock, remove_mock, link_mock): _validate_mock.return_value = self.href_path stat_mock.return_value.st_dev = 'dev1' file_mock = mock.Mock(spec=io.BytesIO) file_mock.name = 'file' self.service.download(self.href, file_mock) _validate_mock.assert_called_once_with(mock.ANY, self.href) self.assertEqual(2, stat_mock.call_count) access_mock.assert_called_once_with(self.href_path, os.R_OK | os.W_OK) remove_mock.assert_called_once_with('file') link_mock.assert_called_once_with(self.href_path, 'file') @mock.patch.object(os, 'sendfile', return_value=42, autospec=True) @mock.patch.object(os.path, 'getsize', return_value=42, autospec=True) @mock.patch.object(builtins, 'open', autospec=True) @mock.patch.object(os, 'access', return_value=False, autospec=True) @mock.patch.object(os, 'stat', autospec=True) @mock.patch.object(image_service.FileImageService, 'validate_href', autospec=True) def test_download_copy(self, _validate_mock, stat_mock, access_mock, open_mock, size_mock, copy_mock): _validate_mock.return_value = self.href_path stat_mock.return_value.st_dev = 'dev1' file_mock = mock.MagicMock(spec=io.BytesIO) file_mock.name = 'file' input_mock = mock.MagicMock(spec=io.BytesIO) open_mock.return_value = input_mock self.service.download(self.href, file_mock) _validate_mock.assert_called_once_with(mock.ANY, self.href) self.assertEqual(2, stat_mock.call_count) access_mock.assert_called_once_with(self.href_path, os.R_OK | os.W_OK) copy_mock.assert_called_once_with(file_mock.fileno(), input_mock.__enter__().fileno(), 0, 42) @mock.patch.object(os, 'sendfile', autospec=True) @mock.patch.object(os.path, 'getsize', return_value=42, autospec=True) @mock.patch.object(builtins, 'open', autospec=True) @mock.patch.object(os, 'access', return_value=False, autospec=True) @mock.patch.object(os, 'stat', autospec=True) @mock.patch.object(image_service.FileImageService, 'validate_href', autospec=True) def test_download_copy_segmented(self, _validate_mock, stat_mock, access_mock, open_mock, size_mock, copy_mock): # Fake a 3G + 1k image chunk_size = image_service.SENDFILE_CHUNK_SIZE fake_image_size = chunk_size * 3 + 1024 fake_chunk_seq = [chunk_size, chunk_size, chunk_size, 1024] _validate_mock.return_value = self.href_path stat_mock.return_value.st_dev = 'dev1' file_mock = mock.MagicMock(spec=io.BytesIO) file_mock.name = 'file' input_mock = mock.MagicMock(spec=io.BytesIO) open_mock.return_value = input_mock size_mock.return_value = fake_image_size copy_mock.side_effect = fake_chunk_seq self.service.download(self.href, file_mock) _validate_mock.assert_called_once_with(mock.ANY, self.href) self.assertEqual(2, stat_mock.call_count) access_mock.assert_called_once_with(self.href_path, os.R_OK | os.W_OK) copy_calls = [mock.call(file_mock.fileno(), input_mock.__enter__().fileno(), chunk_size * i, fake_chunk_seq[i]) for i in range(4)] copy_mock.assert_has_calls(copy_calls) size_mock.assert_called_once_with(self.href_path) @mock.patch.object(os, 'remove', side_effect=OSError, autospec=True) @mock.patch.object(os, 'access', return_value=True, autospec=True) @mock.patch.object(os, 'stat', autospec=True) @mock.patch.object(image_service.FileImageService, 'validate_href', autospec=True) def test_download_hard_link_fail(self, _validate_mock, stat_mock, access_mock, remove_mock): _validate_mock.return_value = self.href_path stat_mock.return_value.st_dev = 'dev1' file_mock = mock.MagicMock(spec=io.BytesIO) file_mock.name = 'file' self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) _validate_mock.assert_called_once_with(mock.ANY, self.href) self.assertEqual(2, stat_mock.call_count) access_mock.assert_called_once_with(self.href_path, os.R_OK | os.W_OK) @mock.patch.object(os, 'sendfile', side_effect=OSError, autospec=True) @mock.patch.object(os.path, 'getsize', return_value=42, autospec=True) @mock.patch.object(builtins, 'open', autospec=True) @mock.patch.object(os, 'access', return_value=False, autospec=True) @mock.patch.object(os, 'stat', autospec=True) @mock.patch.object(image_service.FileImageService, 'validate_href', autospec=True) def test_download_copy_fail(self, _validate_mock, stat_mock, access_mock, open_mock, size_mock, copy_mock): _validate_mock.return_value = self.href_path stat_mock.return_value.st_dev = 'dev1' file_mock = mock.MagicMock(spec=io.BytesIO) file_mock.name = 'file' input_mock = mock.MagicMock(spec=io.BytesIO) open_mock.return_value = input_mock self.assertRaises(exception.ImageDownloadFailed, self.service.download, self.href, file_mock) _validate_mock.assert_called_once_with(mock.ANY, self.href) self.assertEqual(2, stat_mock.call_count) access_mock.assert_called_once_with(self.href_path, os.R_OK | os.W_OK) size_mock.assert_called_once_with(self.href_path) class ServiceGetterTestCase(base.TestCase): @mock.patch.object(glance_v2_service.GlanceImageService, '__init__', return_value=None, autospec=True) def test_get_glance_image_service(self, glance_service_mock): image_href = uuidutils.generate_uuid() image_service.get_image_service(image_href, context=self.context) glance_service_mock.assert_called_once_with(mock.ANY, None, self.context) @mock.patch.object(glance_v2_service.GlanceImageService, '__init__', return_value=None, autospec=True) def test_get_glance_image_service_url(self, glance_service_mock): image_href = 'glance://%s' % uuidutils.generate_uuid() image_service.get_image_service(image_href, context=self.context) glance_service_mock.assert_called_once_with(mock.ANY, None, self.context) @mock.patch.object(image_service.HttpImageService, '__init__', return_value=None, autospec=True) def test_get_http_image_service(self, http_service_mock): image_href = 'http://127.0.0.1/image.qcow2' image_service.get_image_service(image_href) http_service_mock.assert_called_once_with() @mock.patch.object(image_service.HttpImageService, '__init__', return_value=None, autospec=True) def test_get_https_image_service(self, http_service_mock): image_href = 'https://127.0.0.1/image.qcow2' image_service.get_image_service(image_href) http_service_mock.assert_called_once_with() @mock.patch.object(image_service.FileImageService, '__init__', return_value=None, autospec=True) def test_get_file_image_service(self, local_service_mock): image_href = 'file:///home/user/image.qcow2' image_service.get_image_service(image_href) local_service_mock.assert_called_once_with() def test_get_image_service_invalid_image_ref(self): invalid_refs = ( 'usenet://alt.binaries.dvd/image.qcow2', 'no scheme, no uuid') for image_ref in invalid_refs: self.assertRaises(exception.ImageRefValidationFailed, image_service.get_image_service, image_ref)