Browse Source

Remove XML support

XML support has been deprecated for the last two releases, staged to be
removed in Kilo.

implements bp removed-as-of-kilo
Closes-Bug: #1384382
Closes-Bug: #1384789

Change-Id: I95761267b6c0f3052bc52deda6d56fc4b713ea6b
tags/2015.1.0b2
Lance Bragstad 5 years ago
parent
commit
4fdaab3b97
19 changed files with 35 additions and 1549 deletions
  1. +5
    -14
      etc/keystone-paste.ini
  2. +0
    -385
      keystone/common/serializer.py
  3. +0
    -7
      keystone/controllers.py
  4. +13
    -42
      keystone/middleware/core.py
  5. +2
    -10
      keystone/tests/rest.py
  6. +1
    -380
      keystone/tests/test_content_types.py
  7. +0
    -49
      keystone/tests/test_middleware.py
  8. +0
    -337
      keystone/tests/test_serializer.py
  9. +1
    -8
      keystone/tests/test_v3.py
  10. +0
    -21
      keystone/tests/test_v3_assignment.py
  11. +2
    -19
      keystone/tests/test_v3_auth.py
  12. +0
    -15
      keystone/tests/test_v3_catalog.py
  13. +0
    -5
      keystone/tests/test_v3_credential.py
  14. +0
    -29
      keystone/tests/test_v3_federation.py
  15. +0
    -9
      keystone/tests/test_v3_filters.py
  16. +0
    -14
      keystone/tests/test_v3_identity.py
  17. +0
    -5
      keystone/tests/test_v3_policy.py
  18. +11
    -195
      keystone/tests/test_versions.py
  19. +0
    -5
      keystone/token/controllers.py

+ 5
- 14
etc/keystone-paste.ini View File

@@ -12,15 +12,6 @@ paste.filter_factory = keystone.middleware:TokenAuthMiddleware.factory
[filter:admin_token_auth]
paste.filter_factory = keystone.middleware:AdminTokenAuthMiddleware.factory

[filter:xml_body]
paste.filter_factory = keystone.middleware:XmlBodyMiddleware.factory

[filter:xml_body_v2]
paste.filter_factory = keystone.middleware:XmlBodyMiddlewareV2.factory

[filter:xml_body_v3]
paste.filter_factory = keystone.middleware:XmlBodyMiddlewareV3.factory

[filter:json_body]
paste.filter_factory = keystone.middleware:JsonBodyMiddleware.factory

@@ -75,17 +66,17 @@ paste.app_factory = keystone.service:admin_app_factory
[pipeline:public_api]
# The last item in this pipeline must be public_service or an equivalent
# application. It cannot be a filter.
pipeline = sizelimit url_normalize build_auth_context token_auth admin_token_auth xml_body_v2 json_body ec2_extension user_crud_extension public_service
pipeline = sizelimit url_normalize build_auth_context token_auth admin_token_auth json_body ec2_extension user_crud_extension public_service

[pipeline:admin_api]
# The last item in this pipeline must be admin_service or an equivalent
# application. It cannot be a filter.
pipeline = sizelimit url_normalize build_auth_context token_auth admin_token_auth xml_body_v2 json_body ec2_extension s3_extension crud_extension admin_service
pipeline = sizelimit url_normalize build_auth_context token_auth admin_token_auth json_body ec2_extension s3_extension crud_extension admin_service

[pipeline:api_v3]
# The last item in this pipeline must be service_v3 or an equivalent
# application. It cannot be a filter.
pipeline = sizelimit url_normalize build_auth_context token_auth admin_token_auth xml_body_v3 json_body ec2_extension_v3 s3_extension simple_cert_extension revoke_extension service_v3
pipeline = sizelimit url_normalize build_auth_context token_auth admin_token_auth json_body ec2_extension_v3 s3_extension simple_cert_extension revoke_extension service_v3

[app:public_version_service]
paste.app_factory = keystone.service:public_version_app_factory
@@ -94,10 +85,10 @@ paste.app_factory = keystone.service:public_version_app_factory
paste.app_factory = keystone.service:admin_version_app_factory

[pipeline:public_version_api]
pipeline = sizelimit url_normalize xml_body public_version_service
pipeline = sizelimit url_normalize public_version_service

[pipeline:admin_version_api]
pipeline = sizelimit url_normalize xml_body admin_version_service
pipeline = sizelimit url_normalize admin_version_service

[composite:main]
use = egg:Paste#urlmap

+ 0
- 385
keystone/common/serializer.py View File

@@ -1,385 +0,0 @@
# Copyright 2012 OpenStack Foundation
#
# 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.

"""
Dict <--> XML de/serializer.

The identity API prefers attributes over elements, so we serialize that way
by convention, with a few hardcoded exceptions.

"""

import re

import six

try:
from lxml import etree
except ImportError:
etree = None

from keystone import exception
from keystone.i18n import _
from keystone.openstack.common import log as logging


LOG = logging.getLogger(__name__)

DOCTYPE = '<?xml version="1.0" encoding="UTF-8"?>'
XMLNS = 'http://docs.openstack.org/identity/api/v2.0'
XMLNS_LIST = [
{
'value': 'http://docs.openstack.org/identity/api/v2.0'
},
{
'prefix': 'OS-KSADM',
'value': 'http://docs.openstack.org/identity/api/ext/OS-KSADM/v1.0',
},
]
E_LXML_NOT_INSTALLED = _('lxml is not installed.')


def from_xml(xml):
"""Deserialize XML to a dictionary."""
if xml is None:
return None

deserializer = XmlDeserializer()
return deserializer(xml)


def to_xml(d, xmlns=None):
"""Serialize a dictionary to XML."""
if d is None:
return None

serialize = XmlSerializer()
return serialize(d, xmlns)


class XmlDeserializer(object):

def __init__(self):
if etree is None:
LOG.warning(E_LXML_NOT_INSTALLED)
raise exception.UnexpectedError(E_LXML_NOT_INSTALLED)

self.parser = etree.XMLParser(resolve_entities=False,
remove_comments=True,
remove_pis=True)

# NOTE(dolph): lxml.etree.Entity() is just a callable that currently
# returns an lxml.etree._Entity instance, which doesn't appear to be
# part of the public API, so we discover the type dynamically to be
# safe
self.entity_type = type(etree.Entity('x'))

def __call__(self, xml_str):
"""Returns a dictionary populated by decoding the given xml string."""
dom = etree.fromstring(xml_str.strip(), self.parser)
return self.walk_element(dom, True)

def _deserialize_links(self, links):
return dict((x.attrib['rel'], x.attrib['href']) for x in links)

@staticmethod
def _qualified_name(tag, namespace):
"""Returns a qualified tag name.

The tag name may contain the namespace prefix or not, which can
be determined by specifying the parameter namespace.

"""
m = re.search('[^}]+$', tag)
tag_name = m.string[m.start():]
if not namespace:
return tag_name
bracket = re.search('[^{]+$', tag)
ns = m.string[bracket.start():m.start() - 1]
# If the namespace is
# http://docs.openstack.org/identity/api/ext/OS-KSADM/v1.0 for the
# root element, a prefix needs to add in front of the tag name.
prefix = None
for xmlns in XMLNS_LIST:
if xmlns['value'] == ns:
prefix = xmlns.get('prefix')
break
if prefix is not None:
return '%(PREFIX)s:%(tag_name)s' % {
'PREFIX': prefix, 'tag_name': tag_name}
else:
return tag_name

def walk_element(self, element, namespace=False):
"""Populates a dictionary by walking an etree element."""
values = {}
for k, v in six.iteritems(element.attrib):
# boolean-looking attributes become booleans in JSON
if k in ['enabled', 'truncated']:
if v in ['true']:
v = True
elif v in ['false']:
v = False

values[self._qualified_name(k, namespace)] = v

text = None
if element.text is not None:
text = element.text.strip()

# current spec does not have attributes on an element with text
values = values or text or {}
decoded_tag = XmlDeserializer._qualified_name(element.tag, namespace)
list_item_tag = None
if (decoded_tag[-1] == 's' and not values and
decoded_tag != 'access'):
# FIXME(gyee): special-case lists for now unti we
# figure out how to properly handle them.
# If any key ends with an 's', we are assuming it is a list.
# List element have no attributes.
values = list(values)
if decoded_tag == 'policies':
list_item_tag = 'policy'
else:
list_item_tag = decoded_tag[:-1]

