new file mode 100644
@@ -0,0 +1,24 @@
+#
+# Copyright OpenEmbedded Contributors
+#
+# SPDX-License-Identifier: MIT
+#
+
+import sys
+
+from oeqa.core.case import OETestCase
+
+class StreamsTest(OETestCase):
+
+ def testStreams1(self):
+ print("Output from testStreams1")
+ sys.stderr.write("Error output from testStreams1\n")
+
+ def testStreams2(self):
+ print("Output from testStreams2")
+ sys.stderr.write("Error output from testStreams2\n")
+ self.fail("Failed on purpose")
+
+ def testStreams3(self):
+ print("Output from testStreams3")
+ self.fail("Failed on purpose")
@@ -19,7 +19,7 @@ except ImportError:
@unittest.skipIf(ConcurrentTestSuite is None, "concurrencytest needs testtools, subunit and bitbake")
class TestConcurrency(TestBase):
- def _statuses(self, modules, processes=None):
+ def _run(self, modules, processes=None):
tc = self._testLoader(modules=modules)
if processes:
setup = lambda suffix, selftestdir, suite: (None, None)
@@ -27,6 +27,10 @@ class TestConcurrency(TestBase):
# fork_for_tests() only looks up the selftest layer with BUILDDIR set
with mock.patch.dict(os.environ, {'BUILDDIR': '/nonexistent'}):
result = tc.runTests(processes=processes)
+ return tc, result
+
+ def _statuses(self, modules, processes=None):
+ tc, result = self._run(modules, processes)
return {case_id: result._getTestResultDetails(case)[0]
for case_id, case in tc._registry['cases'].items()}
@@ -38,5 +42,16 @@ class TestConcurrency(TestBase):
self.assertNotIn('UNKNOWN', sequential.values())
self.assertEqual(self._statuses(modules, processes=1), sequential)
+ def test_concurrent_output(self):
+ # Each test's results must only have the output from that test, not
+ # everything its process has output so far
+ tc, result = self._run(['streams'], processes=1)
+ for n in (1, 2, 3):
+ stdout, stderr = result.logged_output['streams.StreamsTest.testStreams%d' % n]
+ output = stdout + stderr
+ self.assertIn("Output from testStreams%d" % n, output)
+ for other in {1, 2, 3} - {n}:
+ self.assertNotIn("testStreams%d" % other, output)
+
if __name__ == '__main__':
unittest.main()
@@ -57,6 +57,9 @@ class BBThreadsafeForwardingResult(ThreadsafeForwardingResult):
self.totaltests = totaltests
self.buffer = True
self.outputbuf = output
+ # The process' output accumulates in outputbuf; this is how much of
+ # it was given to earlier results
+ self.outputoffset = 0
self.finalresult = finalresult
self.finalresult.buffer = True
self.target = target
@@ -79,8 +82,13 @@ class BBThreadsafeForwardingResult(ThreadsafeForwardingResult):
test.id())
finally:
self.semaphore.release()
- self.finalresult._stderr_buffer = io.StringIO(initial_value=self.outputbuf.getvalue().decode("utf-8"))
+ # The process runs its tests one at a time, so the output since the
+ # previous result is this test's. stdout and stderr share one pipe so
+ # it is all given as stderr.
+ output = self.outputbuf.getvalue()
+ self.finalresult._stderr_buffer = io.StringIO(initial_value=output[self.outputoffset:].decode("utf-8"))
self.finalresult._stdout_buffer = io.StringIO()
+ self.outputoffset = len(output)
super()._add_result_with_semaphore(method, test, *args, **kwargs)
class ProxyTestResult:
When tests run in separate processes, each process' output is collected in a buffer as it arrives. Each result was given the whole buffer as its output, so a failing test showed the output of every earlier test run by the same process, as did the stdout/stderr saved in the results. Each process runs its tests one at a time, so only give a result the output received since the previous one, and add a test. stdout and stderr still share a pipe and are reported together as stderr. AI-Generated: Uses Claude Opus 5.5 Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org> --- meta/lib/oeqa/core/tests/cases/streams.py | 24 ++++++++++++++++++++ meta/lib/oeqa/core/tests/test_concurrency.py | 17 +++++++++++++- meta/lib/oeqa/core/utils/concurrencytest.py | 10 +++++++- 3 files changed, 49 insertions(+), 2 deletions(-) create mode 100644 meta/lib/oeqa/core/tests/cases/streams.py