"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")