if decoded_tag == 'links':
return {'links': self._deserialize_links(element)}

links = None
truncated = False
for child in [self.walk_element(x) for x in element
if not isinstance(x, self.entity_type)]:
if list_item_tag:
# FIXME(gyee): special-case lists for now until we
# figure out how to properly handle them.
# If any key ends with an 's', we are assuming it is a list.
if list_item_tag in child:
values.append(child[list_item_tag])
else:
if 'links' in child:
links = child['links']
else:
truncated = child['truncated']
else:
values = dict(values.items() + child.items())

# set empty and none-list element to None to align with JSON
if not values:
values = ""

d = {XmlDeserializer._qualified_name(element.tag, namespace): values}

if links:
d['links'] = links
d['links'].setdefault('next')
d['links'].setdefault('previous')

if truncated:
d['truncated'] = truncated['truncated']

return d


class XmlSerializer(object):

def __init__(self):
if etree is None:
LOG.warning(E_LXML_NOT_INSTALLED)
raise exception.UnexpectedError(E_LXML_NOT_INSTALLED)

def __call__(self, d, xmlns=None):
"""Returns an xml etree populated by the given dictionary.

Optionally, namespace the etree by specifying an ``xmlns``.

"""
links = None
truncated = False
# FIXME(dolph): skipping links for now
for key in d.keys():
if '_links' in key:
d.pop(key)
# NOTE(gyee, henry-nash): special-case links and truncation
# attribute in collections
if 'links' == key:
if links:
# we have multiple links
raise Exception('Multiple links found')
links = d.pop(key)
if 'truncated' == key:
if truncated:
# we have multiple attributes
raise Exception(_('Multiple truncation attributes found'))
truncated = d.pop(key)
assert len(d.keys()) == 1, ('Cannot encode more than one root '
'element: %s' % d.keys())

# name the root dom element
name = d.keys()[0]
m = re.search('[^:]+$', name)
root_name = m.string[m.start():]
prefix = m.string[0:m.start() - 1]
for ns in XMLNS_LIST:
if prefix == ns.get('prefix'):
xmlns = ns['value']
break
# only the root dom element gets an xlmns
root = etree.Element(root_name, xmlns=(xmlns or XMLNS))

self.populate_element(root, d[name])

# NOTE(gyee, henry-nash): special-case links and truncation attribute
if links:
self._populate_links(root, links)
if truncated:
self._populate_truncated(root, truncated)

# TODO(dolph): you can get a doctype from lxml, using ElementTrees
return '%s\n%s' % (DOCTYPE, etree.tostring(root, pretty_print=True))

def _populate_links(self, element, links_json):
links = etree.Element('links')
for k, v in six.iteritems(links_json):
if v:
link = etree.Element('link')
link.set('rel', six.text_type(k))
link.set('href', six.text_type(v))
links.append(link)
element.append(links)

def _populate_truncated(self, element, truncated_value):
truncated = etree.Element('truncated')
self._populate_bool(truncated, 'truncated', truncated_value)
element.append(truncated)

def _populate_list(self, element, k, v):
"""Populates an element with a key & list value."""
# spec has a lot of inconsistency here!
container = element

if k == 'media-types':
# xsd compliance: <media-types> contains <media-type>s
# find an existing <media-types> element or make one
container = element.find('media-types')
if container is None:
container = etree.Element(k)
element.append(container)
name = k[:-1]
elif k == 'serviceCatalog' or k == 'catalog':
# xsd compliance: <serviceCatalog> contains <service>s
container = etree.Element(k)
element.append(container)
name = 'service'
elif k == 'roles' and element.tag == 'user':
name = 'role'
elif k == 'endpoints' and element.tag == 'service':
name = 'endpoint'
elif k == 'values' and element.tag[-1] == 's':
# OS convention is to contain lists in a 'values' element,
# so the list itself can have attributes, which is
# unnecessary in XML
name = element.tag[:-1]
elif k[-1] == 's':
container = etree.Element(k)
element.append(container)
if k == 'policies':
# need to special-case policies since policie is not a word
name = 'policy'
else:
name = k[:-1]
else:
name = k

for item in v:
child = etree.Element(name)
self.populate_element(child, item)
container.append(child)

def _populate_dict(self, element, k, v):
"""Populates an element with a key & dictionary value."""
if k == 'links':
# links is a special dict
self._populate_links(element, v)
else:
child = etree.Element(k)
self.populate_element(child, v)
element.append(child)

def _populate_bool(self, element, k, v):
"""Populates an element with a key & boolean value."""
# booleans are 'true' and 'false'
element.set(k, six.text_type(v).lower())

def _populate_str(self, element, k, v):
"""Populates an element with a key & string value."""
if k in ['description']:
# always becomes an element
child = etree.Element(k)
child.text = six.text_type(v)
element.append(child)
else:
# add attributes to the current element
element.set(k, six.text_type(v))

def _populate_number(self, element, k, v):
"""Populates an element with a key & numeric value."""
# numbers can be handled as strings
self._populate_str(element, k, v)

def populate_element(self, element, value):
"""Populates an etree with the given value."""
if isinstance(value, list):
self._populate_sequence(element, value)
elif isinstance(value, dict):
self._populate_tree(element, value)

# NOTE(blk-u): For compatibility with Folsom, when serializing the
# v2.0 version element also add the links to the base element.
if value.get('id') == 'v2.0':
for item in value['links']:
child = etree.Element('link')
self.populate_element(child, item)
element.append(child)

elif isinstance(value, six.string_types):
element.text = six.text_type(value)

def _populate_sequence(self, element, l):
"""Populates an etree with a sequence of elements, given a list."""
# xsd compliance: child elements are singular: <users> has <user>s
name = element.tag
if element.tag[-1] == 's':
name = element.tag[:-1]
if name == 'policie':
name = 'policy'

for item in l:
child = etree.Element(name)
self.populate_element(child, item)
element.append(child)

def _populate_tree(self, element, d):
"""Populates an etree with attributes & elements, given a dict."""
for k, v in six.iteritems(d):
if isinstance(v, dict):
self._populate_dict(element, k, v)
elif isinstance(v, list):
self._populate_list(element, k, v)
elif isinstance(v, bool):
self._populate_bool(element, k, v)
elif isinstance(v, six.string_types):
self._populate_str(element, k, v)
elif type(v) in [int, float, long, complex]:
self._populate_number(element, k, v)

+ 0
- 7
keystone/controllers.py View File

@@ -25,7 +25,6 @@ from keystone.openstack.common import log
LOG = log.getLogger(__name__)

MEDIA_TYPE_JSON = 'application/vnd.openstack.identity-%s+json'
MEDIA_TYPE_XML = 'application/vnd.openstack.identity-%s+xml'

_VERSIONS = []

@@ -141,9 +140,6 @@ class Version(wsgi.Application):
{
'base': 'application/json',
'type': MEDIA_TYPE_JSON % 'v2.0'
}, {
'base': 'application/xml',
'type': MEDIA_TYPE_XML % 'v2.0'
}
]
}
@@ -163,9 +159,6 @@ class Version(wsgi.Application):
{
'base': 'application/json',
'type': MEDIA_TYPE_JSON % 'v3'
}, {
'base': 'application/xml',
'type': MEDIA_TYPE_XML % 'v3'
}
]
}

+ 13
- 42
keystone/middleware/core.py View File

@@ -18,14 +18,12 @@ import webob.dec

from keystone.common import authorization
from keystone.common import config
from keystone.common import serializer
from keystone.common import utils
from keystone.common import wsgi
from keystone import exception
from keystone.i18n import _LW
from keystone.models import token_model
from keystone.openstack.common import log
from keystone.openstack.common import versionutils

CONF = config.CONF
LOG = log.getLogger(__name__)
@@ -145,59 +143,32 @@ class JsonBodyMiddleware(wsgi.Middleware):


class XmlBodyMiddleware(wsgi.Middleware):
"""De/serializes XML to/from JSON."""
"""De/serialize XML to/from JSON."""

