diff options
Diffstat (limited to 'test')
23 files changed, 432 insertions, 490 deletions
diff --git a/test/examples/test_har_dump.py b/test/examples/_test_har_dump.py index e5cfd2e1..e5cfd2e1 100644 --- a/test/examples/test_har_dump.py +++ b/test/examples/_test_har_dump.py diff --git a/test/examples/test_examples.py b/test/examples/test_examples.py index 46fdcd36..4c1631ce 100644 --- a/test/examples/test_examples.py +++ b/test/examples/test_examples.py @@ -1,13 +1,7 @@ -import pytest - -from mitmproxy import options from mitmproxy import contentviews -from mitmproxy import proxy -from mitmproxy import master -from mitmproxy.addons import script - from mitmproxy.test import tflow from mitmproxy.test import tutils +from mitmproxy.test import taddons from mitmproxy.net.http import Headers from ..mitmproxy import tservers @@ -15,106 +9,93 @@ from ..mitmproxy import tservers example_dir = tutils.test_data.push("../examples") -class ScriptError(Exception): - pass - - -class RaiseMaster(master.Master): - def add_log(self, e, level): - if level in ("warn", "error"): - raise ScriptError(e) - - -def tscript(cmd, args=""): - o = options.Options() - cmd = example_dir.path(cmd) + " " + args - m = RaiseMaster(o, proxy.DummyServer()) - sc = script.Script(cmd) - m.addons.add(sc) - return m, sc - - class TestScripts(tservers.MasterTest): def test_add_header(self): - m, _ = tscript("simple/add_header.py") - f = tflow.tflow(resp=tutils.tresp()) - m.addons.handle_lifecycle("response", f) - assert f.response.headers["newheader"] == "foo" + with taddons.context() as tctx: + a = tctx.script(example_dir.path("simple/add_header.py")) + f = tflow.tflow(resp=tutils.tresp()) + a.response(f) + assert f.response.headers["newheader"] == "foo" def test_custom_contentviews(self): - m, sc = tscript("simple/custom_contentview.py") - swapcase = contentviews.get("swapcase") - _, fmt = swapcase(b"<html>Test!</html>") - assert any(b'tEST!' in val[0][1] for val in fmt) + with taddons.context() as tctx: + tctx.script(example_dir.path("simple/custom_contentview.py")) + swapcase = contentviews.get("swapcase") + _, fmt = swapcase(b"<html>Test!</html>") + assert any(b'tEST!' in val[0][1] for val in fmt) def test_iframe_injector(self): - with pytest.raises(ScriptError): - tscript("simple/modify_body_inject_iframe.py") - - m, sc = tscript("simple/modify_body_inject_iframe.py", "http://example.org/evil_iframe") - f = tflow.tflow(resp=tutils.tresp(content=b"<html><body>mitmproxy</body></html>")) - m.addons.handle_lifecycle("response", f) - content = f.response.content - assert b'iframe' in content and b'evil_iframe' in content + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("simple/modify_body_inject_iframe.py")) + tctx.configure( + sc, + iframe = "http://example.org/evil_iframe" + ) + f = tflow.tflow( + resp=tutils.tresp(content=b"<html><body>mitmproxy</body></html>") + ) + tctx.master.addons.invoke_addon(sc, "response", f) + content = f.response.content + assert b'iframe' in content and b'evil_iframe' in content def test_modify_form(self): - m, sc = tscript("simple/modify_form.py") + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("simple/modify_form.py")) - form_header = Headers(content_type="application/x-www-form-urlencoded") - f = tflow.tflow(req=tutils.treq(headers=form_header)) - m.addons.handle_lifecycle("request", f) + form_header = Headers(content_type="application/x-www-form-urlencoded") + f = tflow.tflow(req=tutils.treq(headers=form_header)) + sc.request(f) - assert f.request.urlencoded_form["mitmproxy"] == "rocks" + assert f.request.urlencoded_form["mitmproxy"] == "rocks" - f.request.headers["content-type"] = "" - m.addons.handle_lifecycle("request", f) - assert list(f.request.urlencoded_form.items()) == [("foo", "bar")] + f.request.headers["content-type"] = "" + sc.request(f) + assert list(f.request.urlencoded_form.items()) == [("foo", "bar")] def test_modify_querystring(self): - m, sc = tscript("simple/modify_querystring.py") - f = tflow.tflow(req=tutils.treq(path="/search?q=term")) + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("simple/modify_querystring.py")) + f = tflow.tflow(req=tutils.treq(path="/search?q=term")) - m.addons.handle_lifecycle("request", f) - assert f.request.query["mitmproxy"] == "rocks" + sc.request(f) + assert f.request.query["mitmproxy"] == "rocks" - f.request.path = "/" - m.addons.handle_lifecycle("request", f) - assert f.request.query["mitmproxy"] == "rocks" - - def test_arguments(self): - m, sc = tscript("simple/script_arguments.py", "mitmproxy rocks") - f = tflow.tflow(resp=tutils.tresp(content=b"I <3 mitmproxy")) - m.addons.handle_lifecycle("response", f) - assert f.response.content == b"I <3 rocks" + f.request.path = "/" + sc.request(f) + assert f.request.query["mitmproxy"] == "rocks" def test_redirect_requests(self): - m, sc = tscript("simple/redirect_requests.py") - f = tflow.tflow(req=tutils.treq(host="example.org")) - m.addons.handle_lifecycle("request", f) - assert f.request.host == "mitmproxy.org" + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("simple/redirect_requests.py")) + f = tflow.tflow(req=tutils.treq(host="example.org")) + sc.request(f) + assert f.request.host == "mitmproxy.org" def test_send_reply_from_proxy(self): - m, sc = tscript("simple/send_reply_from_proxy.py") - f = tflow.tflow(req=tutils.treq(host="example.com", port=80)) - m.addons.handle_lifecycle("request", f) - assert f.response.content == b"Hello World" + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("simple/send_reply_from_proxy.py")) + f = tflow.tflow(req=tutils.treq(host="example.com", port=80)) + sc.request(f) + assert f.response.content == b"Hello World" def test_dns_spoofing(self): - m, sc = tscript("complex/dns_spoofing.py") - original_host = "example.com" + with taddons.context() as tctx: + sc = tctx.script(example_dir.path("complex/dns_spoofing.py")) + + original_host = "example.com" - host_header = Headers(host=original_host) - f = tflow.tflow(req=tutils.treq(headers=host_header, port=80)) + host_header = Headers(host=original_host) + f = tflow.tflow(req=tutils.treq(headers=host_header, port=80)) - m.addons.handle_lifecycle("requestheaders", f) + tctx.master.addons.invoke_addon(sc, "requestheaders", f) - # Rewrite by reverse proxy mode - f.request.scheme = "https" - f.request.port = 443 + # Rewrite by reverse proxy mode + f.request.scheme = "https" + f.request.port = 443 - m.addons.handle_lifecycle("request", f) + tctx.master.addons.invoke_addon(sc, "request", f) - assert f.request.scheme == "http" - assert f.request.port == 80 + assert f.request.scheme == "http" + assert f.request.port == 80 - assert f.request.headers["Host"] == original_host + assert f.request.headers["Host"] == original_host diff --git a/test/mitmproxy/addons/test_core_option_validation.py b/test/mitmproxy/addons/test_core_option_validation.py index 0bb2bb0d..6d6d5ba4 100644 --- a/test/mitmproxy/addons/test_core_option_validation.py +++ b/test/mitmproxy/addons/test_core_option_validation.py @@ -11,7 +11,7 @@ def test_simple(): with pytest.raises(exceptions.OptionsError): tctx.configure(sa, body_size_limit = "invalid") tctx.configure(sa, body_size_limit = "1m") - assert tctx.options._processed["body_size_limit"] + assert tctx.master.options._processed["body_size_limit"] with pytest.raises(exceptions.OptionsError, match="mutually exclusive"): tctx.configure( diff --git a/test/mitmproxy/addons/test_onboarding.py b/test/mitmproxy/addons/test_onboarding.py index 63125c23..42a3b574 100644 --- a/test/mitmproxy/addons/test_onboarding.py +++ b/test/mitmproxy/addons/test_onboarding.py @@ -1,4 +1,5 @@ from mitmproxy.addons import onboarding +from mitmproxy.test import taddons from .. import tservers @@ -7,10 +8,14 @@ class TestApp(tservers.HTTPProxyTest): return [onboarding.Onboarding()] def test_basic(self): - assert self.app("/").status_code == 200 + with taddons.context() as tctx: + tctx.configure(self.addons()[0]) + assert self.app("/").status_code == 200 def test_cert(self): - for ext in ["pem", "p12"]: - resp = self.app("/cert/%s" % ext) - assert resp.status_code == 200 - assert resp.content + with taddons.context() as tctx: + tctx.configure(self.addons()[0]) + for ext in ["pem", "p12"]: + resp = self.app("/cert/%s" % ext) + assert resp.status_code == 200 + assert resp.content diff --git a/test/mitmproxy/addons/test_proxyauth.py b/test/mitmproxy/addons/test_proxyauth.py index 513f3f08..86621709 100644 --- a/test/mitmproxy/addons/test_proxyauth.py +++ b/test/mitmproxy/addons/test_proxyauth.py @@ -66,9 +66,6 @@ def test_configure(): with pytest.raises(exceptions.OptionsError): ctx.configure(up, proxyauth="any", mode="socks5") - ctx.configure(up, mode="regular") - assert up.mode == "regular" - def test_check(): up = proxyauth.ProxyAuth() diff --git a/test/mitmproxy/addons/test_script.py b/test/mitmproxy/addons/test_script.py index c68981de..859d99f9 100644 --- a/test/mitmproxy/addons/test_script.py +++ b/test/mitmproxy/addons/test_script.py @@ -1,8 +1,6 @@ import traceback import sys import time -import re -import watchdog.events import pytest from unittest import mock @@ -14,159 +12,95 @@ from mitmproxy import exceptions from mitmproxy import options from mitmproxy import proxy from mitmproxy import master -from mitmproxy import utils from mitmproxy.addons import script -from ...conftest import skip_not_windows - - -def test_scriptenv(): - with taddons.context() as tctx: - with script.scriptenv("path", []): - raise SystemExit - assert tctx.master.has_log("exited", "error") - - tctx.master.clear() - with script.scriptenv("path", []): - raise ValueError("fooo") - assert tctx.master.has_log("fooo", "error") - - -class Called: - def __init__(self): - self.called = False - - def __call__(self, *args, **kwargs): - self.called = True - - -def test_reloadhandler(): - rh = script.ReloadHandler(Called()) - assert not rh.filter(watchdog.events.DirCreatedEvent("path")) - assert not rh.filter(watchdog.events.FileModifiedEvent("/foo/.bar")) - assert not rh.filter(watchdog.events.FileModifiedEvent("/foo/bar")) - assert rh.filter(watchdog.events.FileModifiedEvent("/foo/bar.py")) - - assert not rh.callback.called - rh.on_modified(watchdog.events.FileModifiedEvent("/foo/bar")) - assert not rh.callback.called - rh.on_modified(watchdog.events.FileModifiedEvent("/foo/bar.py")) - assert rh.callback.called - rh.callback.called = False - - rh.on_created(watchdog.events.FileCreatedEvent("foo")) - assert not rh.callback.called - rh.on_created(watchdog.events.FileCreatedEvent("foo.py")) - assert rh.callback.called - - -class TestParseCommand: - def test_empty_command(self): - with pytest.raises(ValueError): - script.parse_command("") - - with pytest.raises(ValueError): - script.parse_command(" ") - - def test_no_script_file(self, tmpdir): - with pytest.raises(Exception, match="not found"): - script.parse_command("notfound") - - with pytest.raises(Exception, match="Not a file"): - script.parse_command(str(tmpdir)) - - def test_parse_args(self): - with utils.chdir(tutils.test_data.dirname): - assert script.parse_command( - "mitmproxy/data/addonscripts/recorder.py" - ) == ("mitmproxy/data/addonscripts/recorder.py", []) - assert script.parse_command( - "mitmproxy/data/addonscripts/recorder.py foo bar" - ) == ("mitmproxy/data/addonscripts/recorder.py", ["foo", "bar"]) - assert script.parse_command( - "mitmproxy/data/addonscripts/recorder.py 'foo bar'" - ) == ("mitmproxy/data/addonscripts/recorder.py", ["foo bar"]) - - @skip_not_windows - def test_parse_windows(self): - with utils.chdir(tutils.test_data.dirname): - assert script.parse_command( - "mitmproxy/data\\addonscripts\\recorder.py" - ) == ("mitmproxy/data\\addonscripts\\recorder.py", []) - assert script.parse_command( - "mitmproxy/data\\addonscripts\\recorder.py 'foo \\ bar'" - ) == ("mitmproxy/data\\addonscripts\\recorder.py", ['foo \\ bar']) - def test_load_script(): - with taddons.context(): + with taddons.context() as tctx: ns = script.load_script( + tctx.ctx(), tutils.test_data.path( - "mitmproxy/data/addonscripts/recorder.py" - ), [] + "mitmproxy/data/addonscripts/recorder/recorder.py" + ) ) - assert ns.load + assert ns.addons + + ns = script.load_script( + tctx.ctx(), + "nonexistent" + ) + assert not ns def test_script_print_stdout(): with taddons.context() as tctx: with mock.patch('mitmproxy.ctx.log.warn') as mock_warn: - with script.scriptenv("path", []): + with addonmanager.safecall(): ns = script.load_script( + tctx.ctx(), tutils.test_data.path( "mitmproxy/data/addonscripts/print.py" - ), [] + ) ) ns.load(addonmanager.Loader(tctx.master)) mock_warn.assert_called_once_with("stdoutprint") class TestScript: + def test_notfound(self): + with taddons.context() as tctx: + sc = script.Script("nonexistent") + tctx.master.addons.add(sc) + def test_simple(self): - with taddons.context(): + with taddons.context() as tctx: sc = script.Script( tutils.test_data.path( - "mitmproxy/data/addonscripts/recorder.py" + "mitmproxy/data/addonscripts/recorder/recorder.py" ) ) - sc.load_script() - assert sc.ns.call_log[0][0:2] == ("solo", "load") + tctx.master.addons.add(sc) + tctx.configure(sc) + sc.tick() + + rec = tctx.master.addons.get("recorder") + + assert rec.call_log[0][0:2] == ("recorder", "load") - sc.ns.call_log = [] + rec.call_log = [] f = tflow.tflow(resp=True) - sc.request(f) + tctx.master.addons.trigger("request", f) - recf = sc.ns.call_log[0] - assert recf[1] == "request" + assert rec.call_log[0][1] == "request" def test_reload(self, tmpdir): with taddons.context() as tctx: f = tmpdir.join("foo.py") f.ensure(file=True) + f.write("\n") sc = script.Script(str(f)) tctx.configure(sc) - for _ in range(100): - f.write(".") + sc.tick() + for _ in range(3): + sc.last_load, sc.last_mtime = 0, 0 sc.tick() time.sleep(0.1) - if tctx.master.logs: - return - raise AssertionError("Change event not detected.") + tctx.master.has_log("Loading") def test_exception(self): with taddons.context() as tctx: sc = script.Script( tutils.test_data.path("mitmproxy/data/addonscripts/error.py") ) - l = addonmanager.Loader(tctx.master) - sc.load(l) + tctx.master.addons.add(sc) + tctx.configure(sc) + sc.tick() + f = tflow.tflow(resp=True) - sc.request(f) - assert tctx.master.logs[0].level == "error" - assert len(tctx.master.logs[0].msg.splitlines()) == 6 - assert re.search(r'addonscripts[\\/]error.py", line \d+, in request', tctx.master.logs[0].msg) - assert re.search(r'addonscripts[\\/]error.py", line \d+, in mkerr', tctx.master.logs[0].msg) - assert tctx.master.logs[0].msg.endswith("ValueError: Error!\n") + tctx.master.addons.trigger("request", f) + + tctx.master.has_log("ValueError: Error!") + tctx.master.has_log("error.py") def test_addon(self): with taddons.context() as tctx: @@ -175,11 +109,11 @@ class TestScript: "mitmproxy/data/addonscripts/addon.py" ) ) - l = addonmanager.Loader(tctx.master) - sc.load(l) + tctx.master.addons.add(sc) tctx.configure(sc) + sc.tick() assert sc.ns.event_log == [ - 'scriptload', 'addonload', 'addonconfigure' + 'scriptload', 'addonload', 'scriptconfigure', 'addonconfigure' ] @@ -194,49 +128,33 @@ class TestCutTraceback: self.raise_(4) except RuntimeError: tb = sys.exc_info()[2] - tb_cut = script.cut_traceback(tb, "test_simple") + tb_cut = addonmanager.cut_traceback(tb, "test_simple") assert len(traceback.extract_tb(tb_cut)) == 5 - tb_cut2 = script.cut_traceback(tb, "nonexistent") + tb_cut2 = addonmanager.cut_traceback(tb, "nonexistent") assert len(traceback.extract_tb(tb_cut2)) == len(traceback.extract_tb(tb)) class TestScriptLoader: - def test_run_once(self): - o = options.Options(scripts=[]) - m = master.Master(o, proxy.DummyServer()) - sl = script.ScriptLoader() - m.addons.add(sl) - - f = tflow.tflow(resp=True) - with m.handlecontext(): - sc = sl.run_once( - tutils.test_data.path( - "mitmproxy/data/addonscripts/recorder.py" - ), [f] - ) - evts = [i[1] for i in sc.ns.call_log] - assert evts == ['load', 'requestheaders', 'request', 'responseheaders', 'response', 'done'] - - f = tflow.tflow(resp=True) - with m.handlecontext(): - with pytest.raises(Exception, match="file not found"): - sl.run_once("nonexistent", [f]) - def test_simple(self): o = options.Options(scripts=[]) m = master.Master(o, proxy.DummyServer()) sc = script.ScriptLoader() + sc.running() m.addons.add(sc) assert len(m.addons) == 1 o.update( scripts = [ - tutils.test_data.path("mitmproxy/data/addonscripts/recorder.py") + tutils.test_data.path( + "mitmproxy/data/addonscripts/recorder/recorder.py" + ) ] ) - assert len(m.addons) == 2 + assert len(m.addons) == 1 + assert len(sc.addons) == 1 o.update(scripts = []) assert len(m.addons) == 1 + assert len(sc.addons) == 0 def test_dupes(self): sc = script.ScriptLoader() @@ -252,65 +170,76 @@ class TestScriptLoader: sc = script.ScriptLoader() with taddons.context() as tctx: tctx.master.addons.add(sc) - with pytest.raises(exceptions.OptionsError): - tctx.configure( - sc, - scripts = ["nonexistent"] - ) + tctx.configure(sc, scripts = ["nonexistent"]) + tctx.master.has_log("nonexistent: file not found") def test_order(self): - rec = tutils.test_data.path("mitmproxy/data/addonscripts/recorder.py") + rec = tutils.test_data.path("mitmproxy/data/addonscripts/recorder") sc = script.ScriptLoader() + sc.is_running = True with taddons.context() as tctx: - tctx.master.addons.add(sc) - sc.running() tctx.configure( sc, scripts = [ - "%s %s" % (rec, "a"), - "%s %s" % (rec, "b"), - "%s %s" % (rec, "c"), + "%s/a.py" % rec, + "%s/b.py" % rec, + "%s/c.py" % rec, ] ) + tctx.master.addons.invoke_addon(sc, "tick") debug = [i.msg for i in tctx.master.logs if i.level == "debug"] assert debug == [ 'a load', - 'a configure', 'a running', + 'a configure', + 'a tick', 'b load', - 'b configure', 'b running', + 'b configure', + 'b tick', 'c load', - 'c configure', 'c running', + 'c configure', + 'c tick', ] + tctx.master.logs = [] tctx.configure( sc, scripts = [ - "%s %s" % (rec, "c"), - "%s %s" % (rec, "a"), - "%s %s" % (rec, "b"), + "%s/c.py" % rec, + "%s/a.py" % rec, + "%s/b.py" % rec, ] ) + debug = [i.msg for i in tctx.master.logs if i.level == "debug"] - assert debug == [] + assert debug == [ + 'c configure', + 'a configure', + 'b configure', + ] tctx.master.logs = [] tctx.configure( sc, scripts = [ - "%s %s" % (rec, "x"), - "%s %s" % (rec, "a"), + "%s/e.py" % rec, + "%s/a.py" % rec, ] ) + tctx.master.addons.invoke_addon(sc, "tick") + debug = [i.msg for i in tctx.master.logs if i.level == "debug"] assert debug == [ 'c done', 'b done', - 'x load', - 'x configure', - 'x running', + 'a configure', + 'e load', + 'e running', + 'e configure', + 'e tick', + 'a tick', ] diff --git a/test/mitmproxy/addons/test_serverplayback.py b/test/mitmproxy/addons/test_serverplayback.py index 7078b66e..29de48a0 100644 --- a/test/mitmproxy/addons/test_serverplayback.py +++ b/test/mitmproxy/addons/test_serverplayback.py @@ -6,7 +6,6 @@ from mitmproxy.test import tflow import mitmproxy.test.tutils from mitmproxy.addons import serverplayback -from mitmproxy import options from mitmproxy import exceptions from mitmproxy import io @@ -39,86 +38,88 @@ def test_tick(): def test_server_playback(): sp = serverplayback.ServerPlayback() - sp.configure(options.Options(), []) - f = tflow.tflow(resp=True) + with taddons.context() as tctx: + tctx.configure(sp) + f = tflow.tflow(resp=True) - assert not sp.flowmap + assert not sp.flowmap - sp.load_flows([f]) - assert sp.flowmap - assert sp.next_flow(f) - assert not sp.flowmap + sp.load_flows([f]) + assert sp.flowmap + assert sp.next_flow(f) + assert not sp.flowmap - sp.load_flows([f]) - assert sp.flowmap - sp.clear() - assert not sp.flowmap + sp.load_flows([f]) + assert sp.flowmap + sp.clear() + assert not sp.flowmap def test_ignore_host(): sp = serverplayback.ServerPlayback() - sp.configure(options.Options(server_replay_ignore_host=True), []) + with taddons.context() as tctx: + tctx.configure(sp, server_replay_ignore_host=True) - r = tflow.tflow(resp=True) - r2 = tflow.tflow(resp=True) + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) - r.request.host = "address" - r2.request.host = "address" - assert sp._hash(r) == sp._hash(r2) - r2.request.host = "wrong_address" - assert sp._hash(r) == sp._hash(r2) + r.request.host = "address" + r2.request.host = "address" + assert sp._hash(r) == sp._hash(r2) + r2.request.host = "wrong_address" + assert sp._hash(r) == sp._hash(r2) def test_ignore_content(): s = serverplayback.ServerPlayback() - s.configure(options.Options(server_replay_ignore_content=False), []) + with taddons.context() as tctx: + tctx.configure(s, server_replay_ignore_content=False) - r = tflow.tflow(resp=True) - r2 = tflow.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) - r2.request.content = b"bar" - assert not s._hash(r) == s._hash(r2) + r.request.content = b"foo" + r2.request.content = b"foo" + assert s._hash(r) == s._hash(r2) + r2.request.content = b"bar" + assert not s._hash(r) == s._hash(r2) - s.configure(options.Options(server_replay_ignore_content=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) - r2.request.content = b"bar" - assert s._hash(r) == s._hash(r2) - r2.request.content = b"" - assert s._hash(r) == s._hash(r2) - r2.request.content = None - assert s._hash(r) == s._hash(r2) + tctx.configure(s, server_replay_ignore_content=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) + r2.request.content = b"bar" + assert s._hash(r) == s._hash(r2) + r2.request.content = b"" + assert s._hash(r) == s._hash(r2) + r2.request.content = None + assert s._hash(r) == s._hash(r2) def test_ignore_content_wins_over_params(): s = serverplayback.ServerPlayback() - s.configure( - options.Options( + with taddons.context() as tctx: + tctx.configure( + s, server_replay_ignore_content=True, server_replay_ignore_payload_params=[ "param1", "param2" ] - ), - [] - ) - # NOTE: parameters are mutually exclusive in options + ) - r = tflow.tflow(resp=True) - r.request.headers["Content-Type"] = "application/x-www-form-urlencoded" - r.request.content = b"paramx=y" + # NOTE: parameters are mutually exclusive in options + r = tflow.tflow(resp=True) + r.request.headers["Content-Type"] = "application/x-www-form-urlencoded" + r.request.content = b"paramx=y" - r2 = tflow.tflow(resp=True) - r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded" - r2.request.content = b"paramx=x" + r2 = tflow.tflow(resp=True) + r2.request.headers["Content-Type"] = "application/x-www-form-urlencoded" + r2.request.content = b"paramx=x" - # same parameters - assert s._hash(r) == s._hash(r2) + # same parameters + assert s._hash(r) == s._hash(r2) def test_ignore_payload_params_other_content_type(): @@ -147,136 +148,139 @@ def test_ignore_payload_params_other_content_type(): def test_hash(): s = serverplayback.ServerPlayback() - s.configure(options.Options(), []) + with taddons.context() as tctx: + tctx.configure(s) - r = tflow.tflow() - r2 = tflow.tflow() + r = tflow.tflow() + r2 = tflow.tflow() - assert s._hash(r) - assert s._hash(r) == s._hash(r2) - r.request.headers["foo"] = "bar" - assert s._hash(r) == s._hash(r2) - r.request.path = "voing" - assert s._hash(r) != s._hash(r2) + assert s._hash(r) + assert s._hash(r) == s._hash(r2) + r.request.headers["foo"] = "bar" + assert s._hash(r) == s._hash(r2) + r.request.path = "voing" + assert s._hash(r) != s._hash(r2) - r.request.path = "path?blank_value" - r2.request.path = "path?" - assert s._hash(r) != s._hash(r2) + r.request.path = "path?blank_value" + r2.request.path = "path?" + assert s._hash(r) != s._hash(r2) def test_headers(): s = serverplayback.ServerPlayback() - s.configure(options.Options(server_replay_use_headers=["foo"]), []) + with taddons.context() as tctx: + tctx.configure(s, server_replay_use_headers=["foo"]) - r = tflow.tflow(resp=True) - r.request.headers["foo"] = "bar" - 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 = tflow.tflow(resp=True) + r.request.headers["foo"] = "bar" + 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 = tflow.tflow(resp=True) - r2 = tflow.tflow(resp=True) - assert s._hash(r) == s._hash(r2) + r = tflow.tflow(resp=True) + r2 = tflow.tflow(resp=True) + assert s._hash(r) == s._hash(r2) def test_load(): s = serverplayback.ServerPlayback() - s.configure(options.Options(), []) + with taddons.context() as tctx: + tctx.configure(s) - r = tflow.tflow(resp=True) - r.request.headers["key"] = "one" + r = tflow.tflow(resp=True) + r.request.headers["key"] = "one" - r2 = tflow.tflow(resp=True) - r2.request.headers["key"] = "two" + r2 = tflow.tflow(resp=True) + r2.request.headers["key"] = "two" - s.load_flows([r, r2]) + s.load_flows([r, r2]) - assert s.count() == 2 + assert s.count() == 2 - n = s.next_flow(r) - assert n.request.headers["key"] == "one" - assert s.count() == 1 + n = s.next_flow(r) + assert n.request.headers["key"] == "one" + assert s.count() == 1 - n = s.next_flow(r) - assert n.request.headers["key"] == "two" - assert not s.flowmap - assert s.count() == 0 + n = s.next_flow(r) + assert n.request.headers["key"] == "two" + assert not s.flowmap + assert s.count() == 0 - assert not s.next_flow(r) + assert not s.next_flow(r) def test_load_with_server_replay_nopop(): s = serverplayback.ServerPlayback() - s.configure(options.Options(server_replay_nopop=True), []) + with taddons.context() as tctx: + tctx.configure(s, server_replay_nopop=True) - r = tflow.tflow(resp=True) - r.request.headers["key"] = "one" + r = tflow.tflow(resp=True) + r.request.headers["key"] = "one" - r2 = tflow.tflow(resp=True) - r2.request.headers["key"] = "two" + r2 = tflow.tflow(resp=True) + r2.request.headers["key"] = "two" - s.load_flows([r, r2]) + s.load_flows([r, r2]) - assert s.count() == 2 - s.next_flow(r) - assert s.count() == 2 + assert s.count() == 2 + s.next_flow(r) + assert s.count() == 2 def test_ignore_params(): s = serverplayback.ServerPlayback() - s.configure( - options.Options( + with taddons.context() as tctx: + tctx.configure( + s, server_replay_ignore_params=["param1", "param2"] - ), - [] - ) + ) - r = tflow.tflow(resp=True) - r.request.path = "/test?param1=1" - r2 = tflow.tflow(resp=True) - r2.request.path = "/test" - assert s._hash(r) == s._hash(r2) - r2.request.path = "/test?param1=2" - assert s._hash(r) == s._hash(r2) - r2.request.path = "/test?param2=1" - assert s._hash(r) == s._hash(r2) - r2.request.path = "/test?param3=2" - assert not s._hash(r) == s._hash(r2) + r = tflow.tflow(resp=True) + r.request.path = "/test?param1=1" + r2 = tflow.tflow(resp=True) + r2.request.path = "/test" + assert s._hash(r) == s._hash(r2) + r2.request.path = "/test?param1=2" + assert s._hash(r) == s._hash(r2) + r2.request.path = "/test?param2=1" + assert s._hash(r) == s._hash(r2) + r2.request.path = "/test?param3=2" + assert not s._hash(r) == s._hash(r2) def thash(r, r2, setter): s = serverplayback.ServerPlayback() - s.configure( - options.Options( + with taddons.context() as tctx: + s = serverplayback.ServerPlayback() + tctx.configure( + s, server_replay_ignore_payload_params=["param1", "param2"] - ), - [] - ) - - setter(r, paramx="x", param1="1") - - setter(r2, paramx="x", param1="1") - # same parameters - assert s._hash(r) == s._hash(r2) - # ignored parameters != - setter(r2, paramx="x", param1="2") - assert s._hash(r) == s._hash(r2) - # missing parameter - setter(r2, paramx="x") - assert s._hash(r) == s._hash(r2) - # ignorable parameter added - setter(r2, paramx="x", param1="2") - assert s._hash(r) == s._hash(r2) - # not ignorable parameter changed - 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) + ) + + setter(r, paramx="x", param1="1") + + setter(r2, paramx="x", param1="1") + # same parameters + assert s._hash(r) == s._hash(r2) + # ignored parameters != + setter(r2, paramx="x", param1="2") + assert s._hash(r) == s._hash(r2) + # missing parameter + setter(r2, paramx="x") + assert s._hash(r) == s._hash(r2) + # ignorable parameter added + setter(r2, paramx="x", param1="2") + assert s._hash(r) == s._hash(r2) + # not ignorable parameter changed + 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(): diff --git a/test/mitmproxy/addons/test_termstatus.py b/test/mitmproxy/addons/test_termstatus.py index 7becc857..2debaff5 100644 --- a/test/mitmproxy/addons/test_termstatus.py +++ b/test/mitmproxy/addons/test_termstatus.py @@ -5,6 +5,7 @@ from mitmproxy.test import taddons def test_configure(): ts = termstatus.TermStatus() with taddons.context() as ctx: + ctx.configure(ts, server=False) ts.running() assert not ctx.master.logs ctx.configure(ts, server=True) diff --git a/test/mitmproxy/data/addonscripts/addon.py b/test/mitmproxy/data/addonscripts/addon.py index beef2ce7..8c834d82 100644 --- a/test/mitmproxy/data/addonscripts/addon.py +++ b/test/mitmproxy/data/addonscripts/addon.py @@ -9,14 +9,16 @@ class Addon: def load(self, opts): event_log.append("addonload") - def configure(self, options, updated): + def configure(self, updated): event_log.append("addonconfigure") -def configure(options, updated): - event_log.append("addonconfigure") +def configure(updated): + event_log.append("scriptconfigure") def load(l): event_log.append("scriptload") - l.boot_into(Addon()) + + +addons = [Addon()] diff --git a/test/mitmproxy/data/addonscripts/concurrent_decorator.py b/test/mitmproxy/data/addonscripts/concurrent_decorator.py index 162c00f4..d1ab6c6c 100644 --- a/test/mitmproxy/data/addonscripts/concurrent_decorator.py +++ b/test/mitmproxy/data/addonscripts/concurrent_decorator.py @@ -1,4 +1,5 @@ import time +import sys from mitmproxy.script import concurrent diff --git a/test/mitmproxy/data/addonscripts/concurrent_decorator_class.py b/test/mitmproxy/data/addonscripts/concurrent_decorator_class.py index 8e6988d4..2a7d300c 100644 --- a/test/mitmproxy/data/addonscripts/concurrent_decorator_class.py +++ b/test/mitmproxy/data/addonscripts/concurrent_decorator_class.py @@ -9,5 +9,4 @@ class ConcurrentClass: time.sleep(0.1) -def load(l): - l.boot_into(ConcurrentClass()) +addons = [ConcurrentClass()] diff --git a/test/mitmproxy/data/addonscripts/concurrent_decorator_err.py b/test/mitmproxy/data/addonscripts/concurrent_decorator_err.py index 7bc28182..4f80e98a 100644 --- a/test/mitmproxy/data/addonscripts/concurrent_decorator_err.py +++ b/test/mitmproxy/data/addonscripts/concurrent_decorator_err.py @@ -2,5 +2,5 @@ from mitmproxy.script import concurrent @concurrent -def start(opts): +def load(v): pass diff --git a/test/mitmproxy/data/addonscripts/recorder/a.py b/test/mitmproxy/data/addonscripts/recorder/a.py new file mode 100644 index 00000000..df81d86b --- /dev/null +++ b/test/mitmproxy/data/addonscripts/recorder/a.py @@ -0,0 +1,3 @@ +import recorder + +addons = [recorder.Recorder("a")] diff --git a/test/mitmproxy/data/addonscripts/recorder/b.py b/test/mitmproxy/data/addonscripts/recorder/b.py new file mode 100644 index 00000000..ccbae705 --- /dev/null +++ b/test/mitmproxy/data/addonscripts/recorder/b.py @@ -0,0 +1,3 @@ +import recorder + +addons = [recorder.Recorder("b")] diff --git a/test/mitmproxy/data/addonscripts/recorder/c.py b/test/mitmproxy/data/addonscripts/recorder/c.py new file mode 100644 index 00000000..b8b0915e --- /dev/null +++ b/test/mitmproxy/data/addonscripts/recorder/c.py @@ -0,0 +1,3 @@ +import recorder + +addons = [recorder.Recorder("c")] diff --git a/test/mitmproxy/data/addonscripts/recorder/e.py b/test/mitmproxy/data/addonscripts/recorder/e.py new file mode 100644 index 00000000..eb5eff5e --- /dev/null +++ b/test/mitmproxy/data/addonscripts/recorder/e.py @@ -0,0 +1,3 @@ +import recorder + +addons = [recorder.Recorder("e")] diff --git a/test/mitmproxy/data/addonscripts/recorder.py b/test/mitmproxy/data/addonscripts/recorder/recorder.py index fe497b05..a962d3df 100644 --- a/test/mitmproxy/data/addonscripts/recorder.py +++ b/test/mitmproxy/data/addonscripts/recorder/recorder.py @@ -1,13 +1,12 @@ from mitmproxy import controller from mitmproxy import eventsequence from mitmproxy import ctx -import sys -class CallLogger: +class Recorder: call_log = [] - def __init__(self, name = "solo"): + def __init__(self, name = "recorder"): self.name = name def __getattr__(self, attr): @@ -22,5 +21,4 @@ class CallLogger: raise AttributeError -def load(l): - l.boot_into(CallLogger(*sys.argv[1:])) +addons = [Recorder()] diff --git a/test/mitmproxy/proxy/test_server.py b/test/mitmproxy/proxy/test_server.py index 447b15a5..b4bb46bb 100644 --- a/test/mitmproxy/proxy/test_server.py +++ b/test/mitmproxy/proxy/test_server.py @@ -296,8 +296,8 @@ class TestHTTP(tservers.HTTPProxyTest, CommonMixin): class TestHTTPAuth(tservers.HTTPProxyTest): def test_auth(self): self.master.addons.add(proxyauth.ProxyAuth()) - self.master.addons.configure_all( - self.master.options, self.master.options.keys() + self.master.addons.trigger( + "configure", self.master.options.keys() ) self.master.options.proxyauth = "test:test" assert self.pathod("202").status_code == 407 diff --git a/test/mitmproxy/script/test_concurrent.py b/test/mitmproxy/script/test_concurrent.py index 86efdfc2..ceff9fb9 100644 --- a/test/mitmproxy/script/test_concurrent.py +++ b/test/mitmproxy/script/test_concurrent.py @@ -2,10 +2,7 @@ from mitmproxy.test import tflow from mitmproxy.test import tutils from mitmproxy.test import taddons -from mitmproxy import addonmanager from mitmproxy import controller -from mitmproxy.addons import script - import time from .. import tservers @@ -20,14 +17,11 @@ class Thing: class TestConcurrent(tservers.MasterTest): def test_concurrent(self): with taddons.context() as tctx: - sc = script.Script( + sc = tctx.script( tutils.test_data.path( "mitmproxy/data/addonscripts/concurrent_decorator.py" ) ) - l = addonmanager.Loader(tctx.master) - sc.load(l) - f1, f2 = tflow.tflow(), tflow.tflow() tctx.cycle(sc, f1) tctx.cycle(sc, f2) @@ -39,25 +33,20 @@ class TestConcurrent(tservers.MasterTest): def test_concurrent_err(self): with taddons.context() as tctx: - sc = script.Script( + tctx.script( tutils.test_data.path( "mitmproxy/data/addonscripts/concurrent_decorator_err.py" ) ) - l = addonmanager.Loader(tctx.master) - sc.load(l) assert tctx.master.has_log("decorator not supported") def test_concurrent_class(self): with taddons.context() as tctx: - sc = script.Script( + sc = tctx.script( tutils.test_data.path( "mitmproxy/data/addonscripts/concurrent_decorator_class.py" ) ) - l = addonmanager.Loader(tctx.master) - sc.load(l) - f1, f2 = tflow.tflow(), tflow.tflow() tctx.cycle(sc, f1) tctx.cycle(sc, f2) diff --git a/test/mitmproxy/test_addonmanager.py b/test/mitmproxy/test_addonmanager.py index e556bf7d..cba40412 100644 --- a/test/mitmproxy/test_addonmanager.py +++ b/test/mitmproxy/test_addonmanager.py @@ -6,14 +6,17 @@ from mitmproxy import exceptions from mitmproxy import options from mitmproxy import master from mitmproxy import proxy +from mitmproxy.test import taddons from mitmproxy.test import tflow class TAddon: - def __init__(self, name): + def __init__(self, name, addons=None): self.name = name self.tick = True self.custom_called = False + if addons: + self.addons = addons def __repr__(self): return "Addon(%s)" % self.name @@ -35,19 +38,6 @@ class AOption: l.add_option("custom_option", bool, False, "help") -class AChain: - def __init__(self, name, next): - self.name = name - self.next = next - - def load(self, l): - if self.next: - l.boot_into(self.next) - - def __repr__(self): - return "<%s>" % self.name - - def test_halt(): o = options.Options() m = master.Master(o, proxy.DummyServer(o)) @@ -71,10 +61,15 @@ def test_lifecycle(): a = addonmanager.AddonManager(m) a.add(TAddon("one")) + with pytest.raises(exceptions.AddonError): + a.add(TAddon("one")) + with pytest.raises(exceptions.AddonError): + a.remove(TAddon("nonexistent")) + f = tflow.tflow() a.handle_lifecycle("request", f) - a.configure_all(o, o.keys()) + a._configure_all(o, o.keys()) def test_defaults(): @@ -82,33 +77,30 @@ def test_defaults(): def test_simple(): - o = options.Options() - m = master.Master(o, proxy.DummyServer(o)) - a = addonmanager.AddonManager(m) - with pytest.raises(exceptions.AddonError): - a.invoke_addon(TAddon("one"), "done") - - assert len(a) == 0 - a.add(TAddon("one")) - assert a.get("one") - assert not a.get("two") - assert len(a) == 1 - a.clear() - assert len(a) == 0 - assert not a.chain - - a.add(TAddon("one")) - a.trigger("done") - with pytest.raises(exceptions.AddonError): + with taddons.context() as tctx: + a = tctx.master.addons + + assert len(a) == 0 + a.add(TAddon("one")) + assert a.get("one") + assert not a.get("two") + assert len(a) == 1 + a.clear() + assert len(a) == 0 + assert not a.chain + + a.add(TAddon("one")) + a.trigger("done") a.trigger("tick") + tctx.master.has_log("not callable") - a.remove(a.get("one")) - assert not a.get("one") + a.remove(a.get("one")) + assert not a.get("one") - ta = TAddon("one") - a.add(ta) - a.trigger("custom") - assert ta.custom_called + ta = TAddon("one") + a.add(ta) + a.trigger("custom") + assert ta.custom_called def test_load_option(): @@ -119,29 +111,47 @@ def test_load_option(): assert "custom_option" in m.options._options -def test_loadchain(): +def test_nesting(): o = options.Options() m = master.Master(o, proxy.DummyServer(o)) a = addonmanager.AddonManager(m) - a.add(AChain("one", None)) + a.add( + TAddon( + "one", + addons=[ + TAddon("two"), + TAddon("three", addons=[TAddon("four")]) + ] + ) + ) + assert len(a.chain) == 1 assert a.get("one") - a.clear() - - a.add(AChain("one", AChain("two", None))) - assert not a.get("one") assert a.get("two") - a.clear() - - a.add(AChain("one", AChain("two", AChain("three", None)))) - assert not a.get("one") - assert not a.get("two") assert a.get("three") - a.clear() + assert a.get("four") - a.add(AChain("one", AChain("two", AChain("three", AChain("four", None))))) - assert not a.get("one") - assert not a.get("two") + a.trigger("custom") + assert a.get("one").custom_called + assert a.get("two").custom_called + assert a.get("three").custom_called + assert a.get("four").custom_called + + a.remove(a.get("three")) assert not a.get("three") - assert a.get("four") - a.clear() + assert not a.get("four") + + +class D: + def __init__(self): + self.w = None + + def log(self, x): + self.w = x + + +def test_streamlog(): + dummy = D() + s = addonmanager.StreamLog(dummy.log) + s.write("foo") + assert dummy.w == "foo" diff --git a/test/mitmproxy/test_taddons.py b/test/mitmproxy/test_taddons.py index 42371cfe..5a4c99fc 100644 --- a/test/mitmproxy/test_taddons.py +++ b/test/mitmproxy/test_taddons.py @@ -1,4 +1,6 @@ +import io from mitmproxy.test import taddons +from mitmproxy.test import tutils from mitmproxy import ctx @@ -9,3 +11,21 @@ def test_recordingmaster(): ctx.log.error("foo") assert not tctx.master.has_log("foo", level="debug") assert tctx.master.has_log("foo", level="error") + + +def test_dumplog(): + with taddons.context() as tctx: + ctx.log.info("testing") + s = io.StringIO() + tctx.master.dump_log(s) + assert s.getvalue() + + +def test_load_script(): + with taddons.context() as tctx: + s = tctx.script( + tutils.test_data.path( + "mitmproxy/data/addonscripts/recorder/recorder.py" + ) + ) + assert s diff --git a/test/mitmproxy/tools/console/test_master.py b/test/mitmproxy/tools/console/test_master.py index 44b9ff3f..c87c9e83 100644 --- a/test/mitmproxy/tools/console/test_master.py +++ b/test/mitmproxy/tools/console/test_master.py @@ -30,7 +30,7 @@ class TestMaster(tservers.MasterTest): opts["verbosity"] = 1 o = options.Options(**opts) m = console.master.ConsoleMaster(o, proxy.DummyServer()) - m.addons.configure_all(o, o.keys()) + m.addons.trigger("configure", o.keys()) return m def test_basic(self): @@ -42,12 +42,6 @@ class TestMaster(tservers.MasterTest): pass assert len(m.view) == i - def test_run_script_once(self): - m = self.mkmaster() - f = tflow.tflow(resp=True) - m.run_script_once("nonexistent", [f]) - assert any("Input error" in str(l) for l in m.logbuffer) - def test_intercept(self): """regression test for https://github.com/mitmproxy/mitmproxy/issues/1605""" m = self.mkmaster(intercept="~b bar") diff --git a/test/mitmproxy/tservers.py b/test/mitmproxy/tservers.py index b737b82a..b8005529 100644 --- a/test/mitmproxy/tservers.py +++ b/test/mitmproxy/tservers.py @@ -74,7 +74,7 @@ class TestMaster(taddons.RecordingMaster): self.state = TestState() self.addons.add(self.state) self.addons.add(*addons) - self.addons.configure_all(self.options, self.options.keys()) + self.addons.trigger("configure", self.options.keys()) self.addons.trigger("running") def reset(self, addons): |