kylefang/mitmproxy
1
0
Fork 0
mirror of https://github.com/zhigang1992/mitmproxy.git synced 2026-04-30 05:05:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

addons.script: convert to test.taddons

Browse Source
This commit is contained in:
Aldo Cortesi
2016-11-03 18:57:34 +13:00
parent d31f2698a5
commit d9538637c3
1 changed files with 97 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
test/mitmproxy/addons/test_script.py
View File

@@ -1,18 +1,17 @@
import traceback import traceback
import sys import sys
import time import time
import re
from mitmproxy.test import tflow from mitmproxy.test import tflow
from mitmproxy.test import tutils from mitmproxy.test import tutils
import re from mitmproxy.test import taddons
from mitmproxy import exceptions from mitmproxy import exceptions
from mitmproxy import options from mitmproxy import options
from mitmproxy import proxy from mitmproxy import proxy
from mitmproxy.addons import script from mitmproxy.addons import script
from mitmproxy import master from mitmproxy import master
from .. import mastertest
from .. import tutils as ttutils from .. import tutils as ttutils
@@ -64,73 +63,68 @@ def test_load_script():
assert ns.start assert ns.start
class TestScript(mastertest.MasterTest): class TestScript:
def test_simple(self): def test_simple(self):
o = options.Options() with taddons.context() as tctx:
m = master.Master(o, proxy.DummyServer()) sc = script.Script(
sc = script.Script( tutils.test_data.path(
tutils.test_data.path( "mitmproxy/data/addonscripts/recorder.py"
"mitmproxy/data/addonscripts/recorder.py" )
) )
) sc.load_script()
m.addons.add(sc) assert sc.ns.call_log == [
assert sc.ns.call_log == [ ("solo", "start", (), {}),
("solo", "start", (), {}), ]
("solo", "configure", (o, o.keys()), {})
]
sc.ns.call_log = [] sc.ns.call_log = []
f = tflow.tflow(resp=True) f = tflow.tflow(resp=True)
m.request(f) sc.request(f)
recf = sc.ns.call_log[0] recf = sc.ns.call_log[0]
assert recf[1] == "request" assert recf[1] == "request"
def test_reload(self): def test_reload(self):
o = options.Options() with taddons.context() as tctx:
m = mastertest.RecordingMaster(o, proxy.DummyServer()) with tutils.tmpdir():
with tutils.tmpdir(): with open("foo.py", "w"):
with open("foo.py", "w"): pass
pass sc = script.Script("foo.py")
sc = script.Script("foo.py") tctx.configure(sc)
m.addons.add(sc) for _ in range(100):
with open("foo.py", "a") as f:
for _ in range(100): f.write(".")
with open("foo.py", "a") as f: sc.tick()
f.write(".") time.sleep(0.1)
m.addons.invoke_with_context(sc, "tick") if tctx.master.event_log:
time.sleep(0.1) return
if m.event_log: raise AssertionError("Change event not detected.")
return
raise AssertionError("Change event not detected.")
def test_exception(self): def test_exception(self):
o = options.Options() with taddons.context() as tctx:
m = mastertest.RecordingMaster(o, proxy.DummyServer()) sc = script.Script(
sc = script.Script( tutils.test_data.path("mitmproxy/data/addonscripts/error.py")
tutils.test_data.path("mitmproxy/data/addonscripts/error.py") )
) sc.start()
m.addons.add(sc) f = tflow.tflow(resp=True)
f = tflow.tflow(resp=True) sc.request(f)
m.request(f) assert tctx.master.event_log[0][0] == "error"
assert m.event_log[0][0] == "error" assert len(tctx.master.event_log[0][1].splitlines()) == 6
assert len(m.event_log[0][1].splitlines()) == 6 assert re.search('addonscripts/error.py", line \d+, in request', tctx.master.event_log[0][1])
assert re.search('addonscripts/error.py", line \d+, in request', m.event_log[0][1]) assert re.search('addonscripts/error.py", line \d+, in mkerr', tctx.master.event_log[0][1])
assert re.search('addonscripts/error.py", line \d+, in mkerr', m.event_log[0][1]) assert tctx.master.event_log[0][1].endswith("ValueError: Error!\n")
assert m.event_log[0][1].endswith("ValueError: Error!\n")
def test_addon(self): def test_addon(self):
o = options.Options() with taddons.context() as tctx:
m = master.Master(o, proxy.DummyServer()) sc = script.Script(
sc = script.Script( tutils.test_data.path(
tutils.test_data.path( "mitmproxy/data/addonscripts/addon.py"
"mitmproxy/data/addonscripts/addon.py" )
) )
) sc.start()
m.addons.add(sc) tctx.configure(sc)
assert sc.ns.event_log == [ assert sc.ns.event_log == [
'scriptstart', 'addonstart', 'addonconfigure' 'scriptstart', 'addonstart', 'addonconfigure'
] ]
class TestCutTraceback: class TestCutTraceback:
@@ -151,7 +145,7 @@ class TestCutTraceback:
assert len(traceback.extract_tb(tb_cut2)) == len(traceback.extract_tb(tb)) assert len(traceback.extract_tb(tb_cut2)) == len(traceback.extract_tb(tb))
class TestScriptLoader(mastertest.MasterTest): class TestScriptLoader:
def test_run_once(self): def test_run_once(self):
o = options.Options(scripts=[]) o = options.Options(scripts=[])
m = master.Master(o, proxy.DummyServer()) m = master.Master(o, proxy.DummyServer())
@@ -199,44 +193,48 @@ class TestScriptLoader(mastertest.MasterTest):
def test_order(self): def test_order(self):
rec = tutils.test_data.path("mitmproxy/data/addonscripts/recorder.py") rec = tutils.test_data.path("mitmproxy/data/addonscripts/recorder.py")
o = options.Options(
scripts = [
"%s %s" % (rec, "a"),
"%s %s" % (rec, "b"),
"%s %s" % (rec, "c"),
]
)
m = mastertest.RecordingMaster(o, proxy.DummyServer())
sc = script.ScriptLoader() sc = script.ScriptLoader()
m.addons.add(sc) with taddons.context() as tctx:
tctx.master.addons.add(sc)
tctx.configure(
sc,
scripts = [
"%s %s" % (rec, "a"),
"%s %s" % (rec, "b"),
"%s %s" % (rec, "c"),
]
)
debug = [(i[0], i[1]) for i in tctx.master.event_log if i[0] == "debug"]
assert debug == [
('debug', 'a start'), ('debug', 'a configure'),
('debug', 'b start'), ('debug', 'b configure'),
('debug', 'c start'), ('debug', 'c configure')
]
tctx.master.event_log = []
tctx.configure(
sc,
scripts = [
"%s %s" % (rec, "c"),
"%s %s" % (rec, "a"),
"%s %s" % (rec, "b"),
]
)
debug = [(i[0], i[1]) for i in tctx.master.event_log if i[0] == "debug"]
# No events, only order has changed
assert debug == []
debug = [(i[0], i[1]) for i in m.event_log if i[0] == "debug"] tctx.master.event_log = []
assert debug == [ tctx.configure(
('debug', 'a start'), ('debug', 'a configure'), sc,
('debug', 'b start'), ('debug', 'b configure'), scripts = [
('debug', 'c start'), ('debug', 'c configure') "%s %s" % (rec, "x"),
] "%s %s" % (rec, "a"),
m.event_log[:] = [] ]
)
o.scripts = [ debug = [(i[0], i[1]) for i in tctx.master.event_log if i[0] == "debug"]
"%s %s" % (rec, "c"), assert debug == [
"%s %s" % (rec, "a"), ('debug', 'c done'),
"%s %s" % (rec, "b"), ('debug', 'b done'),
] ('debug', 'x start'),
debug = [(i[0], i[1]) for i in m.event_log if i[0] == "debug"] ('debug', 'x configure'),
# No events, only order has changed ]
assert debug == []
o.scripts = [
"%s %s" % (rec, "x"),
"%s %s" % (rec, "a"),
]
debug = [(i[0], i[1]) for i in m.event_log if i[0] == "debug"]
assert debug == [
('debug', 'c done'),
('debug', 'b done'),
('debug', 'x start'),
('debug', 'x configure'),
('debug', 'a configure'),
]