def print_warning(self):
LOG.warning(_LW('XML support has been removed as of the Kilo release '
'and should not be referenced or used in deployment. '
'Please remove references to XmlBodyMiddleware from '
'your configuration. This compatibility stub will be '
'removed in the L release'))

@versionutils.deprecated(
what='keystone.middleware.core.XmlBodyMiddleware',
as_of=versionutils.deprecated.ICEHOUSE,
in_favor_of='support for "application/json" only',
remove_in=+2)
def __init__(self, *args, **kwargs):
super(XmlBodyMiddleware, self).__init__(*args, **kwargs)
self.xmlns = None

def process_request(self, request):
"""Transform the request from XML to JSON."""
incoming_xml = 'application/xml' in str(request.content_type)
if incoming_xml and request.body:
request.content_type = 'application/json'
try:
request.body = jsonutils.dumps(
serializer.from_xml(request.body))
except Exception:
LOG.exception('Serializer failed')
e = exception.ValidationError(attribute='valid XML',
target='request body')
return wsgi.render_exception(e, request=request)

def process_response(self, request, response):
"""Transform the response from JSON to XML."""
outgoing_xml = 'application/xml' in str(request.accept)
if outgoing_xml and response.body:
response.content_type = 'application/xml'
try:
body_obj = jsonutils.loads(response.body)
response.body = serializer.to_xml(body_obj, xmlns=self.xmlns)
except Exception:
LOG.exception('Serializer failed')
raise exception.Error(message=response.body)
return response
self.print_warning()


class XmlBodyMiddlewareV2(XmlBodyMiddleware):
"""De/serializes XML to/from JSON for v2.0 API."""
"""De/serialize XML to/from JSON for v2.0 API."""

def __init__(self, *args, **kwargs):
super(XmlBodyMiddlewareV2, self).__init__(*args, **kwargs)
self.xmlns = 'http://docs.openstack.org/identity/api/v2.0'
pass


class XmlBodyMiddlewareV3(XmlBodyMiddleware):
"""De/serializes XML to/from JSON for v3 API."""
"""De/serialize XML to/from JSON for v3 API."""

def __init__(self, *args, **kwargs):
super(XmlBodyMiddlewareV3, self).__init__(*args, **kwargs)
self.xmlns = 'http://docs.openstack.org/identity/api/v3'
pass


class NormalizingFilter(wsgi.Middleware):

+ 2
- 10
keystone/tests/rest.py View File

@@ -14,13 +14,12 @@

import io

from lxml import etree
from oslo.serialization import jsonutils
import six
import webtest

from keystone.auth import controllers as auth_controllers
from keystone.common import serializer
from keystone import tests
from keystone.tests import default_fixtures
from keystone.tests.ksfixtures import database
@@ -152,11 +151,6 @@ class RestfulTestCase(tests.TestCase):
if body:
headers['Content-Type'] = 'application/json'
return jsonutils.dumps(body)
elif content_type == 'xml':
headers['Accept'] = 'application/xml'
if body:
headers['Content-Type'] = 'application/xml'
return serializer.to_xml(body)

def _from_content_type(self, response, content_type=None):
"""Attempt to decode JSON and XML automatically, if detected."""
@@ -169,15 +163,13 @@ class RestfulTestCase(tests.TestCase):

if content_type == 'json':
response.result = jsonutils.loads(response.body)
elif content_type == 'xml':
response.result = etree.fromstring(response.body)
else:
response.result = response.body

def restful_request(self, method='GET', headers=None, body=None,
content_type=None, response_content_type=None,
**kwargs):
"""Serializes/deserializes json/xml as request/response body.
"""Serializes/deserializes json as request/response body.

.. WARNING::


+ 1
- 380
keystone/tests/test_content_types.py View File

@@ -30,30 +30,17 @@ CONF = config.CONF

class CoreApiTests(object):
def assertValidError(self, error):
"""Applicable to XML and JSON."""
self.assertIsNotNone(error.get('code'))
self.assertIsNotNone(error.get('title'))
self.assertIsNotNone(error.get('message'))

def assertValidVersion(self, version):
"""Applicable to XML and JSON.

However, navigating links and media-types differs between content
types so they need to be validated separately.

"""
self.assertIsNotNone(version)
self.assertIsNotNone(version.get('id'))
self.assertIsNotNone(version.get('status'))
self.assertIsNotNone(version.get('updated'))

def assertValidExtension(self, extension):
"""Applicable to XML and JSON.

However, navigating extension links differs between content types.
They need to be validated separately with assertValidExtensionLink.

"""
self.assertIsNotNone(extension)
self.assertIsNotNone(extension.get('name'))
self.assertIsNotNone(extension.get('namespace'))
@@ -61,23 +48,19 @@ class CoreApiTests(object):
self.assertIsNotNone(extension.get('updated'))

def assertValidExtensionLink(self, link):
"""Applicable to XML and JSON."""
self.assertIsNotNone(link.get('rel'))
self.assertIsNotNone(link.get('type'))
self.assertIsNotNone(link.get('href'))

def assertValidTenant(self, tenant):
"""Applicable to XML and JSON."""
self.assertIsNotNone(tenant.get('id'))
self.assertIsNotNone(tenant.get('name'))

def assertValidUser(self, user):
"""Applicable to XML and JSON."""
self.assertIsNotNone(user.get('id'))
self.assertIsNotNone(user.get('name'))

def assertValidRole(self, tenant):
"""Applicable to XML and JSON."""
self.assertIsNotNone(tenant.get('id'))
self.assertIsNotNone(tenant.get('name'))

@@ -315,7 +298,7 @@ class CoreApiTests(object):
self.assertValidUserResponse(r)

def test_create_update_user_invalid_enabled_type(self):
# Enforce usage of boolean for 'enabled' field in JSON and XML
# Enforce usage of boolean for 'enabled' field in JSON
token = self.get_scoped_token()

# Test CREATE request
@@ -326,7 +309,6 @@ class CoreApiTests(object):
'user': {
'name': uuid.uuid4().hex,
'password': uuid.uuid4().hex,
# In XML, only "true|false" are converted to boolean.
'enabled': "False",
},
},
@@ -359,7 +341,6 @@ class CoreApiTests(object):
path=path,
body={
'user': {
# In XML, only "true|false" are converted to boolean.
'enabled': "False",
},
},
@@ -1403,363 +1384,3 @@ class RevokeApiJsonTestCase(JsonTestCase):

def test_fetch_revocation_list_sha256(self):
self.skipTest('Revoke API disables revocation_list.')


class XmlTestCase(RestfulTestCase, CoreApiTests, LegacyV2UsernameTests):
xmlns = 'http://docs.openstack.org/identity/api/v2.0'
content_type = 'xml'

def _get_user_id(self, r):
return r.get('id')

def _get_role_name(self, r):
return r[0].get('name')

def _get_role_id(self, r):
return r[0].get('id')

def _get_project_id(self, r):
return r.get('id')

def assertNoRoles(self, r):
self.assertEqual(0, len(r))

def _get_token_id(self, r):
return r.result.find(self._tag('token')).get('id')

def _tag(self, tag_name, xmlns=None):
"""Helper method to build an namespaced element name."""
return '{%(ns)s}%(tag)s' % {'ns': xmlns or self.xmlns, 'tag': tag_name}

def assertValidErrorResponse(self, r):
xml = r.result
self.assertEqual(self._tag('error'), xml.tag)

self.assertValidError(xml)
self.assertEqual(str(r.status_code), xml.get('code'))

def assertValidExtension(self, extension, expected):
super(XmlTestCase, self).assertValidExtension(extension)

self.assertIsNotNone(extension.find(self._tag('description')))
self.assertTrue(extension.find(self._tag('description')).text)
links = extension.find(self._tag('links'))
self.assertNotEmpty(links.findall(self._tag('link')))
descriptions = [ext['description'] for ext in six.itervalues(expected)]
description = extension.find(self._tag('description')).text
self.assertIn(description, descriptions)
for link in links.findall(self._tag('link')):
self.assertValidExtensionLink(link)

def assertValidExtensionListResponse(self, r, expected):
xml = r.result
self.assertEqual(self._tag('extensions'), xml.tag)
self.assertNotEmpty(xml.findall(self._tag('extension')))
for ext in xml.findall(self._tag('extension')):
self.assertValidExtension(ext, expected)

