Drop unfinshed parts of v2 API

The v2 API will be implemented over time as the spec develops. We
shouldn't implement anything that we aren't 100% behind. This drops
all of the things that we aren't in love with:
* Drop access records and related schemas
* Drop owner attribute from image

Related to bp api-2

Change-Id: Ieef2c141282e7018d56e79aee8f20af0542af25b
This commit is contained in:
Brian Waldon 2012-08-08 13:49:40 -07:00
parent ae6e288072
commit 99d5173afe
12 changed files with 10 additions and 751 deletions

View File

@ -1 +0,0 @@
{}

View File

@ -1,211 +0,0 @@
# 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 webob.exc
import glance.api.v2 as v2
from glance.common import exception
from glance.common import utils
from glance.common import wsgi
import glance.db
import glance.schema
class Controller(object):
def __init__(self, db=None):
self.db_api = db or glance.db.get_api()
self.db_api.configure_db()
def index(self, req, image_id):
#NOTE(bcwaldon): call image_get to ensure user has permission
self.db_api.image_get(req.context, image_id)
members = self.db_api.image_member_find(req.context, image_id=image_id)
#TODO(bcwaldon): We have to filter on non-deleted members
# manually. This should be done for us in the db api
return {
'access_records': filter(lambda m: not m['deleted'], members),
'image_id': image_id,
}
def show(self, req, image_id, tenant_id):
members = self.db_api.image_member_find(req.context,
image_id=image_id,
member=tenant_id)
try:
return members[0]
except IndexError:
raise webob.exc.HTTPNotFound()
@utils.mutating
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 self.db_api.is_image_sharable(req.context, image):
msg = _("No permission to share that image")
raise webob.exc.HTTPForbidden(msg)
access_record['image_id'] = image_id
member = self.db_api.image_member_create(req.context, access_record)
v2.update_image_read_acl(req, self.db_api, image)
return member
@utils.mutating
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()
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()
members = self.db_api.image_member_find(req.context,
image_id=image_id,
member=tenant_id,
session=session)
try:
member = members[0]
except IndexError:
raise webob.exc.HTTPNotFound()
self.db_api.image_member_delete(req.context, member, session=session)
v2.update_image_read_acl(req, self.db_api, image)
class RequestDeserializer(wsgi.JSONRequestDeserializer):
def __init__(self):
super(RequestDeserializer, self).__init__()
self.schema = get_schema()
def create(self, request):
output = super(RequestDeserializer, self).default(request)
body = output.pop('body')
self.schema.validate(body)
body['member'] = body.pop('tenant_id')
output['access_record'] = body
return output
class ResponseSerializer(wsgi.JSONResponseSerializer):
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 _format_access(self, access):
self_link = self._get_access_href(access['image_id'], access['member'])
return {
'tenant_id': access['member'],
'can_share': access['can_share'],
'self': self_link,
'schema': '/v2/schemas/image/access',
'image': '/v2/images/%s' % access['image_id'],
}
def show(self, response, access):
response.body = json.dumps(self._format_access(access))
response.content_type = 'application/json'
def index(self, response, result):
access_records = result['access_records']
first_link = '/v2/images/%s/access' % result['image_id']
body = {
'access_records': [self._format_access(a)
for a in access_records],
'first': first_link,
'schema': '/v2/schemas/image/accesses',
}
response.body = json.dumps(body)
response.content_type = 'application/json'
def create(self, response, access):
response.status_int = 201
response.location = self._get_access_href(access['image_id'],
access['member'])
response.body = json.dumps(self._format_access(access))
response.content_type = 'application/json'
def delete(self, response, result):
response.status_int = 204
def get_schema():
properties = {
'tenant_id': {
'type': 'string',
'description': 'The tenant identifier',
},
'can_share': {
'type': 'boolean',
'description': 'Ability of tenant to share with others',
'default': False,
},
'self': {
'type': 'string',
'description': 'A link to this resource',
},
'schema': {
'type': 'string',
'description': 'A link to the schema describing this resource',
},
'image': {
'type': 'string',
'description': 'A link to the image related to this resource',
},
}
links = [
{'rel': 'self', 'href': '{self}'},
{'rel': 'up', 'href': '{image}'},
{'rel': 'describedby', 'href': '{schema}'},
]
return glance.schema.Schema('access', properties, links)
def get_collection_schema():
access_schema = get_schema()
return glance.schema.CollectionSchema('accesses', access_schema)
def create_resource():
"""Image access resource factory method"""
deserializer = RequestDeserializer()
serializer = ResponseSerializer()
controller = Controller()
return wsgi.Resource(controller, deserializer, serializer)

