diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/mitmproxy/addons/dumperview.py | 57 | ||||
| -rw-r--r-- | test/mitmproxy/addons/test_dumper.py | 101 |
2 files changed, 151 insertions, 7 deletions
diff --git a/test/mitmproxy/addons/dumperview.py b/test/mitmproxy/addons/dumperview.py new file mode 100755 index 00000000..be56fe14 --- /dev/null +++ b/test/mitmproxy/addons/dumperview.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python +import click + +from mitmproxy.addons import dumper +from mitmproxy.test import tflow +from mitmproxy.test import taddons +from mitmproxy.tools import dump + + +def show(flow_detail, flows): + d = dumper.Dumper() + with taddons.context(options=dump.Options()) as ctx: + ctx.configure(d, flow_detail=flow_detail) + for f in flows: + ctx.cycle(d, f) + + +@click.group() +def cli(): + pass + + +@cli.command() +@click.option('--level', default=1, help='Detail level') +def tcp(level): + f1 = tflow.ttcpflow(client_conn=True, server_conn=True) + show(level, [f1]) + + +@cli.command() +@click.option('--level', default=1, help='Detail level') +def large(level): + f1 = tflow.tflow(client_conn=True, server_conn=True, resp=True) + f1.response.headers["content-type"] = "text/html" + f1.response.content = b"foo bar voing\n" * 100 + show(level, [f1]) + + +@cli.command() +@click.option('--level', default=1, help='Detail level') +def small(level): + f1 = tflow.tflow(client_conn=True, server_conn=True, resp=True) + f1.response.headers["content-type"] = "text/html" + f1.response.content = b"<html><body>Hello!</body></html>" + + f2 = tflow.tflow(client_conn=True, server_conn=True, err=True) + + show( + level, + [ + f1, f2, + ] + ) + + +if __name__ == "__main__": + cli() diff --git a/test/mitmproxy/addons/test_dumper.py b/test/mitmproxy/addons/test_dumper.py index ffcef210..0d61f800 100644 --- a/test/mitmproxy/addons/test_dumper.py +++ b/test/mitmproxy/addons/test_dumper.py @@ -1,66 +1,138 @@ import io from mitmproxy.test import tflow from mitmproxy.test import taddons +from mitmproxy.test import tutils from mitmproxy.addons import dumper from mitmproxy import exceptions from mitmproxy.tools import dump from mitmproxy import http -import mitmproxy.test.tutils import mock +def test_configure(): + d = dumper.Dumper() + with taddons.context(options=dump.Options()) as ctx: + ctx.configure(d, filtstr="~b foo") + assert d.filter + + f = tflow.tflow(resp=True) + assert not d.match(f) + f.response.content = b"foo" + assert d.match(f) + + ctx.configure(d, filtstr=None) + assert not d.filter + tutils.raises(exceptions.OptionsError, ctx.configure, d, filtstr="~~") + assert not d.filter + + def test_simple(): d = dumper.Dumper() with taddons.context(options=dump.Options()) as ctx: sio = io.StringIO() ctx.configure(d, tfile = sio, flow_detail = 0) - d.response(tflow.tflow()) + d.response(tflow.tflow(resp=True)) assert not sio.getvalue() + sio.truncate(0) + + ctx.configure(d, tfile = sio, flow_detail = 1) + d.response(tflow.tflow(resp=True)) + assert sio.getvalue() + sio.truncate(0) + + ctx.configure(d, tfile = sio, flow_detail = 1) + d.error(tflow.tflow(err=True)) + assert sio.getvalue() + sio.truncate(0) ctx.configure(d, tfile = sio, flow_detail = 4) - d.response(tflow.tflow()) + d.response(tflow.tflow(resp=True)) assert sio.getvalue() + sio.truncate(0) sio = io.StringIO() ctx.configure(d, tfile = sio, flow_detail = 4) d.response(tflow.tflow(resp=True)) assert "<<" in sio.getvalue() + sio.truncate(0) sio = io.StringIO() ctx.configure(d, tfile = sio, flow_detail = 4) d.response(tflow.tflow(err=True)) assert "<<" in sio.getvalue() + sio.truncate(0) sio = io.StringIO() ctx.configure(d, tfile = sio, flow_detail = 4) flow = tflow.tflow() - flow.request = mitmproxy.test.tutils.treq() + flow.request = tutils.treq() flow.request.stickycookie = True flow.client_conn = mock.MagicMock() flow.client_conn.address.host = "foo" - flow.response = mitmproxy.test.tutils.tresp(content=None) + flow.response = tutils.tresp(content=None) flow.response.is_replay = True flow.response.status_code = 300 d.response(flow) assert sio.getvalue() + sio.truncate(0) sio = io.StringIO() ctx.configure(d, tfile = sio, flow_detail = 4) - flow = tflow.tflow(resp=mitmproxy.test.tutils.tresp(content=b"{")) + flow = tflow.tflow(resp=tutils.tresp(content=b"{")) flow.response.headers["content-type"] = "application/json" flow.response.status_code = 400 d.response(flow) assert sio.getvalue() + sio.truncate(0) sio = io.StringIO() ctx.configure(d, tfile = sio, flow_detail = 4) flow = tflow.tflow() flow.request.content = None - flow.response = http.HTTPResponse.wrap(mitmproxy.test.tutils.tresp()) + flow.response = http.HTTPResponse.wrap(tutils.tresp()) flow.response.content = None d.response(flow) assert "content missing" in sio.getvalue() + sio.truncate(0) + + +def test_echo_body(): + f = tflow.tflow(client_conn=True, server_conn=True, resp=True) + f.response.headers["content-type"] = "text/html" + f.response.content = b"foo bar voing\n" * 100 + + d = dumper.Dumper() + sio = io.StringIO() + with taddons.context(options=dump.Options()) as ctx: + ctx.configure(d, tfile=sio, flow_detail = 3) + d._echo_message(f.response) + t = sio.getvalue() + assert "cut off" in t + + +def test_echo_request_line(): + d = dumper.Dumper() + sio = io.StringIO() + with taddons.context(options=dump.Options()) as ctx: + ctx.configure(d, tfile=sio, flow_detail = 3, showhost = True) + f = tflow.tflow(client_conn=None, server_conn=True, resp=True) + f.request.is_replay = True + d._echo_request_line(f) + assert "[replay]" in sio.getvalue() + sio.truncate(0) + + f = tflow.tflow(client_conn=None, server_conn=True, resp=True) + f.request.is_replay = False + d._echo_request_line(f) + assert "[replay]" not in sio.getvalue() + sio.truncate(0) + + f = tflow.tflow(client_conn=None, server_conn=True, resp=True) + f.request.http_version = "nonstandard" + d._echo_request_line(f) + assert "nonstandard" in sio.getvalue() + sio.truncate(0) class TestContentView: @@ -73,3 +145,18 @@ class TestContentView: ctx.configure(d, flow_detail=4, verbosity=3, tfile=sio) d.response(tflow.tflow()) assert "Content viewer failed" in ctx.master.event_log[0][1] + + +def test_tcp(): + d = dumper.Dumper() + sio = io.StringIO() + with taddons.context(options=dump.Options()) as ctx: + ctx.configure(d, tfile=sio, flow_detail = 3, showhost = True) + f = tflow.ttcpflow(client_conn=True, server_conn=True) + d.tcp_message(f) + assert "it's me" in sio.getvalue() + sio.truncate(0) + + f = tflow.ttcpflow(client_conn=True, err=True) + d.tcp_error(f) + assert "Error in TCP" in sio.getvalue() |