Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-30 03:14:22 +00:00
Code Issues Projects Releases Wiki Activity

100% test coverage

Browse Source
This commit is contained in:
Aldo Cortesi 2015-06-08 16:25:33 +12:00
parent 97fdb9ef0c
commit 332832278f
2 changed files with 22 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
libpathod/pathoc.py
View File

@ -293,35 +293,19 @@ class Pathoc(tcp.TCPClient):
return
yield frm
def websocket_get_frame(self, frame):
"""
Called when a frame is received from the server.
"""
pass
def websocket_send_frame(self, r):
"""
Sends a single websocket frame.
"""
with self.log() as log:
if isinstance(r, basestring):
r = language.parse_pathoc(r).next()
log(">> %s" % r)
try:
language.serve(r, self.wfile, self.settings)
self.wfile.flush()
except tcp.NetLibTimeout:
if self.ignoretimeout:
self.log("Timeout (ignored)")
return None
raise
def websocket_start(self, r, limit=None):
def websocket_start(self, r):
"""
Performs an HTTP request, and attempts to drop into websocket
connection.
limit: Disconnect after receiving N server frames.
"""
resp = self.http(r)
if resp.status_code == 101:
@ -348,8 +332,6 @@ class Pathoc(tcp.TCPClient):
May raise http.HTTPError, tcp.NetLibError
"""
with self.log() as log:
if isinstance(r, basestring):
r = language.parse_pathoc(r).next()
log(">> %s" % r)
resp, req = None, None
try:
@ -395,7 +377,7 @@ class Pathoc(tcp.TCPClient):
r = language.parse_pathoc(r).next()
if isinstance(r, language.http.Request):
if r.ws:
return self.websocket_start(r, self.websocket_get_frame)
return self.websocket_start(r)
else:
return self.http(r)
elif isinstance(r, language.websockets.WebsocketFrame):

18
test/test_pathoc.py
View File

@ -185,6 +185,24 @@ class TestDaemon(_TestDaemon):
def test_conn_err(self):
assert "Invalid server response" in self.tval(["get:'/p/200:d2'"])
def test_websocket_shutdown(self):
c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)
c.connect()
c.request("ws:/")
c.stop()
def test_wait_finish(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ws_read_limit = 1
)
c.connect()
c.request("ws:/")
c.request("wf:f'wf:x100'")
[i for i in c.wait(timeout=0, finish=False)]
[i for i in c.wait(timeout=0)]
def test_connect_fail(self):
to = ("foobar", 80)
c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)