View File

@ -66,7 +66,7 @@ class ImagesController(object):
@utils.mutating
def create(self, req, image):
image.setdefault('owner', req.context.owner)
image['owner'] = req.context.owner
image['status'] = 'queued'
tags = self._extract_tags(image)
@ -166,7 +166,7 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer):
# defined properties contained in a 'properties' dictionary
image = {'properties': body}
for key in ['id', 'name', 'visibility', 'created_at', 'updated_at',
'tags', 'owner', 'status']:
'tags', 'status']:
try:
image[key] = image['properties'].pop(key)
except KeyError:
@ -176,7 +176,6 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer):
image['is_public'] = image.pop('visibility') == 'public'
self._check_readonly(image)
self._check_adminonly(image, request.context)
return {'image': image}
@staticmethod
@ -186,13 +185,6 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer):
msg = "Attribute \'%s\' is read-only." % key
raise webob.exc.HTTPForbidden(explanation=unicode(msg))
@staticmethod
def _check_adminonly(image, context):
for key in ['owner']:
if key in image and not context.is_admin:
msg = "Must be admin to set attribute \'%s\'." % key
raise webob.exc.HTTPForbidden(explanation=unicode(msg))
def create(self, request):
return self._parse_image(request)
@ -271,7 +263,7 @@ class ResponseSerializer(wsgi.JSONResponseSerializer):
def _format_image(self, image):
_image = image['properties']
for key in ['id', 'name', 'created_at', 'updated_at', 'tags', 'size',
'owner', 'checksum', 'status']:
'checksum', 'status']:
_image[key] = image[key]
if CONF.show_image_direct_url and image['location']:
_image['direct_url'] = image['location']
@ -279,7 +271,6 @@ class ResponseSerializer(wsgi.JSONResponseSerializer):
_image = self.schema.filter(_image)
_image['self'] = self._get_image_href(image)
_image['file'] = self._get_image_href(image, 'file')
_image['access'] = self._get_image_href(image, 'access')
_image['schema'] = '/v2/schemas/image'
self._serialize_datetimes(_image)
return _image
@ -342,11 +333,6 @@ _BASE_PROPERTIES = {
'enum': ['queued', 'saving', 'active', 'killed',
'deleted', 'pending_delete'],
},
'owner': {
'type': 'string',
'description': 'Tenant who can modify the image',
'maxLength': 36,
},
'visibility': {
'type': 'string',
'description': 'Scope of image accessibility',
@ -387,14 +373,12 @@ _BASE_PROPERTIES = {
'description': 'URL to access the image file kept in external store',
},
'self': {'type': 'string'},
'access': {'type': 'string'},
'file': {'type': 'string'},
'schema': {'type': 'string'},
}
_BASE_LINKS = [
{'rel': 'self', 'href': '{self}'},
{'rel': 'related', 'href': '{access}'},
{'rel': 'enclosure', 'href': '{file}'},
{'rel': 'describedby', 'href': '{schema}'},
]

View File

@ -15,7 +15,6 @@
# License for the specific language governing permissions and limitations
# under the License.
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
@ -39,14 +38,6 @@ class API(wsgi.Router):
controller=schemas_resource,
action='images',
conditions={'method': ['GET']})
mapper.connect('/schemas/image/access',
controller=schemas_resource,
action='access',
conditions={'method': ['GET']})
mapper.connect('/schemas/image/accesses',
controller=schemas_resource,
action='accesses',
conditions={'method': ['GET']})
images_resource = images.create_resource(custom_image_properties)
mapper.connect('/images',
@ -90,22 +81,4 @@ class API(wsgi.Router):
action='delete',
conditions={'method': ['DELETE']})
image_access_resource = image_access.create_resource()
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)

