refactored multi process test framework, so that test can evaluate to failed/passed and not require someone to read and interpred if it did work

test/mp_sync_test.py

@@ -8,43 +8,24 @@ DYLD_LIBRARY_PATH=/Users/ed/lib/ PYTHONPATH=`pwd`/../build/lib.macosx-10.11-x86_
 
 import multipEp as mp
 
-# unused
-def send_message(from_address, to_address):
-    m = mp.pEp.outgoing_message(Identity(from_address, from_address))
-    m.to = [mp.pEp.Identity(to_address, to_address)]
-    m.shortmsg = "Hello"
-    m.longmsg = "Something\\n"
-    m.encrypt()
-    mp.sent_messages.append(str(m))
 
 scenario0 = [
-   #("instance name", ["func name", [args], {kwargs}]),
+   #("instance name", [func, [args], {kwargs}]),
     ("A", [mp.create_account, ["some.one@some.where", "Some One"]]),
     ("B", [mp.create_account, ["some.one@some.where", "Some One"]]),
-    ("A", []),
+    (mp.cycle_until_no_change, ["A", "B"]),
-    ("B", []),
-    ("A", []),
-    ("B", []),
-    ("A", []),
-    ("B", []),
     ("C", [mp.create_account, ["some.one@some.where", "Some One"]]),
-    ("A", []),
+    (mp.cycle_until_no_change, ["A", "B", "C"]),
-    ("B", []),
+    # force consume messages
-    ("C", []),
-    ("A", []),
-    ("B", []),
-    ("C", []),
-    ("A", []),
-    ("B", []),
-    ("C", []),
-    ("A", []),
-    ("B", []),
-    ("C", []),
-    ("A", []),
-    ("B", []),
     ("C", [None, None, None, -60*15])
 ] 
 
+scenario1 = [
+   #("instance name", [func, [args], {kwargs}]),
+    ("A", [mp.send_message, ["some.one@some.where", "some.other@else.where", "Hey Bro", "Heeeey Brooooo"]]),
+    ("B", [mp.send_message, ["some.other@else.where", "some.one@some.where", "Hey Bro", "Heeeey Brooooo"]]),
+] 
+
 if __name__ == "__main__":
     mp.run_scenario(scenario0)
 

test/multipEp.py

@@ -4,8 +4,13 @@ import multiprocessing
 import importlib
 import tempfile
 import time
+import types
+from copy import deepcopy
 from collections import OrderedDict
 
+# manager globals
+instances = None
+
 # per-instance globals
 pEp = None
 handler = None
@@ -15,12 +20,44 @@ i_name = ""
 handshakes_pending = []
 handshakes_to_accept = []
 
+# both side globals (managed by MP)
+handshakes_seen = []
+handshakes_validated = []
+msgs_folders = None
+
 def create_account(address, name):
     global own_addresses
     i = pEp.Identity(address, name)
     pEp.myself(i)
     own_addresses.append(address)
 
+def _send_message(address, msg):
+    global msgs_folders
+    # list inside dict from MP manager are not proxified.
+    msgs = msgs_folders.get(address,[])
+    msg.sent = int(time.time())
+    msgs.append(str(msg))
+    msgs_folders[address] = msgs
+
+def _encrypted_message(from_address, to_address, shortmsg, longmsg):
+    m = pEp.outgoing_message(Identity(from_address, from_address))
+    m.to = [pEp.Identity(to_address, to_address)]
+    m.shortmsg = shortmsg
+    m.longmsg = longmsg
+    m.encrypt()
+    return msg
+
+def encrypted_message(from_address, to_address, shortmsg, longmsg):
+    return str(_encrypted_message(from_address, to_address, shortmsg, longmsg))
+
+def send_message(from_address, to_address, shortmsg, longmsg):
+    msg = _encrypted_message(from_address, to_address, shortmsg, longmsg)
+    _send_message(to_address, msg)
+
+def decrypt_message(msgstr):
+    msg = pEp.incoming_message(msgstr)
+    msg2, keys, rating, consumed, flags = msg.decrypt()
+
 def printi(*args):
     global indent
     print(i_name + ">" * indent, *args)
@@ -51,48 +88,9 @@ def printmsg(msg):
         pfx = "       "
     printi("attachments : ", msg.attachments)
 
-def pEp_instance_run(iname, conn, msgs_folders, handshakes_seen, handshakes_validated):
+def execute_order(order, handler, conn):
-    global pEp, handler, own_addresses, i_name
+    global handshakes_pending, handshakes_to_accept, handshakes_seen
-
+    global handshakes_validated, msgs_folders
-    i_name = iname
-
-    pEp = importlib.import_module("pEp")
-
-    class Handler(pEp.SyncMixIn):
-        def messageToSend(self, msg):
-            printheader("SYNC MESSAGE to send")
-            printmsg(msg)
-            printheader()
-            for rcpt in msg.to + msg.cc + msg.bcc:
-                # list inside dict from MP manager are not proxified.
-                msgs = msgs_folders.get(rcpt.address,[])
-                msg.sent = int(time.time())
-                msgs.append(str(msg))
-                msgs_folders[rcpt.address] = msgs
-
-        def showHandshake(self, me, partner):
-            printheader("show HANDSHAKE dialog")
-            printi("handshake needed between " + repr(me) + " and " + repr(partner))
-            tw = pEp.trustwords(me, partner, 'en')
-            printi(tw)
-            if tw in handshakes_seen :
-                handshakes_seen.remove(tw)
-                handshakes_to_accept.append((tw,partner))
-                printi("--> TO ACCEPT (already seen)")
-            else:
-                handshakes_pending.append((tw,partner))
-                handshakes_seen.append(tw)
-            printheader()
-
-            # TODO: reject scenario ?
-
-    handler = Handler()
-
-    while True:
-        order = conn.recv()
-        if order is None:
-            break
-
     func, args, kwargs, timeoff = order[0:] + [None, [], {}, 0][len(order):]
 
     printheader("DECRYPT messages")
