"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "glance/tests/unit/v2/test_image_data_resource.py" between
glance-16.0.1.tar.gz and glance-18.0.0.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 "Stein" series (latest release).

test_image_data_resource.py  (glance-16.0.1):test_image_data_resource.py  (glance-18.0.0)
skipping to change at line 34 skipping to change at line 34
import glance.api.policy import glance.api.policy
import glance.api.v2.image_data import glance.api.v2.image_data
from glance.common import exception from glance.common import exception
from glance.common import wsgi from glance.common import wsgi
from glance.tests.unit import base from glance.tests.unit import base
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
class Raise(object): class Raise(object):
def __init__(self, exc): def __init__(self, exc):
self.exc = exc self.exc = exc
def __call__(self, *args, **kwargs): def __call__(self, *args, **kwargs):
raise self.exc raise self.exc
class FakeImage(object): class FakeImage(object):
def __init__(self, image_id=None, data=None, checksum=None, size=0, def __init__(self, image_id=None, data=None, checksum=None, size=0,
virtual_size=0, locations=None, container_format='bear', virtual_size=0, locations=None, container_format='bear',
disk_format='rawr', status=None): disk_format='rawr', status=None):
self.image_id = image_id self.image_id = image_id
self.data = data self.data = data
self.checksum = checksum self.checksum = checksum
self.size = size self.size = size
self.virtual_size = virtual_size self.virtual_size = virtual_size
self.locations = locations self.locations = locations
self.container_format = container_format self.container_format = container_format
skipping to change at line 70 skipping to change at line 72
if isinstance(self._status, BaseException): if isinstance(self._status, BaseException):
raise self._status raise self._status
else: else:
self._status = value self._status = value
def get_data(self, offset=0, chunk_size=None): def get_data(self, offset=0, chunk_size=None):
if chunk_size: if chunk_size:
return self.data[offset:offset + chunk_size] return self.data[offset:offset + chunk_size]
return self.data[offset:] return self.data[offset:]
def set_data(self, data, size=None): def set_data(self, data, size=None, backend=None):
self.data = ''.join(data) self.data = ''.join(data)
self.size = size self.size = size
self.status = 'modified-by-fake' self.status = 'modified-by-fake'
class FakeImageRepo(object): class FakeImageRepo(object):
def __init__(self, result=None): def __init__(self, result=None):
self.result = result self.result = result
def get(self, image_id): def get(self, image_id):
if isinstance(self.result, BaseException): if isinstance(self.result, BaseException):
raise self.result raise self.result
else: else:
return self.result return self.result
def save(self, image, from_state=None): def save(self, image, from_state=None):
self.saved_image = image self.saved_image = image
class FakeGateway(object): class FakeGateway(object):
def __init__(self, db=None, store=None, notifier=None, def __init__(self, db=None, store=None, notifier=None,
policy=None, repo=None): policy=None, repo=None):
self.db = db self.db = db
self.store = store self.store = store
self.notifier = notifier self.notifier = notifier
self.policy = policy self.policy = policy
self.repo = repo self.repo = repo
def get_repo(self, context): def get_repo(self, context):
return self.repo return self.repo
class TestImagesController(base.StoreClearingUnitTest): class TestImagesController(base.StoreClearingUnitTest):
def setUp(self): def setUp(self):
super(TestImagesController, self).setUp() super(TestImagesController, self).setUp()
self.config(debug=True) self.config(debug=True)
self.image_repo = FakeImageRepo() self.image_repo = FakeImageRepo()
db = unit_test_utils.FakeDB() db = unit_test_utils.FakeDB()
policy = unit_test_utils.FakePolicyEnforcer() policy = unit_test_utils.FakePolicyEnforcer()
notifier = unit_test_utils.FakeNotifier() notifier = unit_test_utils.FakeNotifier()
store = unit_test_utils.FakeStoreAPI() store = unit_test_utils.FakeStoreAPI()
self.controller = glance.api.v2.image_data.ImageDataController() self.controller = glance.api.v2.image_data.ImageDataController()
skipping to change at line 155 skipping to change at line 160
request, str(uuid.uuid4())) request, str(uuid.uuid4()))
def test_download_forbidden(self): def test_download_forbidden(self):
request = unit_test_utils.get_fake_request() request = unit_test_utils.get_fake_request()
self.image_repo.result = exception.Forbidden() self.image_repo.result = exception.Forbidden()
self.assertRaises(webob.exc.HTTPForbidden, self.controller.download, self.assertRaises(webob.exc.HTTPForbidden, self.controller.download,
request, str(uuid.uuid4())) request, str(uuid.uuid4()))
def test_download_ok_when_get_image_location_forbidden(self): def test_download_ok_when_get_image_location_forbidden(self):
class ImageLocations(object): class ImageLocations(object):
def __len__(self): def __len__(self):
raise exception.Forbidden() raise exception.Forbidden()
request = unit_test_utils.get_fake_request() request = unit_test_utils.get_fake_request()
image = FakeImage('abcd') image = FakeImage('abcd')
self.image_repo.result = image self.image_repo.result = image
image.locations = ImageLocations() image.locations = ImageLocations()
image = self.controller.download(request, unit_test_utils.UUID1) image = self.controller.download(request, unit_test_utils.UUID1)
self.assertEqual('abcd', image.image_id) self.assertEqual('abcd', image.image_id)
skipping to change at line 395 skipping to change at line 401
# mocking save to raise NotAuthenticated on the second call # mocking save to raise NotAuthenticated on the second call
FakeImageRepo.save = mock_fake_save FakeImageRepo.save = mock_fake_save
self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4) self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4)
# check image data # check image data
self.assertEqual('YYYY', image.data) self.assertEqual('YYYY', image.data)
self.assertEqual(4, image.size) self.assertEqual(4, image.size)
FakeImageRepo.save = temp_save FakeImageRepo.save = temp_save
# check that token has been correctly acquired and deleted # check that token has been correctly acquired and deleted
mock_refresher.assert_called_once_with( mock_refresher.assert_called_once_with(
request.environ['keystone.token_auth'], request.environ['keystone.token_auth'],
request.context.tenant, ['FakeRole']) request.context.project_id, ['FakeRole'])
refresher.refresh_token.assert_called_once_with() refresher.refresh_token.assert_called_once_with()
refresher.release_resources.assert_called_once_with() refresher.release_resources.assert_called_once_with()
self.assertEqual("fake_token", request.context.auth_token) self.assertEqual("fake_token", request.context.auth_token)
@mock.patch("glance.common.trust_auth.TokenRefresher") @mock.patch("glance.common.trust_auth.TokenRefresher")
def test_upload_with_trusts_fails(self, mock_refresher): def test_upload_with_trusts_fails(self, mock_refresher):
"""Test upload with registry if trust was not successfully created""" """Test upload with registry if trust was not successfully created"""
# initialize trust environment # initialize trust environment
self.config(data_api='glance.db.registry.api') self.config(data_api='glance.db.registry.api')
mock_refresher().side_effect = Exception() mock_refresher().side_effect = Exception()
skipping to change at line 668 skipping to change at line 674
self.assertRaises(webob.exc.HTTPUnsupportedMediaType, self.assertRaises(webob.exc.HTTPUnsupportedMediaType,
self.deserializer.upload, request) self.deserializer.upload, request)
request = unit_test_utils.get_fake_request() request = unit_test_utils.get_fake_request()
request.headers['Content-Type'] = 'application/octet-st' request.headers['Content-Type'] = 'application/octet-st'
request.body = b'YYYYY' request.body = b'YYYYY'
self.assertRaises(webob.exc.HTTPUnsupportedMediaType, self.assertRaises(webob.exc.HTTPUnsupportedMediaType,
self.deserializer.upload, request) self.deserializer.upload, request)
def test_stage(self): def test_stage(self):
self.config(enable_image_import=True)
req = unit_test_utils.get_fake_request() req = unit_test_utils.get_fake_request()
req.headers['Content-Type'] = 'application/octet-stream' req.headers['Content-Type'] = 'application/octet-stream'
req.headers['Content-Length'] = 4 req.headers['Content-Length'] = 4
req.body_file = six.BytesIO(b'YYYY') req.body_file = six.BytesIO(b'YYYY')
output = self.deserializer.stage(req) output = self.deserializer.stage(req)
data = output.pop('data') data = output.pop('data')
self.assertEqual(b'YYYY', data.read()) self.assertEqual(b'YYYY', data.read())
def test_stage_if_image_import_is_disabled(self): def test_stage_without_glance_direct(self):
self.config(enable_image_import=False) self.config(enabled_import_methods=['web-download'])
req = unit_test_utils.get_fake_request() req = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPNotFound, self.assertRaises(webob.exc.HTTPNotFound,
self.deserializer.stage, self.deserializer.stage,
req) req)
def test_stage_raises_invalid_content_type(self): def test_stage_raises_invalid_content_type(self):
# TODO(abhishekk): change this when import methods are # TODO(abhishekk): change this when import methods are
# listed in the config file # listed in the config file
self.config(enable_image_import=True)
req = unit_test_utils.get_fake_request() req = unit_test_utils.get_fake_request()
req.headers['Content-Type'] = 'application/json' req.headers['Content-Type'] = 'application/json'
self.assertRaises(webob.exc.HTTPUnsupportedMediaType, self.assertRaises(webob.exc.HTTPUnsupportedMediaType,
self.deserializer.stage, self.deserializer.stage,
req) req)
class TestImageDataSerializer(test_utils.BaseTestCase): class TestImageDataSerializer(test_utils.BaseTestCase):
def setUp(self): def setUp(self):
super(TestImageDataSerializer, self).setUp() super(TestImageDataSerializer, self).setUp()
skipping to change at line 886 skipping to change at line 890
self.assertEqual('application/octet-stream', self.assertEqual('application/octet-stream',
response.headers['Content-Type']) response.headers['Content-Type'])
def test_download_forbidden(self): def test_download_forbidden(self):
"""Make sure the serializer can return 403 forbidden error instead of """Make sure the serializer can return 403 forbidden error instead of
500 internal server error. 500 internal server error.
""" """
def get_data(*args, **kwargs): def get_data(*args, **kwargs):
raise exception.Forbidden() raise exception.Forbidden()
self.stubs.Set(glance.api.policy.ImageProxy, self.mock_object(glance.api.policy.ImageProxy,
'get_data', 'get_data',
get_data) get_data)
request = wsgi.Request.blank('/') request = wsgi.Request.blank('/')
request.environ = {} request.environ = {}
response = webob.Response() response = webob.Response()
response.request = request response.request = request
image = FakeImage(size=3, data=iter('ZZZ')) image = FakeImage(size=3, data=iter('ZZZ'))
image.get_data = get_data image.get_data = get_data
self.assertRaises(webob.exc.HTTPForbidden, self.assertRaises(webob.exc.HTTPForbidden,
self.serializer.download, self.serializer.download,
response, image) response, image)
skipping to change at line 991 skipping to change at line 995
self.assertEqual('0', response.headers['Content-Length']) self.assertEqual('0', response.headers['Content-Length'])
def test_stage(self): def test_stage(self):
request = webob.Request.blank('/') request = webob.Request.blank('/')
request.environ = {} request.environ = {}
response = webob.Response() response = webob.Response()
response.request = request response.request = request
self.serializer.stage(response, {}) self.serializer.stage(response, {})
self.assertEqual(http.NO_CONTENT, response.status_int) self.assertEqual(http.NO_CONTENT, response.status_int)
self.assertEqual('0', response.headers['Content-Length']) self.assertEqual('0', response.headers['Content-Length'])
class TestMultiBackendImagesController(base.MultiStoreClearingUnitTest):
def setUp(self):
super(TestMultiBackendImagesController, self).setUp()
self.config(debug=True)
self.image_repo = FakeImageRepo()
db = unit_test_utils.FakeDB()
policy = unit_test_utils.FakePolicyEnforcer()
notifier = unit_test_utils.FakeNotifier()
store = unit_test_utils.FakeStoreAPI()
self.controller = glance.api.v2.image_data.ImageDataController()
self.controller.gateway = FakeGateway(db, store, notifier, policy,
self.image_repo)
def test_upload(self):
request = unit_test_utils.get_fake_request()
image = FakeImage('abcd')
self.image_repo.result = image
self.controller.upload(request, unit_test_utils.UUID2, 'YYYY', 4)
self.assertEqual('YYYY', image.data)
self.assertEqual(4, image.size)
def test_upload_invalid_backend_in_request_header(self):
request = unit_test_utils.get_fake_request()
request.headers['x-image-meta-store'] = 'dummy'
image = FakeImage('abcd')
self.image_repo.result = image
self.assertRaises(webob.exc.HTTPBadRequest, self.controller.upload,
request, unit_test_utils.UUID2, 'YYYY', 4)
 End of changes. 13 change blocks. 
9 lines changed or deleted 13 lines changed or added

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