# Copyright (C) 2016 Intel Corporation # Released under the MIT license (see COPYING.MIT) import os from oeqa.core.context import OETestContext, OETestContextExecutor from oeqa.core.target.ssh import OESSHTarget from oeqa.core.target.qemu import OEQemuTarget from oeqa.utils.dump import HostDumper from oeqa.runtime.loader import OERuntimeTestLoader class OERuntimeTestContext(OETestContext): loaderClass = OERuntimeTestLoader runtime_files_dir = os.path.join( os.path.dirname(os.path.abspath(__file__)), "files") def __init__(self, td, logger, target, host_dumper, image_packages, extract_dir): super(OERuntimeTestContext, self).__init__(td, logger) self.target = target self.image_packages = image_packages self.host_dumper = host_dumper self.extract_dir = extract_dir self._set_target_cmds() def _set_target_cmds(self): self.target_cmds = {} self.target_cmds['ps'] = 'ps' if 'procps' in self.image_packages: self.target_cmds['ps'] = self.target_cmds['ps'] + ' -ef' class OERuntimeTestContextExecutor(OETestContextExecutor): _context_class = OERuntimeTestContext name = 'runtime' help = 'runtime test component' description = 'executes runtime tests over targets' default_cases = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'cases') default_data = None default_test_data = 'data/testdata.json' default_tests = '' default_target_type = 'simpleremote' default_manifest = 'data/manifest' default_server_ip = '192.168.7.1' default_target_ip = '192.168.7.2' default_host_dumper_dir = '/tmp/oe-saved-tests' default_extract_dir = 'packages/extracted' def register_commands(self, logger, subparsers): super(OERuntimeTestContextExecutor, self).register_commands(logger, subparsers) runtime_group = self.parser.add_argument_group('runtime options') runtime_group.add_argument('--target-type', action='store', default=self.default_target_type, choices=['simpleremote', 'qemu'], help="Target type of device under test, default: %s" \ % self.default_target_type) runtime_group.add_argument('--target-ip', action='store', default=self.default_target_ip, help="IP address of device under test, default: %s" \ % self.default_target_ip) runtime_group.add_argument('--server-ip', action='store', default=self.default_target_ip, help="IP address of device under test, default: %s" \ % self.default_server_ip) runtime_group.add_argument('--host-dumper-dir', action='store', default=self.default_host_dumper_dir, help="Directory where host status is dumped, if tests fails, default: %s" \ % self.default_host_dumper_dir) runtime_group.add_argument('--packages-manifest', action='store', default=self.default_manifest, help="Package manifest of the image under testi, default: %s" \ % self.default_manifest) runtime_group.add_argument('--extract-dir', action='store', default=self.default_extract_dir, help='Directory where extracted packages reside, default: %s' \ % self.default_extract_dir) runtime_group.add_argument('--qemu-boot', action='store', help="Qemu boot configuration, only needed when target_type is QEMU.") @staticmethod def getTarget(target_type, target_modules_path, logger, target_ip, server_ip, **kwargs): target = None if target_type == 'simpleremote': target = OESSHTarget(logger, target_ip, server_ip, **kwargs) elif target_type == 'qemu': target = OEQemuTarget(logger, target_ip, server_ip, **kwargs) else: # Custom target module loading try: controller = OERuntimeTestContextExecutor.getControllerModule(target_type, target_modules_path) target = controller(logger, target_ip, server_ip, **kwargs) except ImportError as e: raise TypeError("Failed to import %s from available controller modules" % target_type) return target # Search oeqa.controllers module directory for and return a controller # corresponding to the given target name. # AttributeError raised if not found. # ImportError raised if a provided module can not be imported. @staticmethod def getControllerModule(target, target_modules_path): controllerslist = OERuntimeTestContextExecutor._getControllerModulenames(target_modules_path) controller = OERuntimeTestContextExecutor._loadControllerFromName(target, controllerslist) return controller # Return a list of all python modules in lib/oeqa/controllers for each # layer in bbpath @staticmethod def _getControllerModulenames(target_modules_path): controllerslist = [] def add_controller_list(path): if not os.path.exists(os.path.join(path, '__init__.py')): raise OSError('Controllers directory %s exists but is missing __init__.py' % path) files = sorted([f for f in os.listdir(path) if f.endswith('.py') and not f.startswith('_')]) for f in files: module = 'oeqa.controllers.' + f[:-3] if module not in controllerslist: controllerslist.append(module) else: raise RuntimeError("Duplicate controller module found for %s. Layers should create unique controller module names" % module) extpath = target_modules_path.split(':') for p in extpath: controllerpath = os.path.join(p, 'lib', 'oeqa', 'controllers') if os.path.exists(controllerpath): add_controller_list(controllerpath) return controllerslist # Search for and return a controller from given target name and # set of module names. # Raise AttributeError if not found. # Raise ImportError if a provided module can not be imported @staticmethod def _loadControllerFromName(target, modulenames): for name in modulenames: obj = OERuntimeTestContextExecutor._loadControllerFromModule(target, name) if obj: return obj raise AttributeError("Unable to load {0} from available modules: {1}".format(target, str(modulenames))) # Search for and return a controller or None from given module name @staticmethod def _loadControllerFromModule(target, modulename): obj = None # import module, allowing it to raise import exception try: module = __import__(modulename, globals(), locals(), [target]) except Exception as e: return obj # look for target class in the module, catching any exceptions as it # is valid that a module may not have the target class. try: obj = getattr(module, target) except: obj = None return obj @staticmethod def readPackagesManifest(manifest): if not manifest or not os.path.exists(manifest): raise OSError("Manifest file not exists: %s" % manifest) image_packages = set() with open(manifest, 'r') as f: for line in f.readlines(): line = line.strip() if line and not line.startswith("#"): image_packages.add(line.split()[0]) return image_packages @staticmethod def getHostDumper(cmds, directory): return HostDumper(cmds, directory) def _process_args(self, logger, args): if not args.packages_manifest: raise TypeError('Manifest file not provided') super(OERuntimeTestContextExecutor, self)._process_args(logger, args) target_kwargs = {} target_kwargs['qemuboot'] = args.qemu_boot self.tc_kwargs['init']['target'] = \ OERuntimeTestContextExecutor.getTarget(args.target_type, None, args.target_ip, args.server_ip, **target_kwargs) self.tc_kwargs['init']['host_dumper'] = \ OERuntimeTestContextExecutor.getHostDumper(None, args.host_dumper_dir) self.tc_kwargs['init']['image_packages'] = \ OERuntimeTestContextExecutor.readPackagesManifest( args.packages_manifest) self.tc_kwargs['init']['extract_dir'] = args.extract_dir _executor_class = OERuntimeTestContextExecutor