def assertValidExtensionResponse(self, r, expected):
xml = r.result
self.assertEqual(self._tag('extension'), xml.tag)

self.assertValidExtension(xml, expected)

def assertValidVersion(self, version):
super(XmlTestCase, self).assertValidVersion(version)

links = version.find(self._tag('links'))
self.assertIsNotNone(links)
self.assertNotEmpty(links.findall(self._tag('link')))
for link in links.findall(self._tag('link')):
self.assertIsNotNone(link.get('rel'))
self.assertIsNotNone(link.get('href'))

media_types = version.find(self._tag('media-types'))
self.assertIsNotNone(media_types)
self.assertNotEmpty(media_types.findall(self._tag('media-type')))
for media in media_types.findall(self._tag('media-type')):
self.assertIsNotNone(media.get('base'))
self.assertIsNotNone(media.get('type'))

def assertValidMultipleChoiceResponse(self, r):
xml = r.result
self.assertEqual(self._tag('versions'), xml.tag)

self.assertNotEmpty(xml.findall(self._tag('version')))
for version in xml.findall(self._tag('version')):
self.assertValidVersion(version)

def assertValidVersionResponse(self, r):
xml = r.result
self.assertEqual(self._tag('version'), xml.tag)

self.assertValidVersion(xml)

def assertValidEndpointListResponse(self, r):
xml = r.result
self.assertEqual(self._tag('endpoints'), xml.tag)

self.assertNotEmpty(xml.findall(self._tag('endpoint')))
for endpoint in xml.findall(self._tag('endpoint')):
self.assertIsNotNone(endpoint.get('id'))
self.assertIsNotNone(endpoint.get('name'))
self.assertIsNotNone(endpoint.get('type'))
self.assertIsNotNone(endpoint.get('publicURL'))
self.assertIsNotNone(endpoint.get('internalURL'))
self.assertIsNotNone(endpoint.get('adminURL'))

def assertValidTenantResponse(self, r):
xml = r.result
self.assertEqual(self._tag('tenant'), xml.tag)

self.assertValidTenant(xml)

def assertValidUserResponse(self, r):
xml = r.result
self.assertEqual(self._tag('user'), xml.tag)

self.assertValidUser(xml)

def assertValidRoleListResponse(self, r):
xml = r.result
self.assertEqual(self._tag('roles'), xml.tag)

self.assertNotEmpty(r.result.findall(self._tag('role')))
for role in r.result.findall(self._tag('role')):
self.assertValidRole(role)

def assertValidAuthenticationResponse(self, r,
require_service_catalog=False):
xml = r.result
self.assertEqual(self._tag('access'), xml.tag)

# validate token
token = xml.find(self._tag('token'))
self.assertIsNotNone(token)
self.assertIsNotNone(token.get('id'))
self.assertIsNotNone(token.get('expires'))
tenant = token.find(self._tag('tenant'))
if tenant is not None:
# validate tenant
self.assertValidTenant(tenant)
self.assertIn(tenant.get('enabled'), ['true', 'false'])

user = xml.find(self._tag('user'))
self.assertIsNotNone(user)
self.assertIsNotNone(user.get('id'))
self.assertIsNotNone(user.get('name'))

if require_service_catalog:
# roles are only provided with a service catalog
roles = user.findall(self._tag('role'))
self.assertNotEmpty(roles)
for role in roles:
self.assertIsNotNone(role.get('name'))

serviceCatalog = xml.find(self._tag('serviceCatalog'))
# validate the serviceCatalog
if require_service_catalog:
self.assertIsNotNone(serviceCatalog)
if serviceCatalog is not None:
services = serviceCatalog.findall(self._tag('service'))
if require_service_catalog:
self.assertNotEmpty(services)
for service in services:
# validate service
self.assertIsNotNone(service.get('name'))
self.assertIsNotNone(service.get('type'))

# services contain at least one endpoint
endpoints = service.findall(self._tag('endpoint'))
self.assertNotEmpty(endpoints)
for endpoint in endpoints:
# validate service endpoint
self.assertIsNotNone(endpoint.get('publicURL'))

def assertValidTenantListResponse(self, r):
xml = r.result
self.assertEqual(self._tag('tenants'), xml.tag)

self.assertNotEmpty(r.result)
for tenant in r.result.findall(self._tag('tenant')):
self.assertValidTenant(tenant)
self.assertIn(tenant.get('enabled'), ['true', 'false'])

def get_user_from_response(self, r):
return r.result

def get_user_attribute_from_response(self, r, attribute_name):
return r.result.get(attribute_name)

def test_authenticate_with_invalid_xml_in_password(self):
# public_request would auto escape the ampersand
self.public_request(
method='POST',
path='/v2.0/tokens',
headers={
'Content-Type': 'application/xml'
},
body="""
<?xml version="1.0" encoding="UTF-8"?>
<auth xmlns="http://docs.openstack.org/identity/api/v2.0"
tenantId="bar">
<passwordCredentials username="FOO" password="&"/>
</auth>
""",
expected_status=400,
convert=False)

def test_add_tenant_xml(self):
"""Create a tenant without providing description field."""
token = self.get_scoped_token()
r = self.admin_request(
method='POST',
path='/v2.0/tenants',
headers={
'Content-Type': 'application/xml',
'X-Auth-Token': token
},
body="""
<?xml version="1.0" encoding="UTF-8"?>
<tenant xmlns="http://docs.openstack.org/identity/api/v2.0"
enabled="true" name="ACME Corp">
<description></description>
</tenant>
""",
convert=False)
self._from_content_type(r, 'json')
self.assertIsNotNone(r.result.get('tenant'))
self.assertValidTenant(r.result['tenant'])
self.assertEqual("", r.result['tenant'].get('description'))

def test_add_tenant_json(self):
"""Create a tenant without providing description field."""
token = self.get_scoped_token()
r = self.admin_request(
method='POST',
path='/v2.0/tenants',
headers={
'Content-Type': 'application/json',
'X-Auth-Token': token
},
body="""
{"tenant":{
"name":"test1",
"description":"",
"enabled":true}
}
""",
convert=False)
self._from_content_type(r, 'json')
self.assertIsNotNone(r.result.get('tenant'))
self.assertValidTenant(r.result['tenant'])
self.assertEqual("", r.result['tenant'].get('description'))

def test_create_project_invalid_enabled_type_string(self):
# Forbidden usage of string for 'enabled' field in JSON and XML
token = self.get_scoped_token()

r = self.admin_request(
method='POST',
path='/v2.0/tenants',
body={
'tenant': {
'name': uuid.uuid4().hex,
# In XML, only "true|false" are converted to boolean.
'enabled': "False",
},
},
token=token,
expected_status=400)
self.assertValidErrorResponse(r)

def test_update_project_invalid_enabled_type_string(self):
# Forbidden usage of string for 'enabled' field in JSON and XML
token = self.get_scoped_token()

path = '/v2.0/tenants/%(tenant_id)s' % {
'tenant_id': self.tenant_bar['id'],
}

r = self.admin_request(
method='PUT',
path=path,
body={
'tenant': {
# In XML, only "true|false" are converted to boolean.
'enabled': "False",
},
},
token=token,
expected_status=400)
self.assertValidErrorResponse(r)

def test_authenticating_a_user_with_an_OSKSADM_password(self):
token = self.get_scoped_token()

xmlns = "http://docs.openstack.org/identity/api/ext/OS-KSADM/v1.0"

username = uuid.uuid4().hex
password = uuid.uuid4().hex

# create the user
self.admin_request(
method='POST',
path='/v2.0/users',
headers={
'Content-Type': 'application/xml'
},
body="""
<?xml version="1.0" encoding="UTF-8"?>
<user xmlns="http://docs.openstack.org/identity/api/v2.0"
xmlns:OS-KSADM="%(xmlns)s"
name="%(username)s"
OS-KSADM:password="%(password)s"
enabled="true"/>
""" % dict(username=username, password=password, xmlns=xmlns),
token=token,
expected_status=200,
convert=False)

