Implement v2 API access resource
* Add functional tests * Implements bp api-v2-image-access Change-Id: If62f4c0c9b387bb1f99306d68b18ad21f5f875d1
This commit is contained in:
parent
f9afe586b6
commit
28d85923fa
@ -16,28 +16,64 @@
|
||||
import json
|
||||
|
||||
import jsonschema
|
||||
import webob.exc
|
||||
|
||||
from glance.api.v2 import base
|
||||
from glance.api.v2 import schemas
|
||||
from glance.common import exception
|
||||
from glance.common import wsgi
|
||||
import glance.registry.db.api
|
||||
|
||||
|
||||
class ImageAccessController(base.Controller):
|
||||
class Controller(base.Controller):
|
||||
def __init__(self, conf, db=None):
|
||||
super(ImageAccessController, self).__init__(conf)
|
||||
super(Controller, self).__init__(conf)
|
||||
self.db_api = db or glance.registry.db.api
|
||||
self.db_api.configure_db(conf)
|
||||
|
||||
def index(self, req, image_id):
|
||||
image = self.db_api.image_get(req.context, image_id)
|
||||
return image['members']
|
||||
#TODO(bcwaldon): We have to filter on non-deleted members
|
||||
# manually. This should be done for us in the db api
|
||||
return filter(lambda m: not m['deleted'], image['members'])
|
||||
|
||||
def show(self, req, image_id, tenant_id):
|
||||
return self.db_api.image_member_find(req.context, image_id, tenant_id)
|
||||
try:
|
||||
return self.db_api.image_member_find(req.context,
|
||||
image_id, tenant_id)
|
||||
except exception.NotFound:
|
||||
raise webob.exc.HTTPNotFound()
|
||||
|
||||
def create(self, req, access):
|
||||
return self.db_api.image_member_create(req.context, access)
|
||||
def create(self, req, image_id, access_record):
|
||||
#TODO(bcwaldon): Refactor these methods so we don't need to
|
||||
# explicitly retrieve a session object here
|
||||
session = self.db_api.get_session()
|
||||
try:
|
||||
image = self.db_api.image_get(req.context, image_id,
|
||||
session=session)
|
||||
except exception.NotFound:
|
||||
raise webob.exc.HTTPNotFound()
|
||||
except exception.Forbidden:
|
||||
# If it's private and doesn't belong to them, don't let on
|
||||
# that it exists
|
||||
raise webob.exc.HTTPNotFound()
|
||||
|
||||
# Image is visible, but authenticated user still may not be able to
|
||||
# share it
|
||||
if not req.context.is_image_sharable(image):
|
||||
msg = _("No permission to share that image")
|
||||
raise webob.exc.HTTPForbidden(msg)
|
||||
|
||||
access_record['image_id'] = image_id
|
||||
return self.db_api.image_member_create(req.context, access_record)
|
||||
|
||||
def delete(self, req, image_id, tenant_id):
|
||||
#TODO(bcwaldon): Refactor these methods so we don't need to explicitly
|
||||
# retrieve a session object here
|
||||
session = self.db_api.get_session()
|
||||
member = self.db_api.image_member_find(req.context, image_id,
|
||||
tenant_id, session=session)
|
||||
self.db_api.image_member_delete(req.context, member, session=session)
|
||||
|
||||
|
||||
class RequestDeserializer(wsgi.JSONRequestDeserializer):
|
||||
@ -54,35 +90,37 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer):
|
||||
body = output.pop('body')
|
||||
self._validate(request, body)
|
||||
body['member'] = body.pop('tenant_id')
|
||||
output['access'] = body
|
||||
output['access_record'] = body
|
||||
return output
|
||||
|
||||
|
||||
class ResponseSerializer(wsgi.JSONResponseSerializer):
|
||||
def _get_access_href(self, image_member):
|
||||
image_id = image_member['image_id']
|
||||
tenant_id = image_member['member']
|
||||
return '/v2/images/%s/access/%s' % (image_id, tenant_id)
|
||||
def _get_access_href(self, image_id, tenant_id=None):
|
||||
link = '/v2/images/%s/access' % image_id
|
||||
if tenant_id:
|
||||
link = '%s/%s' % (link, tenant_id)
|
||||
return link
|
||||
|
||||
def _get_access_links(self, access):
|
||||
self_link = self._get_access_href(access['image_id'], access['member'])
|
||||
return [
|
||||
{'rel': 'self', 'href': self._get_access_href(access)},
|
||||
{'rel': 'self', 'href': self_link},
|
||||
{'rel': 'describedby', 'href': '/v2/schemas/image/access'},
|
||||
]
|
||||
|
||||
def _format_access(self, access):
|
||||
return {
|
||||
'image_id': access['image_id'],
|
||||
'tenant_id': access['member'],
|
||||
'can_share': access['can_share'],
|
||||
'links': self._get_access_links(access),
|
||||
}
|
||||
|
||||
def _get_container_links(self, image_id):
|
||||
return [{'rel': 'self', 'href': '/v2/images/%s/access' % image_id}]
|
||||
return [{'rel': 'self', 'href': self._get_access_href(image_id)}]
|
||||
|
||||
def show(self, response, access):
|
||||
response.body = json.dumps({'access': self._format_access(access)})
|
||||
record = {'access_record': self._format_access(access)}
|
||||
response.body = json.dumps(record)
|
||||
|
||||
def index(self, response, access_records):
|
||||
body = {
|
||||
@ -92,5 +130,19 @@ class ResponseSerializer(wsgi.JSONResponseSerializer):
|
||||
response.body = json.dumps(body)
|
||||
|
||||
def create(self, response, access):
|
||||
response.status_int = 201
|
||||
response.content_type = 'application/json'
|
||||
response.location = self._get_access_href(access['image_id'],
|
||||
access['member'])
|
||||
response.body = json.dumps({'access': self._format_access(access)})
|
||||
response.location = self._get_access_href(access)
|
||||
|
||||
def delete(self, response, result):
|
||||
response.status_int = 204
|
||||
|
||||
|
||||
def create_resource(conf):
|
||||
"""Image access resource factory method"""
|
||||
deserializer = RequestDeserializer(conf)
|
||||
serializer = ResponseSerializer()
|
||||
controller = Controller(conf)
|
||||
return wsgi.Resource(controller, deserializer, serializer)
|
||||
|
@ -19,6 +19,7 @@ import logging
|
||||
|
||||
import routes
|
||||
|
||||
from glance.api.v2 import image_access
|
||||
from glance.api.v2 import image_data
|
||||
from glance.api.v2 import image_tags
|
||||
from glance.api.v2 import images
|
||||
@ -100,4 +101,22 @@ class API(wsgi.Router):
|
||||
action='delete',
|
||||
conditions={'method': ['DELETE']})
|
||||
|
||||
image_access_resource = image_access.create_resource(conf)
|
||||
mapper.connect('/images/{image_id}/access',
|
||||
controller=image_access_resource,
|
||||
action='index',
|
||||
conditions={'method': ['GET']})
|
||||
mapper.connect('/images/{image_id}/access',
|
||||
controller=image_access_resource,
|
||||
action='create',
|
||||
conditions={'method': ['POST']})
|
||||
mapper.connect('/images/{image_id}/access/{tenant_id}',
|
||||
controller=image_access_resource,
|
||||
action='show',
|
||||
conditions={'method': ['GET']})
|
||||
mapper.connect('/images/{image_id}/access/{tenant_id}',
|
||||
controller=image_access_resource,
|
||||
action='delete',
|
||||
conditions={'method': ['DELETE']})
|
||||
|
||||
super(API, self).__init__(mapper)
|
||||
|
@ -38,12 +38,6 @@ IMAGE_SCHEMA = {
|
||||
ACCESS_SCHEMA = {
|
||||
'name': 'access',
|
||||
'properties': {
|
||||
"image_id": {
|
||||
"type": "string",
|
||||
"description": "The image identifier",
|
||||
"required": True,
|
||||
"maxLength": 36,
|
||||
},
|
||||
"tenant_id": {
|
||||
"type": "string",
|
||||
"description": "The tenant identifier",
|
||||
|
@ -60,21 +60,21 @@ class RequestContext(object):
|
||||
return True
|
||||
|
||||
# No owner == image visible
|
||||
if image.owner is None:
|
||||
if image['owner'] is None:
|
||||
return True
|
||||
|
||||
# Image is_public == image visible
|
||||
if image.is_public:
|
||||
if image['is_public']:
|
||||
return True
|
||||
|
||||
# Perform tests based on whether we have an owner
|
||||
if self.owner is not None:
|
||||
if self.owner == image.owner:
|
||||
if self.owner == image['owner']:
|
||||
return True
|
||||
|
||||
# Figure out if this image is shared with that tenant
|
||||
try:
|
||||
tmp = db_api.image_member_find(self, image.id, self.owner)
|
||||
tmp = db_api.image_member_find(self, image['id'], self.owner)
|
||||
return not tmp['deleted']
|
||||
except exception.NotFound:
|
||||
pass
|
||||
@ -89,11 +89,11 @@ class RequestContext(object):
|
||||
return True
|
||||
|
||||
# No owner == image not mutable
|
||||
if image.owner is None or self.owner is None:
|
||||
if image['owner'] is None or self.owner is None:
|
||||
return False
|
||||
|
||||
# Image only mutable by its owner
|
||||
return image.owner == self.owner
|
||||
return image['owner'] == self.owner
|
||||
|
||||
def is_image_sharable(self, image, **kwargs):
|
||||
"""Return True if the image can be shared to others in this context."""
|
||||
@ -106,7 +106,7 @@ class RequestContext(object):
|
||||
return True
|
||||
|
||||
# If we own the image, we can share it
|
||||
if self.owner == image.owner:
|
||||
if self.owner == image['owner']:
|
||||
return True
|
||||
|
||||
# Let's get the membership association
|
||||
@ -117,14 +117,14 @@ class RequestContext(object):
|
||||
return False
|
||||
else:
|
||||
try:
|
||||
membership = db_api.image_member_find(self, image.id,
|
||||
membership = db_api.image_member_find(self, image['id'],
|
||||
self.owner)
|
||||
except exception.NotFound:
|
||||
# Not shared with us anyway
|
||||
return False
|
||||
|
||||
# It's the can_share attribute we're now interested in
|
||||
return membership.can_share
|
||||
return membership['can_share']
|
||||
|
||||
|
||||
class ContextMiddleware(wsgi.Middleware):
|
||||
|
145
glance/tests/functional/v2/test_image_access.py
Normal file
145
glance/tests/functional/v2/test_image_access.py
Normal file
@ -0,0 +1,145 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2012 OpenStack, LLC
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import json
|
||||
|
||||
import requests
|
||||
|
||||
from glance.tests import functional
|
||||
from glance.common import utils
|
||||
|
||||
|
||||
TENANT1 = utils.generate_uuid()
|
||||
TENANT2 = utils.generate_uuid()
|
||||
TENANT3 = utils.generate_uuid()
|
||||
TENANT4 = utils.generate_uuid()
|
||||
|
||||
|
||||
class TestImageAccess(functional.FunctionalTest):
|
||||
def setUp(self):
|
||||
super(TestImageAccess, self).setUp()
|
||||
self.cleanup()
|
||||
self.api_server.deployment_flavor = 'noauth'
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
|
||||
# Create an image for our tests
|
||||
path = 'http://0.0.0.0:%d/v2/images' % self.api_port
|
||||
headers = self._headers({'Content-Type': 'application/json'})
|
||||
data = json.dumps({'name': 'image-1'})
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(200, response.status_code)
|
||||
self.image_url = response.headers['Location']
|
||||
|
||||
def _url(self, path):
|
||||
return '%s%s' % (self.image_url, path)
|
||||
|
||||
def _headers(self, custom_headers=None):
|
||||
base_headers = {
|
||||
'X-Identity-Status': 'Confirmed',
|
||||
'X-Auth-Token': '932c5c84-02ac-4fe5-a9ba-620af0e2bb96',
|
||||
'X-User-Id': 'f9a41d13-0c13-47e9-bee2-ce4e8bfe958e',
|
||||
'X-Tenant-Id': TENANT1,
|
||||
'X-Roles': 'member',
|
||||
}
|
||||
base_headers.update(custom_headers or {})
|
||||
return base_headers
|
||||
|
||||
@functional.runs_sql
|
||||
def test_image_access_lifecycle(self):
|
||||
# Image acccess list should be empty
|
||||
path = self._url('/access')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
self.assertEqual(200, response.status_code)
|
||||
access_records = json.loads(response.text)['access_records']
|
||||
self.assertEqual(0, len(access_records))
|
||||
|
||||
# Other tenants shouldn't be able to share by default, and shouldn't
|
||||
# even know the image exists
|
||||
path = self._url('/access')
|
||||
data = json.dumps({'tenant_id': TENANT3, 'can_share': False})
|
||||
request_headers = {
|
||||
'Content-Type': 'application/json',
|
||||
'X-Tenant-Id': TENANT2,
|
||||
}
|
||||
headers = self._headers(request_headers)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(404, response.status_code)
|
||||
|
||||
# Share the image with another tenant
|
||||
path = self._url('/access')
|
||||
data = json.dumps({'tenant_id': TENANT2, 'can_share': True})
|
||||
headers = self._headers({'Content-Type': 'application/json'})
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(201, response.status_code)
|
||||
access_location = response.headers['Location']
|
||||
|
||||
# Ensure the access record was actually created
|
||||
response = requests.get(access_location, headers=self._headers())
|
||||
self.assertEqual(200, response.status_code)
|
||||
|
||||
# Make sure the sharee can further share the image
|
||||
path = self._url('/access')
|
||||
data = json.dumps({'tenant_id': TENANT3, 'can_share': False})
|
||||
request_headers = {
|
||||
'Content-Type': 'application/json',
|
||||
'X-Tenant-Id': TENANT2,
|
||||
}
|
||||
headers = self._headers(request_headers)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(201, response.status_code)
|
||||
access_location = response.headers['Location']
|
||||
|
||||
# Ensure the access record was actually created
|
||||
response = requests.get(access_location, headers=self._headers())
|
||||
self.assertEqual(200, response.status_code)
|
||||
|
||||
# The third tenant should not be able to share it further
|
||||
path = self._url('/access')
|
||||
data = json.dumps({'tenant_id': TENANT4, 'can_share': False})
|
||||
request_headers = {
|
||||
'Content-Type': 'application/json',
|
||||
'X-Tenant-Id': TENANT3,
|
||||
}
|
||||
headers = self._headers(request_headers)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(403, response.status_code)
|
||||
|
||||
# Image acccess list should now contain 2 entries
|
||||
path = self._url('/access')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
self.assertEqual(200, response.status_code)
|
||||
access_records = json.loads(response.text)['access_records']
|
||||
self.assertEqual(2, len(access_records))
|
||||
print [a['tenant_id'] for a in access_records]
|
||||
|
||||
# Delete an access record
|
||||
response = requests.delete(access_location, headers=self._headers())
|
||||
self.assertEqual(204, response.status_code)
|
||||
|
||||
# Ensure the access record was actually deleted
|
||||
response = requests.get(access_location, headers=self._headers())
|
||||
self.assertEqual(404, response.status_code)
|
||||
|
||||
# Image acccess list should now contain 1 entry
|
||||
path = self._url('/access')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
self.assertEqual(200, response.status_code)
|
||||
access_records = json.loads(response.text)['access_records']
|
||||
print [a['tenant_id'] for a in access_records]
|
||||
self.assertEqual(1, len(access_records))
|
||||
|
||||
self.stop_servers()
|
@ -20,26 +20,16 @@ import unittest
|
||||
from glance.common import context
|
||||
|
||||
|
||||
class FakeImage(object):
|
||||
"""
|
||||
Fake image for providing the image attributes needed for
|
||||
TestContext.
|
||||
"""
|
||||
|
||||
def __init__(self, owner, is_public):
|
||||
self.id = None
|
||||
self.owner = owner
|
||||
self.is_public = is_public
|
||||
def _fake_image(owner, is_public):
|
||||
return {
|
||||
'id': None,
|
||||
'owner': owner,
|
||||
'is_public': is_public,
|
||||
}
|
||||
|
||||
|
||||
class FakeMembership(object):
|
||||
"""
|
||||
Fake membership for providing the membership attributes needed for
|
||||
TestContext.
|
||||
"""
|
||||
|
||||
def __init__(self, can_share=False):
|
||||
self.can_share = can_share
|
||||
def _fake_membership(can_share=False):
|
||||
return {'can_share': can_share}
|
||||
|
||||
|
||||
class TestContext(unittest.TestCase):
|
||||
@ -52,7 +42,7 @@ class TestContext(unittest.TestCase):
|
||||
context.
|
||||
"""
|
||||
|
||||
img = FakeImage(img_owner, img_public)
|
||||
img = _fake_image(img_owner, img_public)
|
||||
ctx = context.RequestContext(**kwargs)
|
||||
|
||||
self.assertEqual(ctx.is_image_visible(img), exp_res)
|
||||
@ -68,7 +58,7 @@ class TestContext(unittest.TestCase):
|
||||
is_image_sharable().
|
||||
"""
|
||||
|
||||
img = FakeImage(img_owner, True)
|
||||
img = _fake_image(img_owner, True)
|
||||
ctx = context.RequestContext(**kwargs)
|
||||
|
||||
sharable_args = {}
|
||||
@ -111,7 +101,7 @@ class TestContext(unittest.TestCase):
|
||||
not share an image, with or without membership.
|
||||
"""
|
||||
self.do_sharable(False, 'pattieblack', None, is_admin=True)
|
||||
self.do_sharable(False, 'pattieblack', FakeMembership(True),
|
||||
self.do_sharable(False, 'pattieblack', _fake_membership(True),
|
||||
is_admin=True)
|
||||
|
||||
def test_anon_public(self):
|
||||
@ -148,7 +138,7 @@ class TestContext(unittest.TestCase):
|
||||
not share an image, with or without membership.
|
||||
"""
|
||||
self.do_sharable(False, 'pattieblack', None)
|
||||
self.do_sharable(False, 'pattieblack', FakeMembership(True))
|
||||
self.do_sharable(False, 'pattieblack', _fake_membership(True))
|
||||
|
||||
def test_auth_public(self):
|
||||
"""
|
||||
@ -227,7 +217,7 @@ class TestContext(unittest.TestCase):
|
||||
False) cannot share an image it does not own even if it is
|
||||
shared with it, but with can_share = False.
|
||||
"""
|
||||
self.do_sharable(False, 'pattieblack', FakeMembership(False),
|
||||
self.do_sharable(False, 'pattieblack', _fake_membership(False),
|
||||
tenant='froggy')
|
||||
|
||||
def test_auth_sharable_can_share(self):
|
||||
@ -236,5 +226,5 @@ class TestContext(unittest.TestCase):
|
||||
False) can share an image it does not own if it is shared with
|
||||
it with can_share = True.
|
||||
"""
|
||||
self.do_sharable(True, 'pattieblack', FakeMembership(True),
|
||||
self.do_sharable(True, 'pattieblack', _fake_membership(True),
|
||||
tenant='froggy')
|
||||
|
@ -78,11 +78,15 @@ class FakeDB(object):
|
||||
def configure_db(*args, **kwargs):
|
||||
pass
|
||||
|
||||
def get_session(self):
|
||||
pass
|
||||
|
||||
def _image_member_format(self, image_id, tenant_id, can_share):
|
||||
return {
|
||||
'image_id': image_id,
|
||||
'member': tenant_id,
|
||||
'can_share': can_share,
|
||||
'deleted': False,
|
||||
}
|
||||
|
||||
def _image_format(self, image_id, **values):
|
||||
@ -96,7 +100,7 @@ class FakeDB(object):
|
||||
image.update(values)
|
||||
return image
|
||||
|
||||
def image_get(self, context, image_id):
|
||||
def image_get(self, context, image_id, session=None):
|
||||
try:
|
||||
image = self.images[image_id]
|
||||
LOG.info('Found image %s: %s' % (image_id, str(image)))
|
||||
|
@ -16,10 +16,9 @@
|
||||
import json
|
||||
import unittest
|
||||
|
||||
import jsonschema
|
||||
import webob
|
||||
|
||||
import glance.api.v2.image_access
|
||||
from glance.api.v2 import image_access
|
||||
from glance.common import exception
|
||||
from glance.common import utils
|
||||
import glance.tests.unit.utils as test_utils
|
||||
@ -30,8 +29,7 @@ class TestImageAccessController(unittest.TestCase):
|
||||
def setUp(self):
|
||||
super(TestImageAccessController, self).setUp()
|
||||
self.db = test_utils.FakeDB()
|
||||
self.controller = \
|
||||
glance.api.v2.image_access.ImageAccessController({}, self.db)
|
||||
self.controller = image_access.Controller({}, self.db)
|
||||
|
||||
def test_index(self):
|
||||
req = test_utils.FakeRequest()
|
||||
@ -41,11 +39,13 @@ class TestImageAccessController(unittest.TestCase):
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT1,
|
||||
'can_share': True,
|
||||
'deleted': False,
|
||||
},
|
||||
{
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT2,
|
||||
'can_share': False,
|
||||
'deleted': False,
|
||||
},
|
||||
]
|
||||
self.assertEqual(expected, output)
|
||||
@ -71,6 +71,7 @@ class TestImageAccessController(unittest.TestCase):
|
||||
'image_id': image_id,
|
||||
'member': tenant_id,
|
||||
'can_share': True,
|
||||
'deleted': False,
|
||||
}
|
||||
self.assertEqual(expected, output)
|
||||
|
||||
@ -78,72 +79,67 @@ class TestImageAccessController(unittest.TestCase):
|
||||
req = test_utils.FakeRequest()
|
||||
image_id = utils.generate_uuid()
|
||||
tenant_id = test_utils.TENANT1
|
||||
self.assertRaises(exception.NotFound,
|
||||
self.assertRaises(webob.exc.HTTPNotFound,
|
||||
self.controller.show, req, image_id, tenant_id)
|
||||
|
||||
def test_show_nonexistant_tenant(self):
|
||||
req = test_utils.FakeRequest()
|
||||
image_id = test_utils.UUID1
|
||||
tenant_id = utils.generate_uuid()
|
||||
self.assertRaises(exception.NotFound,
|
||||
self.assertRaises(webob.exc.HTTPNotFound,
|
||||
self.controller.show, req, image_id, tenant_id)
|
||||
|
||||
def test_create(self):
|
||||
member = utils.generate_uuid()
|
||||
fixture = {
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': utils.generate_uuid(),
|
||||
'member': member,
|
||||
'can_share': True,
|
||||
}
|
||||
expected = {
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': member,
|
||||
'can_share': True,
|
||||
'deleted': False,
|
||||
}
|
||||
req = test_utils.FakeRequest()
|
||||
output = self.controller.create(req, fixture)
|
||||
self.assertEqual(fixture, output)
|
||||
output = self.controller.create(req, test_utils.UUID1, fixture)
|
||||
self.assertEqual(expected, output)
|
||||
|
||||
|
||||
class TestImageAccessDeserializer(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.deserializer = glance.api.v2.image_access.RequestDeserializer({})
|
||||
self.deserializer = image_access.RequestDeserializer({})
|
||||
|
||||
def test_create(self):
|
||||
fixture = {
|
||||
'image_id': test_utils.UUID1,
|
||||
'tenant_id': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
}
|
||||
expected = {
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
'access_record': {
|
||||
'member': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
},
|
||||
}
|
||||
request = test_utils.FakeRequest()
|
||||
request.body = json.dumps(fixture)
|
||||
output = self.deserializer.create(request)
|
||||
self.assertEqual(output, {'access': expected})
|
||||
|
||||
def _test_create_fails(self, fixture):
|
||||
request = test_utils.FakeRequest()
|
||||
request.body = json.dumps(fixture)
|
||||
self.assertRaises(jsonschema.ValidationError,
|
||||
self.deserializer.create, request)
|
||||
|
||||
def test_create_no_image(self):
|
||||
fixture = {'tenant_id': test_utils.TENANT1, 'can_share': True}
|
||||
self._test_create_fails(fixture)
|
||||
self.assertEqual(expected, output)
|
||||
|
||||
|
||||
class TestImageAccessSerializer(unittest.TestCase):
|
||||
serializer = glance.api.v2.image_access.ResponseSerializer()
|
||||
serializer = image_access.ResponseSerializer()
|
||||
|
||||
def test_show(self):
|
||||
fixture = {
|
||||
'member': test_utils.TENANT1,
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
}
|
||||
self_href = ('/v2/images/%s/access/%s' %
|
||||
(test_utils.UUID1, test_utils.TENANT1))
|
||||
expected = {
|
||||
'access': {
|
||||
'image_id': test_utils.UUID1,
|
||||
'access_record': {
|
||||
'tenant_id': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
'links': [
|
||||
@ -159,20 +155,19 @@ class TestImageAccessSerializer(unittest.TestCase):
|
||||
def test_index(self):
|
||||
fixtures = [
|
||||
{
|
||||
'member': test_utils.TENANT1,
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
},
|
||||
{
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT2,
|
||||
'image_id': test_utils.UUID2,
|
||||
'can_share': True,
|
||||
},
|
||||
]
|
||||
expected = {
|
||||
'access_records': [
|
||||
{
|
||||
'image_id': test_utils.UUID1,
|
||||
'tenant_id': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
'links': [
|
||||
@ -188,14 +183,13 @@ class TestImageAccessSerializer(unittest.TestCase):
|
||||
],
|
||||
},
|
||||
{
|
||||
'image_id': test_utils.UUID2,
|
||||
'tenant_id': test_utils.TENANT2,
|
||||
'can_share': True,
|
||||
'links': [
|
||||
{
|
||||
'rel': 'self',
|
||||
'href': ('/v2/images/%s/access/%s' %
|
||||
(test_utils.UUID2, test_utils.TENANT2))
|
||||
(test_utils.UUID1, test_utils.TENANT2))
|
||||
},
|
||||
{
|
||||
'rel': 'describedby',
|
||||
@ -212,15 +206,14 @@ class TestImageAccessSerializer(unittest.TestCase):
|
||||
|
||||
def test_create(self):
|
||||
fixture = {
|
||||
'member': test_utils.TENANT1,
|
||||
'image_id': test_utils.UUID1,
|
||||
'member': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
}
|
||||
self_href = ('/v2/images/%s/access/%s' %
|
||||
(test_utils.UUID1, test_utils.TENANT1))
|
||||
expected = {
|
||||
'access': {
|
||||
'image_id': test_utils.UUID1,
|
||||
'tenant_id': test_utils.TENANT1,
|
||||
'can_share': False,
|
||||
'links': [
|
||||
|
Loading…
Reference in New Issue
Block a user