View File

@ -13,15 +13,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from glance.api.v2 import image_access
from glance.api.v2 import images
from glance.common import wsgi
class Controller(object):
def __init__(self, custom_image_properties=None):
self.access_schema = image_access.get_schema()
self.access_collection_schema = image_access.get_collection_schema()
self.image_schema = images.get_schema(custom_image_properties)
self.image_collection_schema = images.get_collection_schema(
custom_image_properties)
@ -32,12 +29,6 @@ class Controller(object):
def images(self, req):
return self.image_collection_schema.raw()
def access(self, req):
return self.access_schema.raw()
def accesses(self, req):
return self.access_collection_schema.raw()
def create_resource(custom_image_properties=None):
controller = Controller(custom_image_properties)

View File

@ -1 +0,0 @@
{}

View File

@ -421,7 +421,6 @@ class FunctionalTest(test_utils.BaseTestCase):
conf_dir = os.path.join(self.test_dir, 'etc')
utils.safe_mkdirs(conf_dir)
self.copy_data_file('schema-image.json', conf_dir)
self.copy_data_file('schema-access.json', conf_dir)
self.copy_data_file('policy.json', conf_dir)
self.policy_file = os.path.join(conf_dir, 'policy.json')

View File

@ -91,7 +91,6 @@ class TestImages(functional.FunctionalTest):
self.assertTrue(image['created_at'])
self.assertTrue(image['updated_at'])
self.assertEqual(image['updated_at'], image['created_at'])
self.assertEqual(image['owner'], TENANT1)
# The image should be mutable, including adding new properties
path = self._url('/v2/images/%s' % image_id)
@ -234,50 +233,6 @@ class TestImages(functional.FunctionalTest):
response = requests.delete(path, headers=headers)
self.assertEqual(404, response.status_code)
# Share the image with TENANT2
path = self._url('/v2/images/%s/access' % image_id)
data = json.dumps({'tenant_id': TENANT2, 'can_share': False})
request_headers = {'Content-Type': 'application/json'}
headers = self._headers(request_headers)
response = requests.post(path, headers=headers, data=data)
self.assertEqual(201, response.status_code)
# TENANT2 should see the image in their list
path = self._url('/v2/images')
headers = self._headers({'X-Tenant-Id': TENANT2})
response = requests.get(path, headers=headers)
self.assertEqual(200, response.status_code)
images = json.loads(response.text)['images']
self.assertEqual(image_id, images[0]['id'])
# TENANT2 should be able to access the image directly
path = self._url('/v2/images/%s' % image_id)
headers = self._headers({'X-Tenant-Id': TENANT2})
response = requests.get(path, headers=headers)
self.assertEqual(200, response.status_code)
# TENANT2 should not be able to modify the image
path = self._url('/v2/images/%s' % image_id)
headers = self._headers({
'Content-Type': 'application/json',
'X-Tenant-Id': TENANT2,
})
data = json.dumps({'name': 'image-2'})
response = requests.put(path, headers=headers, data=data)
self.assertEqual(404, response.status_code)
# TENANT2 should not be able to delete the image, either
path = self._url('/v2/images/%s' % image_id)
headers = self._headers({'X-Tenant-Id': TENANT2})
response = requests.delete(path, headers=headers)
self.assertEqual(404, response.status_code)
# As an unshared tenant, TENANT3 should not have access to the image
path = self._url('/v2/images/%s' % image_id)
headers = self._headers({'X-Tenant-Id': TENANT3})
response = requests.get(path, headers=headers)
self.assertEqual(404, response.status_code)
# Publicize the image as an admin of TENANT1
path = self._url('/v2/images/%s' % image_id)
headers = self._headers({
@ -320,97 +275,6 @@ class TestImages(functional.FunctionalTest):
self.stop_servers()
def test_access_lifecycle(self):
# Create an image for our tests
path = self._url('/v2/images')
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)
image_id = json.loads(response.text)['id']
# Image acccess list should be empty
path = self._url('/v2/images/%s/access' % image_id)
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('/v2/images/%s/access' % image_id)
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('/v2/images/%s/access' % image_id)
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('/v2/images/%s/access' % image_id)
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('/v2/images/%s/access' % image_id)
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('/v2/images/%s/access' % image_id)
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))
# 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('/v2/images/%s/access' % image_id)
response = requests.get(path, headers=self._headers())
self.assertEqual(200, response.status_code)
access_records = json.loads(response.text)['access_records']
self.assertEqual(1, len(access_records))
self.stop_servers()
def test_tag_lifecycle(self):
# Create an image with a tag
path = self._url('/v2/images')

