Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-23 08:11:00 +00:00
Code Issues Projects Releases Wiki Activity

script: revamp ephemeral script running

Browse Source 
Fixes 2838
This commit is contained in:
Aldo Cortesi 2018-05-12 12:09:48 +12:00
parent 260f729092
commit 88fe26997c
2 changed files with 21 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
mitmproxy/addons/script.py
View File

@ -148,16 +148,26 @@ class ScriptLoader:
@command.command("script.run") @command.command("script.run")
def script_run(self, flows: typing.Sequence[flow.Flow], path: mtypes.Path) -> None: def script_run(self, flows: typing.Sequence[flow.Flow], path: mtypes.Path) -> None:
""" """
Run a script on the specified flows. The script is loaded with Run a script on the specified flows. The script is configured with
default options, and all lifecycle events for each flow are the current options and all lifecycle events for each flow are
simulated. simulated. Note that the load event is not invoked.
""" """
if not os.path.isfile(path):
ctx.log.error('No such script: %s' % path)
return
try: try:
s = Script(path, False) mod = load_script(path)
for f in flows: if mod:
for evt, arg in eventsequence.iterate(f): ctx.master.addons.invoke_addon(mod, "running")
with addonmanager.safecall(): ctx.master.addons.invoke_addon(
ctx.master.addons.invoke_addon(s, evt, arg) mod,
"configure",
ctx.options.keys()
)
for f in flows:
for evt, arg in eventsequence.iterate(f):
with addonmanager.safecall():
ctx.master.addons.invoke_addon(mod, evt, arg)
except exceptions.OptionsError as e: except exceptions.OptionsError as e:
script_error_handler(path, e, msg=str(e)) script_error_handler(path, e, msg=str(e))

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

@ -173,16 +173,14 @@ class TestCutTraceback:
class TestScriptLoader: class TestScriptLoader:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_script_run(self, tdata): async def test_script_run(self, tdata):
rp = tdata.path( rp = tdata.path("mitmproxy/data/addonscripts/recorder/recorder.py")
"mitmproxy/data/addonscripts/recorder/recorder.py"
)
sc = script.ScriptLoader() sc = script.ScriptLoader()
with taddons.context(sc) as tctx: with taddons.context(sc) as tctx:
sc.script_run([tflow.tflow(resp=True)], rp) sc.script_run([tflow.tflow(resp=True)], rp)
await tctx.master.await_log("recorder response") await tctx.master.await_log("recorder response")
debug = [i.msg for i in tctx.master.logs if i.level == "debug"] debug = [i.msg for i in tctx.master.logs if i.level == "debug"]
assert debug == [ assert debug == [
'recorder load', 'recorder running', 'recorder configure', 'recorder running', 'recorder configure',
'recorder requestheaders', 'recorder request', 'recorder requestheaders', 'recorder request',
'recorder responseheaders', 'recorder response' 'recorder responseheaders', 'recorder response'
] ]
@ -192,7 +190,7 @@ class TestScriptLoader:
sc = script.ScriptLoader() sc = script.ScriptLoader()
with taddons.context(sc) as tctx: with taddons.context(sc) as tctx:
sc.script_run([tflow.tflow(resp=True)], "/") sc.script_run([tflow.tflow(resp=True)], "/")
assert await tctx.master.await_log("/: No such script") assert await tctx.master.await_log("No such script")
def test_simple(self, tdata): def test_simple(self, tdata):
sc = script.ScriptLoader() sc = script.ScriptLoader()