Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-27 02:24:18 +00:00
Code Issues Projects Releases Wiki Activity

Change test API to allow multiple pathoc requests

Browse Source 
Add simple unit test for websocket server
This commit is contained in:
Aldo Cortesi 2015-05-31 18:38:11 +12:00
parent 2ebe994375
commit 6122f2da44
4 changed files with 33 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
libpathod/pathod.py
View File

@ -103,8 +103,15 @@ class PathodHandler(tcp.BaseHandler):
while True:
try:
frm = websockets.Frame.from_file(self.rfile)
retlog = dict(
type="wsframe",
frame=dict(
),
cipher=None,
)
self.addlog(retlog)
break
except tcp.NetLibTimeout:
except tcp.NetLibTimeout: # pragma: no cover
pass
lg(frm.human_readable())
return self.handle_websocket, None

2
libpathod/test.py
View File

@ -48,7 +48,7 @@ class Daemon:
l = self.log()
if not l:
return None
return l[-1]
return l[0]
def log(self):
"""

34
test/test_pathod.py
View File

@ -36,7 +36,7 @@ class TestTimeout(tutils.DaemonTests):
# increase test performance
# This is a bodge - we have some platform difference that causes
# different exceptions to be raised here.
tutils.raises(Exception, self.pathoc, "get:/:p1,1")
tutils.raises(Exception, self.pathoc, ["get:/:p1,1"])
assert self.d.last_log()["type"] == "timeout"
@ -58,10 +58,10 @@ class TestNotAfterConnect(tutils.DaemonTests):
def test_connect(self):
r = self.pathoc(
r"get:'http://foo.com/p/202':da",
[r"get:'http://foo.com/p/202':da"],
connect_to=("localhost", self.d.port)
)
assert r.status_code == 202
assert r[0].status_code == 202
class TestCustomCert(tutils.DaemonTests):
@ -71,7 +71,7 @@ class TestCustomCert(tutils.DaemonTests):
)
def test_connect(self):
r = self.pathoc(r"get:/p/202")
r = self.pathoc([r"get:/p/202"])[0]
assert r.status_code == 202
assert r.sslinfo
assert "test.com" in str(r.sslinfo.certchain[0].get_subject())
@ -84,7 +84,7 @@ class TestSSLCN(tutils.DaemonTests):
)
def test_connect(self):
r = self.pathoc(r"get:/p/202")
r = self.pathoc([r"get:/p/202"])[0]
assert r.status_code == 202
assert r.sslinfo
assert r.sslinfo.certchain[0].get_subject().CN == "foo.com"
@ -120,7 +120,7 @@ class CommonTests(tutils.DaemonTests):
assert "too large" in l["response"]["msg"]
def test_preline(self):
r = self.pathoc(r"get:'/p/200':i0,'\r\n'")
r = self.pathoc([r"get:'/p/200':i0,'\r\n'"])[0]
assert r.status_code == 200
def test_info(self):
@ -166,14 +166,14 @@ class CommonTests(tutils.DaemonTests):
tutils.raises(
http.HttpError,
self.pathoc,
"get:/:h'content-length'='foo'"
["get:/:h'content-length'='foo'"]
)
l = self.d.last_log()
assert l["type"] == "error"
assert "Content-Length unknown" in l["msg"]
def test_invalid_headers(self):
tutils.raises(http.HttpError, self.pathoc, "get:/:h'\t'='foo'")
tutils.raises(http.HttpError, self.pathoc, ["get:/:h'\t'='foo'"])
l = self.d.last_log()
assert l["type"] == "error"
assert "Invalid headers" in l["msg"]
@ -188,33 +188,37 @@ class CommonTests(tutils.DaemonTests):
assert "File access denied" in rsp.content
def test_proxy(self):
r = self.pathoc(r"get:'http://foo.com/p/202':da")
r = self.pathoc([r"get:'http://foo.com/p/202':da"])[0]
assert r.status_code == 202
def test_websocket(self):
r = self.pathoc("ws:/p/", ws_read_limit=0)
r = self.pathoc(["ws:/p/"], ws_read_limit=0)[0]
assert r.status_code == 101
r = self.pathoc("ws:/p/ws", ws_read_limit=0)
r = self.pathoc(["ws:/p/ws"], ws_read_limit=0)[0]
assert r.status_code == 101
def test_websocket_frame(self):
r = self.pathoc(["ws:/p/", "wf:b@10"], ws_read_limit=0)
assert self.d.last_log()["type"] == "wsframe"
class TestDaemon(CommonTests):
ssl = False
def test_connect(self):
r = self.pathoc(
r"get:'http://foo.com/p/202':da",
[r"get:'http://foo.com/p/202':da"],
connect_to=("localhost", self.d.port),
ssl=True
)
)[0]
assert r.status_code == 202
def test_connect_err(self):
tutils.raises(
http.HttpError,
self.pathoc,
r"get:'http://foo.com/p/202':da",
[r"get:'http://foo.com/p/202':da"],
connect_to=("localhost", self.d.port)
)
@ -234,6 +238,6 @@ class TestDaemonSSL(CommonTests):
assert "SSL" in l["msg"]
def test_ssl_cipher(self):
r = self.pathoc(r"get:/p/202")
r = self.pathoc([r"get:/p/202"])[0]
assert r.status_code == 202
assert self.d.last_log()["cipher"][1] > 0

7
test/tutils.py
View File

@ -67,7 +67,7 @@ class DaemonTests(object):
def pathoc(
self,
spec,
specs,
timeout=None,
connect_to=None,
ssl=None,
@ -84,7 +84,10 @@ class DaemonTests(object):
c.connect(connect_to)
if timeout:
c.settimeout(timeout)
return c.request(spec)
ret = []
for i in specs:
ret.append(c.request(i))
return ret
@contextmanager