View File

@ -38,7 +38,6 @@ class TestSchemas(functional.FunctionalTest):
expected = set([
'id',
'name',
'owner',
'visibility',
'checksum',
'created_at',
@ -50,19 +49,11 @@ class TestSchemas(functional.FunctionalTest):
'self',
'file',
'status',
'access',
'schema',
'direct_url',
])
self.assertEqual(expected, set(image_schema['properties'].keys()))
# Ensure the access link works
path = 'http://%s:%d/v2/schemas/image/access' % \
('127.0.0.1', self.api_port)
response = requests.get(path)
self.assertEqual(response.status_code, 200)
access_schema = json.loads(response.text)
# Ensure the images link works and agrees with the image schema
path = 'http://%s:%d/v2/schemas/images' % ('127.0.0.1', self.api_port)
response = requests.get(path)
@ -70,12 +61,3 @@ class TestSchemas(functional.FunctionalTest):
images_schema = json.loads(response.text)
item_schema = images_schema['properties']['images']['items']
self.assertEqual(item_schema, image_schema)
# Ensure the accesses schema works and agrees with access schema
path = 'http://%s:%d/v2/schemas/image/accesses' % \
('127.0.0.1', self.api_port)
response = requests.get(path)
self.assertEqual(response.status_code, 200)
accesses_schema = json.loads(response.text)
item_schema = accesses_schema['properties']['accesses']['items']
self.assertEqual(item_schema, access_schema)

View File

