Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2025-02-07 10:40:09 +00:00
Code Issues Projects Releases Wiki Activity

addons.streambodies: 100% test coverage

Browse Source
This commit is contained in:
Aldo Cortesi 2016-11-03 20:16:07 +13:00
parent d7d6edb3d1
commit 5ade93f2ad
2 changed files with 12 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
mitmproxy/addons/streambodies.py
View File

@ -17,7 +17,7 @@ class StreamBodies:
expected_size = http1.expected_http_body_size( expected_size = http1.expected_http_body_size(
f.request, f.response if not is_request else None f.request, f.response if not is_request else None
) )
except exceptions.HTTPException: except exceptions.HttpException:
f.reply.kill() f.reply.kill()
return return
if expected_size and not r.raw_content and not (0 <= expected_size <= self.max_size): if expected_size and not r.raw_content and not (0 <= expected_size <= self.max_size):

24
test/mitmproxy/addons/test_streambodies.py
View File

@ -1,30 +1,28 @@
from mitmproxy.test import tflow from mitmproxy.test import tflow
from mitmproxy.test import taddons
from .. import mastertest
from mitmproxy import master
from mitmproxy import options
from mitmproxy import proxy
from mitmproxy.addons import streambodies from mitmproxy.addons import streambodies
class TestStreamBodies(mastertest.MasterTest): def test_simple():
def test_simple(self): sa = streambodies.StreamBodies()
o = options.Options(stream_large_bodies = 10) with taddons.context() as tctx:
m = master.Master(o, proxy.DummyServer()) tctx.configure(sa, stream_large_bodies = 10)
sa = streambodies.StreamBodies()
m.addons.add(sa)
f = tflow.tflow() f = tflow.tflow()
f.request.content = b"" f.request.content = b""
f.request.headers["Content-Length"] = "1024" f.request.headers["Content-Length"] = "1024"
assert not f.request.stream assert not f.request.stream
m.requestheaders(f) sa.requestheaders(f)
assert f.request.stream assert f.request.stream
f = tflow.tflow(resp=True) f = tflow.tflow(resp=True)
f.response.content = b"" f.response.content = b""
f.response.headers["Content-Length"] = "1024" f.response.headers["Content-Length"] = "1024"
assert not f.response.stream assert not f.response.stream
m.responseheaders(f) sa.responseheaders(f)
assert f.response.stream assert f.response.stream
f = tflow.tflow(resp=True)
f.response.headers["content-length"] = "invalid"
tctx.cycle(sa, f)