# successfully authenticate
self.public_request(
method='POST',
path='/v2.0/tokens',
headers={
'Content-Type': 'application/xml'
},
body="""
<?xml version="1.0" encoding="UTF-8"?>
<auth xmlns="http://docs.openstack.org/identity/api/v2.0"
xmlns:OS-KSADM="%(xmlns)s">
<passwordCredentials
username="%(username)s"
password="%(password)s"/>
</auth>
""" % dict(username=username, password=password, xmlns=xmlns),
token=token,
expected_status=200,
convert=False)

def test_remove_role_revokes_token(self):
self.md_foobar = self.assignment_api.add_role_to_user_and_project(
self.user_foo['id'],
self.tenant_service['id'],
self.role_service['id'])

token = self.get_scoped_token(tenant_id='service')
r = self.admin_request(
path='/v2.0/tokens/%s' % token,
token=token)
self.assertValidAuthenticationResponse(r)

self.assignment_api.remove_role_from_user_and_project(
self.user_foo['id'],
self.tenant_service['id'],
self.role_service['id'])

# TODO(ayoung): test fails due to XML problem
# r = self.admin_request(
# path='/v2.0/tokens/%s' % token,
# token=token,
# expected_status=401)

+ 0
- 49
keystone/tests/test_middleware.py View File

@@ -12,7 +12,6 @@
# License for the specific language governing permissions and limitations
# under the License.

from oslo.serialization import jsonutils
import webob

from keystone import config
@@ -118,51 +117,3 @@ class JsonBodyMiddlewareTest(tests.TestCase):
middleware.JsonBodyMiddleware(None).process_request(req)
params = req.environ.get(middleware.PARAMS_ENV, {})
self.assertEqual({}, params)


class XmlBodyMiddlewareTest(tests.TestCase):
def test_client_wants_xml_back(self):
"""Clients requesting XML should get what they ask for."""
body = '{"container": {"attribute": "value"}}'
req = make_request(body=body, method='POST', accept='application/xml')
middleware.XmlBodyMiddleware(None).process_request(req)
resp = make_response(body=body)
middleware.XmlBodyMiddleware(None).process_response(req, resp)
self.assertEqual('application/xml', resp.content_type)

def test_client_wants_json_back(self):
"""Clients requesting JSON should definitely not get XML back."""
body = '{"container": {"attribute": "value"}}'
req = make_request(body=body, method='POST', accept='application/json')
middleware.XmlBodyMiddleware(None).process_request(req)
resp = make_response(body=body)
middleware.XmlBodyMiddleware(None).process_response(req, resp)
self.assertNotIn('application/xml', resp.content_type)

def test_client_fails_to_specify_accept(self):
"""If client does not specify an Accept header, default to JSON."""
body = '{"container": {"attribute": "value"}}'
req = make_request(body=body, method='POST')
middleware.XmlBodyMiddleware(None).process_request(req)
resp = make_response(body=body)
middleware.XmlBodyMiddleware(None).process_response(req, resp)
self.assertNotIn('application/xml', resp.content_type)

def test_xml_replaced_by_json(self):
"""XML requests should be replaced by JSON requests."""
req = make_request(
body='<container><element attribute="value" /></container>',
content_type='application/xml',
method='POST')
middleware.XmlBodyMiddleware(None).process_request(req)
self.assertEqual('application/json', req.content_type)
self.assertTrue(jsonutils.loads(req.body))

def test_json_unnaffected(self):
"""JSON-only requests should be unaffected by the XML middleware."""
content_type = 'application/json'
body = '{"container": {"attribute": "value"}}'
req = make_request(body=body, content_type=content_type, method='POST')
middleware.XmlBodyMiddleware(None).process_request(req)
self.assertEqual(body, req.body)
self.assertEqual(content_type, req.content_type)

+ 0
- 337
keystone/tests/test_serializer.py View File

@@ -1,337 +0,0 @@
# Copyright 2012 OpenStack Foundation
#
# 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 copy

from lxml import etree
import mock
from testtools import matchers

from keystone.common import serializer
from keystone import exception
from keystone import tests
from keystone.tests import matchers as ksmatchers


class XmlSerializerTestCase(tests.TestCase):
def assertSerializeDeserialize(self, d, xml, xmlns=None):
self.assertThat(
serializer.to_xml(copy.deepcopy(d), xmlns),
ksmatchers.XMLEquals(xml))
self.assertEqual(serializer.from_xml(xml), d)

# operations should be invertible
self.assertEqual(
serializer.from_xml(serializer.to_xml(copy.deepcopy(d), xmlns)),
d)
self.assertThat(
serializer.to_xml(serializer.from_xml(xml), xmlns),
ksmatchers.XMLEquals(xml))

def test_auth_request(self):
d = {
"auth": {
"passwordCredentials": {
"username": "test_user",
"password": "mypass"
},
"tenantName": "customer-x"
}
}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<auth xmlns="http://docs.openstack.org/identity/api/v2.0"
tenantName="customer-x">
<passwordCredentials
username="test_user"
password="mypass"/>
</auth>
"""

self.assertSerializeDeserialize(d, xml)

def test_role_crud(self):
d = {
"role": {
"id": "123",
"name": "Guest",
"description": "Guest Access"
}
}

# TODO(dolph): examples show this description as an attribute?
xml = """
<?xml version="1.0" encoding="UTF-8"?>
<role xmlns="http://docs.openstack.org/identity/api/v2.0"
id="123"
name="Guest">
<description>Guest Access</description>
</role>
"""

self.assertSerializeDeserialize(d, xml)

def test_service_crud(self):
xmlns = "http://docs.openstack.org/identity/api/ext/OS-KSADM/v1.0"

d = {
"OS-KSADM:service": {
"id": "123",
"name": "nova",
"type": "compute",
"description": "OpenStack Compute Service"
}
}

# TODO(dolph): examples show this description as an attribute?
xml = """
<?xml version="1.0" encoding="UTF-8"?>
<service
xmlns="%(xmlns)s"
type="compute"
id="123"
name="nova">
<description>OpenStack Compute Service</description>
</service>
""" % {'xmlns': xmlns}

self.assertSerializeDeserialize(d, xml, xmlns=xmlns)

def test_tenant_crud(self):
d = {
"tenant": {
"id": "1234",
"name": "ACME corp",
"description": "A description...",
"enabled": True
}
}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<tenant
xmlns="http://docs.openstack.org/identity/api/v2.0"
enabled="true"
id="1234"
name="ACME corp">
<description>A description...</description>
</tenant>
"""

self.assertSerializeDeserialize(d, xml)

def test_tenant_crud_no_description(self):
d = {
"tenant": {
"id": "1234",
"name": "ACME corp",
"description": "",
"enabled": True
}
}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<tenant
xmlns="http://docs.openstack.org/identity/api/v2.0"
enabled="true"
id="1234"
name="ACME corp">
<description></description>
</tenant>
"""

self.assertSerializeDeserialize(d, xml)

def test_policy_list(self):
d = {"policies": [{"id": "ab12cd"}]}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<policies xmlns="http://docs.openstack.org/identity/api/v2.0">
<policy id="ab12cd"/>
</policies>
"""
self.assertThat(serializer.to_xml(d), ksmatchers.XMLEquals(xml))

def test_values_list(self):
d = {
"objects": {
"values": [{
"attribute": "value1",
}, {
"attribute": "value2",
}]
}
}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<objects xmlns="http://docs.openstack.org/identity/api/v2.0">
<object attribute="value1"/>
<object attribute="value2"/>
</objects>
"""

self.assertThat(serializer.to_xml(d), ksmatchers.XMLEquals(xml))

def test_collection_list(self):
d = {
"links": {
"next": "http://localhost:5000/v3/objects?page=3",
"previous": None,
"self": "http://localhost:5000/v3/objects"
},
"objects": [{
"attribute": "value1",
"links": {
"self": "http://localhost:5000/v3/objects/abc123def",
"anotherobj": "http://localhost:5000/v3/anotherobjs/123"
}
}, {
"attribute": "value2",
"links": {
"self": "http://localhost:5000/v3/objects/abc456"
}
}]}
xml = """
<?xml version="1.0" encoding="UTF-8"?>
<objects xmlns="http://docs.openstack.org/identity/api/v2.0">
<object attribute="value1">
<links>
<link rel="self"
href="http://localhost:5000/v3/objects/abc123def"/>
<link rel="anotherobj"
href="http://localhost:5000/v3/anotherobjs/123"/>
</links>
</object>
<object attribute="value2">
<links>
<link rel="self"
href="http://localhost:5000/v3/objects/abc456"/>
</links>
</object>
<links>
<link rel="self"
href="http://localhost:5000/v3/objects"/>
<link rel="next"
href="http://localhost:5000/v3/objects?page=3"/>
</links>
</objects>
"""
self.assertSerializeDeserialize(d, xml)

def test_collection_member(self):
d = {
"object": {
"attribute": "value",
"links": {
"self": "http://localhost:5000/v3/objects/abc123def",
"anotherobj": "http://localhost:5000/v3/anotherobjs/123"}}}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<object xmlns="http://docs.openstack.org/identity/api/v2.0"
attribute="value">
<links>
<link rel="self"
href="http://localhost:5000/v3/objects/abc123def"/>
<link rel="anotherobj"
href="http://localhost:5000/v3/anotherobjs/123"/>
</links>
</object>
"""
self.assertSerializeDeserialize(d, xml)

