Start stubbing out a much more powerful script architecture.

libmproxy/plugins.py

@@ -0,0 +1,46 @@
+import imp, os, traceback
+
+
+class Context:
+    def __init__(self, master, state):
+        self.master, self.state = master, state
+
+    def log(self, *args, **kwargs):
+        self.master.log(*args, **kwargs)
+
+
+class Plugin:
+    def __init__(self, path, master):
+        self.path = path
+        self.ctx = Context(master, master.state)
+        self.mod = None
+        self.ns = None
+        self.load()
+
+    def load(self):
+        """
+            Loads a module and runs the start method.
+        """
+        ns = {}
+        self.mod = execfile(os.path.expanduser(self.path), {}, ns)
+        self.ns = ns
+        self.run("start")
+
+    def run(self, name, *args, **kwargs):
+        """
+            Runs a plugin method.
+
+            Returns:
+
+                (True, retval) on success.
+                (False, None) on nonexistent method.
+                (Fals, (exc, traceback string)) if there was an exception.
+        """
+        f = self.ns.get(name)
+        if f:
+            try:
+                return (True, f(self.ctx, *args, **kwargs))
+            except Exception, v:
+                return (False, (v, traceback.format_exc(v)))
+        else:
+            return (False, None)

test/plugins/a.py

@@ -0,0 +1,9 @@
+
+var = 0
+def here(ctx):
+    global var
+    var += 1
+    return var
+
+def errargs():
+    pass

test/plugins/syntaxerr.py

@@ -0,0 +1,3 @@
+
+
+a +

test/test_plugins.py

@@ -0,0 +1,35 @@
+import os
+from libmproxy import plugins, flow
+import libpry
+
+class uPlugin(libpry.AutoTree):
+    def test_simple(self):
+        s = flow.State()
+        fm = flow.FlowMaster(None, s)
+
+        p = plugins.Plugin(os.path.join("plugins", "a.py"), fm)
+        assert "here" in p.ns
+        assert p.run("here") == (True, 1)
+        assert p.run("here") == (True, 2)
+        
+        ret = p.run("errargs")
+        assert not ret[0] 
+        assert len(ret[1]) == 2
+
+        # Check reload
+        p.load()
+        assert p.run("here") == (True, 1)
+
+    def test_err(self):
+        s = flow.State()
+        fm = flow.FlowMaster(None, s)
+
+        libpry.raises(IOError, plugins.Plugin, "nonexistent", fm)
+        libpry.raises(SyntaxError, plugins.Plugin, os.path.join("plugins", "syntaxerr.py"), fm)
+
+
+
+tests = [
+    uPlugin(),
+]
+