build candidate 2019.11.12.20.52.20.14283

qaviton_io/logger.py

@@ -26,6 +26,27 @@ def log(self):
     def clear():
         Log._log = {}
 
+    def send_message(self, name, *args):
+        assert args
+        log = self.log
+        if name not in log:
+            log[name] = {'msg': [*args]}
+        elif 'msg' not in log[name]:
+            log[name]['msg'] = [*args]
+        else:
+            log[name]['msg'].extend(args)
+
+    def has_message(self, name):
+        return 'msg' in self.log[name]
+
+    def receive_messages(self, name) -> list:
+        if self.has_message(name):
+            return self.log[name]['msg']
+
+    def receive_all_messages(self):
+        log = self.log
+        return {name: log[name]['msg'] for name in log if 'msg' in log[name]}
+
     def merge(self, queue: Queue):
         results: List[Dict] = []
         append = results.append
@@ -45,6 +66,11 @@ def merge(self, queue: Queue):
                             self.log[name]["fail"][error] = []
                         self.log[name]["fail"][error].extend(durations)
                     self.log[name]["pass"].extend(log["pass"])
+                    if 'msg' in log:
+                        if 'msg' not in self.log[name]:
+                            self.log[name]['msg'] = log['msg']
+                        else:
+                            self.log[name]['msg'].extend(log['msg'])
         return self
 
     def analyze(

qaviton_io/process_manager.py

@@ -34,6 +34,10 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_val, exc_tb):
         pass
 
+    def merge(self):
+        self.log.merge(self.queue)
+        return self
+
     def analyze(
         self,
         analyze_pass=True,
@@ -87,16 +91,17 @@ def wait_until_tasks_are_done(tasks: List[Task], timeout):
                 for session in tasks:
                     if session.is_finished():
                         finished_sessions += 1
-            try:
-                for session in tasks:
-                    if session.is_finished(timeout=timeout):
-                        finished_sessions += 1
-                    if time()-t > timeout:
-                        raise TimeoutError
-            except TimeoutError as e:
-                for session in tasks:
-                    try:
-                        session.kill()
-                    except:
-                        pass
-                raise e
+            else:
+                try:
+                    for session in tasks:
+                        if session.is_finished(timeout=timeout):
+                            finished_sessions += 1
+                        if time() - t > timeout:
+                            raise TimeoutError
+                except TimeoutError as e:
+                    for session in tasks:
+                        try:
+                            session.kill()
+                        except:
+                            pass
+                    raise e

setup.py

@@ -8,7 +8,7 @@
     with open("README.md", encoding="utf8") as f: long_description = f.read()
     setup(
         name=package_name,
-        version="2019.11.12.17.36.46.68906",
+        version="2019.11.12.20.52.20.14283",
         author="yehonadav",
         author_email="qaviton@gmail.com",
         description="qaviton io",

tests/test_log_messaging.py

@@ -0,0 +1,27 @@
+from time import time
+from uuid import uuid4
+from qaviton_io import AsyncManager, task, Log, ProcessManager
+
+
+@task()
+def st():
+    Log().send_message('st', {'id': f'{uuid4()}', 'timestamp': time()})
+
+
+def test_log_messaging():
+    n = 10
+    with AsyncManager() as m:
+        m.log.clear()
+        m.run([st for _ in range(n)])
+        msgs = m.log.receive_all_messages()['st']
+    assert len(msgs) == n
+
+
+def test_messaging_with_processes():
+    n = 10
+    with ProcessManager() as m:
+        m.log.clear()
+        m.run_until_complete([st for _ in range(n)], timeout=None)
+        m.merge()
+        msgs = m.log.receive_all_messages()['st']
+    assert len(msgs) == n