mirror of
git://git.yoctoproject.org/poky
synced 2026-09-06 01:43:41 +00:00
e9f859106c
Autobuilder type infrastructure can benefit from deletion of certain files as background IO due to the way Linux filesystem priority works. We have problems where build directories as part of oe-selftest being delete starves the running tasks of IO to the point builds take much longer to compelte. Having this option of running the deletion at "idle" helps a lot with that. Use the new option added to bb.utils.prunedir(). (From OE-Core rev: d41e7018be56902b7a1be4590e468cd15e02a3b5) Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
341 lines
14 KiB
Python
341 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
#
|
|
# Modified for use in OE by Richard Purdie, 2018
|
|
#
|
|
# Modified by: Corey Goldberg, 2013
|
|
# License: GPLv2+
|
|
#
|
|
# Original code from:
|
|
# Bazaar (bzrlib.tests.__init__.py, v2.6, copied Jun 01 2013)
|
|
# Copyright (C) 2005-2011 Canonical Ltd
|
|
# License: GPLv2+
|
|
|
|
import os
|
|
import sys
|
|
import traceback
|
|
import unittest
|
|
import subprocess
|
|
import testtools
|
|
import threading
|
|
import time
|
|
import io
|
|
import json
|
|
import subunit
|
|
|
|
from queue import Queue
|
|
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
|
|
import oe.path
|
|
|
|
_all__ = [
|
|
'ConcurrentTestSuite',
|
|
'fork_for_tests',
|
|
'partition_tests',
|
|
]
|
|
|
|
#
|
|
# Patch the version from testtools to allow access to _test_start and allow
|
|
# computation of timing information and threading progress
|
|
#
|
|
class BBThreadsafeForwardingResult(ThreadsafeForwardingResult):
|
|
|
|
def __init__(self, target, semaphore, threadnum, totalinprocess, totaltests):
|
|
super(BBThreadsafeForwardingResult, self).__init__(target, semaphore)
|
|
self.threadnum = threadnum
|
|
self.totalinprocess = totalinprocess
|
|
self.totaltests = totaltests
|
|
|
|
def _add_result_with_semaphore(self, method, test, *args, **kwargs):
|
|
self.semaphore.acquire()
|
|
try:
|
|
if self._test_start:
|
|
self.result.starttime[test.id()] = self._test_start.timestamp()
|
|
self.result.threadprogress[self.threadnum].append(test.id())
|
|
totalprogress = sum(len(x) for x in self.result.threadprogress.values())
|
|
self.result.progressinfo[test.id()] = "%s: %s/%s %s/%s (%ss) (%s)" % (
|
|
self.threadnum,
|
|
len(self.result.threadprogress[self.threadnum]),
|
|
self.totalinprocess,
|
|
totalprogress,
|
|
self.totaltests,
|
|
"{0:.2f}".format(time.time()-self._test_start.timestamp()),
|
|
test.id())
|
|
finally:
|
|
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
|
|
# for a class of tests. This unfortunately has horrible internal knowledge.
|
|
#
|
|
def outSideTestaddError(self, offset, line):
|
|
"""An 'error:' directive has been read."""
|
|
test_name = line[offset:-1].decode('utf8')
|
|
self.parser._current_test = subunit.RemotedTestCase(test_name)
|
|
self.parser.current_test_description = test_name
|
|
self.parser._state = self.parser._reading_error_details
|
|
self.parser._reading_error_details.set_simple()
|
|
self.parser.subunitLineReceived(line)
|
|
|
|
subunit._OutSideTest.addError = outSideTestaddError
|
|
|
|
|
|
#
|
|
# A dummy structure to add to io.StringIO so that the .buffer object
|
|
# is available and accepts writes. This allows unittest with buffer=True
|
|
# to interact ok with subunit which wants to access sys.stdout.buffer.
|
|
#
|
|
class dummybuf(object):
|
|
def __init__(self, parent):
|
|
self.p = parent
|
|
def write(self, data):
|
|
self.p.write(data.decode("utf-8"))
|
|
|
|
#
|
|
# Taken from testtools.ConncurrencyTestSuite but modified for OE use
|
|
#
|
|
class ConcurrentTestSuite(unittest.TestSuite):
|
|
|
|
def __init__(self, suite, processes):
|
|
super(ConcurrentTestSuite, self).__init__([suite])
|
|
self.processes = processes
|
|
|
|
def run(self, result):
|
|
tests, totaltests = fork_for_tests(self.processes, self)
|
|
try:
|
|
threads = {}
|
|
queue = Queue()
|
|
semaphore = threading.Semaphore(1)
|
|
result.threadprogress = {}
|
|
for i, (test, testnum) in enumerate(tests):
|
|
result.threadprogress[i] = []
|
|
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
|
|
# We have to add a buffer object to stdout to keep subunit happy
|
|
process_result._stderr_buffer = io.StringIO()
|
|
process_result._stderr_buffer.buffer = dummybuf(process_result._stderr_buffer)
|
|
process_result._stdout_buffer = io.StringIO()
|
|
process_result._stdout_buffer.buffer = dummybuf(process_result._stdout_buffer)
|
|
reader_thread = threading.Thread(
|
|
target=self._run_test, args=(test, process_result, queue))
|
|
threads[test] = reader_thread, process_result
|
|
reader_thread.start()
|
|
while threads:
|
|
finished_test = queue.get()
|
|
threads[finished_test][0].join()
|
|
del threads[finished_test]
|
|
except:
|
|
for thread, process_result in threads.values():
|
|
process_result.stop()
|
|
raise
|
|
finally:
|
|
for test in tests:
|
|
test[0]._stream.close()
|
|
|
|
def _run_test(self, test, process_result, queue):
|
|
try:
|
|
try:
|
|
test.run(process_result)
|
|
except Exception:
|
|
# The run logic itself failed
|
|
case = testtools.ErrorHolder(
|
|
"broken-runner",
|
|
error=sys.exc_info())
|
|
case.run(process_result)
|
|
finally:
|
|
queue.put(test)
|
|
|
|
def removebuilddir(d):
|
|
delay = 5
|
|
while delay and os.path.exists(d + "/bitbake.lock"):
|
|
time.sleep(1)
|
|
delay = delay - 1
|
|
bb.utils.prunedir(d, ionice=True)
|
|
|
|
def fork_for_tests(concurrency_num, suite):
|
|
result = []
|
|
if 'BUILDDIR' in os.environ:
|
|
selftestdir = get_test_layer()
|
|
|
|
test_blocks = partition_tests(suite, concurrency_num)
|
|
# Clear the tests from the original suite so it doesn't keep them alive
|
|
suite._tests[:] = []
|
|
totaltests = sum(len(x) for x in test_blocks)
|
|
for process_tests in test_blocks:
|
|
numtests = len(process_tests)
|
|
process_suite = unittest.TestSuite(process_tests)
|
|
# Also clear each split list so new suite has only reference
|
|
process_tests[:] = []
|
|
c2pread, c2pwrite = os.pipe()
|
|
# Clear buffers before fork to avoid duplicate output
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
pid = os.fork()
|
|
if pid == 0:
|
|
ourpid = os.getpid()
|
|
try:
|
|
newbuilddir = None
|
|
stream = os.fdopen(c2pwrite, 'wb', 1)
|
|
os.close(c2pread)
|
|
|
|
# Create a new separate BUILDDIR for each group of tests
|
|
if 'BUILDDIR' in os.environ:
|
|
builddir = os.environ['BUILDDIR']
|
|
newbuilddir = builddir + "-st-" + str(ourpid)
|
|
newselftestdir = newbuilddir + "/meta-selftest"
|
|
|
|
bb.utils.mkdirhier(newbuilddir)
|
|
oe.path.copytree(builddir + "/conf", newbuilddir + "/conf")
|
|
oe.path.copytree(builddir + "/cache", newbuilddir + "/cache")
|
|
oe.path.copytree(selftestdir, newselftestdir)
|
|
|
|
for e in os.environ:
|
|
if builddir in os.environ[e]:
|
|
os.environ[e] = os.environ[e].replace(builddir, newbuilddir)
|
|
|
|
subprocess.check_output("git init; git add *; git commit -a -m 'initial'", cwd=newselftestdir, shell=True)
|
|
|
|
# Tried to used bitbake-layers add/remove but it requires recipe parsing and hence is too slow
|
|
subprocess.check_output("sed %s/conf/bblayers.conf -i -e 's#%s#%s#g'" % (newbuilddir, selftestdir, newselftestdir), cwd=newbuilddir, shell=True)
|
|
|
|
os.chdir(newbuilddir)
|
|
|
|
for t in process_suite:
|
|
if not hasattr(t, "tc"):
|
|
continue
|
|
cp = t.tc.config_paths
|
|
for p in cp:
|
|
if selftestdir in cp[p] and newselftestdir not in cp[p]:
|
|
cp[p] = cp[p].replace(selftestdir, newselftestdir)
|
|
if builddir in cp[p] and newbuilddir not in cp[p]:
|
|
cp[p] = cp[p].replace(builddir, newbuilddir)
|
|
|
|
# Leave stderr and stdout open so we can see test noise
|
|
# Close stdin so that the child goes away if it decides to
|
|
# read from stdin (otherwise its a roulette to see what
|
|
# child actually gets keystrokes for pdb etc).
|
|
newsi = os.open(os.devnull, os.O_RDWR)
|
|
os.dup2(newsi, sys.stdin.fileno())
|
|
|
|
subunit_client = TestProtocolClient(stream)
|
|
# Force buffering of stdout/stderr so the console doesn't get corrupted by test output
|
|
# as per default in parent code
|
|
subunit_client.buffer = True
|
|
subunit_result = AutoTimingTestResultDecorator(subunit_client)
|
|
process_suite.run(ExtraResultsEncoderTestResult(subunit_result))
|
|
if ourpid != os.getpid():
|
|
os._exit(0)
|
|
if newbuilddir:
|
|
removebuilddir(newbuilddir)
|
|
except:
|
|
# Don't do anything with process children
|
|
if ourpid != os.getpid():
|
|
os._exit(1)
|
|
# Try and report traceback on stream, but exit with error
|
|
# even if stream couldn't be created or something else
|
|
# goes wrong. The traceback is formatted to a string and
|
|
# written in one go to avoid interleaving lines from
|
|
# multiple failing children.
|
|
try:
|
|
stream.write(traceback.format_exc().encode('utf-8'))
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
finally:
|
|
if newbuilddir:
|
|
removebuilddir(newbuilddir)
|
|
stream.flush()
|
|
os._exit(1)
|
|
stream.flush()
|
|
os._exit(0)
|
|
else:
|
|
os.close(c2pwrite)
|
|
stream = os.fdopen(c2pread, 'rb', 1)
|
|
test = ProtocolTestCase(stream)
|
|
result.append((test, numtests))
|
|
return result, totaltests
|
|
|
|
def partition_tests(suite, count):
|
|
# Keep tests from the same class together but allow tests from modules
|
|
# to go to different processes to aid parallelisation.
|
|
modules = {}
|
|
for test in iterate_tests(suite):
|
|
m = test.__module__ + "." + test.__class__.__name__
|
|
if m not in modules:
|
|
modules[m] = []
|
|
modules[m].append(test)
|
|
|
|
# Simply divide the test blocks between the available processes
|
|
partitions = [list() for _ in range(count)]
|
|
for partition, m in zip(cycle(partitions), modules):
|
|
partition.extend(modules[m])
|
|
|
|
# No point in empty threads so drop them
|
|
return [p for p in partitions if p]
|
|
|