utils: Add multiprocess_launch API and testcase

The current methods of spawning processes for parallel execution have
issues around collection of results or exceptions.

Take the code from package_ipk/deb, make it generic, add a results
collection mechanism, fix the exception handling and for it into a
standard library function.

Also add a test case which tests both the success and failure modes
of operation to stop this functionality regressiing again.

In particular, compared to multiprocess_exec, this fork off the parent
approach means we can pass in the datastore and functions work in the
same scope as the parent. This removes some of the complexities
found trying to scale multiprocess_exec to wider use.

Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>

1 files changed, 45 insertions, 1 deletions

diff --git a/meta/lib/oeqa/selftest/cases/oelib/utils.py b/meta/lib/oeqa/selftest/cases/oelib/utils.py
index 9fb6c1576e..275aeda74e 100644
--- a/meta/lib/oeqa/selftest/cases/oelib/utils.py
+++ b/meta/lib/oeqa/selftest/cases/oelib/utils.py
@@ -1,5 +1,8 @@
+import sys
 from unittest.case import TestCase
-from oe.utils import packages_filter_out_system, trim_version
+from contextlib import contextmanager
+from io import StringIO
+from oe.utils import packages_filter_out_system, trim_version, multiprocess_launch
 
 class TestPackagesFilterOutSystem(TestCase):
     def test_filter(self):
@@ -49,3 +52,44 @@ class TestTrimVersion(TestCase):
         self.assertEqual(trim_version("1.2.3", 2), "1.2")
         self.assertEqual(trim_version("1.2.3", 3), "1.2.3")
         self.assertEqual(trim_version("1.2.3", 4), "1.2.3")
+
+
+class TestMultiprocessLaunch(TestCase):
+
+    def test_multiprocesslaunch(self):
+        import bb
+
+        def testfunction(item, d):
+            if item == "2" or item == "1":
+                raise KeyError("Invalid number %s" % item)
+            return "Found %s" % item
+
+        def dummyerror(msg):
+            print("ERROR: %s" % msg)
+
+        @contextmanager
+        def captured_output():
+            new_out, new_err = StringIO(), StringIO()
+            old_out, old_err = sys.stdout, sys.stderr
+            try:
+                sys.stdout, sys.stderr = new_out, new_err
+                yield sys.stdout, sys.stderr
+            finally:
+                sys.stdout, sys.stderr = old_out, old_err
+
+        d = bb.data_smart.DataSmart()
+        bb.error = dummyerror
+
+        # Assert the function returns the right results
+        result = multiprocess_launch(testfunction, ["3", "4", "5", "6"], d, extraargs=(d,))
+        self.assertIn("Found 3", result)
+        self.assertIn("Found 4", result)
+        self.assertIn("Found 5", result)
+        self.assertIn("Found 6", result)
+        self.assertEqual(len(result), 4)
+
+        # Assert the function prints exceptions
+        with captured_output() as (out, err):
+            self.assertRaises(bb.BBHandledException, multiprocess_launch, testfunction, ["1", "2", "3", "4", "5", "6"], d, extraargs=(d,))
+        self.assertIn("KeyError: 'Invalid number 1'", out.getvalue())
+        self.assertIn("KeyError: 'Invalid number 2'", out.getvalue())