"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "test/unit/common/test_wsgi.py" between
swift-2.19.1.tar.gz and swift-2.21.0.tar.gz

About: OpenStack swift is software for creating redundant, scalable object storage using clusters of commodity servers to store terabytes or even petabytes of accessible data (now supporting storage policies).
The "Stein" series (latest release).

test_wsgi.py  (swift-2.19.1):test_wsgi.py  (swift-2.21.0)
skipping to change at line 30 skipping to change at line 30
import logging import logging
import socket import socket
import unittest import unittest
import os import os
from textwrap import dedent from textwrap import dedent
from collections import defaultdict from collections import defaultdict
import types import types
import eventlet.wsgi import eventlet.wsgi
import six
from six import BytesIO from six import BytesIO
from six import StringIO
from six.moves.urllib.parse import quote from six.moves.urllib.parse import quote
if six.PY2:
import mimetools
import mock import mock
import swift.common.middleware.catch_errors import swift.common.middleware.catch_errors
import swift.common.middleware.gatekeeper import swift.common.middleware.gatekeeper
import swift.proxy.server import swift.proxy.server
import swift.obj.server as obj_server import swift.obj.server as obj_server
import swift.container.server as container_server import swift.container.server as container_server
import swift.account.server as account_server import swift.account.server as account_server
skipping to change at line 70 skipping to change at line 66
obj_ring_path = \ obj_ring_path = \
os.path.join(tmpdir, policy.ring_name + '.ring.gz') os.path.join(tmpdir, policy.ring_name + '.ring.gz')
write_fake_ring(obj_ring_path) write_fake_ring(obj_ring_path)
# make sure there's no other ring cached on this policy # make sure there's no other ring cached on this policy
policy.object_ring = None policy.object_ring = None
@patch_policies @patch_policies
class TestWSGI(unittest.TestCase): class TestWSGI(unittest.TestCase):
"""Tests for swift.common.wsgi""" """Tests for swift.common.wsgi"""
def setUp(self):
utils.HASH_PATH_PREFIX = 'startcap'
if six.PY2:
self._orig_parsetype = mimetools.Message.parsetype
def tearDown(self):
if six.PY2:
mimetools.Message.parsetype = self._orig_parsetype
@unittest.skipIf(six.PY3, "test specific to Python 2")
def test_monkey_patch_mimetools(self):
sio = StringIO('blah')
self.assertEqual(mimetools.Message(sio).type, 'text/plain')
sio = StringIO('blah')
self.assertEqual(mimetools.Message(sio).plisttext, '')
sio = StringIO('blah')
self.assertEqual(mimetools.Message(sio).maintype, 'text')
sio = StringIO('blah')
self.assertEqual(mimetools.Message(sio).subtype, 'plain')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).type, 'text/html')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).plisttext,
'; charset=ISO-8859-4')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).maintype, 'text')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).subtype, 'html')
wsgi.monkey_patch_mimetools()
sio = StringIO('blah')
self.assertIsNone(mimetools.Message(sio).type)
sio = StringIO('blah')
self.assertEqual(mimetools.Message(sio).plisttext, '')
sio = StringIO('blah')
self.assertIsNone(mimetools.Message(sio).maintype)
sio = StringIO('blah')
self.assertIsNone(mimetools.Message(sio).subtype)
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).type, 'text/html')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).plisttext,
'; charset=ISO-8859-4')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).maintype, 'text')
sio = StringIO('Content-Type: text/html; charset=ISO-8859-4')
self.assertEqual(mimetools.Message(sio).subtype, 'html')
def test_init_request_processor(self): def test_init_request_processor(self):
config = """ config = """
[DEFAULT] [DEFAULT]
swift_dir = TEMPDIR swift_dir = TEMPDIR
[pipeline:main] [pipeline:main]
pipeline = proxy-server pipeline = proxy-server
[app:proxy-server] [app:proxy-server]
use = egg:swift#proxy use = egg:swift#proxy
skipping to change at line 405 skipping to change at line 353
expected_socket_opts[socket.IPPROTO_TCP][ expected_socket_opts[socket.IPPROTO_TCP][
socket.TCP_KEEPIDLE] = 600 socket.TCP_KEEPIDLE] = 600
self.assertEqual(sock.opts, expected_socket_opts) self.assertEqual(sock.opts, expected_socket_opts)
# test ssl # test ssl
sock = wsgi.get_socket(ssl_conf) sock = wsgi.get_socket(ssl_conf)
expected_kwargs = { expected_kwargs = {
'certfile': '', 'certfile': '',
'keyfile': '', 'keyfile': '',
} }
self.assertEqual(wsgi.ssl.wrap_socket_called, [expected_kwargs]) self.assertEqual(wsgi.ssl.wrap_socket_called, [expected_kwargs])
# test keep_idle value
keepIdle_value = 700
conf['keep_idle'] = keepIdle_value
sock = wsgi.get_socket(conf)
# assert
if hasattr(socket, 'TCP_KEEPIDLE'):
expected_socket_opts[socket.IPPROTO_TCP][
socket.TCP_KEEPIDLE] = keepIdle_value
self.assertEqual(sock.opts, expected_socket_opts)
# test keep_idle for str -> int conversion
keepIdle_value = '800'
conf['keep_idle'] = keepIdle_value
sock = wsgi.get_socket(conf)
# assert
if hasattr(socket, 'TCP_KEEPIDLE'):
expected_socket_opts[socket.IPPROTO_TCP][
socket.TCP_KEEPIDLE] = int(keepIdle_value)
self.assertEqual(sock.opts, expected_socket_opts)
# test keep_idle for negative value
conf['keep_idle'] = -600
self.assertRaises(wsgi.ConfigFileError, wsgi.get_socket, conf)
# test keep_idle for upperbound value
conf['keep_idle'] = 2 ** 15
self.assertRaises(wsgi.ConfigFileError, wsgi.get_socket, conf)
# test keep_idle for Type mismatch
conf['keep_idle'] = 'foobar'
self.assertRaises(wsgi.ConfigFileError, wsgi.get_socket, conf)
finally: finally:
wsgi.listen = old_listen wsgi.listen = old_listen
wsgi.ssl = old_ssl wsgi.ssl = old_ssl
def test_address_in_use(self): def test_address_in_use(self):
# stubs # stubs
conf = {'bind_port': 54321} conf = {'bind_port': 54321}
# mocks # mocks
def mock_listen(*args, **kwargs): def mock_listen(*args, **kwargs):
skipping to change at line 1779 skipping to change at line 1760
"healthcheck catch_errors tempurl proxy-server") "healthcheck catch_errors tempurl proxy-server")
def test_str_unknown_filter(self): def test_str_unknown_filter(self):
del self.pipe.context.filter_contexts[0].__dict__['name'] del self.pipe.context.filter_contexts[0].__dict__['name']
self.pipe.context.filter_contexts[0].object = 'mysterious' self.pipe.context.filter_contexts[0].object = 'mysterious'
self.assertEqual( self.assertEqual(
str(self.pipe), str(self.pipe),
"<unknown> catch_errors tempurl proxy-server") "<unknown> catch_errors tempurl proxy-server")
@patch_policies @patch_policies
@mock.patch('swift.common.utils.HASH_PATH_SUFFIX', new='endcap')
class TestPipelineModification(unittest.TestCase): class TestPipelineModification(unittest.TestCase):
def pipeline_modules(self, app): def pipeline_modules(self, app):
# This is rather brittle; it'll break if a middleware stores its app # This is rather brittle; it'll break if a middleware stores its app
# anywhere other than an attribute named "app", but it works for now. # anywhere other than an attribute named "app", but it works for now.
pipe = [] pipe = []
for _ in range(1000): for _ in range(1000):
pipe.append(app.__class__.__module__) pipe.append(app.__class__.__module__)
if not hasattr(app, 'app'): if not hasattr(app, 'app'):
break break
app = app.app app = app.app
 End of changes. 6 change blocks. 
53 lines changed or deleted 33 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)