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