mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 16:17:49 +00:00
ce02874e2a
Renamed TestApp class to ExampleApp to prevent py.test from trying to collect it as a test.
107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
from io import BytesIO
|
|
import sys
|
|
from netlib import wsgi
|
|
from netlib.http import Headers
|
|
|
|
|
|
def tflow():
|
|
headers = Headers(test=b"value")
|
|
req = wsgi.Request("http", "GET", "/", "HTTP/1.1", headers, "")
|
|
return wsgi.Flow(("127.0.0.1", 8888), req)
|
|
|
|
|
|
class ExampleApp:
|
|
|
|
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 [b'Hello', b' world!\n']
|
|
|
|
|
|
class TestWSGI:
|
|
|
|
def test_make_environ(self):
|
|
w = wsgi.WSGIAdaptor(None, "foo", 80, "version")
|
|
tf = tflow()
|
|
assert w.make_environ(tf, None)
|
|
|
|
tf.request.path = "/foo?bar=voing"
|
|
r = w.make_environ(tf, None)
|
|
assert r["QUERY_STRING"] == "bar=voing"
|
|
|
|
def test_serve(self):
|
|
ta = ExampleApp()
|
|
w = wsgi.WSGIAdaptor(ta, "foo", 80, "version")
|
|
f = tflow()
|
|
f.request.host = "foo"
|
|
f.request.port = 80
|
|
|
|
wfile = BytesIO()
|
|
err = w.serve(f, wfile)
|
|
assert ta.called
|
|
assert not err
|
|
|
|
val = wfile.getvalue()
|
|
assert b"Hello world" in val
|
|
assert b"Server:" in val
|
|
|
|
def _serve(self, app):
|
|
w = wsgi.WSGIAdaptor(app, "foo", 80, "version")
|
|
f = tflow()
|
|
f.request.host = "foo"
|
|
f.request.port = 80
|
|
wfile = BytesIO()
|
|
w.serve(f, 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:
|
|
sys.exc_info()
|
|
status = '200 OK'
|
|
response_headers = [('Content-type', 'text/plain')]
|
|
start_response(status, response_headers)
|
|
start_response(status, response_headers)
|
|
assert b"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)
|
|
yield b""
|
|
assert b"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 b"aaa"
|
|
start_response(status, response_headers, ei)
|
|
yield b"bbb"
|
|
assert b"Internal Server Error" in self._serve(app)
|