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
|
from __future__ import with_statement
from webob import Request, Response
import sys, logging, threading, random, urllib2, socket, cgi
from contextlib import contextmanager
from nose.tools import assert_raises, eq_ as eq
from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer, ServerHandler
from Queue import Queue, Empty
log = logging.getLogger(__name__)
__test__ = (sys.version >= '2.6') # skip these tests on py2.5
def test_request_reading():
"""
Test actual request/response cycle in the presence of Request.copy()
and other methods that can potentially hang.
"""
with serve(_test_app_req_reading) as server:
for key in _test_ops_req_read:
resp = urllib2.urlopen(server.url+key, timeout=3)
assert resp.read() == "ok"
def _test_app_req_reading(env, sr):
req = Request(env)
log.debug('starting test operation: %s', req.path_info)
test_op = _test_ops_req_read[req.path_info]
test_op(req)
log.debug('done')
r = Response("ok")
return r(env, sr)
_test_ops_req_read = {
'/copy': lambda req: req.copy(),
'/read-all': lambda req: req.body_file.read(),
'/read-0': lambda req: req.body_file.read(0),
'/make-seekable': lambda req: req.make_body_seekable()
}
# TODO: remove server logging for interrupted requests
# TODO: test interrupted body directly
def test_interrupted_request():
with serve(_test_app_req_interrupt) as server:
for path in _test_ops_req_interrupt:
_send_interrupted_req(server, path)
try:
assert _global_res.get(timeout=1)
except Empty:
raise AssertionError("Error during test %s", path)
_global_res = Queue()
def _test_app_req_interrupt(env, sr):
req = Request(env)
assert req.content_length == 100000
op = _test_ops_req_interrupt[req.path_info]
log.info("Running test: %s", req.path_info)
assert_raises(IOError, op, req)
_global_res.put(True)
sr('200 OK', [])
return []
def _req_int_cgi(req):
assert req.body_file.read(0) == ''
#req.environ.setdefault('CONTENT_LENGTH', '0')
d = cgi.FieldStorage(
fp=req.body_file,
environ=req.environ,
)
def _req_int_readline(req):
try:
eq(req.body_file.readline(), 'a=b\n')
req.body_file.readline()
except IOError:
# too early to detect disconnect
raise AssertionError
req.body_file.readline()
_test_ops_req_interrupt = {
'/copy': lambda req: req.copy(),
'/read-body': lambda req: req.body,
'/read-post': lambda req: req.POST,
'/read-all': lambda req: req.body_file.read(),
'/read-too-much': lambda req: req.body_file.read(1<<30),
'/readline': _req_int_readline,
'/readlines': lambda req: req.body_file.readlines(),
'/read-cgi': _req_int_cgi,
'/make-seekable': lambda req: req.make_body_seekable()
}
def _send_interrupted_req(server, path='/'):
sock = socket.socket()
sock.connect(('localhost', server.server_port))
f = sock.makefile('wb')
f.write(_interrupted_req % path)
f.flush()
f.close()
sock.close()
_interrupted_req = (
"POST %s HTTP/1.0\r\n"
"content-type: application/x-www-form-urlencoded\r\n"
"content-length: 100000\r\n"
"\r\n"
)
_interrupted_req += 'a=b\nz='+'x'*10000
@contextmanager
def serve(app):
server = _make_test_server(app)
try:
#worker = threading.Thread(target=server.handle_request)
worker = threading.Thread(target=server.serve_forever)
worker.setDaemon(True)
worker.start()
server.url = "http://localhost:%d" % server.server_port
log.debug("server started on %s", server.url)
yield server
finally:
log.debug("shutting server down")
server.shutdown()
worker.join(1)
if worker.isAlive():
log.warning('worker is hanged')
else:
log.debug("server stopped")
class QuietHanlder(WSGIRequestHandler):
def log_request(self, *args):
pass
ServerHandler.handle_error = lambda: None
class QuietServer(WSGIServer):
def handle_error(self, req, addr):
pass
def _make_test_server(app):
maxport = ((1<<16)-1)
# we'll make 3 attempts to find a free port
for i in range(3, 0, -1):
try:
port = random.randint(maxport/2, maxport)
server = make_server('localhost', port, app,
server_class=QuietServer,
handler_class=QuietHanlder
)
server.timeout = 5
return server
except:
if i == 1:
raise
if __name__ == '__main__':
#test_request_reading()
test_interrupted_request()
|