@ -1,250 +0,0 @@
# 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 webob
import glance.api.v2.image_access
from glance.common import exception
from glance.common import utils
import glance.schema
import glance.store
import glance.tests.unit.utils as unit_test_utils
import glance.tests.utils as test_utils
class TestImageAccessController(test_utils.BaseTestCase):
def setUp(self):
super(TestImageAccessController, self).setUp()
self.db = unit_test_utils.FakeDB()
self.controller = glance.api.v2.image_access.Controller(self.db)
glance.store.create_stores()
def test_index(self):
req = unit_test_utils.get_fake_request()
output = self.controller.index(req, unit_test_utils.UUID1)
expected = {
'access_records': [
{
'image_id': unit_test_utils.UUID1,
'member': unit_test_utils.TENANT1,
'can_share': True,
'deleted': False,
'deleted_at': None,
},
{
'image_id': unit_test_utils.UUID1,
'member': unit_test_utils.TENANT2,
'can_share': False,
'deleted': False,
'deleted_at': None,
},
],
'image_id': unit_test_utils.UUID1,
}
self.assertEqual(expected, output)
def test_index_zero_records(self):
req = unit_test_utils.get_fake_request()
output = self.controller.index(req, unit_test_utils.UUID2)
expected = {
'access_records': [],
'image_id': unit_test_utils.UUID2,
}
self.assertEqual(expected, output)
def test_index_nonexistant_image(self):
req = unit_test_utils.get_fake_request()
image_id = utils.generate_uuid()
self.assertRaises(exception.NotFound,
self.controller.index, req, image_id)
def test_show(self):
req = unit_test_utils.get_fake_request()
image_id = unit_test_utils.UUID1
tenant_id = unit_test_utils.TENANT1
output = self.controller.show(req, image_id, tenant_id)
expected = {
'image_id': image_id,
'member': tenant_id,
'can_share': True,
'deleted': False,
'deleted_at': None,
}
self.assertEqual(expected, output)
def test_show_nonexistant_image(self):
req = unit_test_utils.get_fake_request()
image_id = utils.generate_uuid()
tenant_id = unit_test_utils.TENANT1
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, req, image_id, tenant_id)
def test_show_nonexistant_tenant(self):
req = unit_test_utils.get_fake_request()
image_id = unit_test_utils.UUID1
tenant_id = utils.generate_uuid()
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, req, image_id, tenant_id)
def test_create(self):
member = utils.generate_uuid()
fixture = {
'member': member,
'can_share': True,
}
expected = {
'image_id': unit_test_utils.UUID1,
'member': member,
'can_share': True,
'deleted': False,
'deleted_at': None,
}
req = unit_test_utils.get_fake_request()
output = self.controller.create(req, unit_test_utils.UUID1, fixture)
self.assertEqual(expected, output)
class TestImageAccessDeserializer(test_utils.BaseTestCase):
def setUp(self):
super(TestImageAccessDeserializer, self).setUp()
self.deserializer = glance.api.v2.image_access.RequestDeserializer()
def test_create(self):
fixture = {
'tenant_id': unit_test_utils.TENANT1,
'can_share': False,
}
expected = {
'access_record': {
'member': unit_test_utils.TENANT1,
'can_share': False,
},
}
request = unit_test_utils.get_fake_request()
request.body = json.dumps(fixture)
output = self.deserializer.create(request)
self.assertEqual(expected, output)
class TestImageAccessSerializer(test_utils.BaseTestCase):
serializer = glance.api.v2.image_access.ResponseSerializer()
def test_show(self):
fixture = {
'image_id': unit_test_utils.UUID1,
'member': unit_test_utils.TENANT1,
'can_share': False,
}
self_href = ('/v2/images/%s/access/%s' %
(unit_test_utils.UUID1, unit_test_utils.TENANT1))
expected = {
'tenant_id': unit_test_utils.TENANT1,
'can_share': False,
'self': self_href,
'schema': '/v2/schemas/image/access',
'image': '/v2/images/%s' % unit_test_utils.UUID1,
}
response = webob.Response()
self.serializer.show(response, fixture)
self.assertEqual(expected, json.loads(response.body))
self.assertEqual('application/json', response.content_type)
def test_index(self):
fixtures = [
{
'image_id': unit_test_utils.UUID1,
'member': unit_test_utils.TENANT1,
'can_share': False,
},
{
'image_id': unit_test_utils.UUID1,
'member': unit_test_utils.TENANT2,
'can_share': True,
},
]
result = {
'access_records': fixtures,
'image_id': unit_test_utils.UUID1,
}
expected = {
'access_records': [
{
'tenant_id': unit_test_utils.TENANT1,
'can_share': False,
'self': ('/v2/images/%s/access/%s' %
(unit_test_utils.UUID1,
unit_test_utils.TENANT1)),
'schema': '/v2/schemas/image/access',
'image': '/v2/images/%s' % unit_test_utils.UUID1,
},
{
'tenant_id': unit_test_utils.TENANT2,
'can_share': True,
'self': ('/v2/images/%s/access/%s' %
(unit_test_utils.UUID1,
unit_test_utils.TENANT2)),
'schema': '/v2/schemas/image/access',
'image': '/v2/images/%s' % unit_test_utils.UUID1,
},
],
'first': '/v2/images/%s/access' % unit_test_utils.UUID1,
'schema': '/v2/schemas/image/accesses',
}
response = webob.Response()
self.serializer.index(response, result)
self.assertEqual(expected, json.loads(response.body))
self.assertEqual('application/json', response.content_type)
def test_index_zero_access_records(self):
result = {
'access_records': [],
'image_id': unit_test_utils.UUID1,
}
response = webob.Response()
self.serializer.index(response, result)
first_link = '/v2/images/%s/access' % unit_test_utils.UUID1
expected = {
'access_records': [],
'first': first_link,
'schema': '/v2/schemas/image/accesses',
}
self.assertEqual(expected, json.loads(response.body))
self.assertEqual('application/json', response.content_type)
def test_create(self):
fixture = {
'image_id': unit_test_utils.UUID1,
'member': unit_test_utils.TENANT1,
'can_share': False,
}
self_href = ('/v2/images/%s/access/%s' %
(unit_test_utils.UUID1, unit_test_utils.TENANT1))
expected = {
'tenant_id': unit_test_utils.TENANT1,
'can_share': False,
'self': self_href,
'schema': '/v2/schemas/image/access',
'image': '/v2/images/%s' % unit_test_utils.UUID1,
}
response = webob.Response()
self.serializer.create(response, fixture)
self.assertEqual(expected, json.loads(response.body))
self.assertEqual('application/json', response.content_type)
self.assertEqual(self_href, response.location)

