openstack
/
tacker
Code Issues Proposed changes

Improve UT test_wsgi, reduce code duplication 

Add helpers for cases that require PY2 or PY3.
Reduce code duplication in test_wsgi.

Partially Implements: blueprint test-addition-refactoring

Change-Id: Ibd30063ca1a46c7ea5686734835a030386ba06e5
This commit is contained in:
Yan Xing'an 2018-11-23 00:09:13 +08:00
parent 92ee4eddcd
commit ca437bd223
3 changed files with 91 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
tacker/tests/common/__init__.py Normal file
View File

26
tacker/tests/common/helpers.py Normal file
View File

@ -0,0 +1,26 @@
# Copyright 2015 Red Hat, Inc.
# Copyright (c) 2014-2018 China Mobile (SuZhou) Software Technology Co.,Ltd.
# 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 six
import testtools
def requires_py2(testcase):
return testtools.skipUnless(six.PY2, "requires python 2.x")(testcase)
def requires_py3(testcase):
return testtools.skipUnless(six.PY3, "requires python 3.x")(testcase)

153
tacker/tests/unit/test_wsgi.py
View File

@ -26,6 +26,7 @@ import webob.exc
from tacker.common import exceptions as exception from tacker.common import exceptions as exception
from tacker.tests import base from tacker.tests import base
from tacker.tests.common import helpers
from tacker import wsgi from tacker import wsgi
CONF = cfg.CONF CONF = cfg.CONF
@ -182,9 +183,10 @@ class RequestDeserializerTest(testtools.TestCase):
class JSONDeserializer(object): class JSONDeserializer(object):
def deserialize(self, data, action='default'): def deserialize(self, data, action='default'):
return b'pew_json' return 'pew_json'
self.body_deserializers = {'application/json': JSONDeserializer()} self.body_deserializers = {'application/json': JSONDeserializer()}
self.deserializer = wsgi.RequestDeserializer(self.body_deserializers) self.deserializer = wsgi.RequestDeserializer(self.body_deserializers)
def test_get_deserializer(self): def test_get_deserializer(self):
@ -201,9 +203,8 @@ class RequestDeserializerTest(testtools.TestCase):
request = wsgi.Request.blank('/') request = wsgi.Request.blank('/')
request.headers['Accept'] = 'application/json' request.headers['Accept'] = 'application/json'
self.assertEqual( self.assertEqual('application/json',
'application/json', self.deserializer.get_expected_content_type(request))
self.deserializer.get_expected_content_type(request))
def test_get_action_args(self): def test_get_action_args(self):
"""Test RequestDeserializer.get_action_args.""" """Test RequestDeserializer.get_action_args."""
@ -215,8 +216,8 @@ class RequestDeserializerTest(testtools.TestCase):
'id': 12}]} 'id': 12}]}
expected = {'action': 'update', 'id': 12} expected = {'action': 'update', 'id': 12}
self.assertEqual( self.assertEqual(expected,
expected, self.deserializer.get_action_args(env)) self.deserializer.get_action_args(env))
def test_deserialize(self): def test_deserialize(self):
"""Test RequestDeserializer.deserialize.""" """Test RequestDeserializer.deserialize."""
@ -272,9 +273,8 @@ class ResponseSerializerTest(testtools.TestCase):
def test_get_serializer(self): def test_get_serializer(self):
"""Test ResponseSerializer.get_body_serializer.""" """Test ResponseSerializer.get_body_serializer."""
content_type = 'application/json' content_type = 'application/json'
self.assertEqual( self.assertEqual(self.body_serializers[content_type],
self.serializer.get_body_serializer(content_type), self.serializer.get_body_serializer(content_type))
self.body_serializers[content_type])
def test_serialize_json_response(self): def test_serialize_json_response(self):
response = self.serializer.serialize({}, 'application/json') response = self.serializer.serialize({}, 'application/json')
@ -374,7 +374,7 @@ class RequestTest(base.BaseTestCase):
request.headers["Accept"] = "application/new_type" request.headers["Accept"] = "application/new_type"
result = request.best_match_content_type() result = request.best_match_content_type()
self.assertEqual('application/json', result) self.assertEqual("application/json", result)
def test_best_match_language(self): def test_best_match_language(self):
# Test that we are actually invoking language negotiation by webop # Test that we are actually invoking language negotiation by webop
@ -405,9 +405,8 @@ class ActionDispatcherTest(base.BaseTestCase):
serializer = wsgi.ActionDispatcher() serializer = wsgi.ActionDispatcher()
serializer.create = lambda x: x serializer.create = lambda x: x
self.assertEqual( self.assertEqual('pants',
'pants', serializer.dispatch('pants', action='create'))
serializer.dispatch('pants', action='create'))
def test_dispatch_action_None(self): def test_dispatch_action_None(self):
"""Test ActionDispatcher.dispatch with none action.""" """Test ActionDispatcher.dispatch with none action."""
@ -415,18 +414,16 @@ class ActionDispatcherTest(base.BaseTestCase):
serializer.create = lambda x: x + ' pants' serializer.create = lambda x: x + ' pants'
serializer.default = lambda x: x + ' trousers' serializer.default = lambda x: x + ' trousers'
self.assertEqual( self.assertEqual('Two trousers',
'Two trousers', serializer.dispatch('Two', action=None))
serializer.dispatch('Two', action=None))
def test_dispatch_default(self): def test_dispatch_default(self):
serializer = wsgi.ActionDispatcher() serializer = wsgi.ActionDispatcher()
serializer.create = lambda x: x + ' pants' serializer.create = lambda x: x + ' pants'
serializer.default = lambda x: x + ' trousers' serializer.default = lambda x: x + ' trousers'
self.assertEqual( self.assertEqual('Two trousers',
'Two trousers', serializer.dispatch('Two', action='update'))
serializer.dispatch('Two', action='update'))
class ResponseHeadersSerializerTest(base.BaseTestCase): class ResponseHeadersSerializerTest(base.BaseTestCase):
@ -454,8 +451,8 @@ class DictSerializerTest(base.BaseTestCase):
def test_dispatch_default(self): def test_dispatch_default(self):
serializer = wsgi.DictSerializer() serializer = wsgi.DictSerializer()
self.assertEqual( self.assertEqual('',
'', serializer.serialize({}, 'NonExistentAction')) serializer.serialize({}, 'NonExistentAction'))
class JSONDictSerializerTest(base.BaseTestCase): class JSONDictSerializerTest(base.BaseTestCase):
@ -469,27 +466,34 @@ class JSONDictSerializerTest(base.BaseTestCase):
self.assertEqual(expected_json, result) self.assertEqual(expected_json, result)
# The tested behaviour is only meant to be witnessed in Python 2, so it is
# OK to skip this test with Python 3.
@helpers.requires_py2
def test_json_with_utf8(self): def test_json_with_utf8(self):
data = b'{"a": "\xe7\xbd\x91\xe7\xbb\x9c"}' input_dict = dict(servers=dict(a=(2, '\xe7\xbd\x91\xe7\xbb\x9c')))
as_dict = {'body': {'a': u'\u7f51\u7edc'}} expected_json = b'{"servers":{"a":[2,"\\u7f51\\u7edc"]}}'
deserializer = wsgi.JSONDeserializer() serializer = wsgi.JSONDictSerializer()
self.assertEqual(as_dict, result = serializer.serialize(input_dict)
deserializer.deserialize(data)) result = result.replace(b'\n', b'').replace(b' ', b'')
self.assertEqual(expected_json, result)
def test_json_with_unicode(self): def test_json_with_unicode(self):
data = b'{"a": "\u7f51\u7edc"}' input_dict = dict(servers=dict(a=(2, u'\u7f51\u7edc')))
as_dict = {'body': {'a': u'\u7f51\u7edc'}} expected_json = b'{"servers":{"a":[2,"\\u7f51\\u7edc"]}}'
deserializer = wsgi.JSONDeserializer() serializer = wsgi.JSONDictSerializer()
self.assertEqual(as_dict, result = serializer.serialize(input_dict)
deserializer.deserialize(data)) result = result.replace(b'\n', b'').replace(b' ', b'')
self.assertEqual(expected_json, result)
class TextDeserializerTest(base.BaseTestCase): class TextDeserializerTest(base.BaseTestCase):
def test_dispatch_default(self): def test_dispatch_default(self):
deserializer = wsgi.TextDeserializer() deserializer = wsgi.TextDeserializer()
self.assertEqual( self.assertEqual({},
{}, deserializer.deserialize({}, 'update')) deserializer.deserialize({}, 'update'))
class JSONDeserializerTest(base.BaseTestCase): class JSONDeserializerTest(base.BaseTestCase):
@ -509,8 +513,8 @@ class JSONDeserializerTest(base.BaseTestCase):
'd': {'e': '1'}, 'd': {'e': '1'},
'f': '1'}}} 'f': '1'}}}
deserializer = wsgi.JSONDeserializer() deserializer = wsgi.JSONDeserializer()
self.assertEqual( self.assertEqual(as_dict,
as_dict, deserializer.deserialize(data)) deserializer.deserialize(data))
def test_default_raise_Malformed_Exception(self): def test_default_raise_Malformed_Exception(self):
"""Test JsonDeserializer.default. """Test JsonDeserializer.default.
@ -545,8 +549,8 @@ class RequestHeadersDeserializerTest(base.BaseTestCase):
deserializer = wsgi.RequestHeadersDeserializer() deserializer = wsgi.RequestHeadersDeserializer()
req = wsgi.Request.blank('/') req = wsgi.Request.blank('/')
self.assertEqual( self.assertEqual({},
{}, deserializer.deserialize(req, 'nonExistent')) deserializer.deserialize(req, 'nonExistent'))
def test_custom(self): def test_custom(self):
class Deserializer(wsgi.RequestHeadersDeserializer): class Deserializer(wsgi.RequestHeadersDeserializer):
@ -555,20 +559,23 @@ class RequestHeadersDeserializerTest(base.BaseTestCase):
deserializer = Deserializer() deserializer = Deserializer()
req = wsgi.Request.blank('/') req = wsgi.Request.blank('/')
req.headers['X-Custom-Header'] = 'b' req.headers['X-Custom-Header'] = 'b'
self.assertEqual( self.assertEqual({'a': 'b'},
{'a': 'b'}, deserializer.deserialize(req, 'update')) deserializer.deserialize(req, 'update'))
class ResourceTest(base.BaseTestCase): class ResourceTest(base.BaseTestCase):
def test_dispatch(self):
class Controller(object):
def index(self, request, index=None):
return index
def my_fault_body_function(): @staticmethod
def my_fault_body_function():
return 'off' return 'off'
resource = wsgi.Resource(Controller(), my_fault_body_function) class Controller(object):
def index(self, request, index=None):
return index
def test_dispatch(self):
resource = wsgi.Resource(self.Controller(),
self.my_fault_body_function)
actual = resource.dispatch( actual = resource.dispatch(
resource.controller, 'index', action_args={'index': 'off'}) resource.controller, 'index', action_args={'index': 'off'})
expected = 'off' expected = 'off'
@ -576,23 +583,14 @@ class ResourceTest(base.BaseTestCase):
self.assertEqual(expected, actual) self.assertEqual(expected, actual)
def test_dispatch_unknown_controller_action(self): def test_dispatch_unknown_controller_action(self):
class Controller(object): resource = wsgi.Resource(self.Controller(),
def index(self, request, pants=None): self.my_fault_body_function)
return pants
def my_fault_body_function():
return b'off'
resource = wsgi.Resource(Controller(), my_fault_body_function)
self.assertRaises( self.assertRaises(
AttributeError, resource.dispatch, AttributeError, resource.dispatch,
resource.controller, 'create', {}) resource.controller, 'create', {})
def test_malformed_request_body_throws_bad_request(self): def test_malformed_request_body_throws_bad_request(self):
def my_fault_body_function(): resource = wsgi.Resource(None, self.my_fault_body_function)
return b'off'
resource = wsgi.Resource(None, my_fault_body_function)
request = wsgi.Request.blank( request = wsgi.Request.blank(
"/", body=b"{mal:formed", method='POST', "/", body=b"{mal:formed", method='POST',
headers={'Content-Type': "application/json"}) headers={'Content-Type': "application/json"})
@ -601,9 +599,7 @@ class ResourceTest(base.BaseTestCase):
self.assertEqual(400, response.status_int) self.assertEqual(400, response.status_int)
def test_wrong_content_type_throws_unsupported_media_type_error(self): def test_wrong_content_type_throws_unsupported_media_type_error(self):
def my_fault_body_function(): resource = wsgi.Resource(None, self.my_fault_body_function)
return b'off'
resource = wsgi.Resource(None, my_fault_body_function)
request = wsgi.Request.blank( request = wsgi.Request.blank(
"/", body=b"{some:json}", method='POST', "/", body=b"{some:json}", method='POST',
headers={'Content-Type': "xxx"}) headers={'Content-Type': "xxx"})
@ -612,9 +608,7 @@ class ResourceTest(base.BaseTestCase):
self.assertEqual(400, response.status_int) self.assertEqual(400, response.status_int)
def test_wrong_content_type_server_error(self): def test_wrong_content_type_server_error(self):
def my_fault_body_function(): resource = wsgi.Resource(None, self.my_fault_body_function)
return b'off'
resource = wsgi.Resource(None, my_fault_body_function)
request = wsgi.Request.blank( request = wsgi.Request.blank(
"/", method='POST', headers={'Content-Type': "unknow"}) "/", method='POST', headers={'Content-Type': "unknow"})
@ -622,18 +616,11 @@ class ResourceTest(base.BaseTestCase):
self.assertEqual(500, response.status_int) self.assertEqual(500, response.status_int)
def test_call_resource_class_bad_request(self): def test_call_resource_class_bad_request(self):
class Controller(object):
def index(self, request, index=None):
return index
def my_fault_body_function():
return b'off'
class FakeRequest(object): class FakeRequest(object):
def __init__(self): def __init__(self):
self.url = 'http://where.no' self.url = 'http://where.no'
self.environ = 'environ' self.environ = 'environ'
self.body = b'body' self.body = 'body'
def method(self): def method(self):
pass pass
@ -641,19 +628,15 @@ class ResourceTest(base.BaseTestCase):
def best_match_content_type(self): def best_match_content_type(self):
return 'best_match_content_type' return 'best_match_content_type'
resource = wsgi.Resource(Controller(), my_fault_body_function) resource = wsgi.Resource(self.Controller(),
self.my_fault_body_function)
request = FakeRequest() request = FakeRequest()
result = resource(request) result = resource(request)
self.assertEqual(415, result.status_int) self.assertEqual(415, result.status_int)
def test_type_error(self): def test_type_error(self):
class Controller(object): resource = wsgi.Resource(self.Controller(),
def index(self, request, index=None): self.my_fault_body_function)
return index
def my_fault_body_function():
return b'off'
resource = wsgi.Resource(Controller(), my_fault_body_function)
request = wsgi.Request.blank( request = wsgi.Request.blank(
"/", method='POST', headers={'Content-Type': "json"}) "/", method='POST', headers={'Content-Type': "json"})
@ -662,18 +645,11 @@ class ResourceTest(base.BaseTestCase):
self.assertEqual(400, response.status_int) self.assertEqual(400, response.status_int)
def test_call_resource_class_internal_error(self): def test_call_resource_class_internal_error(self):
class Controller(object):
def index(self, request, index=None):
return index
def my_fault_body_function():
return b'off'
class FakeRequest(object): class FakeRequest(object):
def __init__(self): def __init__(self):
self.url = 'http://where.no' self.url = 'http://where.no'
self.environ = 'environ' self.environ = 'environ'
self.body = b'{"Content-Type": "json"}' self.body = '{"Content-Type": "json"}'
def method(self): def method(self):
pass pass
@ -681,7 +657,8 @@ class ResourceTest(base.BaseTestCase):
def best_match_content_type(self): def best_match_content_type(self):
return 'application/json' return 'application/json'
resource = wsgi.Resource(Controller(), my_fault_body_function) resource = wsgi.Resource(self.Controller(),
self.my_fault_body_function)
request = FakeRequest() request = FakeRequest()
result = resource(request) result = resource(request)
self.assertEqual(500, result.status_int) self.assertEqual(500, result.status_int)