1
from __future__ import generators
4
from twisted.web2.test.test_server import BaseCase
5
from twisted.web2 import resource
6
from twisted.internet import reactor, interfaces
7
from twisted.python import log
9
if interfaces.IReactorThreads(reactor, None) is not None:
10
from twisted.web2.wsgi import WSGIResource as WSGI
14
class TestError(Exception):
17
class TestContainer(BaseCase):
20
def flushErrors(self, result, error):
21
log.flushErrors(error)
24
def test_getContainedResource(self):
25
"""Test that non-blocking WSGI applications render properly."""
26
def application(environ, start_response):
28
response_headers = [('Content-type','text/html')]
29
writer = start_response(status, response_headers)
31
return ['<h1>Some HTML</h1>',
34
return self.assertResponse(
35
(WSGI(application), 'http://host/'),
36
(200, {"Content-Length": None}, '<html><h1>Some HTML</h1></html>'))
38
def test_getBlockingResource(self):
39
"""Test that blocking WSGI applications render properly."""
40
def application(environ, start_response):
41
"""Simplest possible application object"""
43
response_headers = [('Content-type','text/html')]
44
writer = start_response(status, response_headers)
45
writer('<h1>A little bit')
47
writer(' of HTML</h1>')
49
return ['<p>Hello!</p>']
51
return self.assertResponse(
52
(WSGI(application), 'http://host/'),
53
(200, {"Content-Length": None}, '<h1>A little bit of HTML</h1><p>Hello!</p>'))
55
def test_responseCode(self):
56
"""Test that WSGIResource handles strange response codes properly."""
57
def application(environ, start_response):
59
response_headers = [('Content-type','text/html')]
60
writer = start_response(status, response_headers)
63
return self.assertResponse(
64
(WSGI(application), 'http://host/'),
65
(314, {"Content-Length": 0}, ''))
67
def test_errorfulResource(self):
68
def application(environ, start_response):
69
raise TestError("This is an expected error")
71
return self.assertResponse(
72
(WSGI(application), 'http://host/'),
73
(500, {}, None)).addBoth(self.flushErrors, TestError)
75
def test_errorfulResource2(self):
76
def application(environ, start_response):
77
write = start_response("200 OK", {})
79
raise TestError("This is an expected error")
81
return self.assertResponse(
82
(WSGI(application), 'http://host/'),
83
(200, {"Content-Length": None}, "Foo"), failure=True
84
).addBoth(self.flushErrors, TestError)
86
def test_errorfulIterator(self):
88
raise TestError("This is an expected error")
90
def application(environ, start_response):
91
start_response("200 OK", {})
94
return self.assertResponse(
95
(WSGI(application), 'http://host/'),
96
(500, {}, None)).addBoth(self.flushErrors, TestError)
98
def test_errorfulIterator2(self):
102
raise TestError("This is also expected")
104
def application(environ, start_response):
105
start_response("200 OK", {})
108
return self.assertResponse(
109
(WSGI(application), 'http://host/'),
110
(200, {"Content-Length": None}, "FooBar"), failure=True
111
).addBoth(self.flushErrors, TestError)
113
def test_didntCallStartResponse(self):
114
def application(environ, start_response):
117
return self.assertResponse(
118
(WSGI(application), 'http://host/'),
119
(500, {}, None)).addBoth(self.flushErrors, RuntimeError)
121
def test_calledStartResponseLate(self):
122
def application(environ, start_response):
123
start_response("200 OK", {})
126
return self.assertResponse(
127
(WSGI(application), 'http://host/'),
128
(200, {"Content-Length": None}, "Foo"))
130
def test_returnList(self):
131
def application(environ, start_response):
132
write = start_response("200 OK", {})
133
return ["Foo", "Bar"]
135
return self.assertResponse(
136
(WSGI(application), 'http://host/'),
137
(200, {"Content-Length": 6}, "FooBar"))
139
def test_readAllInput(self):
140
def application(environ, start_response):
141
input = environ['wsgi.input']
143
start_response("200 OK", {})
146
return self.assertResponse(
147
(WSGI(application), 'http://host/', {}, None, None, '', "This is some content"),
148
(200, {"Content-Length": 20}, "This is some content"))
150
def test_readInputLines(self):
151
def application(environ, start_response):
152
input = environ['wsgi.input']
153
out = 'X'.join(input.readlines())
154
start_response("200 OK", {})
157
d = self.assertResponse(
158
(WSGI(application), 'http://host/', {}, None, None, '', "a\nb\nc"),
159
(200, {"Content-Length": 7}, "a\nXb\nXc"))
161
d.addCallback(lambda d: self.assertResponse(
162
(WSGI(application), 'http://host/', {}, None, None, '', "a\nb\n"),
163
(200, {"Content-Length": 5}, "a\nXb\n")))
166
def test_readInputLineSizeNegZero(self):
167
"""Test that calling wsgi.input.readline works with -1 and 0 and none."""
168
def application(environ, start_response):
169
input = environ['wsgi.input']
171
out = [input.read(5)] # 'Line '
172
out.extend(["X", input.readline(-1)]) # 'blah blah\n'
173
out.extend(["X", input.readline(0)]) # ''
174
out.extend(["X", input.readline(None)]) # 'Oh Line\n'
175
out.extend(["X", input.readline()]) # ''
177
start_response("200 OK", {})
180
return self.assertResponse(
181
(WSGI(application), 'http://host/', {}, None, None, '',
182
"Line blah blah\nOh Line\n"),
183
(200, {"Content-Length": 27},
184
"Line Xblah blah\nXXOh Line\nX"))
186
def test_readInputLineSize(self):
187
"""Test that readline() with a size works."""
188
def application(environ, start_response):
189
input = environ['wsgi.input']
191
out = [input.read(5)] # 'Line '
192
out.extend(["X", input.readline(5)]) # 'blah '
193
out.extend(["X", input.readline()]) # 'blah\n'
194
out.extend(["X", input.readline(1)]) # 'O'
195
out.extend(["X", input.readline()]) # 'h Line\n'
197
start_response("200 OK", {})
200
return self.assertResponse(
201
(WSGI(application), 'http://host/', {}, None, None, '',
202
"Line blah blah\nOh Line\n"),
203
(200, {"Content-Length": 27},
204
"Line Xblah Xblah\nXOXh Line\n"))
206
def test_readInputMixed(self):
207
def application(environ, start_response):
208
input = environ['wsgi.input']
209
out = [input.read(5)]
210
out.extend(["X", input.readline()])
211
out.extend(["X", input.read(1)])
212
out.extend(["X", input.readline()])
214
start_response("200 OK", {})
217
return self.assertResponse(
218
(WSGI(application), 'http://host/', {}, None, None, '',
219
"Line blah blah\nOh Line\n"),
220
(200, {"Content-Length": 26}, "Line Xblah blah\nXOXh Line\n"))
222
def test_readiter(self):
223
"""Test that using wsgi.input as an iterator works."""
224
def application(environ, start_response):
225
input = environ['wsgi.input']
226
out = 'X'.join(input)
228
start_response("200 OK", {})
231
return self.assertResponse(
232
(WSGI(application), 'http://host/', {}, None, None, '',
233
"Line blah blah\nOh Line\n"),
234
(200, {"Content-Length": 24}, "Line blah blah\nXOh Line\n"))
236
class TestWSGIEnvironment(BaseCase):
238
Test that the WSGI container does everything we expect it to do
239
with the WSGI environment dictionary.
241
def envApp(self, *varnames):
243
Return a WSGI application that writes environment variables.
245
def _app(environ, start_response):
247
response_headers = [('Content-type','text/html')]
248
writer = start_response(status, response_headers)
249
return ['%s=%r;' % (k, environ.get(k, '')) for k in varnames]
252
def assertEnv(self, uri, env, version=None):
255
envstring = ''.join(['%s=%r;' % (k, v) for k, v in env.items()])
257
(WSGI(self.envApp(*keys)), uri, None, None, version),
258
(200, {}, envstring))
260
def test_wsgi_url_scheme(self):
261
"""wsgi.url_scheme"""
262
self.assertEnv('https://host/', {'wsgi.url_scheme': 'https'})
263
self.assertEnv('http://host/', {'wsgi.url_scheme': 'http'})
265
def test_SERVER_PROTOCOL(self):
266
"""SERVER_PROTOCOL"""
267
self.assertEnv('http://host/', {'SERVER_PROTOCOL': 'HTTP/1.1'})
269
def test_SERVER_PORT(self):
271
self.assertEnv('http://host/', {'SERVER_PORT': '80'})
272
self.assertEnv('http://host:523/', {'SERVER_PORT': '523'})
273
self.assertEnv('https://host/', {'SERVER_PORT': '443'})
274
self.assertEnv('https://host:523/', {'SERVER_PORT': '523'})
275
self.assertEnv('/foo', {'SERVER_PORT': '80'}, version=(1,0))
278
for cls in (TestContainer, TestWSGIEnvironment):
279
setattr(cls, 'skip', 'Required thread support is missing, skipping')
added
removed
twisted/web2/test/test_wsgi.py
