diff options
27 files changed, 302 insertions, 264 deletions
diff --git a/mitmproxy/test/tflow.py b/mitmproxy/test/tflow.py new file mode 100644 index 00000000..fcf619b5 --- /dev/null +++ b/mitmproxy/test/tflow.py @@ -0,0 +1,107 @@ +import mitmproxy.test +from mitmproxy import tcp +from mitmproxy import controller +from mitmproxy import http +from mitmproxy import connections +from mitmproxy import flow + + +def ttcpflow(client_conn=True, server_conn=True, messages=True, err=None): + if client_conn is True: + client_conn = tclient_conn() + if server_conn is True: + server_conn = tserver_conn() + if messages is True: + messages = [ + tcp.TCPMessage(True, b"hello"), + tcp.TCPMessage(False, b"it's me"), + ] + if err is True: + err = terr() + + f = tcp.TCPFlow(client_conn, server_conn) + f.messages = messages + f.error = err + f.reply = controller.DummyReply() + return f + + +def tflow(client_conn=True, server_conn=True, req=True, resp=None, err=None): + """ + @type client_conn: bool | None | mitmproxy.proxy.connection.ClientConnection + @type server_conn: bool | None | mitmproxy.proxy.connection.ServerConnection + @type req: bool | None | mitmproxy.proxy.protocol.http.HTTPRequest + @type resp: bool | None | mitmproxy.proxy.protocol.http.HTTPResponse + @type err: bool | None | mitmproxy.proxy.protocol.primitives.Error + @return: mitmproxy.proxy.protocol.http.HTTPFlow + """ + if client_conn is True: + client_conn = tclient_conn() + if server_conn is True: + server_conn = tserver_conn() + if req is True: + req = mitmproxy.test.tutils.treq() + if resp is True: + resp = mitmproxy.test.tutils.tresp() + if err is True: + err = terr() + + if req: + req = http.HTTPRequest.wrap(req) + if resp: + resp = http.HTTPResponse.wrap(resp) + + f = http.HTTPFlow(client_conn, server_conn) + f.request = req + f.response = resp + f.error = err + f.reply = controller.DummyReply() + return f + + +def tclient_conn(): + """ + @return: mitmproxy.proxy.connection.ClientConnection + """ + c = connections.ClientConnection.from_state(dict( + address=dict(address=("address", 22), use_ipv6=True), + clientcert=None, + ssl_established=False, + timestamp_start=1, + timestamp_ssl_setup=2, + timestamp_end=3, + sni="address", + cipher_name="cipher", + tls_version="TLSv1.2", + )) + c.reply = controller.DummyReply() + return c + + +def tserver_conn(): + """ + @return: mitmproxy.proxy.connection.ServerConnection + """ + c = connections.ServerConnection.from_state(dict( + address=dict(address=("address", 22), use_ipv6=True), + source_address=dict(address=("address", 22), use_ipv6=True), + ip_address=None, + cert=None, + timestamp_start=1, + timestamp_tcp_setup=2, + timestamp_ssl_setup=3, + timestamp_end=4, + ssl_established=False, + sni="address", + via=None + )) + c.reply = controller.DummyReply() + return c + + +def terr(content="error"): + """ + @return: mitmproxy.proxy.protocol.primitives.Error + """ + err = flow.Error(content) + return err diff --git a/test/mitmproxy/addons/test_anticache.py b/test/mitmproxy/addons/test_anticache.py index f7418f38..44b96dfe 100644 --- a/test/mitmproxy/addons/test_anticache.py +++ b/test/mitmproxy/addons/test_anticache.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest from mitmproxy.addons import anticache from mitmproxy import master @@ -12,10 +14,10 @@ class TestAntiCache(mastertest.MasterTest): sa = anticache.AntiCache() m.addons.add(sa) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) m.request(f) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.headers["if-modified-since"] = "test" f.request.headers["if-none-match"] = "test" m.request(f) diff --git a/test/mitmproxy/addons/test_anticomp.py b/test/mitmproxy/addons/test_anticomp.py index 5f95013f..0089f97f 100644 --- a/test/mitmproxy/addons/test_anticomp.py +++ b/test/mitmproxy/addons/test_anticomp.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest from mitmproxy.addons import anticomp from mitmproxy import master @@ -12,10 +14,10 @@ class TestAntiComp(mastertest.MasterTest): sa = anticomp.AntiComp() m.addons.add(sa) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) m.request(f) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.headers["Accept-Encoding"] = "foobar" m.request(f) diff --git a/test/mitmproxy/addons/test_clientplayback.py b/test/mitmproxy/addons/test_clientplayback.py index f15b6290..f8fbd4e5 100644 --- a/test/mitmproxy/addons/test_clientplayback.py +++ b/test/mitmproxy/addons/test_clientplayback.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow import mock from mitmproxy.addons import clientplayback @@ -11,7 +12,7 @@ class TestClientPlayback: cp = clientplayback.ClientPlayback() cp.configure(options.Options(), []) assert cp.count() == 0 - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) cp.load([f]) assert cp.count() == 1 RP = "mitmproxy.proxy.protocol.http_replay.RequestReplayThread" diff --git a/test/mitmproxy/addons/test_dumper.py b/test/mitmproxy/addons/test_dumper.py index 5f1b2dbd..71d0583b 100644 --- a/test/mitmproxy/addons/test_dumper.py +++ b/test/mitmproxy/addons/test_dumper.py @@ -1,4 +1,5 @@ import io +from mitmproxy.test import tflow from .. import tutils, mastertest @@ -18,26 +19,26 @@ class TestDumper(mastertest.MasterTest): updated = {"tfile", "flow_detail"} d.configure(dump.Options(tfile = sio, flow_detail = 0), updated) - d.response(tutils.tflow()) + d.response(tflow.tflow()) assert not sio.getvalue() d.configure(dump.Options(tfile = sio, flow_detail = 4), updated) - d.response(tutils.tflow()) + d.response(tflow.tflow()) assert sio.getvalue() sio = io.StringIO() d.configure(dump.Options(tfile = sio, flow_detail = 4), updated) - d.response(tutils.tflow(resp=True)) + d.response(tflow.tflow(resp=True)) assert "<<" in sio.getvalue() sio = io.StringIO() d.configure(dump.Options(tfile = sio, flow_detail = 4), updated) - d.response(tutils.tflow(err=True)) + d.response(tflow.tflow(err=True)) assert "<<" in sio.getvalue() sio = io.StringIO() d.configure(dump.Options(tfile = sio, flow_detail = 4), updated) - flow = tutils.tflow() + flow = tflow.tflow() flow.request = mitmproxy.test.tutils.treq() flow.request.stickycookie = True flow.client_conn = mock.MagicMock() @@ -50,7 +51,7 @@ class TestDumper(mastertest.MasterTest): sio = io.StringIO() d.configure(dump.Options(tfile = sio, flow_detail = 4), updated) - flow = tutils.tflow(resp=mitmproxy.test.tutils.tresp(content=b"{")) + flow = tflow.tflow(resp=mitmproxy.test.tutils.tresp(content=b"{")) flow.response.headers["content-type"] = "application/json" flow.response.status_code = 400 d.response(flow) @@ -58,7 +59,7 @@ class TestDumper(mastertest.MasterTest): sio = io.StringIO() d.configure(dump.Options(tfile = sio), updated) - flow = tutils.tflow() + flow = tflow.tflow() flow.request.content = None flow.response = http.HTTPResponse.wrap(mitmproxy.test.tutils.tresp()) flow.response.content = None @@ -80,5 +81,5 @@ class TestContentView(mastertest.MasterTest): m = mastertest.RecordingMaster(o, proxy.DummyServer()) d = dumper.Dumper() m.addons.add(d) - m.response(tutils.tflow()) + m.response(tflow.tflow()) assert "Content viewer failed" in m.event_log[0][1] diff --git a/test/mitmproxy/addons/test_filestreamer.py b/test/mitmproxy/addons/test_filestreamer.py index 64208f84..49baaf3c 100644 --- a/test/mitmproxy/addons/test_filestreamer.py +++ b/test/mitmproxy/addons/test_filestreamer.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest import os.path @@ -25,7 +27,7 @@ class TestStream(mastertest.MasterTest): sa = filestreamer.FileStreamer() m.addons.add(sa) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) m.request(f) m.response(f) m.addons.remove(sa) @@ -35,7 +37,7 @@ class TestStream(mastertest.MasterTest): m.options.outfile = (p, "ab") m.addons.add(sa) - f = tutils.tflow() + f = tflow.tflow() m.request(f) m.addons.remove(sa) assert not r()[1].response diff --git a/test/mitmproxy/addons/test_replace.py b/test/mitmproxy/addons/test_replace.py index 94c2f832..e57fd295 100644 --- a/test/mitmproxy/addons/test_replace.py +++ b/test/mitmproxy/addons/test_replace.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest, tservers from mitmproxy.addons import replace from mitmproxy import master @@ -40,12 +42,12 @@ class TestReplace(mastertest.MasterTest): sa = replace.Replace() m.addons.add(sa) - f = tutils.tflow() + f = tflow.tflow() f.request.content = b"foo" m.request(f) assert f.request.content == b"bar" - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.content = b"foo" m.response(f) assert f.response.content == b"bar" diff --git a/test/mitmproxy/addons/test_script.py b/test/mitmproxy/addons/test_script.py index 83e60fbd..3ce274d7 100644 --- a/test/mitmproxy/addons/test_script.py +++ b/test/mitmproxy/addons/test_script.py @@ -3,6 +3,7 @@ import traceback import sys import time +from mitmproxy.test import tflow import re from mitmproxy import exceptions from mitmproxy import options @@ -71,7 +72,7 @@ class TestScript(mastertest.MasterTest): ] sc.ns.call_log = [] - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) m.request(f) recf = sc.ns.call_log[0] @@ -102,7 +103,7 @@ class TestScript(mastertest.MasterTest): tutils.test_data.path("data/addonscripts/error.py") ) m.addons.add(sc) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) m.request(f) assert m.event_log[0][0] == "error" assert len(m.event_log[0][1].splitlines()) == 6 @@ -149,7 +150,7 @@ class TestScriptLoader(mastertest.MasterTest): sl = script.ScriptLoader() m.addons.add(sl) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) with m.handlecontext(): sc = sl.run_once( tutils.test_data.path( diff --git a/test/mitmproxy/addons/test_serverplayback.py b/test/mitmproxy/addons/test_serverplayback.py index c8dd8704..7cc5a555 100644 --- a/test/mitmproxy/addons/test_serverplayback.py +++ b/test/mitmproxy/addons/test_serverplayback.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest import mitmproxy.test.tutils @@ -11,7 +13,7 @@ class TestServerPlayback: def test_server_playback(self): sp = serverplayback.ServerPlayback() sp.configure(options.Options(), []) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) assert not sp.flowmap @@ -24,8 +26,8 @@ class TestServerPlayback: sp = serverplayback.ServerPlayback() sp.configure(options.Options(server_replay_ignore_host=True), []) - r = tutils.tflow(resp=True) - r2 = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) r.request.host = "address" r2.request.host = "address" @@ -37,8 +39,8 @@ class TestServerPlayback: s = serverplayback.ServerPlayback() s.configure(options.Options(server_replay_ignore_content=False), []) - r = tutils.tflow(resp=True) - r2 = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) r.request.content = b"foo" r2.request.content = b"foo" @@ -47,8 +49,8 @@ class TestServerPlayback: assert not s._hash(r) == s._hash(r2) s.configure(options.Options(server_replay_ignore_content=True), []) - r = tutils.tflow(resp=True) - r2 = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) r.request.content = b"foo" r2.request.content = b"foo" assert s._hash(r) == s._hash(r2) @@ -72,11 +74,11 @@ class TestServerPlayback: ) # NOTE: parameters are mutually exclusive in options - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.headers["Content-Type"] = "application/x-www-form-urlencoded" r.request.content = b"paramx=y" - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded" r2.request.content = b"paramx=x" @@ -95,10 +97,10 @@ class TestServerPlayback: [] ) - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.headers["Content-Type"] = "application/json" r.request.content = b'{"param1":"1"}' - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) r2.request.headers["Content-Type"] = "application/json" r2.request.content = b'{"param1":"1"}' # same content @@ -111,8 +113,8 @@ class TestServerPlayback: s = serverplayback.ServerPlayback() s.configure(options.Options(), []) - r = tutils.tflow() - r2 = tutils.tflow() + r = tflow.tflow() + r2 = tflow.tflow() assert s._hash(r) assert s._hash(r) == s._hash(r2) @@ -129,27 +131,27 @@ class TestServerPlayback: s = serverplayback.ServerPlayback() s.configure(options.Options(server_replay_use_headers=["foo"]), []) - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.headers["foo"] = "bar" - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) assert not s._hash(r) == s._hash(r2) r2.request.headers["foo"] = "bar" assert s._hash(r) == s._hash(r2) r2.request.headers["oink"] = "bar" assert s._hash(r) == s._hash(r2) - r = tutils.tflow(resp=True) - r2 = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) assert s._hash(r) == s._hash(r2) def test_load(self): s = serverplayback.ServerPlayback() s.configure(options.Options(), []) - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.headers["key"] = "one" - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) r2.request.headers["key"] = "two" s.load([r, r2]) @@ -171,10 +173,10 @@ class TestServerPlayback: s = serverplayback.ServerPlayback() s.configure(options.Options(server_replay_nopop=True), []) - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.headers["key"] = "one" - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) r2.request.headers["key"] = "two" s.load([r, r2]) @@ -192,9 +194,9 @@ class TestServerPlayback: [] ) - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.path = "/test?param1=1" - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) r2.request.path = "/test" assert s._hash(r) == s._hash(r2) r2.request.path = "/test?param1=2" @@ -213,10 +215,10 @@ class TestServerPlayback: [] ) - r = tutils.tflow(resp=True) + r = tflow.tflow(resp=True) r.request.headers["Content-Type"] = "application/x-www-form-urlencoded" r.request.content = b"paramx=x¶m1=1" - r2 = tutils.tflow(resp=True) + r2 = tflow.tflow(resp=True) r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded" r2.request.content = b"paramx=x¶m1=1" # same parameters @@ -243,16 +245,16 @@ class TestServerPlayback: m = mastertest.RecordingMaster(o, proxy.DummyServer()) m.addons.add(s) - f = tutils.tflow() + f = tflow.tflow() f.response = mitmproxy.test.tutils.tresp(content=f.request.content) s.load([f, f]) - tf = tutils.tflow() + tf = tflow.tflow() assert not tf.response m.request(tf) assert tf.response == f.response - tf = tutils.tflow() + tf = tflow.tflow() tf.request.content = b"gibble" assert not tf.response m.request(tf) @@ -262,8 +264,8 @@ class TestServerPlayback: s.tick() assert not s.stop - tf = tutils.tflow() - m.request(tutils.tflow()) + tf = tflow.tflow() + m.request(tflow.tflow()) assert s.stop def test_server_playback_kill(self): @@ -272,11 +274,11 @@ class TestServerPlayback: m = mastertest.RecordingMaster(o, proxy.DummyServer()) m.addons.add(s) - f = tutils.tflow() + f = tflow.tflow() f.response = mitmproxy.test.tutils.tresp(content=f.request.content) s.load([f]) - f = tutils.tflow() + f = tflow.tflow() f.request.host = "nonexistent" m.request(f) assert f.reply.value == exceptions.Kill diff --git a/test/mitmproxy/addons/test_setheaders.py b/test/mitmproxy/addons/test_setheaders.py index 0fbbd4cb..4cdd8e8e 100644 --- a/test/mitmproxy/addons/test_setheaders.py +++ b/test/mitmproxy/addons/test_setheaders.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest from mitmproxy.addons import setheaders @@ -30,12 +32,12 @@ class TestSetHeaders(mastertest.MasterTest): ("~s", "one", "three") ] ) - f = tutils.tflow() + f = tflow.tflow() f.request.headers["one"] = "xxx" m.request(f) assert f.request.headers["one"] == "two" - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.headers["one"] = "xxx" m.response(f) assert f.response.headers["one"] == "three" @@ -46,7 +48,7 @@ class TestSetHeaders(mastertest.MasterTest): ("~s", "one", "three") ] ) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.headers["one"] = "xxx" f.response.headers["one"] = "xxx" m.response(f) @@ -58,7 +60,7 @@ class TestSetHeaders(mastertest.MasterTest): ("~q", "one", "three") ] ) - f = tutils.tflow() + f = tflow.tflow() f.request.headers["one"] = "xxx" m.request(f) assert f.request.headers.get_all("one") == ["two", "three"] diff --git a/test/mitmproxy/addons/test_state.py b/test/mitmproxy/addons/test_state.py index 71c46dcb..deb417f6 100644 --- a/test/mitmproxy/addons/test_state.py +++ b/test/mitmproxy/addons/test_state.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow from mitmproxy import proxy from mitmproxy import master from mitmproxy.addons import state @@ -10,7 +11,7 @@ class TestState: s = state.State() fm = master.Master(None, proxy.DummyServer()) fm.addons.add(s) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) fm.load_flow(f) assert s.flow_count() == 1 diff --git a/test/mitmproxy/addons/test_stickyauth.py b/test/mitmproxy/addons/test_stickyauth.py index 459dc5f7..4121507d 100644 --- a/test/mitmproxy/addons/test_stickyauth.py +++ b/test/mitmproxy/addons/test_stickyauth.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest from mitmproxy.addons import stickyauth from mitmproxy import master @@ -12,12 +14,12 @@ class TestStickyAuth(mastertest.MasterTest): sa = stickyauth.StickyAuth() m.addons.add(sa) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.headers["authorization"] = "foo" m.request(f) assert "address" in sa.hosts - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) m.request(f) assert f.request.headers["authorization"] == "foo" diff --git a/test/mitmproxy/addons/test_stickycookie.py b/test/mitmproxy/addons/test_stickycookie.py index 28c35b28..56d1ab5d 100644 --- a/test/mitmproxy/addons/test_stickycookie.py +++ b/test/mitmproxy/addons/test_stickycookie.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest from mitmproxy.addons import stickycookie from mitmproxy import master @@ -31,7 +33,7 @@ class TestStickyCookie(mastertest.MasterTest): m, sc = self.mk() m.addons.add(sc) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.headers["set-cookie"] = "foo=bar" m.request(f) @@ -47,7 +49,7 @@ class TestStickyCookie(mastertest.MasterTest): assert f.request.headers["cookie"] == "foo=bar" def _response(self, m, sc, cookie, host): - f = tutils.tflow(req=ntutils.treq(host=host, port=80), resp=True) + f = tflow.tflow(req=ntutils.treq(host=host, port=80), resp=True) f.response.headers["Set-Cookie"] = cookie m.response(f) return f @@ -86,7 +88,7 @@ class TestStickyCookie(mastertest.MasterTest): m, sc = self.mk() # Test setting of weird cookie keys - f = tutils.tflow(req=ntutils.treq(host="www.google.com", port=80), resp=True) + f = tflow.tflow(req=ntutils.treq(host="www.google.com", port=80), resp=True) cs = [ "foo/bar=hello", "foo:bar=world", diff --git a/test/mitmproxy/addons/test_streambodies.py b/test/mitmproxy/addons/test_streambodies.py index cf416c86..26f6fb59 100644 --- a/test/mitmproxy/addons/test_streambodies.py +++ b/test/mitmproxy/addons/test_streambodies.py @@ -1,3 +1,5 @@ +from mitmproxy.test import tflow + from .. import tutils, mastertest from mitmproxy import master from mitmproxy import options @@ -13,14 +15,14 @@ class TestStreamBodies(mastertest.MasterTest): sa = streambodies.StreamBodies() m.addons.add(sa) - f = tutils.tflow() + f = tflow.tflow() f.request.content = b"" f.request.headers["Content-Length"] = "1024" assert not f.request.stream m.requestheaders(f) assert f.request.stream - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.content = b"" f.response.headers["Content-Length"] = "1024" assert not f.response.stream diff --git a/test/mitmproxy/addons/test_view.py b/test/mitmproxy/addons/test_view.py index 15cf534e..2715fd6a 100644 --- a/test/mitmproxy/addons/test_view.py +++ b/test/mitmproxy/addons/test_view.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow from mitmproxy.addons import view from mitmproxy import flowfilter @@ -6,7 +7,7 @@ from .. import tutils def test_simple(): v = view.View() - f = tutils.tflow() + f = tflow.tflow() f.request.timestamp_start = 1 v.request(f) assert list(v) == [f] @@ -14,7 +15,7 @@ def test_simple(): assert list(v) == [f] assert len(v._store) == 1 - f2 = tutils.tflow() + f2 = tflow.tflow() f2.request.timestamp_start = 3 v.request(f2) assert list(v) == [f, f2] @@ -22,7 +23,7 @@ def test_simple(): assert list(v) == [f, f2] assert len(v._store) == 2 - f3 = tutils.tflow() + f3 = tflow.tflow() f3.request.timestamp_start = 2 v.request(f3) assert list(v) == [f, f3, f2] @@ -32,7 +33,7 @@ def test_simple(): def tft(*, method="get", start=0): - f = tutils.tflow() + f = tflow.tflow() f.request.method = method f.request.timestamp_start = start return f diff --git a/test/mitmproxy/console/test_common.py b/test/mitmproxy/console/test_common.py index 1a12d360..49ce35b7 100644 --- a/test/mitmproxy/console/test_common.py +++ b/test/mitmproxy/console/test_common.py @@ -1,10 +1,11 @@ +from mitmproxy.test import tflow from mitmproxy.tools.console import common from .. import tutils @tutils.skip_appveyor def test_format_flow(): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) assert common.format_flow(f, True) assert common.format_flow(f, True, hostheader=True) assert common.format_flow(f, True, extended=True) diff --git a/test/mitmproxy/console/test_master.py b/test/mitmproxy/console/test_master.py index 6d4bb7f9..78ee6175 100644 --- a/test/mitmproxy/console/test_master.py +++ b/test/mitmproxy/console/test_master.py @@ -1,5 +1,6 @@ import gc +from mitmproxy.test import tflow import mitmproxy.test.tutils from mitmproxy.tools import console from mitmproxy import proxy @@ -54,7 +55,7 @@ class TestConsoleState: assert c.get_focus() == (None, None) def _add_request(self, state): - f = tutils.tflow() + f = tflow.tflow() return state.add_flow(f) def _add_response(self, state): @@ -127,12 +128,12 @@ class TestMaster(mastertest.MasterTest): def test_intercept(self): """regression test for https://github.com/mitmproxy/mitmproxy/issues/1605""" m = self.mkmaster(intercept="~b bar") - f = tutils.tflow(req=mitmproxy.test.tutils.treq(content=b"foo")) + f = tflow.tflow(req=mitmproxy.test.tutils.treq(content=b"foo")) m.request(f) assert not m.state.flows[0].intercepted - f = tutils.tflow(req=mitmproxy.test.tutils.treq(content=b"bar")) + f = tflow.tflow(req=mitmproxy.test.tutils.treq(content=b"bar")) m.request(f) assert m.state.flows[1].intercepted - f = tutils.tflow(resp=mitmproxy.test.tutils.tresp(content=b"bar")) + f = tflow.tflow(resp=mitmproxy.test.tutils.tresp(content=b"bar")) m.request(f) assert m.state.flows[2].intercepted diff --git a/test/mitmproxy/mastertest.py b/test/mitmproxy/mastertest.py index 0672add9..10c7fbc2 100644 --- a/test/mitmproxy/mastertest.py +++ b/test/mitmproxy/mastertest.py @@ -1,5 +1,7 @@ import contextlib +from mitmproxy.test import tflow + from . import tutils import mitmproxy.test.tutils @@ -13,7 +15,7 @@ from mitmproxy import options class MasterTest: def cycle(self, master, content): - f = tutils.tflow(req=mitmproxy.test.tutils.treq(content=content)) + f = tflow.tflow(req=mitmproxy.test.tutils.treq(content=content)) master.clientconnect(f.client_conn) master.serverconnect(f.server_conn) master.request(f) @@ -33,7 +35,7 @@ class MasterTest: def flowfile(self, path): f = open(path, "wb") fw = io.FlowWriter(f) - t = tutils.tflow(resp=True) + t = tflow.tflow(resp=True) fw.add(t) f.close() diff --git a/test/mitmproxy/protocol/test_http1.py b/test/mitmproxy/protocol/test_http1.py index 43c903f3..a382867f 100644 --- a/test/mitmproxy/protocol/test_http1.py +++ b/test/mitmproxy/protocol/test_http1.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow from mitmproxy.net.http import http1 from mitmproxy.net.tcp import TCPClient from mitmproxy.test.tutils import treq @@ -7,7 +8,7 @@ from .. import tutils, tservers class TestHTTPFlow: def test_repr(self): - f = tutils.tflow(resp=True, err=True) + f = tflow.tflow(resp=True, err=True) assert repr(f) diff --git a/test/mitmproxy/script/test_concurrent.py b/test/mitmproxy/script/test_concurrent.py index 51c36abf..fec0d1b7 100644 --- a/test/mitmproxy/script/test_concurrent.py +++ b/test/mitmproxy/script/test_concurrent.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow from test.mitmproxy import tutils, mastertest from mitmproxy import controller from mitmproxy.addons import script @@ -23,7 +24,7 @@ class TestConcurrent(mastertest.MasterTest): ) ) m.addons.add(sc) - f1, f2 = tutils.tflow(), tutils.tflow() + f1, f2 = tflow.tflow(), tflow.tflow() m.request(f1) m.request(f2) start = time.time() diff --git a/test/mitmproxy/test_dump.py b/test/mitmproxy/test_dump.py index 33d2423c..8d699984 100644 --- a/test/mitmproxy/test_dump.py +++ b/test/mitmproxy/test_dump.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow import os import io @@ -45,7 +46,7 @@ class TestDumpMaster(mastertest.MasterTest): flow_detail=1 ) m = dump.DumpMaster(o, proxy.DummyServer()) - f = tutils.tflow(err=True) + f = tflow.tflow(err=True) m.error(f) assert "error" in o.tfile.getvalue() diff --git a/test/mitmproxy/test_examples.py b/test/mitmproxy/test_examples.py index 3cd29a50..4275b3c8 100644 --- a/test/mitmproxy/test_examples.py +++ b/test/mitmproxy/test_examples.py @@ -1,4 +1,6 @@ import json + +from mitmproxy.test import tflow import os import shlex @@ -40,7 +42,7 @@ def tscript(cmd, args=""): class TestScripts(mastertest.MasterTest): def test_add_header(self): m, _ = tscript("add_header.py") - f = tutils.tflow(resp=netutils.tresp()) + f = tflow.tflow(resp=netutils.tresp()) m.response(f) assert f.response.headers["newheader"] == "foo" @@ -56,7 +58,7 @@ class TestScripts(mastertest.MasterTest): tscript("iframe_injector.py") m, sc = tscript("iframe_injector.py", "http://example.org/evil_iframe") - f = tutils.tflow(resp=netutils.tresp(content=b"<html>mitmproxy</html>")) + f = tflow.tflow(resp=netutils.tresp(content=b"<html>mitmproxy</html>")) m.response(f) content = f.response.content assert b'iframe' in content and b'evil_iframe' in content @@ -65,7 +67,7 @@ class TestScripts(mastertest.MasterTest): m, sc = tscript("modify_form.py") form_header = Headers(content_type="application/x-www-form-urlencoded") - f = tutils.tflow(req=netutils.treq(headers=form_header)) + f = tflow.tflow(req=netutils.treq(headers=form_header)) m.request(f) assert f.request.urlencoded_form[b"mitmproxy"] == b"rocks" @@ -76,7 +78,7 @@ class TestScripts(mastertest.MasterTest): def test_modify_querystring(self): m, sc = tscript("modify_querystring.py") - f = tutils.tflow(req=netutils.treq(path="/search?q=term")) + f = tflow.tflow(req=netutils.treq(path="/search?q=term")) m.request(f) assert f.request.query["mitmproxy"] == "rocks" @@ -87,13 +89,13 @@ class TestScripts(mastertest.MasterTest): def test_arguments(self): m, sc = tscript("arguments.py", "mitmproxy rocks") - f = tutils.tflow(resp=netutils.tresp(content=b"I <3 mitmproxy")) + f = tflow.tflow(resp=netutils.tresp(content=b"I <3 mitmproxy")) m.response(f) assert f.response.content == b"I <3 rocks" def test_redirect_requests(self): m, sc = tscript("redirect_requests.py") - f = tutils.tflow(req=netutils.treq(host="example.org")) + f = tflow.tflow(req=netutils.treq(host="example.org")) m.request(f) assert f.request.host == "mitmproxy.org" @@ -107,7 +109,7 @@ class TestHARDump: ) # Create a dummy flow for testing - return tutils.tflow( + return tflow.tflow( req=netutils.treq(method=b'GET', **times), resp=netutils.tresp(content=resp_content, **times) ) diff --git a/test/mitmproxy/test_flow.py b/test/mitmproxy/test_flow.py index ed15a766..6be2bf13 100644 --- a/test/mitmproxy/test_flow.py +++ b/test/mitmproxy/test_flow.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow import mock import io @@ -20,7 +21,7 @@ from . import tutils class TestHTTPFlow: def test_copy(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.get_state() f2 = f.copy() a = f.get_state() @@ -37,7 +38,7 @@ class TestHTTPFlow: assert f.response.get_state() == f2.response.get_state() assert f.response is not f2.response - f = tutils.tflow(err=True) + f = tflow.tflow(err=True) f2 = f.copy() assert f is not f2 assert f.request is not f2.request @@ -47,18 +48,18 @@ class TestHTTPFlow: assert f.error is not f2.error def test_match(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) assert not flowfilter.match("~b test", f) assert flowfilter.match(None, f) assert not flowfilter.match("~b test", f) - f = tutils.tflow(err=True) + f = tflow.tflow(err=True) assert flowfilter.match("~e", f) tutils.raises(ValueError, flowfilter.match, "~", f) def test_backup(self): - f = tutils.tflow() + f = tflow.tflow() f.response = http.HTTPResponse.wrap(mitmproxy.test.tutils.tresp()) f.request.content = b"foo" assert not f.modified() @@ -69,14 +70,14 @@ class TestHTTPFlow: assert f.request.content == b"foo" def test_backup_idempotence(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.backup() f.revert() f.backup() f.revert() def test_getset_state(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) state = f.get_state() assert f.get_state() == http.HTTPFlow.from_state( state).get_state() @@ -98,7 +99,7 @@ class TestHTTPFlow: def test_kill(self): fm = mock.Mock() - f = tutils.tflow() + f = tflow.tflow() f.reply.handle() f.intercept(fm) assert f.killable @@ -112,7 +113,7 @@ class TestHTTPFlow: fm = master.Master(None, srv) fm.addons.add(s) - f = tutils.tflow() + f = tflow.tflow() f.reply.handle() f.intercept(fm) @@ -121,7 +122,7 @@ class TestHTTPFlow: assert "killed" in str(i.error) def test_resume(self): - f = tutils.tflow() + f = tflow.tflow() f.reply.handle() f.intercept(mock.Mock()) assert f.reply.state == "taken" @@ -129,17 +130,17 @@ class TestHTTPFlow: assert f.reply.state == "committed" def test_replace_unicode(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.content = b"\xc2foo" f.replace(b"foo", u"bar") def test_replace_no_content(self): - f = tutils.tflow() + f = tflow.tflow() f.request.content = None assert f.replace("foo", "bar") == 0 def test_replace(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.headers["foo"] = "foo" f.request.content = b"afoob" @@ -154,7 +155,7 @@ class TestHTTPFlow: assert f.response.content == b"abarb" def test_replace_encoded(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.content = b"afoob" f.request.encode("gzip") f.response.content = b"afoob" @@ -174,12 +175,12 @@ class TestHTTPFlow: class TestTCPFlow: def test_match(self): - f = tutils.ttcpflow() + f = tflow.ttcpflow() assert not flowfilter.match("~b nonexistent", f) assert flowfilter.match(None, f) assert not flowfilter.match("~b nonexistent", f) - f = tutils.ttcpflow(err=True) + f = tflow.ttcpflow(err=True) assert flowfilter.match("~e", f) tutils.raises(ValueError, flowfilter.match, "~", f) @@ -189,7 +190,7 @@ class TestState: def test_backup(self): c = state.State() - f = tutils.tflow() + f = tflow.tflow() c.add_flow(f) f.backup() c.revert(f) @@ -201,13 +202,13 @@ class TestState: connect -> request -> response """ c = state.State() - f = tutils.tflow() + f = tflow.tflow() c.add_flow(f) assert f assert c.flow_count() == 1 assert c.active_flow_count() == 1 - newf = tutils.tflow() + newf = tflow.tflow() assert c.add_flow(newf) assert c.active_flow_count() == 2 @@ -225,24 +226,24 @@ class TestState: def test_err(self): c = state.State() - f = tutils.tflow() + f = tflow.tflow() c.add_flow(f) f.error = flow.Error("message") assert c.update_flow(f) c = state.State() - f = tutils.tflow() + f = tflow.tflow() c.add_flow(f) c.set_view_filter("~e") assert not c.view - f.error = tutils.terr() + f.error = tflow.terr() assert c.update_flow(f) assert c.view def test_set_view_filter(self): c = state.State() - f = tutils.tflow() + f = tflow.tflow() assert len(c.view) == 0 c.add_flow(f) @@ -257,7 +258,7 @@ class TestState: c.set_view_filter(None) assert len(c.view) == 1 - f = tutils.tflow() + f = tflow.tflow() c.add_flow(f) assert len(c.view) == 2 c.set_view_filter("~q") @@ -276,18 +277,18 @@ class TestState: assert c.intercept_txt is None def _add_request(self, state): - f = tutils.tflow() + f = tflow.tflow() state.add_flow(f) return f def _add_response(self, state): - f = tutils.tflow() + f = tflow.tflow() state.add_flow(f) f.response = http.HTTPResponse.wrap(mitmproxy.test.tutils.tresp()) state.update_flow(f) def _add_error(self, state): - f = tutils.tflow(err=True) + f = tflow.tflow(err=True) state.add_flow(f) def test_clear(self): @@ -328,14 +329,14 @@ class TestSerialize: sio = io.BytesIO() w = mitmproxy.io.FlowWriter(sio) for i in range(3): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) w.add(f) for i in range(3): - f = tutils.tflow(err=True) + f = tflow.tflow(err=True) w.add(f) - f = tutils.ttcpflow() + f = tflow.ttcpflow() w.add(f) - f = tutils.ttcpflow(err=True) + f = tflow.ttcpflow(err=True) w.add(f) sio.seek(0) @@ -343,7 +344,7 @@ class TestSerialize: def test_roundtrip(self): sio = io.BytesIO() - f = tutils.tflow() + f = tflow.tflow() f.marked = True f.request.content = bytes(range(256)) w = mitmproxy.io.FlowWriter(sio) @@ -385,11 +386,11 @@ class TestSerialize: flt = flowfilter.parse("~c 200") w = mitmproxy.io.FilteredFlowWriter(sio, flt) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.status_code = 200 w.add(f) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.response.status_code = 201 w.add(f) @@ -408,7 +409,7 @@ class TestSerialize: assert str(f) == "foo" def test_versioncheck(self): - f = tutils.tflow() + f = tflow.tflow() d = f.get_state() d["version"] = (0, 0) sio = io.BytesIO() @@ -423,7 +424,7 @@ class TestFlowMaster: def test_replay(self): fm = master.Master(None, DummyServer()) - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) f.request.content = None tutils.raises("missing", fm.replay_request, f) @@ -441,7 +442,7 @@ class TestFlowMaster: s = state.State() fm = master.Master(None, DummyServer()) fm.addons.add(s) - f = tutils.tflow(req=None) + f = tflow.tflow(req=None) fm.clientconnect(f.client_conn) f.request = http.HTTPRequest.wrap(mitmproxy.test.tutils.treq()) fm.request(f) @@ -462,7 +463,7 @@ class TestFlowMaster: class TestRequest: def test_simple(self): - f = tutils.tflow() + f = tflow.tflow() r = f.request u = r.url r.url = u @@ -521,7 +522,7 @@ class TestRequest: class TestResponse: def test_simple(self): - f = tutils.tflow(resp=True) + f = tflow.tflow(resp=True) resp = f.response resp2 = resp.copy() assert resp2.get_state() == resp.get_state() @@ -564,11 +565,11 @@ class TestError: class TestClientConnection: def test_state(self): - c = tutils.tclient_conn() + c = tflow.tclient_conn() assert connections.ClientConnection.from_state(c.get_state()).get_state() == \ c.get_state() - c2 = tutils.tclient_conn() + c2 = tflow.tclient_conn() c2.address.address = (c2.address.host, 4242) assert not c == c2 diff --git a/test/mitmproxy/test_flow_export.py b/test/mitmproxy/test_flow_export.py index 86234616..bf675d50 100644 --- a/test/mitmproxy/test_flow_export.py +++ b/test/mitmproxy/test_flow_export.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow import re import mitmproxy.test.tutils @@ -33,74 +34,74 @@ def req_patch(): class TestExportCurlCommand: def test_get(self): - flow = tutils.tflow(req=req_get()) + flow = tflow.tflow(req=req_get()) result = """curl -H 'header:qvalue' -H 'content-length:7' 'http://address:22/path?a=foo&a=bar&b=baz'""" assert export.curl_command(flow) == result def test_post(self): - flow = tutils.tflow(req=req_post()) + flow = tflow.tflow(req=req_post()) result = """curl -X POST 'http://address:22/path' --data-binary 'content'""" assert export.curl_command(flow) == result def test_patch(self): - flow = tutils.tflow(req=req_patch()) + flow = tflow.tflow(req=req_patch()) result = """curl -H 'header:qvalue' -H 'content-length:7' -X PATCH 'http://address:22/path?query=param' --data-binary 'content'""" assert export.curl_command(flow) == result class TestExportPythonCode: def test_get(self): - flow = tutils.tflow(req=req_get()) + flow = tflow.tflow(req=req_get()) python_equals("data/test_flow_export/python_get.py", export.python_code(flow)) def test_post(self): - flow = tutils.tflow(req=req_post()) + flow = tflow.tflow(req=req_post()) python_equals("data/test_flow_export/python_post.py", export.python_code(flow)) def test_post_json(self): p = req_post() p.content = b'{"name": "example", "email": "example@example.com"}' p.headers = Headers(content_type="application/json") - flow = tutils.tflow(req=p) + flow = tflow.tflow(req=p) python_equals("data/test_flow_export/python_post_json.py", export.python_code(flow)) def test_patch(self): - flow = tutils.tflow(req=req_patch()) + flow = tflow.tflow(req=req_patch()) python_equals("data/test_flow_export/python_patch.py", export.python_code(flow)) class TestExportLocustCode: def test_get(self): - flow = tutils.tflow(req=req_get()) + flow = tflow.tflow(req=req_get()) python_equals("data/test_flow_export/locust_get.py", export.locust_code(flow)) def test_post(self): p = req_post() p.content = b'content' p.headers = '' - flow = tutils.tflow(req=p) + flow = tflow.tflow(req=p) python_equals("data/test_flow_export/locust_post.py", export.locust_code(flow)) def test_patch(self): - flow = tutils.tflow(req=req_patch()) + flow = tflow.tflow(req=req_patch()) python_equals("data/test_flow_export/locust_patch.py", export.locust_code(flow)) class TestExportLocustTask: def test_get(self): - flow = tutils.tflow(req=req_get()) + flow = tflow.tflow(req=req_get()) python_equals("data/test_flow_export/locust_task_get.py", export.locust_task(flow)) def test_post(self): - flow = tutils.tflow(req=req_post()) + flow = tflow.tflow(req=req_post()) python_equals("data/test_flow_export/locust_task_post.py", export.locust_task(flow)) def test_patch(self): - flow = tutils.tflow(req=req_patch()) + flow = tflow.tflow(req=req_patch()) python_equals("data/test_flow_export/locust_task_patch.py", export.locust_task(flow)) class TestURL: def test_url(self): - flow = tutils.tflow() + flow = tflow.tflow() assert export.url(flow) == "http://address:22/path" diff --git a/test/mitmproxy/test_flowfilter.py b/test/mitmproxy/test_flowfilter.py index 8fa30900..16d61d25 100644 --- a/test/mitmproxy/test_flowfilter.py +++ b/test/mitmproxy/test_flowfilter.py @@ -1,4 +1,5 @@ import io +from mitmproxy.test import tflow from mock import patch from mitmproxy import flowfilter @@ -78,13 +79,13 @@ class TestParsing: class TestMatchingHTTPFlow: def req(self): - return tutils.tflow() + return tflow.tflow() def resp(self): - return tutils.tflow(resp=True) + return tflow.tflow(resp=True) def err(self): - return tutils.tflow(err=True) + return tflow.tflow(err=True) def q(self, q, o): return flowfilter.parse(q)(o) @@ -223,7 +224,7 @@ class TestMatchingHTTPFlow: def test_dst(self): q = self.req() - q.server_conn = tutils.tserver_conn() + q.server_conn = tflow.tserver_conn() assert self.q("~dst address", q) assert not self.q("~dst foobar", q) assert self.q("~dst :22", q) @@ -257,10 +258,10 @@ class TestMatchingHTTPFlow: class TestMatchingTCPFlow: def flow(self): - return tutils.ttcpflow() + return tflow.ttcpflow() def err(self): - return tutils.ttcpflow(err=True) + return tflow.ttcpflow(err=True) def q(self, q, o): return flowfilter.parse(q)(o) @@ -302,7 +303,7 @@ class TestMatchingTCPFlow: def test_dst(self): f = self.flow() - f.server_conn = tutils.tserver_conn() + f.server_conn = tflow.tserver_conn() assert self.q("~dst address", f) assert not self.q("~dst foobar", f) assert self.q("~dst :22", f) @@ -311,7 +312,7 @@ class TestMatchingTCPFlow: def test_and(self): f = self.flow() - f.server_conn = tutils.tserver_conn() + f.server_conn = tflow.tserver_conn() assert self.q("~b hello & ~b me", f) assert not self.q("~src wrongaddress & ~b hello", f) assert self.q("(~src :22 & ~dst :22) & ~b hello", f) @@ -320,7 +321,7 @@ class TestMatchingTCPFlow: def test_or(self): f = self.flow() - f.server_conn = tutils.tserver_conn() + f.server_conn = tflow.tserver_conn() assert self.q("~b hello | ~b me", f) assert self.q("~src :22 | ~b me", f) assert not self.q("~src :99 | ~dst :99", f) @@ -392,7 +393,7 @@ class TestMatchingDummyFlow: def test_filters(self): e = self.err() f = self.flow() - f.server_conn = tutils.tserver_conn() + f.server_conn = tflow.tserver_conn() assert not self.q("~a", f) diff --git a/test/mitmproxy/test_proxy.py b/test/mitmproxy/test_proxy.py index a9d9bb87..4ac6c177 100644 --- a/test/mitmproxy/test_proxy.py +++ b/test/mitmproxy/test_proxy.py @@ -1,3 +1,4 @@ +from mitmproxy.test import tflow import os import mock from OpenSSL import SSL @@ -20,7 +21,7 @@ class TestServerConnection: self.d = test.Daemon() sc = connections.ServerConnection((self.d.IFACE, self.d.port)) sc.connect() - f = tutils.tflow() + f = tflow.tflow() f.server_conn = sc f.request.path = "/p/200:da" @@ -45,7 +46,7 @@ class TestServerConnection: self.d.shutdown() def test_repr(self): - sc = tutils.tserver_conn() + sc = tflow.tserver_conn() assert "address:22" in repr(sc) assert "ssl" not in repr(sc) sc.ssl_established = True diff --git a/test/mitmproxy/tutils.py b/test/mitmproxy/tutils.py index 5f1dda53..e3611e27 100644 --- a/test/mitmproxy/tutils.py +++ b/test/mitmproxy/tutils.py @@ -1,20 +1,16 @@ -import os -import shutil -import tempfile import argparse import sys - from contextlib import contextmanager from unittest.case import SkipTest import io - import mitmproxy.test.tutils +import os +import shutil +import tempfile from mitmproxy import controller -from mitmproxy import connections from mitmproxy import flow -from mitmproxy import http -from mitmproxy import tcp +import mitmproxy.test.tflow from mitmproxy.utils import data @@ -56,11 +52,11 @@ class DummyFlow(flow.Flow): def tdummyflow(client_conn=True, server_conn=True, err=None): if client_conn is True: - client_conn = tclient_conn() + client_conn = mitmproxy.test.tflow.tclient_conn() if server_conn is True: - server_conn = tserver_conn() + server_conn = mitmproxy.test.tflow.tserver_conn() if err is True: - err = terr() + err = mitmproxy.test.tflow.terr() f = DummyFlow(client_conn, server_conn) f.error = err @@ -68,107 +64,6 @@ def tdummyflow(client_conn=True, server_conn=True, err=None): return f -def ttcpflow(client_conn=True, server_conn=True, messages=True, err=None): - if client_conn is True: - client_conn = tclient_conn() - if server_conn is True: - server_conn = tserver_conn() - if messages is True: - messages = [ - tcp.TCPMessage(True, b"hello"), - tcp.TCPMessage(False, b"it's me"), - ] - if err is True: - err = terr() - - f = tcp.TCPFlow(client_conn, server_conn) - f.messages = messages - f.error = err - f.reply = controller.DummyReply() - return f - - -def tflow(client_conn=True, server_conn=True, req=True, resp=None, err=None): - """ - @type client_conn: bool | None | mitmproxy.proxy.connection.ClientConnection - @type server_conn: bool | None | mitmproxy.proxy.connection.ServerConnection - @type req: bool | None | mitmproxy.proxy.protocol.http.HTTPRequest - @type resp: bool | None | mitmproxy.proxy.protocol.http.HTTPResponse - @type err: bool | None | mitmproxy.proxy.protocol.primitives.Error - @return: mitmproxy.proxy.protocol.http.HTTPFlow - """ - if client_conn is True: - client_conn = tclient_conn() - if server_conn is True: - server_conn = tserver_conn() - if req is True: - req = mitmproxy.test.tutils.treq() - if resp is True: - resp = mitmproxy.test.tutils.tresp() - if err is True: - err = terr() - - if req: - req = http.HTTPRequest.wrap(req) - if resp: - resp = http.HTTPResponse.wrap(resp) - - f = http.HTTPFlow(client_conn, server_conn) - f.request = req - f.response = resp - f.error = err - f.reply = controller.DummyReply() - return f - - -def tclient_conn(): - """ - @return: mitmproxy.proxy.connection.ClientConnection - """ - c = connections.ClientConnection.from_state(dict( - address=dict(address=("address", 22), use_ipv6=True), - clientcert=None, - ssl_established=False, - timestamp_start=1, - timestamp_ssl_setup=2, - timestamp_end=3, - sni="address", - cipher_name="cipher", - tls_version="TLSv1.2", - )) - c.reply = controller.DummyReply() - return c - - -def tserver_conn(): - """ - @return: mitmproxy.proxy.connection.ServerConnection - """ - c = connections.ServerConnection.from_state(dict( - address=dict(address=("address", 22), use_ipv6=True), - source_address=dict(address=("address", 22), use_ipv6=True), - ip_address=None, - cert=None, - timestamp_start=1, - timestamp_tcp_setup=2, - timestamp_ssl_setup=3, - timestamp_end=4, - ssl_established=False, - sni="address", - via=None - )) - c.reply = controller.DummyReply() - return c - - -def terr(content="error"): - """ - @return: mitmproxy.proxy.protocol.primitives.Error - """ - err = flow.Error(content) - return err - - def get_body_line(last_displayed_body, line_nb): return last_displayed_body.contents()[line_nb + 2] |