diff options
Diffstat (limited to 'import-layers/yocto-poky/meta/lib/oeqa/runtime/context.py')
| -rw-r--r-- | import-layers/yocto-poky/meta/lib/oeqa/runtime/context.py | 226 |
1 files changed, 0 insertions, 226 deletions
diff --git a/import-layers/yocto-poky/meta/lib/oeqa/runtime/context.py b/import-layers/yocto-poky/meta/lib/oeqa/runtime/context.py deleted file mode 100644 index 0294003fc..000000000 --- a/import-layers/yocto-poky/meta/lib/oeqa/runtime/context.py +++ /dev/null @@ -1,226 +0,0 @@ -# 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, logger, target_ip, server_ip, **kwargs): - target = None - - if target_ip: - target_ip_port = target_ip.split(':') - if len(target_ip_port) == 2: - target_ip = target_ip_port[0] - kwargs['port'] = target_ip_port[1] - - 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: - # XXX: This code uses the old naming convention for controllers and - # targets, the idea it is to leave just targets as the controller - # most of the time was just a wrapper. - # XXX: This code tries to import modules from lib/oeqa/controllers - # directory and treat them as controllers, it will less error prone - # to use introspection to load such modules. - # XXX: Don't base your targets on this code it will be refactored - # in the near future. - # Custom target module loading - try: - target_modules_path = kwargs.get('target_modules_path', '') - 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 |