From ca437bd223d3d14a31952aa624adc971a21e816b Mon Sep 17 00:00:00 2001 From: Yan Xing'an Date: Fri, 23 Nov 2018 00:09:13 +0800 Subject: [PATCH] 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 --- tacker/tests/common/__init__.py | 0 tacker/tests/common/helpers.py | 26 ++++++ tacker/tests/unit/test_wsgi.py | 153 ++++++++++++++------------------ 3 files changed, 91 insertions(+), 88 deletions(-) create mode 100644 tacker/tests/common/__init__.py create mode 100644 tacker/tests/common/helpers.py diff --git a/tacker/tests/common/__init__.py b/tacker/tests/common/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tacker/tests/common/helpers.py b/tacker/tests/common/helpers.py new file mode 100644 index 000000000..2da373c88 --- /dev/null +++ b/tacker/tests/common/helpers.py @@ -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) diff --git a/tacker/tests/unit/test_wsgi.py b/tacker/tests/unit/test_wsgi.py index 36441cb82..931617501 100644 --- a/tacker/tests/unit/test_wsgi.py +++ b/tacker/tests/unit/test_wsgi.py @@ -26,6 +26,7 @@ import webob.exc from tacker.common import exceptions as exception from tacker.tests import base +from tacker.tests.common import helpers from tacker import wsgi CONF = cfg.CONF @@ -182,9 +183,10 @@ class RequestDeserializerTest(testtools.TestCase): class JSONDeserializer(object): def deserialize(self, data, action='default'): - return b'pew_json' + return 'pew_json' self.body_deserializers = {'application/json': JSONDeserializer()} + self.deserializer = wsgi.RequestDeserializer(self.body_deserializers) def test_get_deserializer(self): @@ -201,9 +203,8 @@ class RequestDeserializerTest(testtools.TestCase): request = wsgi.Request.blank('/') request.headers['Accept'] = 'application/json' - self.assertEqual( - 'application/json', - self.deserializer.get_expected_content_type(request)) + self.assertEqual('application/json', + self.deserializer.get_expected_content_type(request)) def test_get_action_args(self): """Test RequestDeserializer.get_action_args.""" @@ -215,8 +216,8 @@ class RequestDeserializerTest(testtools.TestCase): 'id': 12}]} expected = {'action': 'update', 'id': 12} - self.assertEqual( - expected, self.deserializer.get_action_args(env)) + self.assertEqual(expected, + self.deserializer.get_action_args(env)) def test_deserialize(self): """Test RequestDeserializer.deserialize.""" @@ -272,9 +273,8 @@ class ResponseSerializerTest(testtools.TestCase): def test_get_serializer(self): """Test ResponseSerializer.get_body_serializer.""" content_type = 'application/json' - self.assertEqual( - self.serializer.get_body_serializer(content_type), - self.body_serializers[content_type]) + self.assertEqual(self.body_serializers[content_type], + self.serializer.get_body_serializer(content_type)) def test_serialize_json_response(self): response = self.serializer.serialize({}, 'application/json') @@ -374,7 +374,7 @@ class RequestTest(base.BaseTestCase): request.headers["Accept"] = "application/new_type" result = request.best_match_content_type() - self.assertEqual('application/json', result) + self.assertEqual("application/json", result) def test_best_match_language(self): # Test that we are actually invoking language negotiation by webop @@ -405,9 +405,8 @@ class ActionDispatcherTest(base.BaseTestCase): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: x - self.assertEqual( - 'pants', - serializer.dispatch('pants', action='create')) + self.assertEqual('pants', + serializer.dispatch('pants', action='create')) def test_dispatch_action_None(self): """Test ActionDispatcher.dispatch with none action.""" @@ -415,18 +414,16 @@ class ActionDispatcherTest(base.BaseTestCase): serializer.create = lambda x: x + ' pants' serializer.default = lambda x: x + ' trousers' - self.assertEqual( - 'Two trousers', - serializer.dispatch('Two', action=None)) + self.assertEqual('Two trousers', + serializer.dispatch('Two', action=None)) def test_dispatch_default(self): serializer = wsgi.ActionDispatcher() serializer.create = lambda x: x + ' pants' serializer.default = lambda x: x + ' trousers' - self.assertEqual( - 'Two trousers', - serializer.dispatch('Two', action='update')) + self.assertEqual('Two trousers', + serializer.dispatch('Two', action='update')) class ResponseHeadersSerializerTest(base.BaseTestCase): @@ -454,8 +451,8 @@ class DictSerializerTest(base.BaseTestCase): def test_dispatch_default(self): serializer = wsgi.DictSerializer() - self.assertEqual( - '', serializer.serialize({}, 'NonExistentAction')) + self.assertEqual('', + serializer.serialize({}, 'NonExistentAction')) class JSONDictSerializerTest(base.BaseTestCase): @@ -469,27 +466,34 @@ class JSONDictSerializerTest(base.BaseTestCase): 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): - data = b'{"a": "\xe7\xbd\x91\xe7\xbb\x9c"}' - as_dict = {'body': {'a': u'\u7f51\u7edc'}} - deserializer = wsgi.JSONDeserializer() - self.assertEqual(as_dict, - deserializer.deserialize(data)) + input_dict = dict(servers=dict(a=(2, '\xe7\xbd\x91\xe7\xbb\x9c'))) + expected_json = b'{"servers":{"a":[2,"\\u7f51\\u7edc"]}}' + serializer = wsgi.JSONDictSerializer() + result = serializer.serialize(input_dict) + result = result.replace(b'\n', b'').replace(b' ', b'') + + self.assertEqual(expected_json, result) def test_json_with_unicode(self): - data = b'{"a": "\u7f51\u7edc"}' - as_dict = {'body': {'a': u'\u7f51\u7edc'}} - deserializer = wsgi.JSONDeserializer() - self.assertEqual(as_dict, - deserializer.deserialize(data)) + input_dict = dict(servers=dict(a=(2, u'\u7f51\u7edc'))) + expected_json = b'{"servers":{"a":[2,"\\u7f51\\u7edc"]}}' + serializer = wsgi.JSONDictSerializer() + result = serializer.serialize(input_dict) + result = result.replace(b'\n', b'').replace(b' ', b'') + + self.assertEqual(expected_json, result) class TextDeserializerTest(base.BaseTestCase): def test_dispatch_default(self): deserializer = wsgi.TextDeserializer() - self.assertEqual( - {}, deserializer.deserialize({}, 'update')) + self.assertEqual({}, + deserializer.deserialize({}, 'update')) class JSONDeserializerTest(base.BaseTestCase): @@ -509,8 +513,8 @@ class JSONDeserializerTest(base.BaseTestCase): 'd': {'e': '1'}, 'f': '1'}}} deserializer = wsgi.JSONDeserializer() - self.assertEqual( - as_dict, deserializer.deserialize(data)) + self.assertEqual(as_dict, + deserializer.deserialize(data)) def test_default_raise_Malformed_Exception(self): """Test JsonDeserializer.default. @@ -545,8 +549,8 @@ class RequestHeadersDeserializerTest(base.BaseTestCase): deserializer = wsgi.RequestHeadersDeserializer() req = wsgi.Request.blank('/') - self.assertEqual( - {}, deserializer.deserialize(req, 'nonExistent')) + self.assertEqual({}, + deserializer.deserialize(req, 'nonExistent')) def test_custom(self): class Deserializer(wsgi.RequestHeadersDeserializer): @@ -555,20 +559,23 @@ class RequestHeadersDeserializerTest(base.BaseTestCase): deserializer = Deserializer() req = wsgi.Request.blank('/') req.headers['X-Custom-Header'] = 'b' - self.assertEqual( - {'a': 'b'}, deserializer.deserialize(req, 'update')) + self.assertEqual({'a': 'b'}, + deserializer.deserialize(req, 'update')) 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' - 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( resource.controller, 'index', action_args={'index': 'off'}) expected = 'off' @@ -576,23 +583,14 @@ class ResourceTest(base.BaseTestCase): self.assertEqual(expected, actual) def test_dispatch_unknown_controller_action(self): - class Controller(object): - def index(self, request, pants=None): - return pants - - def my_fault_body_function(): - return b'off' - - resource = wsgi.Resource(Controller(), my_fault_body_function) + resource = wsgi.Resource(self.Controller(), + self.my_fault_body_function) self.assertRaises( AttributeError, resource.dispatch, resource.controller, 'create', {}) def test_malformed_request_body_throws_bad_request(self): - def my_fault_body_function(): - return b'off' - - resource = wsgi.Resource(None, my_fault_body_function) + resource = wsgi.Resource(None, self.my_fault_body_function) request = wsgi.Request.blank( "/", body=b"{mal:formed", method='POST', headers={'Content-Type': "application/json"}) @@ -601,9 +599,7 @@ class ResourceTest(base.BaseTestCase): self.assertEqual(400, response.status_int) def test_wrong_content_type_throws_unsupported_media_type_error(self): - def my_fault_body_function(): - return b'off' - resource = wsgi.Resource(None, my_fault_body_function) + resource = wsgi.Resource(None, self.my_fault_body_function) request = wsgi.Request.blank( "/", body=b"{some:json}", method='POST', headers={'Content-Type': "xxx"}) @@ -612,9 +608,7 @@ class ResourceTest(base.BaseTestCase): self.assertEqual(400, response.status_int) def test_wrong_content_type_server_error(self): - def my_fault_body_function(): - return b'off' - resource = wsgi.Resource(None, my_fault_body_function) + resource = wsgi.Resource(None, self.my_fault_body_function) request = wsgi.Request.blank( "/", method='POST', headers={'Content-Type': "unknow"}) @@ -622,18 +616,11 @@ class ResourceTest(base.BaseTestCase): self.assertEqual(500, response.status_int) 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): def __init__(self): self.url = 'http://where.no' self.environ = 'environ' - self.body = b'body' + self.body = 'body' def method(self): pass @@ -641,19 +628,15 @@ class ResourceTest(base.BaseTestCase): def best_match_content_type(self): 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() result = resource(request) self.assertEqual(415, result.status_int) def test_type_error(self): - class Controller(object): - def index(self, request, index=None): - return index - - def my_fault_body_function(): - return b'off' - resource = wsgi.Resource(Controller(), my_fault_body_function) + resource = wsgi.Resource(self.Controller(), + self.my_fault_body_function) request = wsgi.Request.blank( "/", method='POST', headers={'Content-Type': "json"}) @@ -662,18 +645,11 @@ class ResourceTest(base.BaseTestCase): self.assertEqual(400, response.status_int) 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): def __init__(self): self.url = 'http://where.no' self.environ = 'environ' - self.body = b'{"Content-Type": "json"}' + self.body = '{"Content-Type": "json"}' def method(self): pass @@ -681,7 +657,8 @@ class ResourceTest(base.BaseTestCase): def best_match_content_type(self): return 'application/json' - resource = wsgi.Resource(Controller(), my_fault_body_function) + resource = wsgi.Resource(self.Controller(), + self.my_fault_body_function) request = FakeRequest() result = resource(request) self.assertEqual(500, result.status_int)