"Fossies" - the Fresh Open Source Software Archive

Member "ironic-16.0.3/ironic/tests/unit/drivers/modules/test_image_utils.py" (18 Jan 2021, 20752 Bytes) of package /linux/misc/openstack/ironic-16.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_image_utils.py": 16.0.2_vs_16.0.3.

    1 # Copyright 2019 Red Hat, Inc.
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 import os
   17 from unittest import mock
   18 
   19 from oslo_utils import importutils
   20 
   21 from ironic.common import images
   22 from ironic.conductor import task_manager
   23 from ironic.drivers.modules import image_utils
   24 from ironic.tests.unit.db import base as db_base
   25 from ironic.tests.unit.db import utils as db_utils
   26 from ironic.tests.unit.objects import utils as obj_utils
   27 
   28 sushy = importutils.try_import('sushy')
   29 
   30 INFO_DICT = db_utils.get_test_redfish_info()
   31 
   32 
   33 class RedfishImageHandlerTestCase(db_base.DbTestCase):
   34 
   35     def setUp(self):
   36         super(RedfishImageHandlerTestCase, self).setUp()
   37         self.config(enabled_hardware_types=['redfish'],
   38                     enabled_power_interfaces=['redfish'],
   39                     enabled_boot_interfaces=['redfish-virtual-media'],
   40                     enabled_management_interfaces=['redfish'],
   41                     enabled_inspect_interfaces=['redfish'],
   42                     enabled_bios_interfaces=['redfish'])
   43         self.node = obj_utils.create_test_node(
   44             self.context, driver='redfish', driver_info=INFO_DICT)
   45 
   46     def test__append_filename_param_without_qs(self):
   47         img_handler_obj = image_utils.ImageHandler(self.node.driver)
   48         res = img_handler_obj._append_filename_param(
   49             'http://a.b/c', 'b.img')
   50         expected = 'http://a.b/c?filename=b.img'
   51         self.assertEqual(expected, res)
   52 
   53     def test__append_filename_param_with_qs(self):
   54         img_handler_obj = image_utils.ImageHandler(self.node.driver)
   55         res = img_handler_obj._append_filename_param(
   56             'http://a.b/c?d=e&f=g', 'b.img')
   57         expected = 'http://a.b/c?d=e&f=g&filename=b.img'
   58         self.assertEqual(expected, res)
   59 
   60     def test__append_filename_param_with_filename(self):
   61         img_handler_obj = image_utils.ImageHandler(self.node.driver)
   62         res = img_handler_obj._append_filename_param(
   63             'http://a.b/c?filename=bootme.img', 'b.img')
   64         expected = 'http://a.b/c?filename=bootme.img'
   65         self.assertEqual(expected, res)
   66 
   67     @mock.patch.object(image_utils, 'swift', autospec=True)
   68     def test_publish_image_swift(self, mock_swift):
   69         img_handler_obj = image_utils.ImageHandler(self.node.driver)
   70         mock_swift_api = mock_swift.SwiftAPI.return_value
   71         mock_swift_api.get_temp_url.return_value = 'https://a.b/c.f?e=f'
   72 
   73         url = img_handler_obj.publish_image('file.iso', 'boot.iso')
   74 
   75         self.assertEqual(
   76             'https://a.b/c.f?e=f&filename=file.iso', url)
   77 
   78         mock_swift.SwiftAPI.assert_called_once_with()
   79 
   80         mock_swift_api.create_object.assert_called_once_with(
   81             mock.ANY, mock.ANY, mock.ANY, mock.ANY)
   82 
   83         mock_swift_api.get_temp_url.assert_called_once_with(
   84             mock.ANY, mock.ANY, mock.ANY)
   85 
   86     @mock.patch.object(image_utils, 'swift', autospec=True)
   87     def test_unpublish_image_swift(self, mock_swift):
   88         with task_manager.acquire(self.context, self.node.uuid,
   89                                   shared=True) as task:
   90             img_handler_obj = image_utils.ImageHandler(self.node.driver)
   91             object_name = 'image-%s' % task.node.uuid
   92 
   93             img_handler_obj.unpublish_image(object_name)
   94 
   95             mock_swift.SwiftAPI.assert_called_once_with()
   96             mock_swift_api = mock_swift.SwiftAPI.return_value
   97 
   98             mock_swift_api.delete_object.assert_called_once_with(
   99                 'ironic_redfish_container', object_name)
  100 
  101     @mock.patch.object(image_utils.ImageHandler, '_is_swift_enabled',
  102                        autospec=True)
  103     @mock.patch.object(os, 'chmod', autospec=True)
  104     @mock.patch.object(image_utils, 'shutil', autospec=True)
  105     @mock.patch.object(os, 'link', autospec=True)
  106     @mock.patch.object(os, 'mkdir', autospec=True)
  107     def test_publish_image_local_link(
  108             self, mock_mkdir, mock_link, mock_shutil, mock_chmod,
  109             mock__is_swift):
  110         img_handler_obj = image_utils.ImageHandler(self.node.driver)
  111         mock__is_swift.return_value = False
  112         self.config(use_swift=False, group='redfish')
  113         self.config(http_url='http://localhost', group='deploy')
  114 
  115         url = img_handler_obj.publish_image('file.iso', 'boot.iso')
  116 
  117         self.assertEqual(
  118             'http://localhost/redfish/boot.iso?filename=file.iso', url)
  119 
  120         mock_mkdir.assert_called_once_with('/httpboot/redfish', 0o755)
  121         mock_link.assert_called_once_with(
  122             'file.iso', '/httpboot/redfish/boot.iso')
  123         mock_chmod.assert_called_once_with('file.iso', 0o644)
  124 
  125     @mock.patch.object(image_utils.ImageHandler, '_is_swift_enabled',
  126                        autospec=True)
  127     @mock.patch.object(os, 'chmod', autospec=True)
  128     @mock.patch.object(image_utils, 'shutil', autospec=True)
  129     @mock.patch.object(os, 'link', autospec=True)
  130     @mock.patch.object(os, 'mkdir', autospec=True)
  131     def test_publish_image_local_copy(self, mock_mkdir, mock_link,
  132                                       mock_shutil, mock_chmod,
  133                                       mock__is_swift):
  134         mock__is_swift.return_value = False
  135         self.config(use_swift=False, group='redfish')
  136         self.config(http_url='http://localhost', group='deploy')
  137         img_handler_obj = image_utils.ImageHandler(self.node.driver)
  138 
  139         mock_link.side_effect = OSError()
  140 
  141         url = img_handler_obj.publish_image('file.iso', 'boot.iso')
  142 
  143         self.assertEqual(
  144             'http://localhost/redfish/boot.iso?filename=file.iso', url)
  145 
  146         mock_mkdir.assert_called_once_with('/httpboot/redfish', 0o755)
  147 
  148         mock_shutil.copyfile.assert_called_once_with(
  149             'file.iso', '/httpboot/redfish/boot.iso')
  150         mock_chmod.assert_called_once_with('/httpboot/redfish/boot.iso',
  151                                            0o644)
  152 
  153     @mock.patch.object(image_utils.ImageHandler, '_is_swift_enabled',
  154                        autospec=True)
  155     @mock.patch.object(image_utils, 'ironic_utils', autospec=True)
  156     def test_unpublish_image_local(self, mock_ironic_utils, mock__is_swift):
  157         self.config(use_swift=False, group='redfish')
  158         mock__is_swift.return_value = False
  159         with task_manager.acquire(self.context, self.node.uuid,
  160                                   shared=True) as task:
  161             img_handler_obj = image_utils.ImageHandler(self.node.driver)
  162             object_name = 'image-%s' % task.node.uuid
  163 
  164             expected_file = '/httpboot/redfish/' + object_name
  165 
  166             img_handler_obj.unpublish_image(object_name)
  167 
  168             mock_ironic_utils.unlink_without_raise.assert_called_once_with(
  169                 expected_file)
  170 
  171 
  172 class RedfishImageUtilsTestCase(db_base.DbTestCase):
  173 
  174     def setUp(self):
  175         super(RedfishImageUtilsTestCase, self).setUp()
  176         self.config(enabled_hardware_types=['redfish'],
  177                     enabled_power_interfaces=['redfish'],
  178                     enabled_boot_interfaces=['redfish-virtual-media'],
  179                     enabled_management_interfaces=['redfish'],
  180                     enabled_inspect_interfaces=['redfish'],
  181                     enabled_bios_interfaces=['redfish'])
  182         self.node = obj_utils.create_test_node(
  183             self.context, driver='redfish', driver_info=INFO_DICT)
  184 
  185     @mock.patch.object(image_utils.ImageHandler, 'unpublish_image',
  186                        autospec=True)
  187     def test_cleanup_floppy_image(self, mock_unpublish):
  188         with task_manager.acquire(self.context, self.node.uuid,
  189                                   shared=True) as task:
  190             image_utils.cleanup_floppy_image(task)
  191 
  192             object_name = 'image-%s' % task.node.uuid
  193 
  194             mock_unpublish.assert_called_once_with(mock.ANY, object_name)
  195 
  196     @mock.patch.object(image_utils.ImageHandler, 'publish_image',
  197                        autospec=True)
  198     @mock.patch.object(images, 'create_vfat_image', autospec=True)
  199     def test_prepare_floppy_image(
  200             self, mock_create_vfat_image, mock_publish_image):
  201         with task_manager.acquire(self.context, self.node.uuid,
  202                                   shared=True) as task:
  203             expected_url = 'https://a.b/c.f?e=f'
  204 
  205             mock_publish_image.return_value = expected_url
  206 
  207             url = image_utils.prepare_floppy_image(task)
  208 
  209             object_name = 'image-%s' % task.node.uuid
  210 
  211             mock_publish_image.assert_called_once_with(mock.ANY,
  212                                                        mock.ANY, object_name)
  213 
  214             mock_create_vfat_image.assert_called_once_with(
  215                 mock.ANY, parameters=mock.ANY)
  216 
  217             self.assertEqual(expected_url, url)
  218 
  219     @mock.patch.object(image_utils.ImageHandler, 'unpublish_image',
  220                        autospec=True)
  221     def test_cleanup_iso_image(self, mock_unpublish):
  222         with task_manager.acquire(self.context, self.node.uuid,
  223                                   shared=True) as task:
  224             image_utils.cleanup_iso_image(task)
  225 
  226             object_name = 'boot-%s.iso' % task.node.uuid
  227 
  228             mock_unpublish.assert_called_once_with(mock.ANY, object_name)
  229 
  230     @mock.patch.object(image_utils.ImageHandler, 'publish_image',
  231                        autospec=True)
  232     @mock.patch.object(images, 'create_boot_iso', autospec=True)
  233     def test__prepare_iso_image_uefi(
  234             self, mock_create_boot_iso, mock_publish_image):
  235         with task_manager.acquire(self.context, self.node.uuid,
  236                                   shared=True) as task:
  237             task.node.instance_info.update(deploy_boot_mode='uefi')
  238 
  239             expected_url = 'https://a.b/c.f?e=f'
  240 
  241             mock_publish_image.return_value = expected_url
  242 
  243             url = image_utils._prepare_iso_image(
  244                 task, 'http://kernel/img', 'http://ramdisk/img',
  245                 'http://bootloader/img', root_uuid=task.node.uuid)
  246 
  247             object_name = 'boot-%s.iso' % task.node.uuid
  248 
  249             mock_publish_image.assert_called_once_with(
  250                 mock.ANY, mock.ANY, object_name)
  251 
  252             mock_create_boot_iso.assert_called_once_with(
  253                 mock.ANY, mock.ANY, 'http://kernel/img', 'http://ramdisk/img',
  254                 boot_mode='uefi', esp_image_href='http://bootloader/img',
  255                 kernel_params='nofb nomodeset vga=normal',
  256                 root_uuid='1be26c0b-03f2-4d2e-ae87-c02d7f33c123',
  257                 base_iso=None, inject_files=None)
  258 
  259             self.assertEqual(expected_url, url)
  260 
  261     @mock.patch.object(image_utils.ImageHandler, 'publish_image',
  262                        autospec=True)
  263     @mock.patch.object(images, 'create_boot_iso', autospec=True)
  264     def test__prepare_iso_image_default_boot_mode(
  265             self, mock_create_boot_iso, mock_publish_image):
  266         self.config(default_boot_mode='uefi', group='deploy')
  267         with task_manager.acquire(self.context, self.node.uuid,
  268                                   shared=True) as task:
  269             image_utils._prepare_iso_image(
  270                 task, 'http://kernel/img', 'http://ramdisk/img',
  271                 bootloader_href=None, root_uuid=task.node.uuid,
  272                 base_iso='/path/to/baseiso')
  273 
  274             mock_create_boot_iso.assert_called_once_with(
  275                 mock.ANY, mock.ANY, 'http://kernel/img', 'http://ramdisk/img',
  276                 boot_mode='uefi', esp_image_href=None,
  277                 kernel_params='nofb nomodeset vga=normal',
  278                 root_uuid='1be26c0b-03f2-4d2e-ae87-c02d7f33c123',
  279                 base_iso='/path/to/baseiso', inject_files=None)
  280 
  281     @mock.patch.object(image_utils.ImageHandler, 'publish_image',
  282                        autospec=True)
  283     @mock.patch.object(images, 'create_boot_iso', autospec=True)
  284     def test__prepare_iso_image_bios(
  285             self, mock_create_boot_iso, mock_publish_image):
  286         with task_manager.acquire(self.context, self.node.uuid,
  287                                   shared=True) as task:
  288 
  289             expected_url = 'https://a.b/c.f?e=f'
  290 
  291             mock_publish_image.return_value = expected_url
  292 
  293             url = image_utils._prepare_iso_image(
  294                 task, 'http://kernel/img', 'http://ramdisk/img',
  295                 bootloader_href=None, root_uuid=task.node.uuid, base_iso=None)
  296 
  297             object_name = 'boot-%s.iso' % task.node.uuid
  298 
  299             mock_publish_image.assert_called_once_with(
  300                 mock.ANY, mock.ANY, object_name)
  301 
  302             mock_create_boot_iso.assert_called_once_with(
  303                 mock.ANY, mock.ANY, 'http://kernel/img', 'http://ramdisk/img',
  304                 boot_mode='bios', esp_image_href=None,
  305                 kernel_params='nofb nomodeset vga=normal',
  306                 root_uuid='1be26c0b-03f2-4d2e-ae87-c02d7f33c123',
  307                 base_iso=None, inject_files=None)
  308 
  309             self.assertEqual(expected_url, url)
  310 
  311     @mock.patch.object(image_utils.ImageHandler, 'publish_image',
  312                        autospec=True)
  313     @mock.patch.object(images, 'create_boot_iso', autospec=True)
  314     def test__prepare_iso_image_kernel_params(
  315             self, mock_create_boot_iso, mock_publish_image):
  316         with task_manager.acquire(self.context, self.node.uuid,
  317                                   shared=True) as task:
  318             kernel_params = 'network-config=base64-cloudinit-blob'
  319 
  320             task.node.instance_info.update(kernel_append_params=kernel_params)
  321 
  322             image_utils._prepare_iso_image(
  323                 task, 'http://kernel/img', 'http://ramdisk/img',
  324                 bootloader_href=None, root_uuid=task.node.uuid,
  325                 base_iso='/path/to/baseiso')
  326 
  327             mock_create_boot_iso.assert_called_once_with(
  328                 mock.ANY, mock.ANY, 'http://kernel/img', 'http://ramdisk/img',
  329                 boot_mode='bios', esp_image_href=None,
  330                 kernel_params=kernel_params,
  331                 root_uuid='1be26c0b-03f2-4d2e-ae87-c02d7f33c123',
  332                 base_iso='/path/to/baseiso', inject_files=None)
  333 
  334     def test__find_param(self):
  335         param_dict = {
  336             'ilo_deploy_kernel': 'kernel',
  337             'ilo_deploy_ramdisk': 'ramdisk',
  338             'ilo_bootloader': 'bootloader'
  339         }
  340         param_str = "deploy_kernel"
  341         expected = "kernel"
  342 
  343         actual = image_utils._find_param(param_str, param_dict)
  344         self.assertEqual(actual, expected)
  345 
  346     def test__find_param_not_found(self):
  347         param_dict = {
  348             'ilo_deploy_ramdisk': 'ramdisk',
  349             'ilo_bootloader': 'bootloader'
  350         }
  351         param_str = "deploy_kernel"
  352         expected = None
  353         actual = image_utils._find_param(param_str, param_dict)
  354         self.assertEqual(actual, expected)
  355 
  356     @mock.patch.object(image_utils, '_find_param', autospec=True)
  357     @mock.patch.object(image_utils, '_prepare_iso_image', autospec=True)
  358     def test_prepare_deploy_iso(self, mock__prepare_iso_image,
  359                                 find_mock):
  360         with task_manager.acquire(self.context, self.node.uuid,
  361                                   shared=True) as task:
  362 
  363             d_info = {
  364                 'ilo_deploy_kernel': 'kernel',
  365                 'ilo_deploy_ramdisk': 'ramdisk',
  366                 'ilo_bootloader': 'bootloader'
  367             }
  368             task.node.driver_info.update(d_info)
  369 
  370             find_call_list = [
  371                 mock.call('deploy_kernel', d_info),
  372                 mock.call('deploy_ramdisk', d_info),
  373                 mock.call('bootloader', d_info)
  374             ]
  375             find_mock.side_effect = [
  376                 'kernel', 'ramdisk', 'bootloader'
  377             ]
  378 
  379             task.node.instance_info.update(deploy_boot_mode='uefi')
  380 
  381             image_utils.prepare_deploy_iso(task, {}, 'deploy', d_info)
  382 
  383             mock__prepare_iso_image.assert_called_once_with(
  384                 task, 'kernel', 'ramdisk', 'bootloader', params={},
  385                 inject_files={})
  386 
  387             find_mock.assert_has_calls(find_call_list)
  388 
  389     @mock.patch.object(image_utils, '_find_param', autospec=True)
  390     @mock.patch.object(image_utils, '_prepare_iso_image', autospec=True)
  391     def test_prepare_deploy_iso_network_data(
  392             self, mock__prepare_iso_image, find_mock):
  393         with task_manager.acquire(self.context, self.node.uuid,
  394                                   shared=True) as task:
  395 
  396             d_info = {
  397                 'ilo_deploy_kernel': 'kernel',
  398                 'ilo_deploy_ramdisk': 'ramdisk'
  399             }
  400             task.node.driver_info.update(d_info)
  401 
  402             task.node.instance_info.update()
  403 
  404             find_call_list = [
  405                 mock.call('deploy_kernel', d_info),
  406                 mock.call('deploy_ramdisk', d_info),
  407                 mock.call('bootloader', d_info)
  408             ]
  409             find_mock.side_effect = [
  410                 'kernel', 'ramdisk', None
  411             ]
  412             network_data = {'a': ['b']}
  413 
  414             mock_get_node_nw_data = mock.MagicMock(return_value=network_data)
  415             task.driver.network.get_node_network_data = mock_get_node_nw_data
  416 
  417             image_utils.prepare_deploy_iso(task, {}, 'deploy', d_info)
  418 
  419             expected_files = {
  420                 b"""{
  421   "a": [
  422     "b"
  423   ]
  424 }""": 'openstack/latest/network_data.json'
  425             }
  426 
  427             mock__prepare_iso_image.assert_called_once_with(
  428                 task, 'kernel', 'ramdisk', bootloader_href=None,
  429                 params={}, inject_files=expected_files)
  430 
  431             find_mock.assert_has_calls(find_call_list)
  432 
  433     @mock.patch.object(image_utils, '_find_param', autospec=True)
  434     @mock.patch.object(image_utils, '_prepare_iso_image', autospec=True)
  435     @mock.patch.object(images, 'create_boot_iso', autospec=True)
  436     def test_prepare_boot_iso(self, mock_create_boot_iso,
  437                               mock__prepare_iso_image, find_mock):
  438         with task_manager.acquire(self.context, self.node.uuid,
  439                                   shared=True) as task:
  440             d_info = {
  441                 'ilo_deploy_kernel': 'kernel',
  442                 'ilo_deploy_ramdisk': 'ramdisk',
  443                 'ilo_bootloader': 'bootloader'
  444             }
  445             task.node.driver_info.update(d_info)
  446 
  447             task.node.instance_info.update(
  448                 {'image_source': 'http://boot/iso',
  449                  'kernel': 'http://kernel/img',
  450                  'ramdisk': 'http://ramdisk/img'})
  451 
  452             find_call_list = [
  453                 mock.call('bootloader', d_info)
  454             ]
  455 
  456             find_mock.side_effect = [
  457                 'bootloader'
  458             ]
  459 
  460             image_utils.prepare_boot_iso(
  461                 task, d_info, root_uuid=task.node.uuid)
  462 
  463             mock__prepare_iso_image.assert_called_once_with(
  464                 mock.ANY, 'http://kernel/img', 'http://ramdisk/img',
  465                 'bootloader', root_uuid=task.node.uuid,
  466                 base_iso=None)
  467 
  468             find_mock.assert_has_calls(find_call_list)
  469 
  470     @mock.patch.object(image_utils, '_find_param', autospec=True)
  471     @mock.patch.object(image_utils, '_prepare_iso_image', autospec=True)
  472     @mock.patch.object(images, 'create_boot_iso', autospec=True)
  473     def test_prepare_boot_iso_user_supplied(self, mock_create_boot_iso,
  474                                             mock__prepare_iso_image,
  475                                             find_mock):
  476         with task_manager.acquire(self.context, self.node.uuid,
  477                                   shared=True) as task:
  478             d_info = {
  479                 'deploy_kernel': 'kernel',
  480                 'deploy_ramdisk': 'ramdisk',
  481                 'bootloader': 'bootloader'
  482             }
  483             task.node.driver_info.update(d_info)
  484 
  485             task.node.instance_info.update(
  486                 {'boot_iso': 'http://boot/iso'})
  487 
  488             find_call_list = [
  489                 mock.call('bootloader', d_info)
  490             ]
  491 
  492             find_mock.side_effect = [
  493                 'bootloader'
  494             ]
  495             image_utils.prepare_boot_iso(
  496                 task, d_info, root_uuid=task.node.uuid)
  497 
  498             mock__prepare_iso_image.assert_called_once_with(
  499                 mock.ANY, None, None,
  500                 'bootloader', root_uuid=task.node.uuid,
  501                 base_iso='http://boot/iso')
  502 
  503             find_mock.assert_has_calls(find_call_list)