def test_v2_links_special_case(self):
# There's special-case code (for backward compatibility) where if the
# data is the v2 version data, the link elements are also added to the
# main element.

d = {
"object": {
"id": "v2.0",
"status": "deprecated",
"updated": "2014-04-17T00:00:00Z",
"links": [{"href": "http://localhost:5000/v2.0/",
"rel": "self"},
{"href": "http://docs.openstack.org/api/openstack-"
"identity-service/2.0/content/",
"type": "text/html", "rel": "describedby"},
{"href": "http://docs.openstack.org/api/openstack-"
"identity-service/2.0/"
"identity-dev-guide-2.0.pdf",
"type": "application/pdf", "rel": "describedby"}]
}}

xml = """
<?xml version="1.0" encoding="UTF-8"?>
<object xmlns="http://docs.openstack.org/identity/api/v2.0"
id="v2.0" status="deprecated" updated="2014-04-17T00:00:00Z">
<links>
<link rel="self" href="http://localhost:5000/v2.0/"/>
<link rel="describedby"
href="http://docs.openstack.org/api/openstack-\
identity-service/2.0/content/" type="text/html"/>
<link rel="describedby"
href="http://docs.openstack.org/api/openstack-\
identity-service/2.0/identity-dev-guide-2.0.pdf" type="application/pdf"/>
</links>
<link rel="self" href="http://localhost:5000/v2.0/"/>
<link rel="describedby"
href="http://docs.openstack.org/api/openstack-\
identity-service/2.0/content/" type="text/html"/>
<link rel="describedby"
href="http://docs.openstack.org/api/openstack-\
identity-service/2.0/identity-dev-guide-2.0.pdf" type="application/pdf"/>
</object>
"""
self.assertThat(serializer.to_xml(d), ksmatchers.XMLEquals(xml))

def test_xml_with_namespaced_attribute_to_dict(self):
expected = {
"user": {
"username": "test_user",
"OS-KSADM:password": "mypass",
},
}

xmlns = 'http://docs.openstack.org/identity/api/ext/OS-KSADM/v1.0'
xml = """
<?xml version="1.0" encoding="UTF-8"?>
<user xmlns="http://docs.openstack.org/identity/api/v2.0"
xmlns:OS-KSADM="%(xmlns)s"
username="test_user"
OS-KSADM:password="mypass"/>
""" % dict(xmlns=xmlns)
self.assertThat(serializer.from_xml(xml), matchers.Equals(expected))

@mock.patch('keystone.common.serializer.etree', new=etree)
def test_XmlDeserializer_with_installed_succeeds(self):
serializer.XmlDeserializer()

@mock.patch('keystone.common.serializer.etree', new=None)
def test_XmlDeserializer_without_etree_installed_fails(self):
self.assertRaises(exception.UnexpectedError,
serializer.XmlDeserializer)

@mock.patch('keystone.common.serializer.etree', new=etree)
def test_XmlSerializer_with_installed_succeeds(self):
serializer.XmlSerializer()

@mock.patch('keystone.common.serializer.etree', new=None)
def test_XmlSerializer_without_etree_installed_fails(self):
self.assertRaises(exception.UnexpectedError,
serializer.XmlSerializer)

+ 1
- 8
keystone/tests/test_v3.py View File

@@ -15,7 +15,6 @@
import datetime
import uuid

