Rooba
/
mitmproxy
mirror of https://github.com/mitmproxy/mitmproxy.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Internal error page for WSGI. 

Also, 100% test coverage.
This commit is contained in:
Aldo Cortesi 2012-04-27 15:56:42 +12:00
parent 2fe54d17df
commit 116fcfcf7a
2 changed files with 81 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
libmproxy/wsgi.py
View File

@ -55,6 +55,23 @@ class WSGIAdaptor:
environ[key] = value
return environ
def error_page(self, soc, headers_sent):
"""
Make a best-effort attempt to write an error page. If headers are
already sent, we just bung the error into the page.
"""
c = """
<html>
<h1>Internal Server Error</h1>
</html>
"""
if not headers_sent:
soc.write("HTTP/1.1 500 Internal Server Error%s\r\n")
soc.write("Content-Type: text/html\r\n")
soc.write("Content-Length: %s\r\n"%len(c))
soc.write("\r\n")
soc.write(c)
def serve(self, request, soc):
state = dict(
response_started = False,
@ -97,12 +114,12 @@ class WSGIAdaptor:
if not state["headers_sent"]:
write("")
except Exception, v:
print v
try:
# Serve internal server error page
pass
self.error_page(soc, state["headers_sent"])
# begin nocover
except Exception, v:
pass
# end nocover
return errs.getvalue()

66
test/test_wsgi.py
View File

@ -1,4 +1,4 @@
import cStringIO
import cStringIO, sys
import libpry
from libmproxy import wsgi
import tutils
@ -19,10 +19,12 @@ class TestApp:
class uWSGIAdaptor(libpry.AutoTree):
def test_make_environ(self):
w = wsgi.WSGIAdaptor(None, "foo", 80)
assert w.make_environ(
tutils.treq(),
None
)
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()
@ -40,6 +42,60 @@ class uWSGIAdaptor(libpry.AutoTree):
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):