1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
# 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 logging
import mimetools
import os
import socket
import sys
import unittest
from getpass import getuser
from shutil import rmtree
from StringIO import StringIO
from collections import defaultdict
from eventlet import sleep
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
if __name__ == '__main__':
unittest.main()
|