@@ -153,6 +151,52 @@ def pEp_instance_run(iname, conn, msgs_folders, handshakes_seen, handshakes_vali
 
     conn.send(res)
 
+def pEp_instance_run(iname, conn, _msgs_folders, _handshakes_seen, _handshakes_validated):
+    global pEp, handler, own_addresses, i_name, msgs_folders
+    global handshakes_pending, handshakes_to_accept
+    global handshakes_seen, handshakes_validated
+
+    # assign instance globals
+    msgs_folders = _msgs_folders
+    handshakes_seen = _handshakes_seen
+    handshakes_validated = _handshakes_validated
+    i_name = iname
+
+    pEp = importlib.import_module("pEp")
+
+    class Handler(pEp.SyncMixIn):
+        def messageToSend(self, msg):
+            printheader("SYNC MESSAGE to send")
+            printmsg(msg)
+            printheader()
+            for rcpt in msg.to + msg.cc + msg.bcc:
+                _send_message(rcpt.address, msg)
+
+        def showHandshake(self, me, partner):
+            printheader("show HANDSHAKE dialog")
+            printi("handshake needed between " + repr(me) + " and " + repr(partner))
+            tw = pEp.trustwords(me, partner, 'en')
+            printi(tw)
+            if tw in handshakes_seen :
+                handshakes_seen.remove(tw)
+                handshakes_to_accept.append((tw,partner))
+                printi("--> TO ACCEPT (already seen)")
+            else:
+                handshakes_pending.append((tw,partner))
+                handshakes_seen.append(tw)
+            printheader()
+
+    handler = Handler()
+
+    while True:
+        order = conn.recv()
+        if order is None:
+            break
+
+        execute_order(order, handler, conn)
+
+    msgs_folders = None
+
 def pEp_instance_main(iname, *args):
     # run with a dispensable $HOME to get fresh DB and PGP keyrings
     with tempfile.TemporaryDirectory() as tmpdirname:
@@ -161,13 +205,10 @@ def pEp_instance_main(iname, *args):
         pEp_instance_run(iname, *args)
         print(iname + " exiting")
 
-def run_scenario(scenario):
+def run_instance_action(action):
-    instances = OrderedDict()
+    global handshakes_seen, handshakes_validated, msgs_folders
-    with multiprocessing.Manager() as manager:
+    global instances
-        msgs_folders = manager.dict()
+    iname, order = action
-        handshakes_seen = manager.list()
-        handshakes_validated = manager.list()
-        for iname, order in scenario:
     if iname not in instances:
         conn, child_conn = multiprocessing.Pipe()
         proc = multiprocessing.Process(
@@ -193,15 +234,63 @@ def run_scenario(scenario):
         proc, conn = instances[iname]
 
     conn.send(order)
-            res = conn.recv()
+    return conn.recv()
+
+def purge_instance():
+    global instances
+    for iname, (proc, conn) in instances.items():
+        # tell process to terminate
+        conn.send(None)
+        proc.join()
+
+def run_manager_action(action):
+    func, args, kwargs = action[0:] + (None, [], {})[len(action):]
+    return func(*args, **kwargs)
+
+def run_scenario(scenario):
+    global handshakes_seen, handshakes_validated, msgs_folders, instances
+    instances = OrderedDict()
+    with multiprocessing.Manager() as manager:
+        msgs_folders = manager.dict()
+        handshakes_seen = manager.list()
+        handshakes_validated = manager.list()
+
+        res = None
+        for action in scenario:
+            output = None
+            if type(action[-1]) == types.FunctionType:
+                output = action[-1]
+                action = action[:-1]
+
+            if type(action[0]) == str:
+                res = run_instance_action(action)
+            else:
+                res = run_manager_action(action)
+
+            if output is not None:
+                output(res, action)
 
         if "wait_for_debug" in sys.argv:
             input("#"*80 + "\n" +
                   "Press ENTER to cleanup\n" +
                   "#"*80 + "\n")
 
-        for iname, (proc, conn) in instances.items():
+        purge_instance()
-            # tell process to terminate
+
-            conn.send(None)
+def cycle_until_no_change(*instancelist, maxcycles=20):
-            proc.join()
+    count = 0
+    while True:
+        global msgs_folders
+        tmp = deepcopy(dict(msgs_folders))
+        for iname in instancelist:
+            action = (iname, [])
+            run_instance_action(action)
+        count += 1
+
+        if dict(msgs_folders) == tmp:
+            return
+
+        if count >= maxcycles:
+            raise Exception("Too many cycles waiting for stability") 
+