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
|
import cStringIO, sys
import libpry
from libmproxy import wsgi
import tutils
class TestApp:
def __init__(self):
self.called = False
def __call__(self, environ, start_response):
self.called = True
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return ['Hello', ' world!\n']
class uWSGIAdaptor(libpry.AutoTree):
def test_make_environ(self):
w = wsgi.WSGIAdaptor(None, "foo", 80)
tr = tutils.treq()
assert w.make_environ(tr, None)
tr.path = "/foo?bar=voing"
r = w.make_environ(tr, None)
assert r["QUERY_STRING"] == "bar=voing"
def test_serve(self):
ta = TestApp()
w = wsgi.WSGIAdaptor(ta, "foo", 80)
r = tutils.treq()
r.host = "foo"
r.port = 80
wfile = cStringIO.StringIO()
err = w.serve(r, wfile)
assert ta.called
assert not err
val = wfile.getvalue()
assert "Hello world" in val
assert "Server:" in val
def _serve(self, app):
w = wsgi.WSGIAdaptor(app, "foo", 80)
r = tutils.treq()
r.host = "foo"
r.port = 80
wfile = cStringIO.StringIO()
err = w.serve(r, wfile)
return wfile.getvalue()
def test_serve_empty_body(self):
def app(environ, start_response):
status = '200 OK'
response_headers = [('Foo', 'bar')]
start_response(status, response_headers)
return []
assert self._serve(app)
def test_serve_double_start(self):
def app(environ, start_response):
try:
raise ValueError("foo")
except:
ei = sys.exc_info()
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
start_response(status, response_headers)
assert "Internal Server Error" in self._serve(app)
def test_serve_single_err(self):
def app(environ, start_response):
try:
raise ValueError("foo")
except:
ei = sys.exc_info()
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers, ei)
assert "Internal Server Error" in self._serve(app)
def test_serve_double_err(self):
def app(environ, start_response):
try:
raise ValueError("foo")
except:
ei = sys.exc_info()
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
yield "aaa"
start_response(status, response_headers, ei)
yield "bbb"
assert "Internal Server Error" in self._serve(app)
class uAppRegistry(libpry.AutoTree):
def test_add_get(self):
ar = wsgi.AppRegistry()
ar.add("foo", "domain", 80)
r = tutils.treq()
r.host = "domain"
r.port = 80
assert ar.get(r)
r.port = 81
assert not ar.get(r)
tests = [
uWSGIAdaptor(),
uAppRegistry()
]
|
