blob: 95d3bf72fc0ee32436d0ca4a4f20140057db462d [file] [log] [blame]
# Copyright (C) 2013 Intel Corporation
#
# Released under the MIT license (see COPYING.MIT)
# Main unittest module used by testimage.bbclass
# This provides the oeRuntimeTest base class which is inherited by all tests in meta/lib/oeqa/runtime.
# It also has some helper functions and it's responsible for actually starting the tests
import os, re, mmap, sys
import unittest
import inspect
import subprocess
import signal
import shutil
import functools
try:
import bb
except ImportError:
pass
import logging
import oeqa.runtime
# Exported test doesn't require sdkext
try:
import oeqa.sdkext
except ImportError:
pass
from oeqa.utils.decorators import LogResults, gettag, getResults
from oeqa.utils import avoid_paths_in_environ
logger = logging.getLogger("BitBake")
def getVar(obj):
#extend form dict, if a variable didn't exists, need find it in testcase
class VarDict(dict):
def __getitem__(self, key):
return gettag(obj, key)
return VarDict()
def checkTags(tc, tagexp):
return eval(tagexp, None, getVar(tc))
def filterByTagExp(testsuite, tagexp):
if not tagexp:
return testsuite
caseList = []
for each in testsuite:
if not isinstance(each, unittest.BaseTestSuite):
if checkTags(each, tagexp):
caseList.append(each)
else:
caseList.append(filterByTagExp(each, tagexp))
return testsuite.__class__(caseList)
@LogResults
class oeTest(unittest.TestCase):
pscmd = "ps"
longMessage = True
@classmethod
def hasPackage(self, pkg):
"""
True if the full package name exists in the manifest, False otherwise.
"""
return pkg in oeTest.tc.pkgmanifest
@classmethod
def hasPackageMatch(self, match):
"""
True if match exists in the manifest as a regular expression substring,
False otherwise.
"""
for s in oeTest.tc.pkgmanifest:
if re.match(match, s):
return True
return False
@classmethod
def hasFeature(self,feature):
if feature in oeTest.tc.imagefeatures or \
feature in oeTest.tc.distrofeatures:
return True
else:
return False
class oeRuntimeTest(oeTest):
def __init__(self, methodName='runTest'):
self.target = oeRuntimeTest.tc.target
super(oeRuntimeTest, self).__init__(methodName)
def setUp(self):
# Install packages in the DUT
self.tc.install_uninstall_packages(self.id())
# Check if test needs to run
if self.tc.sigterm:
self.fail("Got SIGTERM")
elif (type(self.target).__name__ == "QemuTarget"):
self.assertTrue(self.target.check(), msg = "Qemu not running?")
self.setUpLocal()
# a setup method before tests but after the class instantiation
def setUpLocal(self):
pass
def tearDown(self):
# Unistall packages in the DUT
self.tc.install_uninstall_packages(self.id(), False)
res = getResults()
# If a test fails or there is an exception dump
# for QemuTarget only
if (type(self.target).__name__ == "QemuTarget" and
(self.id() in res.getErrorList() or
self.id() in res.getFailList())):
self.tc.host_dumper.create_dir(self._testMethodName)
self.tc.host_dumper.dump_host()
self.target.target_dumper.dump_target(
self.tc.host_dumper.dump_dir)
print ("%s dump data stored in %s" % (self._testMethodName,
self.tc.host_dumper.dump_dir))
self.tearDownLocal()
# Method to be run after tearDown and implemented by child classes
def tearDownLocal(self):
pass
#TODO: use package_manager.py to install packages on any type of image
def install_packages(self, packagelist):
for package in packagelist:
(status, result) = self.target.run("smart install -y "+package)
if status != 0:
return status
class OETestCalledProcessError(subprocess.CalledProcessError):
def __str__(self):
if hasattr(self, "stderr"):
return "Command '%s' returned non-zero exit status %d with output %s and stderr %s" % (self.cmd, self.returncode, self.output, self.stderr)
else:
return "Command '%s' returned non-zero exit status %d with output %s" % (self.cmd, self.returncode, self.output)
subprocess.CalledProcessError = OETestCalledProcessError
class oeSDKTest(oeTest):
def __init__(self, methodName='runTest'):
self.sdktestdir = oeSDKTest.tc.sdktestdir
super(oeSDKTest, self).__init__(methodName)
@classmethod
def hasHostPackage(self, pkg):
if re.search(pkg, oeTest.tc.hostpkgmanifest):
return True
return False
def _run(self, cmd):
return subprocess.check_output(". %s > /dev/null; %s;" % (self.tc.sdkenv, cmd), shell=True, stderr=subprocess.STDOUT).decode("utf-8")
class oeSDKExtTest(oeSDKTest):
def _run(self, cmd):
# extensible sdk shows a warning if found bitbake in the path
# because can cause contamination, i.e. use devtool from
# poky/scripts instead of eSDK one.
env = os.environ.copy()
paths_to_avoid = ['bitbake/bin', 'poky/scripts']
env['PATH'] = avoid_paths_in_environ(paths_to_avoid)
return subprocess.check_output(". %s > /dev/null;"\
" %s;" % (self.tc.sdkenv, cmd), stderr=subprocess.STDOUT, shell=True, env=env).decode("utf-8")
def getmodule(pos=2):
# stack returns a list of tuples containg frame information
# First element of the list the is current frame, caller is 1
frameinfo = inspect.stack()[pos]
modname = inspect.getmodulename(frameinfo[1])
#modname = inspect.getmodule(frameinfo[0]).__name__
return modname
def skipModule(reason, pos=2):
modname = getmodule(pos)
if modname not in oeTest.tc.testsrequired:
raise unittest.SkipTest("%s: %s" % (modname, reason))
else:
raise Exception("\nTest %s wants to be skipped.\nReason is: %s" \
"\nTest was required in TEST_SUITES, so either the condition for skipping is wrong" \
"\nor the image really doesn't have the required feature/package when it should." % (modname, reason))
def skipModuleIf(cond, reason):
if cond:
skipModule(reason, 3)
def skipModuleUnless(cond, reason):
if not cond:
skipModule(reason, 3)
_buffer_logger = ""
def custom_verbose(msg, *args, **kwargs):
global _buffer_logger
if msg[-1] != "\n":
_buffer_logger += msg
else:
_buffer_logger += msg
try:
bb.plain(_buffer_logger.rstrip("\n"), *args, **kwargs)
except NameError:
logger.info(_buffer_logger.rstrip("\n"), *args, **kwargs)
_buffer_logger = ""
class TestContext(object):
def __init__(self, d, exported=False):
self.d = d
self.testsuites = self._get_test_suites()
if exported:
path = [os.path.dirname(os.path.abspath(__file__))]
extrapath = ""
else:
path = d.getVar("BBPATH", True).split(':')
extrapath = "lib/oeqa"
self.testslist = self._get_tests_list(path, extrapath)
self.testsrequired = self._get_test_suites_required()
self.filesdir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "runtime/files")
self.imagefeatures = d.getVar("IMAGE_FEATURES", True).split()
self.distrofeatures = d.getVar("DISTRO_FEATURES", True).split()
# get testcase list from specified file
# if path is a relative path, then relative to build/conf/
def _read_testlist(self, fpath, builddir):
if not os.path.isabs(fpath):
fpath = os.path.join(builddir, "conf", fpath)
if not os.path.exists(fpath):
bb.fatal("No such manifest file: ", fpath)
tcs = []
for line in open(fpath).readlines():
line = line.strip()
if line and not line.startswith("#"):
tcs.append(line)
return " ".join(tcs)
# return test list by type also filter if TEST_SUITES is specified
def _get_tests_list(self, bbpath, extrapath):
testslist = []
type = self._get_test_namespace()
# This relies on lib/ under each directory in BBPATH being added to sys.path
# (as done by default in base.bbclass)
for testname in self.testsuites:
if testname != "auto":
if testname.startswith("oeqa."):
testslist.append(testname)
continue
found = False
for p in bbpath:
if os.path.exists(os.path.join(p, extrapath, type, testname + ".py")):
testslist.append("oeqa." + type + "." + testname)
found = True
break
elif os.path.exists(os.path.join(p, extrapath, type, testname.split(".")[0] + ".py")):
testslist.append("oeqa." + type + "." + testname)
found = True
break
if not found:
bb.fatal('Test %s specified in TEST_SUITES could not be found in lib/oeqa/runtime under BBPATH' % testname)
if "auto" in self.testsuites:
def add_auto_list(path):
files = sorted([f for f in os.listdir(path) if f.endswith('.py') and not f.startswith('_')])
for f in files:
module = 'oeqa.' + type + '.' + f[:-3]
if module not in testslist:
testslist.append(module)
for p in bbpath:
testpath = os.path.join(p, 'lib', 'oeqa', type)
bb.debug(2, 'Searching for tests in %s' % testpath)
if os.path.exists(testpath):
add_auto_list(testpath)
return testslist
def getTestModules(self):
"""
Returns all the test modules in the testlist.
"""
import pkgutil
modules = []
for test in self.testslist:
if re.search("\w+\.\w+\.test_\S+", test):
test = '.'.join(t.split('.')[:3])
module = pkgutil.get_loader(test)
modules.append(module)
return modules
def getModulefromID(self, test_id):
"""
Returns the test module based on a test id.
"""
module_name = ".".join(test_id.split(".")[:3])
modules = self.getTestModules()
for module in modules:
if module.name == module_name:
return module
return None
def getTests(self, test):
'''Return all individual tests executed when running the suite.'''
# Unfortunately unittest does not have an API for this, so we have
# to rely on implementation details. This only needs to work
# for TestSuite containing TestCase.
method = getattr(test, '_testMethodName', None)
if method:
# leaf case: a TestCase
yield test
else:
# Look into TestSuite.
tests = getattr(test, '_tests', [])
for t1 in tests:
for t2 in self.getTests(t1):
yield t2
def loadTests(self):
setattr(oeTest, "tc", self)
testloader = unittest.TestLoader()
testloader.sortTestMethodsUsing = None
suites = [testloader.loadTestsFromName(name) for name in self.testslist]
suites = filterByTagExp(suites, getattr(self, "tagexp", None))
# Determine dependencies between suites by looking for @skipUnlessPassed
# method annotations. Suite A depends on suite B if any method in A
# depends on a method on B.
for suite in suites:
suite.dependencies = []
suite.depth = 0
for test in self.getTests(suite):
methodname = getattr(test, '_testMethodName', None)
if methodname:
method = getattr(test, methodname)
depends_on = getattr(method, '_depends_on', None)
if depends_on:
for dep_suite in suites:
if depends_on in [getattr(t, '_testMethodName', None) for t in self.getTests(dep_suite)]:
if dep_suite not in suite.dependencies and \
dep_suite is not suite:
suite.dependencies.append(dep_suite)
break
else:
logger.warning("Test %s was declared as @skipUnlessPassed('%s') but that test is either not defined or not active. Will run the test anyway." %
(test, depends_on))
# Use brute-force topological sort to determine ordering. Sort by
# depth (higher depth = must run later), with original ordering to
# break ties.
def set_suite_depth(suite):
for dep in suite.dependencies:
new_depth = set_suite_depth(dep) + 1
if new_depth > suite.depth:
suite.depth = new_depth
return suite.depth
for index, suite in enumerate(suites):
set_suite_depth(suite)
suite.index = index
def cmp(a, b):
return (a > b) - (a < b)
def cmpfunc(a, b):
return cmp((a.depth, a.index), (b.depth, b.index))
suites.sort(key=functools.cmp_to_key(cmpfunc))
self.suite = testloader.suiteClass(suites)
return self.suite
def runTests(self):
logger.info("Test modules %s" % self.testslist)
if hasattr(self, "tagexp") and self.tagexp:
logger.info("Filter test cases by tags: %s" % self.tagexp)
logger.info("Found %s tests" % self.suite.countTestCases())
runner = unittest.TextTestRunner(verbosity=2)
if 'bb' in sys.modules:
runner.stream.write = custom_verbose
return runner.run(self.suite)
class RuntimeTestContext(TestContext):
def __init__(self, d, target, exported=False):
super(RuntimeTestContext, self).__init__(d, exported)
self.target = target
self.pkgmanifest = {}
manifest = os.path.join(d.getVar("DEPLOY_DIR_IMAGE", True),
d.getVar("IMAGE_LINK_NAME", True) + ".manifest")
nomanifest = d.getVar("IMAGE_NO_MANIFEST", True)
if nomanifest is None or nomanifest != "1":
try:
with open(manifest) as f:
for line in f:
(pkg, arch, version) = line.strip().split()
self.pkgmanifest[pkg] = (version, arch)
except IOError as e:
bb.fatal("No package manifest file found. Did you build the image?\n%s" % e)
def _get_test_namespace(self):
return "runtime"
def _get_test_suites(self):
testsuites = []
manifests = (self.d.getVar("TEST_SUITES_MANIFEST", True) or '').split()
if manifests:
for manifest in manifests:
testsuites.extend(self._read_testlist(manifest,
self.d.getVar("TOPDIR", True)).split())
else:
testsuites = self.d.getVar("TEST_SUITES", True).split()
return testsuites
def _get_test_suites_required(self):
return [t for t in self.d.getVar("TEST_SUITES", True).split() if t != "auto"]
def loadTests(self):
super(RuntimeTestContext, self).loadTests()
if oeTest.hasPackage("procps"):
oeRuntimeTest.pscmd = "ps -ef"
def extract_packages(self):
"""
Find packages that will be needed during runtime.
"""
modules = self.getTestModules()
bbpaths = self.d.getVar("BBPATH", True).split(":")
shutil.rmtree(self.d.getVar("TEST_EXTRACTED_DIR", True))
shutil.rmtree(self.d.getVar("TEST_PACKAGED_DIR", True))
for module in modules:
json_file = self._getJsonFile(module)
if json_file:
needed_packages = self._getNeededPackages(json_file)
self._perform_package_extraction(needed_packages)
def _perform_package_extraction(self, needed_packages):
"""
Extract packages that will be needed during runtime.
"""
import oe.path
extracted_path = self.d.getVar("TEST_EXTRACTED_DIR", True)
packaged_path = self.d.getVar("TEST_PACKAGED_DIR", True)
for key,value in needed_packages.items():
packages = ()
if isinstance(value, dict):
packages = (value, )
elif isinstance(value, list):
packages = value
else:
bb.fatal("Failed to process needed packages for %s; "
"Value must be a dict or list" % key)
for package in packages:
pkg = package["pkg"]
rm = package.get("rm", False)
extract = package.get("extract", True)
if extract:
dst_dir = os.path.join(extracted_path, pkg)
else:
dst_dir = os.path.join(packaged_path)
# Extract package and copy it to TEST_EXTRACTED_DIR
pkg_dir = self._extract_in_tmpdir(pkg)
if extract:
# Same package used for more than one test,
# don't need to extract again.
if os.path.exists(dst_dir):
continue
oe.path.copytree(pkg_dir, dst_dir)
shutil.rmtree(pkg_dir)
# Copy package to TEST_PACKAGED_DIR
else:
self._copy_package(pkg)
def _getJsonFile(self, module):
"""
Returns the path of the JSON file for a module, empty if doesn't exitst.
"""
module_file = module.path
json_file = "%s.json" % module_file.rsplit(".", 1)[0]
if os.path.isfile(module_file) and os.path.isfile(json_file):
return json_file
else:
return ""
def _getNeededPackages(self, json_file, test=None):
"""
Returns a dict with needed packages based on a JSON file.
If a test is specified it will return the dict just for that test.
"""
import json
needed_packages = {}
with open(json_file) as f:
test_packages = json.load(f)
for key,value in test_packages.items():
needed_packages[key] = value
if test:
if test in needed_packages:
needed_packages = needed_packages[test]
else:
needed_packages = {}
return needed_packages
def _extract_in_tmpdir(self, pkg):
""""
Returns path to a temp directory where the package was
extracted without dependencies.
"""
from oeqa.utils.package_manager import get_package_manager
pkg_path = os.path.join(self.d.getVar("TEST_INSTALL_TMP_DIR", True), pkg)
pm = get_package_manager(self.d, pkg_path)
extract_dir = pm.extract(pkg)
shutil.rmtree(pkg_path)
return extract_dir
def _copy_package(self, pkg):
"""
Copy the RPM, DEB or IPK package to dst_dir
"""
from oeqa.utils.package_manager import get_package_manager
pkg_path = os.path.join(self.d.getVar("TEST_INSTALL_TMP_DIR", True), pkg)
dst_dir = self.d.getVar("TEST_PACKAGED_DIR", True)
pm = get_package_manager(self.d, pkg_path)
pkg_info = pm.package_info(pkg)
file_path = pkg_info[pkg]["filepath"]
shutil.copy2(file_path, dst_dir)
shutil.rmtree(pkg_path)
def install_uninstall_packages(self, test_id, pkg_dir, install):
"""
Check if the test requires a package and Install/Unistall it in the DUT
"""
test = test_id.split(".")[4]
module = self.getModulefromID(test_id)
json = self._getJsonFile(module)
if json:
needed_packages = self._getNeededPackages(json, test)
if needed_packages:
self._install_uninstall_packages(needed_packages, pkg_dir, install)
def _install_uninstall_packages(self, needed_packages, pkg_dir, install=True):
"""
Install/Unistall packages in the DUT without using a package manager
"""
if isinstance(needed_packages, dict):
packages = [needed_packages]
elif isinstance(needed_packages, list):
packages = needed_packages
for package in packages:
pkg = package["pkg"]
rm = package.get("rm", False)
extract = package.get("extract", True)
src_dir = os.path.join(pkg_dir, pkg)
# Install package
if install and extract:
self.target.connection.copy_dir_to(src_dir, "/")
# Unistall package
elif not install and rm:
self.target.connection.delete_dir_structure(src_dir, "/")
class ImageTestContext(RuntimeTestContext):
def __init__(self, d, target, host_dumper):
super(ImageTestContext, self).__init__(d, target)
self.tagexp = d.getVar("TEST_SUITES_TAGS", True)
self.host_dumper = host_dumper
self.sigterm = False
self.origsigtermhandler = signal.getsignal(signal.SIGTERM)
signal.signal(signal.SIGTERM, self._sigterm_exception)
def _sigterm_exception(self, signum, stackframe):
bb.warn("TestImage received SIGTERM, shutting down...")
self.sigterm = True
self.target.stop()
def install_uninstall_packages(self, test_id, install=True):
"""
Check if the test requires a package and Install/Unistall it in the DUT
"""
pkg_dir = self.d.getVar("TEST_EXTRACTED_DIR", True)
super(ImageTestContext, self).install_uninstall_packages(test_id, pkg_dir, install)
class ExportTestContext(RuntimeTestContext):
def __init__(self, d, target, exported=False, parsedArgs={}):
"""
This class is used when exporting tests and when are executed outside OE environment.
parsedArgs can contain the following:
- tag: Filter test by tag.
"""
super(ExportTestContext, self).__init__(d, target, exported)
tag = parsedArgs.get("tag", None)
self.tagexp = tag if tag != None else d.getVar("TEST_SUITES_TAGS", True)
self.sigterm = None
def install_uninstall_packages(self, test_id, install=True):
"""
Check if the test requires a package and Install/Unistall it in the DUT
"""
export_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
extracted_dir = self.d.getVar("TEST_EXPORT_EXTRACTED_DIR", True)
pkg_dir = os.path.join(export_dir, extracted_dir)
super(ExportTestContext, self).install_uninstall_packages(test_id, pkg_dir, install)
class SDKTestContext(TestContext):
def __init__(self, d, sdktestdir, sdkenv, tcname, *args):
super(SDKTestContext, self).__init__(d)
self.sdktestdir = sdktestdir
self.sdkenv = sdkenv
self.tcname = tcname
if not hasattr(self, 'target_manifest'):
self.target_manifest = d.getVar("SDK_TARGET_MANIFEST", True)
try:
self.pkgmanifest = {}
with open(self.target_manifest) as f:
for line in f:
(pkg, arch, version) = line.strip().split()
self.pkgmanifest[pkg] = (version, arch)
except IOError as e:
bb.fatal("No package manifest file found. Did you build the sdk image?\n%s" % e)
if not hasattr(self, 'host_manifest'):
self.host_manifest = d.getVar("SDK_HOST_MANIFEST", True)
try:
with open(self.host_manifest) as f:
self.hostpkgmanifest = f.read()
except IOError as e:
bb.fatal("No host package manifest file found. Did you build the sdk image?\n%s" % e)
def _get_test_namespace(self):
return "sdk"
def _get_test_suites(self):
return (self.d.getVar("TEST_SUITES_SDK", True) or "auto").split()
def _get_test_suites_required(self):
return [t for t in (self.d.getVar("TEST_SUITES_SDK", True) or \
"auto").split() if t != "auto"]
class SDKExtTestContext(SDKTestContext):
def __init__(self, d, sdktestdir, sdkenv, tcname, *args):
self.target_manifest = d.getVar("SDK_EXT_TARGET_MANIFEST", True)
self.host_manifest = d.getVar("SDK_EXT_HOST_MANIFEST", True)
if args:
self.cm = args[0] # Compatibility mode for run SDK tests
else:
self.cm = False
super(SDKExtTestContext, self).__init__(d, sdktestdir, sdkenv, tcname)
self.sdkextfilesdir = os.path.join(os.path.dirname(os.path.abspath(
oeqa.sdkext.__file__)), "files")
def _get_test_namespace(self):
if self.cm:
return "sdk"
else:
return "sdkext"
def _get_test_suites(self):
return (self.d.getVar("TEST_SUITES_SDK_EXT", True) or "auto").split()
def _get_test_suites_required(self):
return [t for t in (self.d.getVar("TEST_SUITES_SDK_EXT", True) or \
"auto").split() if t != "auto"]