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

Convert parse_pathod to generator

Browse Source
This commit is contained in:
Aldo Cortesi 2015-06-07 16:11:32 +12:00
parent 0da3e51e1c
commit df962f8e7b
6 changed files with 48 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
libpathod/app.py
View File

@ -137,7 +137,7 @@ def make_app(noapi, debug):
if is_request: if is_request:
r = language.parse_pathoc(spec).next() r = language.parse_pathoc(spec).next()
else: else:
r = language.parse_pathod(spec) r = language.parse_pathod(spec).next()
except language.ParseException as v: except language.ParseException as v:
args["syntaxerror"] = str(v) args["syntaxerror"] = str(v)
args["marked"] = v.marked() args["marked"] = v.marked()

28
libpathod/language/__init__.py
View File

@ -10,6 +10,15 @@ from base import Settings
assert Settings # prevent pyflakes from messing with this assert Settings # prevent pyflakes from messing with this
def expand(msg):
times = getattr(msg, "times", None)
if times:
for j in xrange(int(times.value)):
yield msg.strike_token("times")
else:
yield msg
def parse_pathod(s): def parse_pathod(s):
""" """
May raise ParseException May raise ParseException
@ -19,28 +28,18 @@ def parse_pathod(s):
except UnicodeError: except UnicodeError:
raise exceptions.ParseException("Spec must be valid ASCII.", 0, 0) raise exceptions.ParseException("Spec must be valid ASCII.", 0, 0)
try: try:
return pp.Or( reqs = pp.Or(
[ [
websockets.WebsocketFrame.expr(), websockets.WebsocketFrame.expr(),
http.Response.expr(), http.Response.expr(),
] ]
).parseString(s, parseAll=True)[0] ).parseString(s, parseAll=True)
except pp.ParseException as v: except pp.ParseException as v:
raise exceptions.ParseException(v.msg, v.line, v.col) raise exceptions.ParseException(v.msg, v.line, v.col)
return itertools.chain(*[expand(i) for i in reqs])
def expand(req):
if req.times:
for j in xrange(int(req.times.value)):
yield req.strike_token("times")
else:
yield req
def parse_pathoc(s): def parse_pathoc(s):
"""
May raise ParseException
"""
try: try:
s = s.decode("ascii") s = s.decode("ascii")
except UnicodeError: except UnicodeError:
@ -60,6 +59,9 @@ def parse_pathoc(s):
def parse_websocket_frame(s): def parse_websocket_frame(s):
"""
May raise ParseException
"""
try: try:
return websockets.WebsocketFrame.expr().parseString( return websockets.WebsocketFrame.expr().parseString(
s, s,

6
libpathod/pathod.py
View File

@ -295,17 +295,17 @@ class PathodHandler(tcp.BaseHandler):
anchor_spec = language.parse_pathod(spec) anchor_spec = language.parse_pathod(spec)
except language.ParseException as v: except language.ParseException as v:
lg("Parse error: %s" % v.msg) lg("Parse error: %s" % v.msg)
anchor_spec = language.http.make_error_response( anchor_spec = iter([language.http.make_error_response(
"Parse Error", "Parse Error",
"Error parsing response spec: %s\n" % ( "Error parsing response spec: %s\n" % (
v.msg + v.marked() v.msg + v.marked()
) )
) )])
if anchor_spec: if anchor_spec:
lg("crafting spec: %s" % anchor_spec) lg("crafting spec: %s" % anchor_spec)
nexthandler, retlog["response"] = self.http_serve_crafted( nexthandler, retlog["response"] = self.http_serve_crafted(
anchor_spec anchor_spec.next()
) )
if nexthandler and websocket_key: if nexthandler and websocket_key:
return self.handle_websocket, retlog return self.handle_websocket, retlog

12
test/test_language_actions.py
View File

@ -15,9 +15,9 @@ def test_unique_name():
class TestDisconnects: class TestDisconnects:
def test_parse_pathod(self): def test_parse_pathod(self):
a = language.parse_pathod("400:d0").actions[0] a = language.parse_pathod("400:d0").next().actions[0]
assert a.spec() == "d0" assert a.spec() == "d0"
a = language.parse_pathod("400:dr").actions[0] a = language.parse_pathod("400:dr").next().actions[0]
assert a.spec() == "dr" assert a.spec() == "dr"
def test_at(self): def test_at(self):
@ -40,12 +40,12 @@ class TestDisconnects:
class TestInject: class TestInject:
def test_parse_pathod(self): def test_parse_pathod(self):
a = language.parse_pathod("400:ir,@100").actions[0] a = language.parse_pathod("400:ir,@100").next().actions[0]
assert a.offset == "r" assert a.offset == "r"
assert a.value.datatype == "bytes" assert a.value.datatype == "bytes"
assert a.value.usize == 100 assert a.value.usize == 100
a = language.parse_pathod("400:ia,@100").actions[0] a = language.parse_pathod("400:ia,@100").next().actions[0]
assert a.offset == "a" assert a.offset == "a"
def test_at(self): def test_at(self):
@ -60,7 +60,7 @@ class TestInject:
def test_serve(self): def test_serve(self):
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:i0,'foo'") r = language.parse_pathod("400:i0,'foo'").next()
assert language.serve(r, s, {}) assert language.serve(r, s, {})
def test_spec(self): def test_spec(self):
@ -93,7 +93,7 @@ class TestPauses:
assert v.offset == "a" assert v.offset == "a"
def test_request(self): def test_request(self):
r = language.parse_pathod('400:p10,10') r = language.parse_pathod('400:p10,10').next()
assert r.actions[0].spec() == "p10,10" assert r.actions[0].spec() == "p10,10"
def test_spec(self): def test_spec(self):

38
test/test_language_http.py
View File

@ -143,42 +143,42 @@ class TestRequest:
class TestResponse: class TestResponse:
def dummy_response(self): def dummy_response(self):
return language.parse_pathod("400'msg'") return language.parse_pathod("400'msg'").next()
def test_response(self): def test_response(self):
r = language.parse_pathod("400:m'msg'") r = language.parse_pathod("400:m'msg'").next()
assert r.code.string() == "400" assert r.code.string() == "400"
assert r.reason.string() == "msg" assert r.reason.string() == "msg"
r = language.parse_pathod("400:m'msg':b@100b") r = language.parse_pathod("400:m'msg':b@100b").next()
assert r.reason.string() == "msg" assert r.reason.string() == "msg"
assert r.body.values({}) assert r.body.values({})
assert str(r) assert str(r)
r = language.parse_pathod("200") r = language.parse_pathod("200").next()
assert r.code.string() == "200" assert r.code.string() == "200"
assert not r.reason assert not r.reason
assert "OK" in [i[:] for i in r.preamble({})] assert "OK" in [i[:] for i in r.preamble({})]
def test_render(self): def test_render(self):
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:m'msg'") r = language.parse_pathod("400:m'msg'").next()
assert language.serve(r, s, {}) assert language.serve(r, s, {})
r = language.parse_pathod("400:p0,100:dr") r = language.parse_pathod("400:p0,100:dr").next()
assert "p0" in r.spec() assert "p0" in r.spec()
s = r.preview_safe() s = r.preview_safe()
assert "p0" not in s.spec() assert "p0" not in s.spec()
def test_raw(self): def test_raw(self):
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:b'foo'") r = language.parse_pathod("400:b'foo'").next()
language.serve(r, s, {}) language.serve(r, s, {})
v = s.getvalue() v = s.getvalue()
assert "Content-Length" in v assert "Content-Length" in v
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:b'foo':r") r = language.parse_pathod("400:b'foo':r").next()
language.serve(r, s, {}) language.serve(r, s, {})
v = s.getvalue() v = s.getvalue()
assert "Content-Length" not in v assert "Content-Length" not in v
@ -186,6 +186,7 @@ class TestResponse:
def test_length(self): def test_length(self):
def testlen(x): def testlen(x):
s = cStringIO.StringIO() s = cStringIO.StringIO()
x = x.next()
language.serve(x, s, language.Settings()) language.serve(x, s, language.Settings())
assert x.length(language.Settings()) == len(s.getvalue()) assert x.length(language.Settings()) == len(s.getvalue())
testlen(language.parse_pathod("400:m'msg':r")) testlen(language.parse_pathod("400:m'msg':r"))
@ -194,6 +195,7 @@ class TestResponse:
def test_maximum_length(self): def test_maximum_length(self):
def testlen(x): def testlen(x):
x = x.next()
s = cStringIO.StringIO() s = cStringIO.StringIO()
m = x.maximum_length({}) m = x.maximum_length({})
language.serve(x, s, {}) language.serve(x, s, {})
@ -222,19 +224,19 @@ class TestResponse:
tutils.raises("ascii", language.parse_pathod, "foo:b\xf0") tutils.raises("ascii", language.parse_pathod, "foo:b\xf0")
def test_parse_header(self): def test_parse_header(self):
r = language.parse_pathod('400:h"foo"="bar"') r = language.parse_pathod('400:h"foo"="bar"').next()
assert http.get_header("foo", r.headers) assert http.get_header("foo", r.headers)
def test_parse_pause_before(self): def test_parse_pause_before(self):
r = language.parse_pathod("400:p0,10") r = language.parse_pathod("400:p0,10").next()
assert r.actions[0].spec() == "p0,10" assert r.actions[0].spec() == "p0,10"
def test_parse_pause_after(self): def test_parse_pause_after(self):
r = language.parse_pathod("400:pa,10") r = language.parse_pathod("400:pa,10").next()
assert r.actions[0].spec() == "pa,10" assert r.actions[0].spec() == "pa,10"
def test_parse_pause_random(self): def test_parse_pause_random(self):
r = language.parse_pathod("400:pr,10") r = language.parse_pathod("400:pr,10").next()
assert r.actions[0].spec() == "pr,10" assert r.actions[0].spec() == "pr,10"
def test_parse_stress(self): def test_parse_stress(self):
@ -242,19 +244,19 @@ class TestResponse:
# returns an int and a python 2.7 int on windows has 32bit precision. # returns an int and a python 2.7 int on windows has 32bit precision.
# Therefore, we should keep the body length < 2147483647 bytes in our # Therefore, we should keep the body length < 2147483647 bytes in our
# tests. # tests.
r = language.parse_pathod("400:b@1g") r = language.parse_pathod("400:b@1g").next()
assert r.length({}) assert r.length({})
def test_spec(self): def test_spec(self):
def rt(s): def rt(s):
s = language.parse_pathod(s).spec() s = language.parse_pathod(s).next().spec()
assert language.parse_pathod(s).spec() == s assert language.parse_pathod(s).next().spec() == s
rt("400:b@100g") rt("400:b@100g")
rt("400") rt("400")
rt("400:da") rt("400:da")
def test_websockets(self): def test_websockets(self):
r = language.parse_pathod("ws") r = language.parse_pathod("ws").next()
tutils.raises("no websocket key", r.resolve, language.Settings()) tutils.raises("no websocket key", r.resolve, language.Settings())
res = r.resolve(language.Settings(websocket_key="foo")) res = r.resolve(language.Settings(websocket_key="foo"))
assert res.code.string() == "101" assert res.code.string() == "101"
@ -294,9 +296,9 @@ def test_location_shortcut():
def test_shortcuts(): def test_shortcuts():
assert language.parse_pathod( assert language.parse_pathod(
"400:c'foo'").headers[0].key.val == "Content-Type" "400:c'foo'").next().headers[0].key.val == "Content-Type"
assert language.parse_pathod( assert language.parse_pathod(
"400:l'foo'").headers[0].key.val == "Location" "400:l'foo'").next().headers[0].key.val == "Location"
assert "Android" in tutils.render(parse_request("get:/:ua")) assert "Android" in tutils.render(parse_request("get:/:ua"))
assert "User-Agent" in tutils.render(parse_request("get:/:ua")) assert "User-Agent" in tutils.render(parse_request("get:/:ua"))

6
test/test_language_writer.py
View File

@ -78,14 +78,14 @@ def test_write_values_pauses():
def test_write_values_after(): def test_write_values_after():
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:da") r = language.parse_pathod("400:da").next()
language.serve(r, s, {}) language.serve(r, s, {})
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:pa,0") r = language.parse_pathod("400:pa,0").next()
language.serve(r, s, {}) language.serve(r, s, {})
s = cStringIO.StringIO() s = cStringIO.StringIO()
r = language.parse_pathod("400:ia,'xx'") r = language.parse_pathod("400:ia,'xx'").next()
language.serve(r, s, {}) language.serve(r, s, {})
assert s.getvalue().endswith('xx') assert s.getvalue().endswith('xx')