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