# Copyright (c) 2010 OpenStack, LLC. # # 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. """ Tests for swift.common.utils """ from __future__ import with_statement import errno import mimetools import socket import unittest from StringIO import StringIO from collections import defaultdict from urllib import quote from swift.common.swob import Request from swift.common import wsgi class TestWSGI(unittest.TestCase): """ Tests for swift.common.wsgi """ def test_monkey_patch_mimetools(self): sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).type, 'text/plain') sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).plisttext, '') sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).maintype, 'text') sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).subtype, 'plain') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).type, 'text/html') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).plisttext, '; charset=ISO-8859-4') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).maintype, 'text') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).subtype, 'html') wsgi.monkey_patch_mimetools() sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).type, None) sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).plisttext, '') sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).maintype, None) sio = StringIO('blah') self.assertEquals(mimetools.Message(sio).subtype, None) sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).type, 'text/html') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).plisttext, '; charset=ISO-8859-4') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).maintype, 'text') sio = StringIO('Content-Type: text/html; charset=ISO-8859-4') self.assertEquals(mimetools.Message(sio).subtype, 'html') def test_get_socket(self): # stubs conf = {} ssl_conf = { 'cert_file': '', 'key_file': '', } # mocks class MockSocket(): def __init__(self): self.opts = defaultdict(dict) def setsockopt(self, level, optname, value): self.opts[level][optname] = value def mock_listen(*args, **kwargs): return MockSocket() class MockSsl(): def __init__(self): self.wrap_socket_called = [] def wrap_socket(self, sock, **kwargs): self.wrap_socket_called.append(kwargs) return sock # patch old_listen = wsgi.listen old_ssl = wsgi.ssl try: wsgi.listen = mock_listen wsgi.ssl = MockSsl() # test sock = wsgi.get_socket(conf) # assert self.assert_(isinstance(sock, MockSocket)) expected_socket_opts = { socket.SOL_SOCKET: { socket.SO_REUSEADDR: 1, socket.SO_KEEPALIVE: 1, }, socket.IPPROTO_TCP: { socket.TCP_KEEPIDLE: 600, }, } self.assertEquals(sock.opts, expected_socket_opts) # test ssl sock = wsgi.get_socket(ssl_conf) expected_kwargs = { 'certfile': '', 'keyfile': '', } self.assertEquals(wsgi.ssl.wrap_socket_called, [expected_kwargs]) finally: wsgi.listen = old_listen wsgi.ssl = old_ssl def test_address_in_use(self): # stubs conf = {} # mocks def mock_listen(*args, **kwargs): raise socket.error(errno.EADDRINUSE) def value_error_listen(*args, **kwargs): raise ValueError('fake') def mock_sleep(*args): pass class MockTime(): """Fast clock advances 10 seconds after every call to time """ def __init__(self): self.current_time = old_time.time() def time(self, *args, **kwargs): rv = self.current_time # advance for next call self.current_time += 10 return rv old_listen = wsgi.listen old_sleep = wsgi.sleep old_time = wsgi.time try: wsgi.listen = mock_listen wsgi.sleep = mock_sleep wsgi.time = MockTime() # test error self.assertRaises(Exception, wsgi.get_socket, conf) # different error wsgi.listen = value_error_listen self.assertRaises(ValueError, wsgi.get_socket, conf) finally: wsgi.listen = old_listen wsgi.sleep = old_sleep wsgi.time = old_time def test_pre_auth_wsgi_input(self): oldenv = {} newenv = wsgi.make_pre_authed_env(oldenv) self.assertTrue('wsgi.input' in newenv) self.assertEquals(newenv['wsgi.input'].read(), '') oldenv = {'wsgi.input': StringIO('original wsgi.input')} newenv = wsgi.make_pre_authed_env(oldenv) self.assertTrue('wsgi.input' in newenv) self.assertEquals(newenv['wsgi.input'].read(), '') oldenv = {'swift.source': 'UT'} newenv = wsgi.make_pre_authed_env(oldenv) self.assertEquals(newenv['swift.source'], 'UT') oldenv = {'swift.source': 'UT'} newenv = wsgi.make_pre_authed_env(oldenv, swift_source='SA') self.assertEquals(newenv['swift.source'], 'SA') def test_pre_auth_req(self): class FakeReq(object): @classmethod def fake_blank(cls, path, environ={}, body='', headers={}): self.assertEquals(environ['swift.authorize']('test'), None) self.assertFalse('HTTP_X_TRANS_ID' in environ) was_blank = Request.blank Request.blank = FakeReq.fake_blank wsgi.make_pre_authed_request({'HTTP_X_TRANS_ID': '1234'}, 'PUT', '/', body='tester', headers={}) wsgi.make_pre_authed_request({'HTTP_X_TRANS_ID': '1234'}, 'PUT', '/', headers={}) Request.blank = was_blank def test_pre_auth_req_with_quoted_path(self): r = wsgi.make_pre_authed_request( {'HTTP_X_TRANS_ID': '1234'}, 'PUT', path=quote('/a space'), body='tester', headers={}) self.assertEquals(r.path, quote('/a space')) def test_pre_auth_req_drops_query(self): r = wsgi.make_pre_authed_request( {'QUERY_STRING': 'original'}, 'GET', 'path') self.assertEquals(r.query_string, 'original') r = wsgi.make_pre_authed_request( {'QUERY_STRING': 'original'}, 'GET', 'path?replacement') self.assertEquals(r.query_string, 'replacement') r = wsgi.make_pre_authed_request( {'QUERY_STRING': 'original'}, 'GET', 'path?') self.assertEquals(r.query_string, '') def test_pre_auth_req_with_body(self): r = wsgi.make_pre_authed_request( {'QUERY_STRING': 'original'}, 'GET', 'path', 'the body') self.assertEquals(r.body, 'the body') def test_pre_auth_creates_script_name(self): e = wsgi.make_pre_authed_env({}) self.assertTrue('SCRIPT_NAME' in e) def test_pre_auth_copies_script_name(self): e = wsgi.make_pre_authed_env({'SCRIPT_NAME': '/script_name'}) self.assertEquals(e['SCRIPT_NAME'], '/script_name') def test_pre_auth_copies_script_name_unless_path_overridden(self): e = wsgi.make_pre_authed_env({'SCRIPT_NAME': '/script_name'}, path='/override') self.assertEquals(e['SCRIPT_NAME'], '') self.assertEquals(e['PATH_INFO'], '/override') def test_pre_auth_req_swift_source(self): r = wsgi.make_pre_authed_request( {'QUERY_STRING': 'original'}, 'GET', 'path', 'the body', swift_source='UT') self.assertEquals(r.body, 'the body') self.assertEquals(r.environ['swift.source'], 'UT') class TestWSGIContext(unittest.TestCase): def test_app_call(self): statuses = ['200 Ok', '404 Not Found'] def app(env, start_response): start_response(statuses.pop(0), [('Content-Length', '3')]) yield 'Ok\n' wc = wsgi.WSGIContext(app) r = Request.blank('/') it = wc._app_call(r.environ) self.assertEquals(wc._response_status, '200 Ok') self.assertEquals(''.join(it), 'Ok\n') r = Request.blank('/') it = wc._app_call(r.environ) self.assertEquals(wc._response_status, '404 Not Found') self.assertEquals(''.join(it), 'Ok\n') if __name__ == '__main__': unittest.main()