View File

@ -175,12 +175,12 @@ class TestImagesController(test_utils.BaseTestCase):
filters={'size_max': 'blah'})
def test_index_with_filters_return_many(self):
path = '/images?owner=%s' % TENANT1
path = '/images?status=queued'
request = unit_test_utils.get_fake_request(path)
output = self.controller.index(request, filters={'owner': TENANT1})
self.assertEqual(2, len(output['images']))
output = self.controller.index(request, filters={'status': 'queued'})
self.assertEqual(3, len(output['images']))
actual = set([image['id'] for image in output['images']])
expected = set([UUID1, UUID2])
expected = set([UUID1, UUID2, UUID3])
self.assertEqual(actual, expected)
def test_index_with_nonexistant_name_filter(self):
@ -203,10 +203,10 @@ class TestImagesController(test_utils.BaseTestCase):
self.assertEqual(2, len(output['images']))
def test_index_with_many_filters(self):
request = unit_test_utils.get_fake_request('/images?owner=%s&name=%s' %
(TENANT1, '1'))
url = '/images?status=queued&name=2'
request = unit_test_utils.get_fake_request(url)
output = self.controller.index(request,
filters={'owner': TENANT1, 'name': '2'})
filters={'status': 'queued', 'name': '2'})
self.assertEqual(1, len(output['images']))
actual = set([image['id'] for image in output['images']])
expected = set([UUID2])
@ -330,12 +330,6 @@ class TestImagesController(test_utils.BaseTestCase):
}
self.assertEqual(expected, output)
def test_create_with_owner_as_admin(self):
request = unit_test_utils.get_fake_request(is_admin=True)
image = {'name': 'image-1', 'owner': utils.generate_uuid()}
output = self.controller.create(request, image)
self.assertEqual(image['owner'], output['owner'])
def test_create_public_image_as_admin(self):
request = unit_test_utils.get_fake_request()
image = {'name': 'image-1', 'is_public': True}
@ -407,19 +401,6 @@ class TestImagesDeserializer(test_utils.BaseTestCase):
expected = {'image': {'name': 'image-1', 'properties': {}}}
self.assertEqual(expected, output)
def test_create_with_owner_forbidden(self):
request = unit_test_utils.get_fake_request()
request.body = json.dumps({'owner': TENANT2})
self.assertRaises(webob.exc.HTTPForbidden,
self.deserializer.create, request)
def test_create_with_owner_admin(self):
request = unit_test_utils.get_fake_request(is_admin=True)
request.body = json.dumps({'owner': TENANT2})
output = self.deserializer.create(request)
expected = {'image': {'owner': TENANT2, 'properties': {}}}
self.assertEqual(expected, output)
def test_create_public(self):
request = unit_test_utils.get_fake_request()
request.body = json.dumps({'visibility': 'public'})
@ -737,7 +718,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
{
'id': unit_test_utils.UUID1,
'name': 'image-1',
'owner': TENANT1,
'status': 'queued',
'visibility': 'public',
'checksum': None,
@ -747,13 +727,11 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID1,
'file': '/v2/images/%s/file' % unit_test_utils.UUID1,
'access': '/v2/images/%s/access' % unit_test_utils.UUID1,
'schema': '/v2/schemas/image',
},
{
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': None,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -763,7 +741,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': None,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
},
],
@ -812,7 +789,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
{
'id': unit_test_utils.UUID1,
'name': 'image-1',
'owner': TENANT1,
'status': 'queued',
'visibility': 'public',
'checksum': None,
@ -822,13 +798,11 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID1,
'file': '/v2/images/%s/file' % unit_test_utils.UUID1,
'access': '/v2/images/%s/access' % unit_test_utils.UUID1,
'schema': '/v2/schemas/image',
},
{
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -838,7 +812,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': None,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
},
],
@ -888,7 +861,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
{
'id': unit_test_utils.UUID1,
'name': 'image-1',
'owner': TENANT1,
'status': 'queued',
'visibility': 'public',
'checksum': None,
@ -898,13 +870,11 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID1,
'file': '/v2/images/%s/file' % unit_test_utils.UUID1,
'access': '/v2/images/%s/access' % unit_test_utils.UUID1,
'schema': '/v2/schemas/image',
},
{
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -914,7 +884,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': None,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
},
],
@ -963,7 +932,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
{
'id': unit_test_utils.UUID1,
'name': 'image-1',
'owner': TENANT1,
'status': 'queued',
'visibility': 'public',
'checksum': None,
@ -973,13 +941,11 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID1,
'file': '/v2/images/%s/file' % unit_test_utils.UUID1,
'access': '/v2/images/%s/access' % unit_test_utils.UUID1,
'schema': '/v2/schemas/image',
},
{
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -989,7 +955,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': None,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
},
],
@ -1024,7 +989,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'public',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1034,7 +998,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1060,7 +1023,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1070,7 +1032,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': self_link,
'file': '%s/file' % self_link,
'access': '%s/access' % self_link,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1097,7 +1058,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'public',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1107,7 +1067,6 @@ class TestImagesSerializer(test_utils.BaseTestCase):
'size': 1024,
'self': self_link,
'file': '%s/file' % self_link,
'access': '%s/access' % self_link,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1149,7 +1108,6 @@ class TestImagesSerializerWithExtendedSchema(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1160,7 +1118,6 @@ class TestImagesSerializerWithExtendedSchema(test_utils.BaseTestCase):
'color': 'green',
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1172,7 +1129,6 @@ class TestImagesSerializerWithExtendedSchema(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1183,7 +1139,6 @@ class TestImagesSerializerWithExtendedSchema(test_utils.BaseTestCase):
'color': 'invalid',
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1217,7 +1172,6 @@ class TestImagesSerializerWithAdditionalProperties(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1228,7 +1182,6 @@ class TestImagesSerializerWithAdditionalProperties(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1244,7 +1197,6 @@ class TestImagesSerializerWithAdditionalProperties(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1255,7 +1207,6 @@ class TestImagesSerializerWithAdditionalProperties(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
}
response = webob.Response()
@ -1268,7 +1219,6 @@ class TestImagesSerializerWithAdditionalProperties(test_utils.BaseTestCase):
expected = {
'id': unit_test_utils.UUID2,
'name': 'image-2',
'owner': TENANT2,
'status': 'queued',
'visibility': 'private',
'checksum': 'ca425b88f047ce8ec45ee90e813ada91',
@ -1278,7 +1228,6 @@ class TestImagesSerializerWithAdditionalProperties(test_utils.BaseTestCase):
'size': 1024,
'self': '/v2/images/%s' % unit_test_utils.UUID2,
'file': '/v2/images/%s/file' % unit_test_utils.UUID2,
'access': '/v2/images/%s/access' % unit_test_utils.UUID2,
'schema': '/v2/schemas/image',
}
response = webob.Response()

View File

@ -40,23 +40,3 @@ class TestSchemasController(test_utils.BaseTestCase):
expected = set(['{schema}', '{first}', '{next}'])
actual = set([link['href'] for link in output['links']])
self.assertEqual(actual, expected)
def test_access(self):
req = unit_test_utils.get_fake_request()
output = self.controller.access(req)
self.assertEqual(output['name'], 'access')
expected = set(['tenant_id', 'can_share', 'schema', 'self', 'image'])
self.assertEqual(set(output['properties'].keys()), expected)
expected = set(['{schema}', '{self}', '{image}'])
actual = set([link['href'] for link in output['links']])
self.assertEqual(actual, expected)
def test_accesses(self):
req = unit_test_utils.get_fake_request()
output = self.controller.accesses(req)
self.assertEqual(output['name'], 'accesses')
expected = set(['accesses', 'schema', 'first', 'next'])
self.assertEqual(set(output['properties'].keys()), expected)
expected = set(['{schema}', '{first}', '{next}'])
actual = set([link['href'] for link in output['links']])
self.assertEqual(actual, expected)