1# 2# Copyright (C) 2016 Intel Corporation 3# 4# SPDX-License-Identifier: MIT 5# 6 7import os 8import sys 9 10from oeqa.core.context import OETestContext, OETestContextExecutor 11from oeqa.core.target.serial import OESerialTarget 12from oeqa.core.target.ssh import OESSHTarget 13from oeqa.core.target.qemu import OEQemuTarget 14 15from oeqa.runtime.loader import OERuntimeTestLoader 16 17class OERuntimeTestContext(OETestContext): 18 loaderClass = OERuntimeTestLoader 19 runtime_files_dir = os.path.join( 20 os.path.dirname(os.path.abspath(__file__)), "files") 21 22 def __init__(self, td, logger, target, 23 image_packages, extract_dir): 24 super(OERuntimeTestContext, self).__init__(td, logger) 25 26 self.target = target 27 self.image_packages = image_packages 28 self.extract_dir = extract_dir 29 self._set_target_cmds() 30 31 def _set_target_cmds(self): 32 self.target_cmds = {} 33 34 self.target_cmds['ps'] = 'ps' 35 if 'procps' in self.image_packages: 36 self.target_cmds['ps'] = self.target_cmds['ps'] + ' -ef' 37 38class OERuntimeTestContextExecutor(OETestContextExecutor): 39 _context_class = OERuntimeTestContext 40 41 name = 'runtime' 42 help = 'runtime test component' 43 description = 'executes runtime tests over targets' 44 45 default_cases = os.path.join(os.path.abspath(os.path.dirname(__file__)), 46 'cases') 47 default_data = None 48 default_test_data = 'data/testdata.json' 49 default_tests = '' 50 default_json_result_dir = '%s-results' % name 51 52 default_target_type = 'simpleremote' 53 default_manifest = 'data/manifest' 54 default_server_ip = '192.168.7.1' 55 default_target_ip = '192.168.7.2' 56 default_extract_dir = 'packages/extracted' 57 58 def register_commands(self, logger, subparsers): 59 super(OERuntimeTestContextExecutor, self).register_commands(logger, subparsers) 60 61 runtime_group = self.parser.add_argument_group('runtime options') 62 63 runtime_group.add_argument('--target-type', action='store', 64 default=self.default_target_type, choices=['simpleremote', 'qemu', 'serial'], 65 help="Target type of device under test, default: %s" \ 66 % self.default_target_type) 67 runtime_group.add_argument('--target-ip', action='store', 68 default=self.default_target_ip, 69 help="IP address and optionally ssh port (default 22) of device under test, for example '192.168.0.7:22'. Default: %s" \ 70 % self.default_target_ip) 71 runtime_group.add_argument('--server-ip', action='store', 72 default=self.default_target_ip, 73 help="IP address of the test host from test target machine, default: %s" \ 74 % self.default_server_ip) 75 76 runtime_group.add_argument('--host-dumper-dir', action='store', 77 help="Directory where host status is dumped, if tests fails") 78 79 runtime_group.add_argument('--packages-manifest', action='store', 80 default=self.default_manifest, 81 help="Package manifest of the image under test, default: %s" \ 82 % self.default_manifest) 83 84 runtime_group.add_argument('--extract-dir', action='store', 85 default=self.default_extract_dir, 86 help='Directory where extracted packages reside, default: %s' \ 87 % self.default_extract_dir) 88 89 runtime_group.add_argument('--qemu-boot', action='store', 90 help="Qemu boot configuration, only needed when target_type is QEMU.") 91 92 @staticmethod 93 def getTarget(target_type, logger, target_ip, server_ip, **kwargs): 94 target = None 95 96 if target_ip: 97 target_ip_port = target_ip.split(':') 98 if len(target_ip_port) == 2: 99 target_ip = target_ip_port[0] 100 kwargs['port'] = target_ip_port[1] 101 102 if server_ip: 103 server_ip_port = server_ip.split(':') 104 if len(server_ip_port) == 2: 105 server_ip = server_ip_port[0] 106 kwargs['server_port'] = int(server_ip_port[1]) 107 108 if target_type == 'simpleremote': 109 target = OESSHTarget(logger, target_ip, server_ip, **kwargs) 110 elif target_type == 'qemu': 111 target = OEQemuTarget(logger, server_ip, **kwargs) 112 elif target_type == 'serial': 113 target = OESerialTarget(logger, target_ip, server_ip, **kwargs) 114 else: 115 # XXX: This code uses the old naming convention for controllers and 116 # targets, the idea it is to leave just targets as the controller 117 # most of the time was just a wrapper. 118 # XXX: This code tries to import modules from lib/oeqa/controllers 119 # directory and treat them as controllers, it will less error prone 120 # to use introspection to load such modules. 121 # XXX: Don't base your targets on this code it will be refactored 122 # in the near future. 123 # Custom target module loading 124 controller = OERuntimeTestContextExecutor.getControllerModule(target_type) 125 target = controller(logger, target_ip, server_ip, **kwargs) 126 127 return target 128 129 # Search oeqa.controllers module directory for and return a controller 130 # corresponding to the given target name. 131 # AttributeError raised if not found. 132 # ImportError raised if a provided module can not be imported. 133 @staticmethod 134 def getControllerModule(target): 135 controllerslist = OERuntimeTestContextExecutor._getControllerModulenames() 136 controller = OERuntimeTestContextExecutor._loadControllerFromName(target, controllerslist) 137 return controller 138 139 # Return a list of all python modules in lib/oeqa/controllers for each 140 # layer in bbpath 141 @staticmethod 142 def _getControllerModulenames(): 143 144 controllerslist = [] 145 146 def add_controller_list(path): 147 if not os.path.exists(os.path.join(path, '__init__.py')): 148 raise OSError('Controllers directory %s exists but is missing __init__.py' % path) 149 files = sorted([f for f in os.listdir(path) if f.endswith('.py') and not f.startswith('_') and not f.startswith('.#')]) 150 for f in files: 151 module = 'oeqa.controllers.' + f[:-3] 152 if module not in controllerslist: 153 controllerslist.append(module) 154 else: 155 raise RuntimeError("Duplicate controller module found for %s. Layers should create unique controller module names" % module) 156 157 # sys.path can contain duplicate paths, but because of the login in 158 # add_controller_list this doesn't work and causes testimage to abort. 159 # Remove duplicates using an intermediate dictionary to ensure this 160 # doesn't happen. 161 for p in list(dict.fromkeys(sys.path)): 162 controllerpath = os.path.join(p, 'oeqa', 'controllers') 163 if os.path.exists(controllerpath): 164 add_controller_list(controllerpath) 165 return controllerslist 166 167 # Search for and return a controller from given target name and 168 # set of module names. 169 # Raise AttributeError if not found. 170 # Raise ImportError if a provided module can not be imported 171 @staticmethod 172 def _loadControllerFromName(target, modulenames): 173 for name in modulenames: 174 obj = OERuntimeTestContextExecutor._loadControllerFromModule(target, name) 175 if obj: 176 return obj 177 raise AttributeError("Unable to load {0} from available modules: {1}".format(target, str(modulenames))) 178 179 # Search for and return a controller or None from given module name 180 @staticmethod 181 def _loadControllerFromModule(target, modulename): 182 try: 183 import importlib 184 module = importlib.import_module(modulename) 185 return getattr(module, target) 186 except AttributeError: 187 return None 188 189 @staticmethod 190 def readPackagesManifest(manifest): 191 if not manifest or not os.path.exists(manifest): 192 raise OSError("Manifest file not exists: %s" % manifest) 193 194 image_packages = set() 195 with open(manifest, 'r') as f: 196 for line in f.readlines(): 197 line = line.strip() 198 if line and not line.startswith("#"): 199 image_packages.add(line.split()[0]) 200 201 return image_packages 202 203 def _process_args(self, logger, args): 204 if not args.packages_manifest: 205 raise TypeError('Manifest file not provided') 206 207 super(OERuntimeTestContextExecutor, self)._process_args(logger, args) 208 209 td = self.tc_kwargs['init']['td'] 210 211 target_kwargs = {} 212 target_kwargs['machine'] = td.get("MACHINE") or None 213 target_kwargs['qemuboot'] = args.qemu_boot 214 target_kwargs['serialcontrol_cmd'] = td.get("TEST_SERIALCONTROL_CMD") or None 215 target_kwargs['serialcontrol_extra_args'] = td.get("TEST_SERIALCONTROL_EXTRA_ARGS") or "" 216 target_kwargs['serialcontrol_ps1'] = td.get("TEST_SERIALCONTROL_PS1") or None 217 target_kwargs['serialcontrol_connect_timeout'] = td.get("TEST_SERIALCONTROL_CONNECT_TIMEOUT") or None 218 219 self.tc_kwargs['init']['target'] = \ 220 OERuntimeTestContextExecutor.getTarget(args.target_type, 221 None, args.target_ip, args.server_ip, **target_kwargs) 222 self.tc_kwargs['init']['image_packages'] = \ 223 OERuntimeTestContextExecutor.readPackagesManifest( 224 args.packages_manifest) 225 self.tc_kwargs['init']['extract_dir'] = args.extract_dir 226 227_executor_class = OERuntimeTestContextExecutor 228