from lxml import etree
from oslo.serialization import jsonutils
from oslo.utils import timeutils
import six
@@ -24,7 +23,6 @@ from testtools import matchers
from keystone import auth
from keystone.common import authorization
from keystone.common import cache
from keystone.common import serializer
from keystone import config
from keystone import exception
from keystone import middleware
@@ -400,8 +398,6 @@ class RestfulTestCase(tests.SQLDriverOverrides, rest.RestfulTestCase,

"""
r = super(RestfulTestCase, self).admin_request(*args, **kwargs)
if r.headers.get('Content-Type') == 'application/xml':
r.result = serializer.from_xml(etree.tostring(r.result))
return r

def get_scoped_token(self):
@@ -505,10 +501,7 @@ class RestfulTestCase(tests.SQLDriverOverrides, rest.RestfulTestCase,
return r

def assertValidErrorResponse(self, r):
if r.headers.get('Content-Type') == 'application/xml':
resp = serializer.from_xml(etree.tostring(r.result))
else:
resp = r.result
resp = r.result
self.assertIsNotNone(resp.get('error'))
self.assertIsNotNone(resp['error'].get('code'))
self.assertIsNotNone(resp['error'].get('title'))

+ 0
- 21
keystone/tests/test_v3_assignment.py View File

@@ -134,13 +134,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase):
self.assertValidDomainListResponse(r, ref=self.domain,
resource_url=resource_url)

def test_list_domains_xml(self):
"""Call ``GET /domains (xml data)``."""
resource_url = '/domains'
r = self.get(resource_url, content_type='xml')
self.assertValidDomainListResponse(r, ref=self.domain,
resource_url=resource_url)

def test_get_domain(self):
"""Call ``GET /domains/{domain_id}``."""
r = self.get('/domains/%(domain_id)s' % {
@@ -470,13 +463,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase):
self.assertValidProjectListResponse(r, ref=self.project,
resource_url=resource_url)

def test_list_projects_xml(self):
"""Call ``GET /projects`` (xml data)."""
resource_url = '/projects'
r = self.get(resource_url, content_type='xml')
self.assertValidProjectListResponse(r, ref=self.project,
resource_url=resource_url)

def test_create_project(self):
"""Call ``POST /projects``."""
ref = self.new_project_ref(domain_id=self.domain_id)
@@ -678,13 +664,6 @@ class AssignmentTestCase(test_v3.RestfulTestCase):
self.assertValidRoleListResponse(r, ref=self.role,
resource_url=resource_url)

def test_list_roles_xml(self):
"""Call ``GET /roles`` (xml data)."""
resource_url = '/roles'
r = self.get(resource_url, content_type='xml')
self.assertValidRoleListResponse(r, ref=self.role,
resource_url=resource_url)

def test_get_role(self):
"""Call ``GET /roles/{role_id}``."""
r = self.get('/roles/%(role_id)s' % {

+ 2
- 19
keystone/tests/test_v3_auth.py View File

@@ -1675,8 +1675,7 @@ class TestAuthKerberos(TestAuthExternalDomain):
token='keystone.auth.plugins.token.Token')


class TestAuthJSON(test_v3.RestfulTestCase):
content_type = 'json'
class TestAuth(test_v3.RestfulTestCase):

def test_unscoped_token_with_user_id(self):
auth_data = self.build_authentication_request(
@@ -1777,9 +1776,6 @@ class TestAuthJSON(test_v3.RestfulTestCase):
self.project['id'])

def test_auth_catalog_attributes(self):
if self.content_type == 'xml':
self.skipTest('XML catalog parsing is just broken')

auth_data = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'],
@@ -1823,8 +1819,7 @@ class TestAuthJSON(test_v3.RestfulTestCase):
project_id=self.project['id'])
r = self.v3_authenticate_token(auth_data)

# In JSON, this is an empty list. In XML, this is an empty string.
self.assertFalse(r.result['token']['catalog'])
self.assertEqual([], r.result['token']['catalog'])

def test_auth_catalog_disabled_endpoint(self):
"""On authenticate, get a catalog that excludes disabled endpoints."""
@@ -2563,18 +2558,6 @@ class TestAuthJSONExternal(test_v3.RestfulTestCase):
auth_context)


class TestAuthXML(TestAuthJSON):
content_type = 'xml'

def _check_disabled_endpoint_result(self, catalog, disabled_endpoint_id):
# FIXME(blk-u): As far as I can tell the catalog in the XML result is
# broken. Looks like it includes only one endpoint or the other, and
# which one is included is random.

endpoint = catalog['service']['endpoint']
self.assertEqual(self.endpoint_id, endpoint['id'])


class TestTrustOptional(test_v3.RestfulTestCase):
def config_overrides(self):
super(TestTrustOptional, self).config_overrides()

+ 0
- 15
keystone/tests/test_v3_catalog.py View File

@@ -209,11 +209,6 @@ class CatalogTestCase(test_v3.RestfulTestCase):
for region in r.result['regions']:
self.assertEqual(parent_id, region['parent_region_id'])

def test_list_regions_xml(self):
"""Call ``GET /regions (xml data)``."""
r = self.get('/regions', content_type='xml')
self.assertValidRegionListResponse(r, ref=self.region)

def test_get_region(self):
"""Call ``GET /regions/{region_id}``."""
r = self.get('/regions/%(region_id)s' % {
@@ -389,11 +384,6 @@ class CatalogTestCase(test_v3.RestfulTestCase):
filtered_service = filtered_service_list[0]
self.assertEqual(target_ref['name'], filtered_service['name'])

def test_list_services_xml(self):
"""Call ``GET /services (xml data)``."""
r = self.get('/services', content_type='xml')
self.assertValidServiceListResponse(r, ref=self.service)

def test_get_service(self):
"""Call ``GET /services/{service_id}``."""
r = self.get('/services/%(service_id)s' % {
@@ -421,11 +411,6 @@ class CatalogTestCase(test_v3.RestfulTestCase):
r = self.get('/endpoints')
self.assertValidEndpointListResponse(r, ref=self.endpoint)

def test_list_endpoints_xml(self):
"""Call ``GET /endpoints`` (xml data)."""
r = self.get('/endpoints', content_type='xml')
self.assertValidEndpointListResponse(r, ref=self.endpoint)

def test_create_endpoint_no_enabled(self):
"""Call ``POST /endpoints``."""
ref = self.new_endpoint_ref(service_id=self.service_id)

+ 0
- 5
keystone/tests/test_v3_credential.py View File

@@ -85,11 +85,6 @@ class CredentialTestCase(CredentialBaseTestCase):
r = self.get('/credentials')
self.assertValidCredentialListResponse(r, ref=self.credential)

def test_list_credentials_xml(self):
"""Call ``GET /credentials`` (xml data)."""
r = self.get('/credentials', content_type='xml')
self.assertValidCredentialListResponse(r, ref=self.credential)

def test_list_credentials_filtered_by_user_id(self):
"""Call ``GET /credentials?user_id={user_id}``."""
credential = self.new_credential_ref(

+ 0
- 29
keystone/tests/test_v3_federation.py View File

@@ -26,7 +26,6 @@ import xmldsig

from keystone.auth import controllers as auth_controllers
from keystone.common import dependency
from keystone.common import serializer
from keystone import config
from keystone.contrib.federation import controllers as federation_controllers
from keystone.contrib.federation import idp as keystone_idp
@@ -841,17 +840,6 @@ class FederatedTokenTests(FederationTests):
'rules': rules or self.rules['rules']
}

def _assertSerializeToXML(self, json_body):
"""Serialize JSON body to XML.

Serialize JSON body to XML, then deserialize to JSON
again. Expect both JSON dictionaries to be equal.

"""
xml_body = serializer.to_xml(json_body)
json_deserialized = serializer.from_xml(xml_body)
self.assertDictEqual(json_deserialized, json_body)

def _scope_request(self, unscoped_token_id, scope, scope_id):
return {
'auth': {
@@ -932,23 +920,6 @@ class FederatedTokenTests(FederationTests):
r = self._issue_unscoped_token(environment={'REMOTE_USER': ''})
self.assertIsNotNone(r.headers.get('X-Subject-Token'))

def test_issue_unscoped_token_serialize_to_xml(self):
"""Issue unscoped token and serialize to XML.

Make sure common.serializer doesn't complain about
the response structure and tag names.

"""
r = self._issue_unscoped_token()
token_resp = r.json_body
# Remove 'extras' if empty or None,
# as JSON and XML (de)serializers treat
# them differently, making dictionaries
# comparisons fail.
if not token_resp['token'].get('extras'):
token_resp['token'].pop('extras')
self._assertSerializeToXML(token_resp)

def test_issue_unscoped_token_no_groups(self):
self.assertRaises(exception.Unauthorized,
self._issue_unscoped_token,

+ 0
- 9
keystone/tests/test_v3_filters.py View File

@@ -30,7 +30,6 @@ CONF = config.CONF
class IdentityTestFilteredCase(filtering.FilterTests,
test_v3.RestfulTestCase):
"""Test filter enforcement on the v3 Identity API."""
content_type = 'json'

def setUp(self):
"""Setup for Identity Filter Test Cases."""
@@ -435,11 +434,3 @@ class IdentityTestListLimitCase(IdentityTestFilteredCase):
r = self.get('/services', auth=self.auth)
self.assertEqual(len(r.result.get('services')), 10)
self.assertIsNone(r.result.get('truncated'))


class IdentityTestFilteredCaseXML(IdentityTestFilteredCase):
content_type = 'xml'


class IdentityTestListLimitCaseXML(IdentityTestListLimitCase):
content_type = 'xml'

+ 0
- 14
keystone/tests/test_v3_identity.py View File

@@ -177,13 +177,6 @@ class IdentityTestCase(test_v3.RestfulTestCase):
self.assertValidUserListResponse(r, ref=user,
resource_url=resource_url)

def test_list_users_xml(self):
"""Call ``GET /users`` (xml data)."""
resource_url = '/users'
r = self.get(resource_url, content_type='xml')
self.assertValidUserListResponse(r, ref=self.user,
resource_url=resource_url)

def test_get_user(self):
"""Call ``GET /users/{user_id}``."""
r = self.get('/users/%(user_id)s' % {
@@ -370,13 +363,6 @@ class IdentityTestCase(test_v3.RestfulTestCase):
self.assertValidGroupListResponse(r, ref=self.group,
resource_url=resource_url)

def test_list_groups_xml(self):
"""Call ``GET /groups`` (xml data)."""
resource_url = '/groups'
r = self.get(resource_url, content_type='xml')
self.assertValidGroupListResponse(r, ref=self.group,
resource_url=resource_url)

def test_get_group(self):
"""Call ``GET /groups/{group_id}``."""
r = self.get('/groups/%(group_id)s' % {

+ 0
- 5
keystone/tests/test_v3_policy.py View File

@@ -44,11 +44,6 @@ class PolicyTestCase(test_v3.RestfulTestCase):
r = self.get('/policies')
self.assertValidPolicyListResponse(r, ref=self.policy)

def test_list_policies_xml(self):
"""Call ``GET /policies (xml data)``."""
r = self.get('/policies', content_type='xml')
self.assertValidPolicyListResponse(r, ref=self.policy)

def test_get_policy(self):
"""Call ``GET /policies/{policy_id}``."""
r = self.get(

+ 11
- 195
keystone/tests/test_versions.py View File

@@ -25,7 +25,6 @@ from keystone.common import json_home
from keystone import config
from keystone import controllers
from keystone import tests
from keystone.tests import matchers


CONF = config.CONF
@@ -35,10 +34,6 @@ v2_MEDIA_TYPES = [
"base": "application/json",
"type": "application/"
"vnd.openstack.identity-v2.0+json"
}, {
"base": "application/xml",
"type": "application/"
"vnd.openstack.identity-v2.0+xml"
}
]

@@ -72,10 +67,6 @@ v3_MEDIA_TYPES = [
"base": "application/json",
"type": "application/"
"vnd.openstack.identity-v3+json"
}, {
"base": "application/xml",
"type": "application/"
"vnd.openstack.identity-v3+xml"
}
]

@@ -629,6 +620,17 @@ class VersionTestCase(tests.TestCase):
# If request some unknown mime-type, get JSON.
self.assertThat(make_request(self.getUniqueString()), JSON_MATCHER)

@mock.patch.object(controllers, '_VERSIONS', [])
def test_no_json_home_document_returned_when_v3_disabled(self):
json_home_document = controllers.request_v3_json_home('some_prefix')
expected_document = {'resources': {}}
self.assertEqual(expected_document, json_home_document)

def test_extension_property_method_returns_none(self):
extension_obj = controllers.Extensions()
extensions_property = extension_obj.extensions
self.assertIsNone(extensions_property)


class VersionSingleAppTestCase(tests.TestCase):
"""Tests running with a single application loaded.
@@ -713,189 +715,3 @@ class VersionInheritEnabledTestCase(tests.TestCase):

self.assertThat(jsonutils.loads(resp.body),
tt_matchers.Equals(exp_json_home_data))


class XmlVersionTestCase(tests.TestCase):

REQUEST_HEADERS = {'Accept': 'application/xml'}

DOC_INTRO = '<?xml version="1.0" encoding="UTF-8"?>'
XML_NAMESPACE_ATTR = 'xmlns="http://docs.openstack.org/identity/api/v2.0"'
XML_NAMESPACE_V3 = 'xmlns="http://docs.openstack.org/identity/api/v3"'

v2_VERSION_DATA = """
<version %(v2_namespace)s status="stable" updated="2014-04-17T00:00:00Z"
id="v2.0">
<media-types>
<media-type base="application/json" type="application/\
vnd.openstack.identity-v2.0+json"/>
<media-type base="application/xml" type="application/\
vnd.openstack.identity-v2.0+xml"/>
</media-types>
<links>
<link href="http://localhost:%%(port)s/v2.0/" rel="self"/>
<link href="http://docs.openstack.org/" type="text/html" \
rel="describedby"/>
</links>
<link href="http://localhost:%%(port)s/v2.0/" rel="self"/>
<link href="http://docs.openstack.org/" type="text/html" \
rel="describedby"/>
</version>
"""

v2_VERSION_RESPONSE = ((DOC_INTRO + v2_VERSION_DATA) %
dict(v2_namespace=XML_NAMESPACE_ATTR))

v3_VERSION_DATA = """
<version %(v3_namespace)s status="stable" updated="2013-03-06T00:00:00Z"
id="v3.0">
<media-types>
<media-type base="application/json" type="application/\
vnd.openstack.identity-v3+json"/>
<media-type base="application/xml" type="application/\
vnd.openstack.identity-v3+xml"/>
</media-types>
<links>
<link href="http://localhost:%%(port)s/v3/" rel="self"/>
</links>
</version>
"""

v3_VERSION_RESPONSE = ((DOC_INTRO + v3_VERSION_DATA) %
dict(v3_namespace=XML_NAMESPACE_V3))

VERSIONS_RESPONSE = ((DOC_INTRO + """
<versions %(namespace)s>
""" +
v3_VERSION_DATA +
v2_VERSION_DATA + """
</versions>
""") % dict(namespace=XML_NAMESPACE_ATTR, v3_namespace='', v2_namespace=''))

def setUp(self):
super(XmlVersionTestCase, self).setUp()
self.load_backends()
self.public_app = self.loadapp('keystone', 'main')
self.admin_app = self.loadapp('keystone', 'admin')

self.config_fixture.config(
public_endpoint='http://localhost:%(public_port)d',
admin_endpoint='http://localhost:%(admin_port)d')

def config_overrides(self):
super(XmlVersionTestCase, self).config_overrides()
port = random.randint(10000, 30000)
self.config_fixture.config(public_port=port, admin_port=port)

def test_public_versions(self):
client = self.client(self.public_app)
resp = client.get('/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 300)
data = resp.body
expected = self.VERSIONS_RESPONSE % dict(port=CONF.public_port)
self.assertThat(data, matchers.XMLEquals(expected))

def test_admin_versions(self):
client = self.client(self.admin_app)
resp = client.get('/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 300)
data = resp.body
expected = self.VERSIONS_RESPONSE % dict(port=CONF.admin_port)
self.assertThat(data, matchers.XMLEquals(expected))

def test_use_site_url_if_endpoint_unset(self):
client = self.client(self.public_app)
resp = client.get('/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 300)
data = resp.body
expected = self.VERSIONS_RESPONSE % dict(port=CONF.public_port)
self.assertThat(data, matchers.XMLEquals(expected))

def test_public_version_v2(self):
client = self.client(self.public_app)
resp = client.get('/v2.0/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 200)
data = resp.body
expected = self.v2_VERSION_RESPONSE % dict(port=CONF.public_port)
self.assertThat(data, matchers.XMLEquals(expected))

def test_admin_version_v2(self):
client = self.client(self.admin_app)
resp = client.get('/v2.0/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 200)
data = resp.body
expected = self.v2_VERSION_RESPONSE % dict(port=CONF.admin_port)
self.assertThat(data, matchers.XMLEquals(expected))

def test_public_version_v3(self):
client = self.client(self.public_app)
resp = client.get('/v3/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 200)
data = resp.body
expected = self.v3_VERSION_RESPONSE % dict(port=CONF.public_port)
self.assertThat(data, matchers.XMLEquals(expected))

def test_admin_version_v3(self):
client = self.client(self.public_app)
resp = client.get('/v3/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 200)
data = resp.body
expected = self.v3_VERSION_RESPONSE % dict(port=CONF.admin_port)
self.assertThat(data, matchers.XMLEquals(expected))

@mock.patch.object(controllers, '_VERSIONS', ['v3'])
def test_v2_disabled(self):
client = self.client(self.public_app)

# request to /v3 should pass
resp = client.get('/v3/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 200)
data = resp.body
expected = self.v3_VERSION_RESPONSE % dict(port=CONF.public_port)
self.assertThat(data, matchers.XMLEquals(expected))

# only v3 information should be displayed by requests to /
v3_only_response = ((self.DOC_INTRO + '<versions %(namespace)s>' +
self.v3_VERSION_DATA + '</versions>') %
dict(namespace=self.XML_NAMESPACE_ATTR,
v3_namespace='') %
dict(port=CONF.public_port))

resp = client.get('/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 300)
data = resp.body
self.assertThat(data, matchers.XMLEquals(v3_only_response))

@mock.patch.object(controllers, '_VERSIONS', ['v2.0'])
def test_v3_disabled(self):
client = self.client(self.public_app)

# request to /v2.0 should pass
resp = client.get('/v2.0/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 200)
data = resp.body
expected = self.v2_VERSION_RESPONSE % dict(port=CONF.public_port)
self.assertThat(data, matchers.XMLEquals(expected))

# only v2 information should be displayed by requests to /
v2_only_response = ((self.DOC_INTRO + '<versions %(namespace)s>' +
self.v2_VERSION_DATA + '</versions>') %
dict(namespace=self.XML_NAMESPACE_ATTR,
v2_namespace='') %
dict(port=CONF.public_port))

resp = client.get('/', headers=self.REQUEST_HEADERS)
self.assertEqual(resp.status_int, 300)
data = resp.body
self.assertThat(data, matchers.XMLEquals(v2_only_response))

@mock.patch.object(controllers, '_VERSIONS', [])
def test_no_json_home_document_returned_when_v3_disabled(self):
json_home_document = controllers.request_v3_json_home('some_prefix')
expected_document = {'resources': {}}
self.assertEqual(expected_document, json_home_document)

def test_extension_property_method_returns_none(self):
extension_obj = controllers.Extensions()
extensions_property = extension_obj.extensions
self.assertIsNone(extensions_property)

+ 0
- 5
keystone/token/controllers.py View File

@@ -314,11 +314,6 @@ class Auth(controller.V2Controller):
if not environment.get('REMOTE_USER'):
raise ExternalAuthNotApplicable()

# NOTE(jamielennox): xml and json differ and get confused about what
# empty auth should look like so just reset it.
if not auth:
auth = {}

username = environment['REMOTE_USER']
try:
user_ref = self.identity_api.get_user_by_name(

Loading…
Cancel
Save