1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
#
# Copyright (C) 2016 Intel Corporation
#
# SPDX-License-Identifier: MIT
#
import os
import sys
from oeqa.core.context import OETestContext, OETestContextExecutor
from oeqa.core.target.ssh import OESSHTarget
from oeqa.core.target.qemu import OEQemuTarget
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,
image_packages, extract_dir):
super(OERuntimeTestContext, self).__init__(td, logger)
self.target = target
self.image_packages = image_packages
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_json_result_dir = '%s-results' % name
default_target_type = 'simpleremote'
default_manifest = 'data/manifest'
default_server_ip = '192.168.7.1'
default_target_ip = '192.168.7.2'
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 and optionally ssh port (default 22) of device under test, for example '192.168.0.7:22'. Default: %s" \
% self.default_target_ip)
runtime_group.add_argument('--server-ip', action='store',
default=self.default_target_ip,
help="IP address of the test host from test target machine, default: %s" \
% self.default_server_ip)
runtime_group.add_argument('--host-dumper-dir', action='store',
help="Directory where host status is dumped, if tests fails")
runtime_group.add_argument('--packages-manifest', action='store',
default=self.default_manifest,
help="Package manifest of the image under test, 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 server_ip:
server_ip_port = server_ip.split(':')
if len(server_ip_port) == 2:
server_ip = server_ip_port[0]
kwargs['server_port'] = int(server_ip_port[1])
if target_type == 'simpleremote':
target = OESSHTarget(logger, target_ip, server_ip, **kwargs)
elif target_type == 'qemu':
target = OEQemuTarget(logger, 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
controller = OERuntimeTestContextExecutor.getControllerModule(target_type)
target = controller(logger, target_ip, server_ip, **kwargs)
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):
controllerslist = OERuntimeTestContextExecutor._getControllerModulenames()
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():
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('_') 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)
# sys.path can contain duplicate paths, but because of the login in
# add_controller_list this doesn't work and causes testimage to abort.
# Remove duplicates using an intermediate dictionary to ensure this
# doesn't happen.
for p in list(dict.fromkeys(sys.path)):
controllerpath = os.path.join(p, '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):
try:
import importlib
module = importlib.import_module(modulename)
return getattr(module, target)
except AttributeError:
return None
@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
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']['image_packages'] = \
OERuntimeTestContextExecutor.readPackagesManifest(
args.packages_manifest)
self.tc_kwargs['init']['extract_dir'] = args.extract_dir
_executor_class = OERuntimeTestContextExecutor
|
