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