"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "glance/tests/unit/async_/flows/test_copy_image.py" between
glance-20.0.0.tar.gz and glance-20.0.1.tar.gz

About: OpenStack Glance (Core Service: Image Service) stores and retrieves virtual machine disk images. OpenStack Compute makes use of this during instance provisioning.
The "Ussuri" series (latest release).

test_copy_image.py  (glance-20.0.0):test_copy_image.py  (glance-20.0.1)
skipping to change at line 17 skipping to change at line 17
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import datetime import datetime
import os
from unittest import mock from unittest import mock
import glance_store as store_api import glance_store as store_api
from oslo_config import cfg from oslo_config import cfg
from glance.async_.flows._internal_plugins import copy_image from glance.async_.flows._internal_plugins import copy_image
import glance.common.exception as exception import glance.common.exception as exception
from glance import domain from glance import domain
import glance.tests.unit.utils as unit_test_utils import glance.tests.unit.utils as unit_test_utils
import glance.tests.utils as test_utils import glance.tests.utils as test_utils
skipping to change at line 139 skipping to change at line 140
locations=self.images[0]['locations'], locations=self.images[0]['locations'],
status=self.images[0]['status'] status=self.images[0]['status']
) )
with mock.patch.object(store_api, 'get') as get_data: with mock.patch.object(store_api, 'get') as get_data:
get_data.return_value = (b"dddd", 4) get_data.return_value = (b"dddd", 4)
copy_image_task.execute() copy_image_task.execute()
self.staging_store.add.assert_called_once() self.staging_store.add.assert_called_once()
mock_store_api.assert_called_once_with( mock_store_api.assert_called_once_with(
"os_glance_staging_store") "os_glance_staging_store")
@mock.patch.object(os, 'unlink')
@mock.patch.object(os.path, 'getsize')
@mock.patch.object(os.path, 'exists')
@mock.patch.object(store_api, 'get_store_from_store_identifier')
def test_copy_image_to_staging_store_partial_data_exists(
self, mock_store_api, mock_exists, mock_getsize, mock_unlink):
mock_store_api.return_value = self.staging_store
mock_exists.return_value = True
mock_getsize.return_value = 3
copy_image_task = copy_image._CopyImage(
self.task.task_id, self.task_type, self.image_repo,
self.image_id)
with mock.patch.object(self.image_repo, 'get') as get_mock:
get_mock.return_value = mock.MagicMock(
image_id=self.images[0]['id'],
locations=self.images[0]['locations'],
status=self.images[0]['status'],
size=4
)
with mock.patch.object(store_api, 'get') as get_data:
get_data.return_value = (b"dddd", 4)
copy_image_task.execute()
mock_exists.assert_called_once()
mock_getsize.assert_called_once()
mock_unlink.assert_called_once()
self.staging_store.add.assert_called_once()
mock_store_api.assert_called_once_with(
"os_glance_staging_store")
@mock.patch.object(os, 'unlink')
@mock.patch.object(os.path, 'getsize')
@mock.patch.object(os.path, 'exists')
@mock.patch.object(store_api, 'get_store_from_store_identifier')
def test_copy_image_to_staging_store_data_exists(
self, mock_store_api, mock_exists, mock_getsize, mock_unlink):
mock_store_api.return_value = self.staging_store
mock_exists.return_value = True
mock_getsize.return_value = 4
copy_image_task = copy_image._CopyImage(
self.task.task_id, self.task_type, self.image_repo,
self.image_id)
with mock.patch.object(self.image_repo, 'get') as get_mock:
get_mock.return_value = mock.MagicMock(
image_id=self.images[0]['id'],
locations=self.images[0]['locations'],
status=self.images[0]['status'],
size=4
)
copy_image_task.execute()
mock_exists.assert_called_once()
mock_store_api.assert_called_once_with(
"os_glance_staging_store")
mock_getsize.assert_called_once()
# As valid image data already exists in staging area
# it does not remove it and also does not download
# it again to staging area
mock_unlink.assert_not_called()
self.staging_store.add.assert_not_called()
@mock.patch.object(store_api, 'get_store_from_store_identifier') @mock.patch.object(store_api, 'get_store_from_store_identifier')
def test_copy_non_existing_image_to_staging_store_(self, mock_store_api): def test_copy_non_existing_image_to_staging_store_(self, mock_store_api):
mock_store_api.return_value = self.staging_store mock_store_api.return_value = self.staging_store
copy_image_task = copy_image._CopyImage( copy_image_task = copy_image._CopyImage(
self.task.task_id, self.task_type, self.image_repo, self.task.task_id, self.task_type, self.image_repo,
self.image_id) self.image_id)
with mock.patch.object(self.image_repo, 'get') as get_mock: with mock.patch.object(self.image_repo, 'get') as get_mock:
get_mock.side_effect = exception.NotFound() get_mock.side_effect = exception.NotFound()
self.assertRaises(exception.NotFound, copy_image_task.execute) self.assertRaises(exception.NotFound, copy_image_task.execute)
 End of changes. 2 change blocks. 
0 lines changed or deleted 62 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)