oeqa/core: Implement proper extra result collection and serialization

Implement handling of extra result (e.g. ptestresult) collection with
the addition of a "extraresults" extraction function in OETestResult. In
order to be able to serialize and deserialize the extraresults data,
allow OETestResult add* calls to take a details kwarg. The subunit
module can handle cross-process transfer of binary data for the details
kwarg. With a TestResult proxy class to sit inbetween to encode and
decode to and from json.

Signed-off-by: Nathan Rossi <nathan@nathanrossi.com>
Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>

2 files changed, 96 insertions, 6 deletions

diff --git a/meta/lib/oeqa/core/runner.py b/meta/lib/oeqa/core/runner.py
index 930620ea19..3060a00fbf 100644
--- a/meta/lib/oeqa/core/runner.py
+++ b/meta/lib/oeqa/core/runner.py
@@ -43,6 +43,7 @@ class OETestResult(_TestResult):
         self.starttime = {}
         self.endtime = {}
         self.progressinfo = {}
+        self.extraresults = {}
 
         # Inject into tc so that TestDepends decorator can see results
         tc.results = self
@@ -129,19 +130,51 @@ class OETestResult(_TestResult):
 
         return 'UNKNOWN', None
 
-    def addSuccess(self, test):
+    def extractExtraResults(self, test, details = None):
+        extraresults = None
+        if details is not None and "extraresults" in details:
+            extraresults = details.get("extraresults", {})
+        elif hasattr(test, "extraresults"):
+            extraresults = test.extraresults
+
+        if extraresults is not None:
+            for k, v in extraresults.items():
+                # handle updating already existing entries (e.g. ptestresults.sections)
+                if k in self.extraresults:
+                    self.extraresults[k].update(v)
+                else:
+                    self.extraresults[k] = v
+
+    def addError(self, test, *args, details = None):
+        self.extractExtraResults(test, details = details)
+        return super(OETestResult, self).addError(test, *args)
+
+    def addFailure(self, test, *args, details = None):
+        self.extractExtraResults(test, details = details)
+        return super(OETestResult, self).addFailure(test, *args)
+
+    def addSuccess(self, test, details = None):
         #Added so we can keep track of successes too
         self.successes.append((test, None))
-        super(OETestResult, self).addSuccess(test)
+        self.extractExtraResults(test, details = details)
+        return super(OETestResult, self).addSuccess(test)
+
+    def addExpectedFailure(self, test, *args, details = None):
+        self.extractExtraResults(test, details = details)
+        return super(OETestResult, self).addExpectedFailure(test, *args)
+
+    def addUnexpectedSuccess(self, test, details = None):
+        self.extractExtraResults(test, details = details)
+        return super(OETestResult, self).addUnexpectedSuccess(test)
 
     def logDetails(self, json_file_dir=None, configuration=None, result_id=None,
             dump_streams=False):
         self.tc.logger.info("RESULTS:")
 
-        result = {}
+        result = self.extraresults
         logs = {}
         if hasattr(self.tc, "extraresults"):
-            result = self.tc.extraresults
+            result.update(self.tc.extraresults)
 
         for case_name in self.tc._registry['cases']:
             case = self.tc._registry['cases'][case_name]
diff --git a/meta/lib/oeqa/core/utils/concurrencytest.py b/meta/lib/oeqa/core/utils/concurrencytest.py
index 6bf7718863..fa6fa34b0e 100644
--- a/meta/lib/oeqa/core/utils/concurrencytest.py
+++ b/meta/lib/oeqa/core/utils/concurrencytest.py
@@ -21,6 +21,7 @@ import testtools
 import threading
 import time
 import io
+import json
 import subunit
 
 from queue import Queue
@@ -28,6 +29,8 @@ from itertools import cycle
 from subunit import ProtocolTestCase, TestProtocolClient
 from subunit.test_results import AutoTimingTestResultDecorator
 from testtools import ThreadsafeForwardingResult, iterate_tests
