xref: /openbmc/openbmc/poky/meta/lib/oeqa/core/context.py (revision 393846f1) 
1## Copyright (C) 2016 Intel Corporation
2#
3# SPDX-License-Identifier: MIT
4#
5
6import os
7import sys
8import json
9import time
10import logging
11import collections
12
13from oeqa.core.loader import OETestLoader
14from oeqa.core.runner import OETestRunner
15from oeqa.core.exception import OEQAMissingManifest, OEQATestNotFound
16
17class OETestContext(object):
18    loaderClass = OETestLoader
19    runnerClass = OETestRunner
20
21    files_dir = os.path.abspath(os.path.join(os.path.dirname(
22        os.path.abspath(__file__)), "../files"))
23
24    def __init__(self, td=None, logger=None):
25        if not type(td) is dict:
26            raise TypeError("td isn't dictionary type")
27
28        self.td = td
29        self.logger = logger
30        self._registry = {}
31        self._registry['cases'] = collections.OrderedDict()
32
33    def _read_modules_from_manifest(self, manifest):
34        if not os.path.exists(manifest):
35            raise OEQAMissingManifest("Manifest does not exist on %s" % manifest)
36
37        modules = []
38        for line in open(manifest).readlines():
39            line = line.strip()
40            if line and not line.startswith("#"):
41                modules.append(line)
42
43        return modules
44
45    def skipTests(self, skips):
46        if not skips:
47            return
48        for test in self.suites:
49            for skip in skips:
50                if test.id().startswith(skip):
51                    setattr(test, 'setUp', lambda: test.skipTest('Skip by the command line argument "%s"' % skip))
52
53    def loadTests(self, module_paths, modules=[], tests=[],
54            modules_manifest="", modules_required=[], filters={}):
55        if modules_manifest:
56            modules = self._read_modules_from_manifest(modules_manifest)
57
58        self.loader = self.loaderClass(self, module_paths, modules, tests,
59                modules_required, filters)
60        self.suites = self.loader.discover()
61
62    def runTests(self, processes=None, skips=[]):
63        self.runner = self.runnerClass(self, descriptions=False, verbosity=2)
64
65        # Dinamically skip those tests specified though arguments
66        self.skipTests(skips)
67
68        self._run_start_time = time.time()
69        if processes:
70            from oeqa.core.utils.concurrencytest import ConcurrentTestSuite
71
72            concurrent_suite = ConcurrentTestSuite(self.suites, processes)
73            result = self.runner.run(concurrent_suite)
74        else:
75            self.runner.buffer = True
76            result = self.runner.run(self.suites)
77        self._run_end_time = time.time()
78
79        return result
80
81    def listTests(self, display_type):
82        self.runner = self.runnerClass(self, verbosity=2)
83        return self.runner.list_tests(self.suites, display_type)
84
85class OETestContextExecutor(object):
86    _context_class = OETestContext
87    _script_executor = 'oe-test'
88
89    name = 'core'
90    help = 'core test component example'
91    description = 'executes core test suite example'
92
93    default_cases = [os.path.join(os.path.abspath(os.path.dirname(__file__)),
94            'cases/example')]
95    default_test_data = os.path.join(default_cases[0], 'data.json')
96    default_tests = None
97
98    def register_commands(self, logger, subparsers):
99        self.parser = subparsers.add_parser(self.name, help=self.help,
100                description=self.description, group='components')
101
102        self.default_output_log = '%s-results-%s.log' % (self.name,
103                time.strftime("%Y%m%d%H%M%S"))
104        self.parser.add_argument('--output-log', action='store',
105                default=self.default_output_log,
106                help="results output log, default: %s" % self.default_output_log)
107
108        group = self.parser.add_mutually_exclusive_group()
109        group.add_argument('--run-tests', action='store', nargs='+',
110                default=self.default_tests,
111                help="tests to run in <module>[.<class>[.<name>]]")
112        group.add_argument('--list-tests', action='store',
113                choices=('module', 'class', 'name'),
114                help="lists available tests")
115
116        if self.default_test_data:
117            self.parser.add_argument('--test-data-file', action='store',
118                    default=self.default_test_data,
119                    help="data file to load, default: %s" % self.default_test_data)
120        else:
121            self.parser.add_argument('--test-data-file', action='store',
122                    help="data file to load")
123
124        if self.default_cases:
125            self.parser.add_argument('CASES_PATHS', action='store',
126                    default=self.default_cases, nargs='*',
127                    help="paths to directories with test cases, default: %s"\
128                            % self.default_cases)
129        else:
130            self.parser.add_argument('CASES_PATHS', action='store',
131                    nargs='+', help="paths to directories with test cases")
132
133        self.parser.set_defaults(func=self.run)
134
135    def _setup_logger(self, logger, args):
136        formatter = logging.Formatter('%(asctime)s - ' + self.name + \
137                ' - %(levelname)s - %(message)s')
138        sh = logger.handlers[0]
139        sh.setFormatter(formatter)
140        fh = logging.FileHandler(args.output_log)
141        fh.setFormatter(formatter)
142        logger.addHandler(fh)
143
144        return logger
145
146    def _process_args(self, logger, args):
147        self.tc_kwargs = {}
148        self.tc_kwargs['init'] = {}
149        self.tc_kwargs['load'] = {}
150        self.tc_kwargs['list'] = {}
151        self.tc_kwargs['run']  = {}
152
153        self.tc_kwargs['init']['logger'] = self._setup_logger(logger, args)
154        if args.test_data_file:
155            self.tc_kwargs['init']['td'] = json.load(
156                    open(args.test_data_file, "r"))
157        else:
158            self.tc_kwargs['init']['td'] = {}
159
160        if args.run_tests:
161            self.tc_kwargs['load']['modules'] = args.run_tests
162            self.tc_kwargs['load']['modules_required'] = args.run_tests
163        else:
164            self.tc_kwargs['load']['modules'] = []
165
166        self.tc_kwargs['run']['skips'] = []
167
168        self.module_paths = args.CASES_PATHS
169
170    def _pre_run(self):
171        pass
172
173    def run(self, logger, args):
174        self._process_args(logger, args)
175
176        self.tc = self._context_class(**self.tc_kwargs['init'])
177        try:
178            self.tc.loadTests(self.module_paths, **self.tc_kwargs['load'])
179        except OEQATestNotFound as ex:
180            logger.error(ex)
181            sys.exit(1)
182
183        if args.list_tests:
184            rc = self.tc.listTests(args.list_tests, **self.tc_kwargs['list'])
185        else:
186            self._pre_run()
187            rc = self.tc.runTests(**self.tc_kwargs['run'])
188            rc.logDetails()
189            rc.logSummary(self.name)
190
191        output_link = os.path.join(os.path.dirname(args.output_log),
192                "%s-results.log" % self.name)
193        if os.path.exists(output_link):
194            os.remove(output_link)
195        os.symlink(args.output_log, output_link)
196
197        return rc
198
199_executor_class = OETestContextExecutor
200