"Fossies" - the Fresh Open Source Software Archive

Member "glance-20.0.1/glance/tests/unit/async_/test_async.py" (12 Aug 2020, 7140 Bytes) of package /linux/misc/openstack/glance-20.0.1.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_async.py": 20.0.0_vs_20.0.1.

    1 # Copyright 2014 OpenStack Foundation
    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 
   17 from unittest import mock
   18 
   19 import glance_store as store
   20 from oslo_config import cfg
   21 from taskflow.patterns import linear_flow
   22 
   23 import glance.async_
   24 from glance.async_.flows import api_image_import
   25 import glance.tests.utils as test_utils
   26 
   27 CONF = cfg.CONF
   28 
   29 
   30 class TestTaskExecutor(test_utils.BaseTestCase):
   31 
   32     def setUp(self):
   33         super(TestTaskExecutor, self).setUp()
   34         self.context = mock.Mock()
   35         self.task_repo = mock.Mock()
   36         self.image_repo = mock.Mock()
   37         self.image_factory = mock.Mock()
   38         self.executor = glance.async_.TaskExecutor(self.context,
   39                                                    self.task_repo,
   40                                                    self.image_repo,
   41                                                    self.image_factory)
   42 
   43     def test_begin_processing(self):
   44         # setup
   45         task_id = mock.ANY
   46         task_type = mock.ANY
   47         task = mock.Mock()
   48 
   49         with mock.patch.object(
   50                 glance.async_.TaskExecutor,
   51                 '_run') as mock_run:
   52             self.task_repo.get.return_value = task
   53             self.executor.begin_processing(task_id)
   54 
   55         # assert the call
   56         mock_run.assert_called_once_with(task_id, task_type)
   57 
   58 
   59 class TestImportTaskFlow(test_utils.BaseTestCase):
   60 
   61     def setUp(self):
   62         super(TestImportTaskFlow, self).setUp()
   63         store.register_opts(CONF)
   64         self.config(default_store='file',
   65                     stores=['file', 'http'],
   66                     filesystem_store_datadir=self.test_dir,
   67                     group="glance_store")
   68         self.config(enabled_import_methods=[
   69             'glance-direct', 'web-download', 'copy-image'])
   70         self.config(node_staging_uri='file:///tmp/staging')
   71         store.create_stores(CONF)
   72         self.base_flow = ['ConfigureStaging', 'ImportToStore',
   73                           'DeleteFromFS', 'VerifyImageState',
   74                           'CompleteTask']
   75         self.import_plugins = ['Convert_Image',
   76                                'Decompress_Image',
   77                                'InjectMetadataProperties']
   78 
   79     def _get_flow(self, import_req=None):
   80         inputs = {
   81             'task_id': mock.MagicMock(),
   82             'task_type': mock.MagicMock(),
   83             'task_repo': mock.MagicMock(),
   84             'image_repo': mock.MagicMock(),
   85             'image_id': mock.MagicMock(),
   86             'import_req': import_req or mock.MagicMock()
   87         }
   88         flow = api_image_import.get_flow(**inputs)
   89         return flow
   90 
   91     def _get_flow_tasks(self, flow):
   92         flow_comp = []
   93         for c, p in flow.iter_nodes():
   94             if isinstance(c, linear_flow.Flow):
   95                 flow_comp += self._get_flow_tasks(c)
   96             else:
   97                 name = str(c).split('-')
   98                 if len(name) > 1:
   99                     flow_comp.append(name[1])
  100         return flow_comp
  101 
  102     def test_get_default_flow(self):
  103         # This test will ensure that without import plugins
  104         # and without internal plugins flow builds with the
  105         # base_flow components
  106         flow = self._get_flow()
  107 
  108         flow_comp = self._get_flow_tasks(flow)
  109         # assert flow has 5 tasks
  110         self.assertEqual(5, len(flow_comp))
  111         for c in self.base_flow:
  112             self.assertIn(c, flow_comp)
  113 
  114     def test_get_flow_web_download_enabled(self):
  115         # This test will ensure that without import plugins
  116         # and with web-download plugin flow builds with
  117         # base_flow components and '_WebDownload'
  118         import_req = {
  119             'method': {
  120                 'name': 'web-download',
  121                 'uri': 'http://cloud.foo/image.qcow2'
  122             }
  123         }
  124 
  125         flow = self._get_flow(import_req=import_req)
  126 
  127         flow_comp = self._get_flow_tasks(flow)
  128         # assert flow has 6 tasks
  129         self.assertEqual(6, len(flow_comp))
  130         for c in self.base_flow:
  131             self.assertIn(c, flow_comp)
  132         self.assertIn('WebDownload', flow_comp)
  133 
  134     @mock.patch.object(store, 'get_store_from_store_identifier')
  135     def test_get_flow_copy_image_enabled(self, mock_store):
  136         # This test will ensure that without import plugins
  137         # and with copy-image plugin flow builds with
  138         # base_flow components and '_CopyImage'
  139         import_req = {
  140             'method': {
  141                 'name': 'copy-image',
  142                 'stores': ['fake-store']
  143             }
  144         }
  145 
  146         mock_store.return_value = mock.Mock()
  147         flow = self._get_flow(import_req=import_req)
  148 
  149         flow_comp = self._get_flow_tasks(flow)
  150         # assert flow has 6 tasks
  151         self.assertEqual(6, len(flow_comp))
  152         for c in self.base_flow:
  153             self.assertIn(c, flow_comp)
  154         self.assertIn('CopyImage', flow_comp)
  155 
  156     def test_get_flow_with_all_plugins_enabled(self):
  157         # This test will ensure that flow includes import plugins
  158         # and base flow
  159         self.config(image_import_plugins=['image_conversion',
  160                                           'image_decompression',
  161                                           'inject_image_metadata'],
  162                     group='image_import_opts')
  163 
  164         flow = self._get_flow()
  165 
  166         flow_comp = self._get_flow_tasks(flow)
  167         # assert flow has 8 tasks (base_flow + plugins)
  168         self.assertEqual(8, len(flow_comp))
  169         for c in self.base_flow:
  170             self.assertIn(c, flow_comp)
  171         for c in self.import_plugins:
  172             self.assertIn(c, flow_comp)
  173 
  174     @mock.patch.object(store, 'get_store_from_store_identifier')
  175     def test_get_flow_copy_image_not_includes_import_plugins(
  176             self, mock_store):
  177         # This test will ensure that flow does not includes import
  178         # plugins as import method is copy image
  179         self.config(image_import_plugins=['image_conversion',
  180                                           'image_decompression',
  181                                           'inject_image_metadata'],
  182                     group='image_import_opts')
  183 
  184         mock_store.return_value = mock.Mock()
  185         import_req = {
  186             'method': {
  187                 'name': 'copy-image',
  188                 'stores': ['fake-store']
  189             }
  190         }
  191 
  192         flow = self._get_flow(import_req=import_req)
  193 
  194         flow_comp = self._get_flow_tasks(flow)
  195         # assert flow has 6 tasks
  196         self.assertEqual(6, len(flow_comp))
  197         for c in self.base_flow:
  198             self.assertIn(c, flow_comp)
  199         self.assertIn('CopyImage', flow_comp)