¡@
OpenStack Study: test_wsgi.py**** CubicPower OpenStack Study ****
# Copyright 2011 United States Government as represented by the # Administrator of the National Aeronautics and Space Administration. # 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. """Unit tests for `cinder.wsgi`.""" import mock import os.path import tempfile import urllib2 from oslo.config import cfg import testtools import webob import webob.dec from cinder import exception from cinder.openstack.common import gettextutils from cinder import test import cinder.wsgi CONF = cfg.CONF TEST_VAR_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), 'var'))
**** CubicPower OpenStack Study **** class TestLoaderNothingExists(test.TestCase): """Loader tests where os.path.exists always returns False."""
**** CubicPower OpenStack Study **** def setUp(self): super(TestLoaderNothingExists, self).setUp() self.stubs.Set(os.path, 'exists', lambda _: False)
**** CubicPower OpenStack Study **** def test_config_not_found(self): self.assertRaises( cinder.exception.ConfigNotFound, cinder.wsgi.Loader, )
**** CubicPower OpenStack Study **** class TestLoaderNormalFilesystem(test.TestCase): """Loader tests with normal filesystem (unmodified os.path module).""" _paste_config = """ [app:test_app] use = egg:Paste#static document_root = /tmp """
**** CubicPower OpenStack Study **** def setUp(self): super(TestLoaderNormalFilesystem, self).setUp() self.config = tempfile.NamedTemporaryFile(mode="w+t") self.config.write(self._paste_config.lstrip()) self.config.seek(0) self.config.flush() self.loader = cinder.wsgi.Loader(self.config.name) self.addCleanup(self.config.close)
**** CubicPower OpenStack Study **** def test_config_found(self): self.assertEqual(self.config.name, self.loader.config_path)
**** CubicPower OpenStack Study **** def test_app_not_found(self): self.assertRaises( cinder.exception.PasteAppNotFound, self.loader.load_app, "non-existent app", )
**** CubicPower OpenStack Study **** def test_app_found(self): url_parser = self.loader.load_app("test_app") self.assertEqual("/tmp", url_parser.directory)
**** CubicPower OpenStack Study **** class TestWSGIServer(test.TestCase): """WSGI server tests."""
**** CubicPower OpenStack Study **** def _ipv6_configured(): try: with file('/proc/net/if_inet6') as f: return len(f.read()) > 0 except IOError: return False
**** CubicPower OpenStack Study **** def test_no_app(self): server = cinder.wsgi.Server("test_app", None) self.assertEqual("test_app", server.name)
**** CubicPower OpenStack Study **** def test_start_random_port(self): server = cinder.wsgi.Server("test_random_port", None, host="127.0.0.1") self.assertEqual(0, server.port) server.start() self.assertNotEqual(0, server.port) server.stop() server.wait() @testtools.skipIf(not _ipv6_configured(), "Test requires an IPV6 configured interface")
**** CubicPower OpenStack Study **** def test_start_random_port_with_ipv6(self): server = cinder.wsgi.Server("test_random_port", None, host="::1") server.start() self.assertEqual("::1", server.host) self.assertNotEqual(0, server.port) server.stop() server.wait()
**** CubicPower OpenStack Study **** def test_app(self): greetings = 'Hello, World!!!' def hello_world(env, start_response): if env['PATH_INFO'] != '/': start_response('404 Not Found', [('Content-Type', 'text/plain')]) return ['Not Found\r\n'] start_response('200 OK', [('Content-Type', 'text/plain')]) return [greetings] server = cinder.wsgi.Server("test_app", hello_world) server.start() response = urllib2.urlopen('http://127.0.0.1:%d/' % server.port) self.assertEqual(greetings, response.read()) server.stop()
**** CubicPower OpenStack Study **** def hello_world(env, start_response): if env['PATH_INFO'] != '/': start_response('404 Not Found', [('Content-Type', 'text/plain')]) return ['Not Found\r\n'] start_response('200 OK', [('Content-Type', 'text/plain')]) return [greetings] server = cinder.wsgi.Server("test_app", hello_world) server.start() response = urllib2.urlopen('http://127.0.0.1:%d/' % server.port) self.assertEqual(greetings, response.read()) server.stop()
**** CubicPower OpenStack Study **** def test_app_using_ssl(self): CONF.set_default("ssl_cert_file", os.path.join(TEST_VAR_DIR, 'certificate.crt')) CONF.set_default("ssl_key_file", os.path.join(TEST_VAR_DIR, 'privatekey.key')) greetings = 'Hello, World!!!' @webob.dec.wsgify def hello_world(req): return greetings server = cinder.wsgi.Server("test_app", hello_world) server.start() response = urllib2.urlopen('https://127.0.0.1:%d/' % server.port) self.assertEqual(greetings, response.read()) server.stop() @testtools.skipIf(not _ipv6_configured(), "Test requires an IPV6 configured interface")
**** CubicPower OpenStack Study **** def hello_world(req): return greetings server = cinder.wsgi.Server("test_app", hello_world) server.start() response = urllib2.urlopen('https://127.0.0.1:%d/' % server.port) self.assertEqual(greetings, response.read()) server.stop() @testtools.skipIf(not _ipv6_configured(), "Test requires an IPV6 configured interface")
**** CubicPower OpenStack Study **** def test_app_using_ipv6_and_ssl(self): CONF.set_default("ssl_cert_file", os.path.join(TEST_VAR_DIR, 'certificate.crt')) CONF.set_default("ssl_key_file", os.path.join(TEST_VAR_DIR, 'privatekey.key')) greetings = 'Hello, World!!!' @webob.dec.wsgify def hello_world(req): return greetings server = cinder.wsgi.Server("test_app", hello_world, host="::1", port=0) server.start() response = urllib2.urlopen('https://[::1]:%d/' % server.port) self.assertEqual(greetings, response.read()) server.stop()
**** CubicPower OpenStack Study **** def hello_world(req): return greetings server = cinder.wsgi.Server("test_app", hello_world, host="::1", port=0) server.start() response = urllib2.urlopen('https://[::1]:%d/' % server.port) self.assertEqual(greetings, response.read()) server.stop()
**** CubicPower OpenStack Study **** class ExceptionTest(test.TestCase):
**** CubicPower OpenStack Study **** def _wsgi_app(self, inner_app): # NOTE(luisg): In order to test localization, we need to # make sure the lazy _() is installed in the 'fault' module # also we don't want to install the _() system-wide and # potentially break other test cases, so we do it here for this # test suite only. gettextutils.install('', lazy=True) from cinder.api.middleware import fault return fault.FaultWrapper(inner_app)
**** CubicPower OpenStack Study **** def _do_test_exception_safety_reflected_in_faults(self, expose): class ExceptionWithSafety(exception.CinderException): safe = expose @webob.dec.wsgify def fail(req): raise ExceptionWithSafety('some explanation') api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertIn('{"computeFault', resp.body) expected = ('ExceptionWithSafety: some explanation' if expose else 'The server has either erred or is incapable ' 'of performing the requested operation.') self.assertIn(expected, resp.body) self.assertEqual(resp.status_int, 500, resp.body)
**** CubicPower OpenStack Study **** def fail(req): raise ExceptionWithSafety('some explanation') api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertIn('{"computeFault', resp.body) expected = ('ExceptionWithSafety: some explanation' if expose else 'The server has either erred or is incapable ' 'of performing the requested operation.') self.assertIn(expected, resp.body) self.assertEqual(resp.status_int, 500, resp.body)
**** CubicPower OpenStack Study **** def test_safe_exceptions_are_described_in_faults(self): self._do_test_exception_safety_reflected_in_faults(True)
**** CubicPower OpenStack Study **** def test_unsafe_exceptions_are_not_described_in_faults(self): self._do_test_exception_safety_reflected_in_faults(False)
**** CubicPower OpenStack Study **** def _do_test_exception_mapping(self, exception_type, msg): @webob.dec.wsgify def fail(req): raise exception_type(msg) api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertIn(msg, resp.body) self.assertEqual(resp.status_int, exception_type.code, resp.body) if hasattr(exception_type, 'headers'): for (key, value) in exception_type.headers.iteritems(): self.assertIn(key, resp.headers) self.assertEqual(resp.headers[key], value)
**** CubicPower OpenStack Study **** def fail(req): raise exception_type(msg) api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertIn(msg, resp.body) self.assertEqual(resp.status_int, exception_type.code, resp.body) if hasattr(exception_type, 'headers'): for (key, value) in exception_type.headers.iteritems(): self.assertIn(key, resp.headers) self.assertEqual(resp.headers[key], value)
**** CubicPower OpenStack Study **** def test_quota_error_mapping(self): self._do_test_exception_mapping(exception.QuotaError, 'too many used')
**** CubicPower OpenStack Study **** def test_non_cinder_notfound_exception_mapping(self): class ExceptionWithCode(Exception): code = 404 self._do_test_exception_mapping(ExceptionWithCode, 'NotFound')
**** CubicPower OpenStack Study **** def test_non_cinder_exception_mapping(self): class ExceptionWithCode(Exception): code = 417 self._do_test_exception_mapping(ExceptionWithCode, 'Expectation failed')
**** CubicPower OpenStack Study **** def test_exception_with_none_code_throws_500(self): class ExceptionWithNoneCode(Exception): code = None msg = 'Internal Server Error' @webob.dec.wsgify def fail(req): raise ExceptionWithNoneCode() api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertEqual(500, resp.status_int) @mock.patch('cinder.openstack.common.gettextutils.translate')
**** CubicPower OpenStack Study **** def fail(req): raise ExceptionWithNoneCode() api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertEqual(500, resp.status_int) @mock.patch('cinder.openstack.common.gettextutils.translate')
**** CubicPower OpenStack Study **** def test_cinder_exception_with_localized_explanation(self, mock_t9n): msg = 'My Not Found' msg_translation = 'Mi No Encontrado' message = gettextutils.Message(msg, '') @webob.dec.wsgify def fail(req): class MyVolumeNotFound(exception.NotFound): def __init__(self): self.msg = message self.safe = True raise MyVolumeNotFound() # Test response without localization def mock_get_non_localized_message(msgid, locale): return msg mock_t9n.side_effect = mock_get_non_localized_message api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertEqual(404, resp.status_int) self.assertIn(msg, resp.body) # Test response with localization def mock_translate(msgid, locale): if isinstance(msgid, gettextutils.Message): return msg_translation return msgid mock_t9n.side_effect = mock_translate api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertEqual(404, resp.status_int) self.assertIn(msg_translation, resp.body)
**** CubicPower OpenStack Study **** def fail(req): class MyVolumeNotFound(exception.NotFound): def __init__(self): self.msg = message self.safe = True raise MyVolumeNotFound() # Test response without localization
**** CubicPower OpenStack Study **** def mock_get_non_localized_message(msgid, locale): return msg mock_t9n.side_effect = mock_get_non_localized_message api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertEqual(404, resp.status_int) self.assertIn(msg, resp.body) # Test response with localization
**** CubicPower OpenStack Study **** def mock_translate(msgid, locale): if isinstance(msgid, gettextutils.Message): return msg_translation return msgid mock_t9n.side_effect = mock_translate api = self._wsgi_app(fail) resp = webob.Request.blank('/').get_response(api) self.assertEqual(404, resp.status_int) self.assertIn(msg_translation, resp.body)
|