From 356c4987a2a43816c59f5cd070097d2e777b2272 Mon Sep 17 00:00:00 2001 From: Aldo Cortesi Date: Sun, 29 Jan 2017 11:37:51 +1300 Subject: [PATCH] addons: unit tests to 100% This patch pushes the coverage of ./tests/mitproxy/addons to 100% of the addons module. --- test/mitmproxy/addons/test_serverplayback.py | 61 +++++++++++++++----- test/mitmproxy/addons/test_stickyauth.py | 10 +++- test/mitmproxy/addons/test_stickycookie.py | 15 +++-- test/mitmproxy/addons/test_streamfile.py | 4 ++ test/mitmproxy/addons/test_view.py | 3 + 5 files changed, 73 insertions(+), 20 deletions(-) diff --git a/test/mitmproxy/addons/test_serverplayback.py b/test/mitmproxy/addons/test_serverplayback.py index 613a290c3..37766dfae 100644 --- a/test/mitmproxy/addons/test_serverplayback.py +++ b/test/mitmproxy/addons/test_serverplayback.py @@ -1,4 +1,6 @@ import os +import urllib + from mitmproxy.test import tutils from mitmproxy.test import tflow from mitmproxy.test import taddons @@ -22,8 +24,13 @@ def test_config(): with taddons.context() as tctx: fpath = os.path.join(p, "flows") tdump(fpath, [tflow.tflow(resp=True)]) - tctx.configure(s, server_replay = [fpath]) - tutils.raises(exceptions.OptionsError, tctx.configure, s, server_replay = [p]) + tctx.configure(s, server_replay=[fpath]) + tutils.raises( + exceptions.OptionsError, + tctx.configure, + s, + server_replay=[p] + ) def test_tick(): @@ -246,7 +253,7 @@ def test_ignore_params(): assert not s._hash(r) == s._hash(r2) -def test_ignore_payload_params(): +def thash(r, r2, setter): s = serverplayback.ServerPlayback() s.configure( options.Options( @@ -255,31 +262,59 @@ def test_ignore_payload_params(): [] ) - r = tflow.tflow(resp=True) - r.request.headers["Content-Type"] = "application/x-www-form-urlencoded" - r.request.content = b"paramx=x¶m1=1" - r2 = tflow.tflow(resp=True) - r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded" - r2.request.content = b"paramx=x¶m1=1" + setter(r, paramx="x", param1="1") + + setter(r2, paramx="x", param1="1") # same parameters assert s._hash(r) == s._hash(r2) # ignored parameters != - r2.request.content = b"paramx=x¶m1=2" + setter(r2, paramx="x", param1="2") assert s._hash(r) == s._hash(r2) # missing parameter - r2.request.content = b"paramx=x" + setter(r2, paramx="x") assert s._hash(r) == s._hash(r2) # ignorable parameter added - r2.request.content = b"paramx=x¶m1=2" + setter(r2, paramx="x", param1="2") assert s._hash(r) == s._hash(r2) # not ignorable parameter changed - r2.request.content = b"paramx=y¶m1=1" + setter(r2, paramx="y", param1="1") assert not s._hash(r) == s._hash(r2) # not ignorable parameter missing + setter(r2, param1="1") r2.request.content = b"param1=1" assert not s._hash(r) == s._hash(r2) +def test_ignore_payload_params(): + def urlencode_setter(r, **kwargs): + r.request.content = urllib.parse.urlencode(kwargs).encode() + + r = tflow.tflow(resp=True) + r.request.headers["Content-Type"] = "application/x-www-form-urlencoded" + r2 = tflow.tflow(resp=True) + r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded" + thash(r, r2, urlencode_setter) + + boundary = 'somefancyboundary' + + def multipart_setter(r, **kwargs): + b = "--{0}\n".format(boundary) + parts = [] + for k, v in kwargs.items(): + parts.append( + "Content-Disposition: form-data; name=\"%s\"\n\n" + "%s\n" % (k, v) + ) + c = b + b.join(parts) + b + r.request.content = c.encode() + r.request.headers["content-type"] = 'multipart/form-data; boundary=' +\ + boundary + + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) + thash(r, r2, multipart_setter) + + def test_server_playback_full(): s = serverplayback.ServerPlayback() with taddons.context() as tctx: diff --git a/test/mitmproxy/addons/test_stickyauth.py b/test/mitmproxy/addons/test_stickyauth.py index df74f44d9..a21ad38fa 100644 --- a/test/mitmproxy/addons/test_stickyauth.py +++ b/test/mitmproxy/addons/test_stickyauth.py @@ -10,7 +10,15 @@ def test_configure(): r = stickyauth.StickyAuth() with taddons.context() as tctx: tctx.configure(r, stickyauth="~s") - tutils.raises(exceptions.OptionsError, tctx.configure, r, stickyauth="~~") + tutils.raises( + exceptions.OptionsError, + tctx.configure, + r, + stickyauth="~~" + ) + + tctx.configure(r, stickyauth=None) + assert not r.flt def test_simple(): diff --git a/test/mitmproxy/addons/test_stickycookie.py b/test/mitmproxy/addons/test_stickycookie.py index df5d0221b..4a1ede8eb 100644 --- a/test/mitmproxy/addons/test_stickycookie.py +++ b/test/mitmproxy/addons/test_stickycookie.py @@ -3,7 +3,6 @@ from mitmproxy.test import tutils from mitmproxy.test import taddons from mitmproxy.addons import stickycookie -from mitmproxy import options from mitmproxy.test import tutils as ntutils @@ -15,11 +14,15 @@ def test_domain_match(): class TestStickyCookie: def test_config(self): sc = stickycookie.StickyCookie() - o = options.Options(stickycookie = "~b") - tutils.raises( - "invalid filter", - sc.configure, o, o.keys() - ) + with taddons.context() as tctx: + tutils.raises( + "invalid filter", tctx.configure, sc, stickycookie="~b" + ) + + tctx.configure(sc, stickycookie="foo") + assert sc.flt + tctx.configure(sc, stickycookie=None) + assert not sc.flt def test_simple(self): sc = stickycookie.StickyCookie() diff --git a/test/mitmproxy/addons/test_streamfile.py b/test/mitmproxy/addons/test_streamfile.py index 82a4345b4..4c60cd513 100644 --- a/test/mitmproxy/addons/test_streamfile.py +++ b/test/mitmproxy/addons/test_streamfile.py @@ -22,6 +22,10 @@ def test_configure(): "invalid filter", tctx.configure, sa, streamfile=p, filtstr="~~" ) + tctx.configure(sa, filtstr="foo") + assert sa.filt + tctx.configure(sa, filtstr=None) + assert not sa.filt def rd(p): diff --git a/test/mitmproxy/addons/test_view.py b/test/mitmproxy/addons/test_view.py index 96f213e26..58991ffd5 100644 --- a/test/mitmproxy/addons/test_view.py +++ b/test/mitmproxy/addons/test_view.py @@ -73,12 +73,15 @@ def test_simple(): assert v.store_count() == 0 v.request(f) assert list(v) == [f] + assert v.get_by_id(f.id) + assert not v.get_by_id("nonexistent") # These all just call udpate v.error(f) v.response(f) v.intercept(f) v.resume(f) + v.kill(f) assert list(v) == [f] v.request(f)