diff options
Diffstat (limited to 'test/test_script.py')
-rw-r--r-- | test/test_script.py | 132 |
1 files changed, 0 insertions, 132 deletions
diff --git a/test/test_script.py b/test/test_script.py deleted file mode 100644 index f3a6499e..00000000 --- a/test/test_script.py +++ /dev/null @@ -1,132 +0,0 @@ -import os -import time -import mock -from libmproxy import script, flow -from . import tutils - - -def test_simple(): - s = flow.State() - fm = flow.FlowMaster(None, s) - sp = tutils.test_data.path("scripts/a.py") - p = script.Script("%s --var 40" % sp, script.ScriptContext(fm)) - - assert "here" in p.ns - assert p.run("here") == 41 - assert p.run("here") == 42 - - tutils.raises(script.ScriptException, p.run, "errargs") - - # Check reload - p.load() - assert p.run("here") == 41 - - -def test_duplicate_flow(): - s = flow.State() - fm = flow.FlowMaster(None, s) - fm.load_script(tutils.test_data.path("scripts/duplicate_flow.py")) - f = tutils.tflow() - fm.handle_request(f) - assert fm.state.flow_count() == 2 - assert not fm.state.view[0].request.is_replay - assert fm.state.view[1].request.is_replay - - -def test_err(): - s = flow.State() - fm = flow.FlowMaster(None, s) - sc = script.ScriptContext(fm) - - tutils.raises( - "not found", - script.Script, "nonexistent", sc - ) - - tutils.raises( - "not a file", - script.Script, tutils.test_data.path("scripts"), sc - ) - - tutils.raises( - script.ScriptException, - script.Script, tutils.test_data.path("scripts/syntaxerr.py"), sc - ) - - tutils.raises( - script.ScriptException, - script.Script, tutils.test_data.path("scripts/loaderr.py"), sc - ) - - scr = script.Script(tutils.test_data.path("scripts/unloaderr.py"), sc) - tutils.raises(script.ScriptException, scr.unload) - - -@tutils.skip_appveyor -def test_concurrent(): - s = flow.State() - fm = flow.FlowMaster(None, s) - fm.load_script(tutils.test_data.path("scripts/concurrent_decorator.py")) - - with mock.patch("libmproxy.controller.DummyReply.__call__") as m: - f1, f2 = tutils.tflow(), tutils.tflow() - t_start = time.time() - fm.handle_request(f1) - f1.reply() - fm.handle_request(f2) - f2.reply() - - # Two instantiations - assert m.call_count == 0 # No calls yet. - assert (time.time() - t_start) < 0.1 - - -def test_concurrent2(): - s = flow.State() - fm = flow.FlowMaster(None, s) - s = script.Script( - tutils.test_data.path("scripts/concurrent_decorator.py"), - script.ScriptContext(fm)) - s.load() - m = mock.Mock() - - class Dummy: - - def __init__(self): - self.response = self - self.error = self - self.reply = m - - t_start = time.time() - - for hook in ("clientconnect", - "serverconnect", - "response", - "error", - "clientconnect"): - d = Dummy() - s.run(hook, d) - d.reply() - while (time.time() - t_start) < 20 and m.call_count <= 5: - if m.call_count == 5: - return - time.sleep(0.001) - assert False - - -def test_concurrent_err(): - s = flow.State() - fm = flow.FlowMaster(None, s) - tutils.raises( - "Concurrent decorator not supported for 'start' method", - script.Script, - tutils.test_data.path("scripts/concurrent_decorator_err.py"), - fm) - - -def test_command_parsing(): - s = flow.State() - fm = flow.FlowMaster(None, s) - absfilepath = os.path.normcase(tutils.test_data.path("scripts/a.py")) - s = script.Script(absfilepath, script.ScriptContext(fm)) - assert os.path.isfile(s.args[0]) |