"Fossies" - the Fresh Open Source Software Archive

Member "glance-20.0.1/glance/tests/unit/async_/flows/test_copy_image.py" (12 Aug 2020, 8518 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_copy_image.py": 20.0.0_vs_20.0.1.

    1 # Copyright 2020 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 datetime
   17 import os
   18 from unittest import mock
   19 
   20 import glance_store as store_api
   21 from oslo_config import cfg
   22 
   23 from glance.async_.flows._internal_plugins import copy_image
   24 import glance.common.exception as exception
   25 from glance import domain
   26 import glance.tests.unit.utils as unit_test_utils
   27 import glance.tests.utils as test_utils
   28 
   29 CONF = cfg.CONF
   30 
   31 DATETIME = datetime.datetime(2012, 5, 16, 15, 27, 36, 325355)
   32 
   33 
   34 TENANT1 = '6838eb7b-6ded-434a-882c-b344c77fe8df'
   35 UUID1 = 'c80a1a6c-bd1f-41c5-90ee-81afedb1d58d'
   36 FAKEHASHALGO = 'fake-name-for-sha512'
   37 CHKSUM = '93264c3edf5972c9f1cb309543d38a5c'
   38 RESERVED_STORES = {
   39     'os_glance_staging_store': 'file',
   40 }
   41 
   42 
   43 def _db_fixture(id, **kwargs):
   44     obj = {
   45         'id': id,
   46         'name': None,
   47         'visibility': 'shared',
   48         'properties': {},
   49         'checksum': None,
   50         'os_hash_algo': FAKEHASHALGO,
   51         'os_hash_value': None,
   52         'owner': None,
   53         'status': 'queued',
   54         'tags': [],
   55         'size': None,
   56         'virtual_size': None,
   57         'locations': [],
   58         'protected': False,
   59         'disk_format': None,
   60         'container_format': None,
   61         'deleted': False,
   62         'min_ram': None,
   63         'min_disk': None,
   64     }
   65     obj.update(kwargs)
   66     return obj
   67 
   68 
   69 class TestCopyImageTask(test_utils.BaseTestCase):
   70 
   71     def setUp(self):
   72         super(TestCopyImageTask, self).setUp()
   73 
   74         self.db = unit_test_utils.FakeDB(initialize=False)
   75         self._create_images()
   76         self.image_repo = mock.MagicMock()
   77         self.task_repo = mock.MagicMock()
   78         self.image_id = UUID1
   79         self.staging_store = mock.MagicMock()
   80         self.task_factory = domain.TaskFactory()
   81 
   82         task_input = {
   83             "import_req": {
   84                 'method': {
   85                     'name': 'copy-image',
   86                 },
   87                 'stores': ['fast']
   88             }
   89         }
   90         task_ttl = CONF.task.task_time_to_live
   91 
   92         self.task_type = 'import'
   93         self.task = self.task_factory.new_task(self.task_type, TENANT1,
   94                                                task_time_to_live=task_ttl,
   95                                                task_input=task_input)
   96 
   97         stores = {'cheap': 'file', 'fast': 'file'}
   98         self.config(enabled_backends=stores)
   99         store_api.register_store_opts(CONF, reserved_stores=RESERVED_STORES)
  100         self.config(default_backend='fast', group='glance_store')
  101         store_api.create_multi_stores(CONF, reserved_stores=RESERVED_STORES)
  102 
  103     def _create_images(self):
  104         self.images = [
  105             _db_fixture(UUID1, owner=TENANT1, checksum=CHKSUM,
  106                         name='1', size=512, virtual_size=2048,
  107                         visibility='public',
  108                         disk_format='raw',
  109                         container_format='bare',
  110                         status='active',
  111                         tags=['redhat', '64bit', 'power'],
  112                         properties={'hypervisor_type': 'kvm', 'foo': 'bar',
  113                                     'bar': 'foo'},
  114                         locations=[{'url': 'file://%s/%s' % (self.test_dir,
  115                                                              UUID1),
  116                                     'metadata': {'store': 'fast'},
  117                                     'status': 'active'}],
  118                         created_at=DATETIME + datetime.timedelta(seconds=1)),
  119         ]
  120         [self.db.image_create(None, image) for image in self.images]
  121 
  122         self.db.image_tag_set_all(None, UUID1, ['ping', 'pong'])
  123 
  124     @mock.patch.object(store_api, 'get_store_from_store_identifier')
  125     def test_copy_image_to_staging_store(self, mock_store_api):
  126         mock_store_api.return_value = self.staging_store
  127         copy_image_task = copy_image._CopyImage(
  128             self.task.task_id, self.task_type, self.image_repo,
  129             self.image_id)
  130         with mock.patch.object(self.image_repo, 'get') as get_mock:
  131             get_mock.return_value = mock.MagicMock(
  132                 image_id=self.images[0]['id'],
  133                 locations=self.images[0]['locations'],
  134                 status=self.images[0]['status']
  135             )
  136             with mock.patch.object(store_api, 'get') as get_data:
  137                 get_data.return_value = (b"dddd", 4)
  138                 copy_image_task.execute()
  139                 self.staging_store.add.assert_called_once()
  140                 mock_store_api.assert_called_once_with(
  141                     "os_glance_staging_store")
  142 
  143     @mock.patch.object(os, 'unlink')
  144     @mock.patch.object(os.path, 'getsize')
  145     @mock.patch.object(os.path, 'exists')
  146     @mock.patch.object(store_api, 'get_store_from_store_identifier')
  147     def test_copy_image_to_staging_store_partial_data_exists(
  148             self, mock_store_api, mock_exists, mock_getsize, mock_unlink):
  149         mock_store_api.return_value = self.staging_store
  150         mock_exists.return_value = True
  151         mock_getsize.return_value = 3
  152 
  153         copy_image_task = copy_image._CopyImage(
  154             self.task.task_id, self.task_type, self.image_repo,
  155             self.image_id)
  156         with mock.patch.object(self.image_repo, 'get') as get_mock:
  157             get_mock.return_value = mock.MagicMock(
  158                 image_id=self.images[0]['id'],
  159                 locations=self.images[0]['locations'],
  160                 status=self.images[0]['status'],
  161                 size=4
  162             )
  163             with mock.patch.object(store_api, 'get') as get_data:
  164                 get_data.return_value = (b"dddd", 4)
  165                 copy_image_task.execute()
  166                 mock_exists.assert_called_once()
  167                 mock_getsize.assert_called_once()
  168                 mock_unlink.assert_called_once()
  169                 self.staging_store.add.assert_called_once()
  170                 mock_store_api.assert_called_once_with(
  171                     "os_glance_staging_store")
  172 
  173     @mock.patch.object(os, 'unlink')
  174     @mock.patch.object(os.path, 'getsize')
  175     @mock.patch.object(os.path, 'exists')
  176     @mock.patch.object(store_api, 'get_store_from_store_identifier')
  177     def test_copy_image_to_staging_store_data_exists(
  178             self, mock_store_api, mock_exists, mock_getsize, mock_unlink):
  179         mock_store_api.return_value = self.staging_store
  180         mock_exists.return_value = True
  181         mock_getsize.return_value = 4
  182 
  183         copy_image_task = copy_image._CopyImage(
  184             self.task.task_id, self.task_type, self.image_repo,
  185             self.image_id)
  186         with mock.patch.object(self.image_repo, 'get') as get_mock:
  187             get_mock.return_value = mock.MagicMock(
  188                 image_id=self.images[0]['id'],
  189                 locations=self.images[0]['locations'],
  190                 status=self.images[0]['status'],
  191                 size=4
  192             )
  193             copy_image_task.execute()
  194             mock_exists.assert_called_once()
  195             mock_store_api.assert_called_once_with(
  196                 "os_glance_staging_store")
  197             mock_getsize.assert_called_once()
  198             # As valid image data already exists in staging area
  199             # it does not remove it and also does not download
  200             # it again to staging area
  201             mock_unlink.assert_not_called()
  202             self.staging_store.add.assert_not_called()
  203 
  204     @mock.patch.object(store_api, 'get_store_from_store_identifier')
  205     def test_copy_non_existing_image_to_staging_store_(self, mock_store_api):
  206         mock_store_api.return_value = self.staging_store
  207         copy_image_task = copy_image._CopyImage(
  208             self.task.task_id, self.task_type, self.image_repo,
  209             self.image_id)
  210         with mock.patch.object(self.image_repo, 'get') as get_mock:
  211             get_mock.side_effect = exception.NotFound()
  212 
  213             self.assertRaises(exception.NotFound, copy_image_task.execute)
  214             mock_store_api.assert_called_once_with(
  215                 "os_glance_staging_store")