+from testtools.content import Content
+from testtools.content_type import ContentType
 from oeqa.utils.commands import get_test_layer
 
 import bb.utils
@@ -70,6 +73,58 @@ class BBThreadsafeForwardingResult(ThreadsafeForwardingResult):
             self.semaphore.release()
         super(BBThreadsafeForwardingResult, self)._add_result_with_semaphore(method, test, *args, **kwargs)
 
+class ProxyTestResult:
+    # a very basic TestResult proxy, in order to modify add* calls
+    def __init__(self, target):
+        self.result = target
+
+    def _addResult(self, method, test, *args, **kwargs):
+        return method(test, *args, **kwargs)
+
+    def addError(self, test, *args, **kwargs):
+        self._addResult(self.result.addError, test, *args, **kwargs)
+
+    def addFailure(self, test, *args, **kwargs):
+        self._addResult(self.result.addFailure, test, *args, **kwargs)
+
+    def addSuccess(self, test, *args, **kwargs):
+        self._addResult(self.result.addSuccess, test, *args, **kwargs)
+
+    def addExpectedFailure(self, test, *args, **kwargs):
+        self._addResult(self.result.addExpectedFailure, test, *args, **kwargs)
+
+    def addUnexpectedSuccess(self, test, *args, **kwargs):
+        self._addResult(self.result.addUnexpectedSuccess, test, *args, **kwargs)
+
+    def __getattr__(self, attr):
+        return getattr(self.result, attr)
+
+class ExtraResultsDecoderTestResult(ProxyTestResult):
+    def _addResult(self, method, test, *args, **kwargs):
+        if "details" in kwargs and "extraresults" in kwargs["details"]:
+            if isinstance(kwargs["details"]["extraresults"], Content):
+                kwargs = kwargs.copy()
+                kwargs["details"] = kwargs["details"].copy()
+                extraresults = kwargs["details"]["extraresults"]
+                data = bytearray()
+                for b in extraresults.iter_bytes():
+                    data += b
+                extraresults = json.loads(data.decode())
+                kwargs["details"]["extraresults"] = extraresults
+        return method(test, *args, **kwargs)
+
+class ExtraResultsEncoderTestResult(ProxyTestResult):
+    def _addResult(self, method, test, *args, **kwargs):
+        if hasattr(test, "extraresults"):
+            extras = lambda : [json.dumps(test.extraresults).encode()]
+            kwargs = kwargs.copy()
+            if "details" not in kwargs:
+                kwargs["details"] = {}
+            else:
+                kwargs["details"] = kwargs["details"].copy()
+            kwargs["details"]["extraresults"] = Content(ContentType("application", "json", {'charset': 'utf8'}), extras)
+        return method(test, *args, **kwargs)
+
 #
 # We have to patch subunit since it doesn't understand how to handle addError
 # outside of a running test case. This can happen if classSetUp() fails
@@ -116,7 +171,9 @@ class ConcurrentTestSuite(unittest.TestSuite):
             result.threadprogress = {}
             for i, (test, testnum) in enumerate(tests):
                 result.threadprogress[i] = []
-                process_result = BBThreadsafeForwardingResult(result, semaphore, i, testnum, totaltests)
+                process_result = BBThreadsafeForwardingResult(
+                        ExtraResultsDecoderTestResult(result),
+                        semaphore, i, testnum, totaltests)
                 # Force buffering of stdout/stderr so the console doesn't get corrupted by test output
                 # as per default in parent code
                 process_result.buffer = True
@@ -231,7 +288,7 @@ def fork_for_tests(concurrency_num, suite):
                 # as per default in parent code
                 subunit_client.buffer = True
                 subunit_result = AutoTimingTestResultDecorator(subunit_client)
-                process_suite.run(subunit_result)
+                process_suite.run(ExtraResultsEncoderTestResult(subunit_result))
                 if ourpid != os.getpid():
                     os._exit(0)
                 if newbuilddir: