# # RTEMS Tools Project (http://www.rtems.org/) # Copyright 2013-2020 Chris Johns (chrisj@rtems.org) # All rights reserved. # # This file is part of the RTEMS Tools package in 'rtems-tools'. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # # 1. Redistributions of source code must retain the above copyright notice, # this list of conditions and the following disclaimer. # # 2. Redistributions in binary form must reproduce the above copyright notice, # this list of conditions and the following disclaimer in the documentation # and/or other materials provided with the distribution. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. # # # RTEMS Testing Config # from __future__ import print_function import datetime import os import re import threading from rtemstoolkit import configuration from rtemstoolkit import config from rtemstoolkit import error from rtemstoolkit import execute from rtemstoolkit import log from rtemstoolkit import path from rtemstoolkit import stacktraces import tester.rt.console import tester.rt.exe import tester.rt.gdb import tester.rt.tftp timeout = 15 class file(config.file): """RTEMS Testing configuration.""" _directives = ['%execute', '%gdb', '%tftp', '%console'] def __init__(self, index, total, report, name, opts, console_prefix = ']', _directives = _directives): super(file, self).__init__(name, opts, directives = _directives) self.lock = threading.Lock() self.realtime_trace = self.exe_trace('output') self.console_trace = self.exe_trace('console') self.console_prefix = console_prefix self.show_header = not self.defined('test_disable_header') self.process = None self.console = None self.output = None self.index = index self.total = total self.report = report self.name = name self.timedout = False self.test_too_long = False self.test_started = False self.kill_good = False self.kill_on_end = False self.test_label = None self.target_reset_regx = None self.target_start_regx = None def __del__(self): if self.console: del self.console super(file, self).__del__() def _lock(self): self.lock.acquire() def _unlock(self): self.lock.release() def _timeout(self): self._lock() self.timedout = True self._unlock() self.capture('*** TIMEOUT TIMEOUT') def _test_too_long(self): self._lock() self.test_too_long = True self._unlock() self.capture('*** TEST TOO LONG') def _ok_kill(self): self._lock() self.kill_good = True self._unlock() try: self.process.kill() except: pass def _target_regex(self, label): regex = None if self.defined(label): r = self.expand('%%{%s}' % (label)) try: regex = re.compile(r, re.MULTILINE) except: msg = 'invalid target regex: %s: %s' % (label, r) raise error.general(msg) return regex def _target_command(self, command, bsp_arch = None, bsp = None, exe = None, fexe = None): if self.defined('target_%s_command' % (command)): cmd = self.expand('%%{target_%s_command}' % (command)).strip() if bsp_arch is not None and '@ARCH@' in cmd: cmd = cmd.replace('@ARCH@', bsp_arch) if bsp is not None and '@BSP@' in cmd: cmd = cmd.replace('@BSP@', bsp) if exe is not None and '@EXE@' in cmd: cmd = cmd.replace('@EXE@', exe) if fexe is not None and '@FEXE@' in cmd: cmd = cmd.replace('@FEXE@', exe) if len(cmd) > 0: output = '' if not self.opts.dry_run(): rs_proc = execute.capture_execution() ec, proc, output = rs_proc.open(cmd, shell = True) self._capture_console('target %s: %s' % (command, cmd)) if len(output) > 0: output = os.linesep.join([' ' + l for l in output.splitlines()]) self._capture_console(output) def _target_exe_filter(self, exe): if self.defined('target_exe_filter'): f = self.expand('%{target_exe_filter}').strip() # Be like sed and use the first character as the delmiter. if len(f) > 0: delimiter = f[0] pat = '' repl = '' repl_not_pat = False esc = False for c in f[1:]: add = True if not esc and c == '\\': esc = True add = False elif esc: if c == delimiter: c = delimiter else: c = '\\' + c esc = False elif c == delimiter: if repl_not_pat: exe = re.sub(r'%s' % (pat), repl, exe) self._capture_console('target exe filter: find:%s subst:%s -> %s' % \ (pat, repl, exe)) return exe repl_not_pat = True add = False if add: if repl_not_pat: repl += c else: pat += c raise error.general('invalid exe filter: %s' % (f)) return exe def _output_length(self): self._lock() if self.test_started: l = len(self.output) self._unlock() return l self._unlock() return 0 def _capture_console(self, text): text = [('=> ', l) for l in text.replace(chr(13), '').splitlines()] if self.output is not None: if self.console_trace: self._realtime_trace(text) self.output += text def _dir_console(self, data): if self.console is not None: raise error.general(self._name_line_msg('console already configured')) if len(data) == 0: raise error.general(self._name_line_msg('no console configuration provided')) console_trace = trace = self.exe_trace('console') if not self.opts.dry_run(): if data[0] == 'stdio': self.console = tester.rt.console.stdio(trace = console_trace) elif data[0] == 'tty': if len(data) < 2 or len(data) >3: raise error.general(self._name_line_msg('no tty configuration provided')) if len(data) == 3: settings = data[2] else: settings = None self.console = tester.rt.console.tty(data[1], output = self.capture, setup = settings, trace = console_trace) else: raise error.general(self._name_line_msg('invalid console type')) def _dir_execute(self, data, total, index, rexe, bsp_arch, bsp): self.process = tester.rt.exe.exe(bsp_arch, bsp, trace = self.exe_trace('exe')) if not self.in_error: if self.console: self.console.open() if not self.opts.dry_run(): self.process.open(data, ignore_exit_code = self.defined('exe_ignore_ret'), output = self.capture, console = self.capture_console, timeout = (int(self.expand('%{timeout}')), int(self.expand('%{max_test_period}')), self._timeout, self._test_too_long)) if self.console: self.console.close() def _dir_gdb(self, data, total, index, exe, bsp_arch, bsp): if len(data) < 3 or len(data) > 4: raise error.general('invalid %gdb arguments') self.process = tester.rt.gdb.gdb(bsp_arch, bsp, trace = self.exe_trace('gdb'), mi_trace = self.exe_trace('gdb-mi')) script = self.expand('%%{%s}' % data[2]) if script: script = [l.strip() for l in script.splitlines()] if not self.in_error: if self.console: self.console.open() if not self.opts.dry_run(): self.process.open(data[0], data[1], script = script, output = self.capture, gdb_console = self.capture_console, timeout = (int(self.expand('%{timeout}')), int(self.expand('%{max_test_period}')), self._timeout, self._test_too_long)) if self.console: self.console.close() def _dir_tftp(self, data, total, index, exe, bsp_arch, bsp): if len(data) != 2: raise error.general('invalid %tftp arguments') try: port = int(data[1]) except: raise error.general('invalid %tftp port') self.kill_on_end = True if not self.opts.dry_run(): self.process = tester.rt.tftp.tftp(bsp_arch, bsp, trace = self.exe_trace('tftp')) if not self.in_error: if self.console: self.console.open() self.process.open(executable = exe, port = port, output_length = self._output_length, console = self.capture_console, timeout = (int(self.expand('%{timeout}')), int(self.expand('%{max_test_period}')), self._timeout, self._test_too_long)) if self.console: self.console.close() def _directive_filter(self, results, directive, info, data): if results[0] == 'directive': _directive = results[1] _data = results[2] ds = [] if len(_data): ds = [_data[0]] if len(_data) > 1: ds += _data[1].split() ds = self.expand(ds) if _directive == '%console': self._dir_console(ds) else: self._lock() try: self.output = [] total = int(self.expand('%{test_total}')) index = int(self.expand('%{test_index}')) exe = self.expand('%{test_executable}') bsp_arch = self.expand('%{arch}') bsp = self.expand('%{bsp}') fexe = self._target_exe_filter(exe) self.report.start(index, total, exe, fexe, bsp_arch, bsp, self.show_header) if self.index == 1: self._target_command('on', bsp_arch, bsp, exe, fexe) self._target_command('pretest', bsp_arch, bsp, exe, fexe) finally: self._unlock() if _directive == '%execute': self._dir_execute(ds, total, index, fexe, bsp_arch, bsp) elif _directive == '%gdb': self._dir_gdb(ds, total, index, fexe, bsp_arch, bsp) elif _directive == '%tftp': self._dir_tftp(ds, total, index, fexe, bsp_arch, bsp) else: raise error.general(self._name_line_msg('invalid directive')) self._lock() if self.index == self.total: self._target_command('off', bsp_arch, bsp, exe, fexe) self._target_command('posttest', bsp_arch, bsp, exe, fexe) try: status = self.report.end(exe, self.output, self.console_prefix) version = self.report.get_config('version', not_found = 'n/p') build = self.report.get_config('build', not_found = 'n/p') tools = self.report.get_config('tools', not_found = 'n/p') self._capture_console('test result: %s' % (status)) self._capture_console('test version: %s' % (version)) self._capture_console('test build: %s' % (build)) self._capture_console('test tools: %s' % (tools)) if status == 'timeout': if self.index != self.total: self._target_command('reset', bsp_arch, bsp, exe, fexe) self.process = None self.output = None finally: self._unlock() return None, None, None def _realtime_trace(self, text): for l in text: self._unlock() try: print(''.join(l)) except: stacktraces.trace() raise finally: self._lock() def run(self): self.target_start_regx = self._target_regex('target_start_regex') self.target_reset_regx = self._target_regex('target_reset_regex') self.load(self.name) def capture(self, text): self._lock() try: if not self.test_started: s = text.find('*** BEGIN OF TEST ') if s >= 0: self.test_started = True e = text[s + 3:].find('***') if e >= 0: self.test_label = text[s + len('*** BEGIN OF TEST '):s + e + 3 - 1] self._capture_console('test start: %s' % (self.test_label)) ok_to_kill = '*** TEST STATE: USER_INPUT' in text or \ '*** TEST STATE: BENCHMARK' in text if ok_to_kill: reset_target = True else: reset_target = False if self.test_started and self.target_start_regx is not None: if self.target_start_regx.match(text): self._capture_console('target start detected') ok_to_kill = True if not reset_target and self.target_reset_regx is not None: if self.target_reset_regx.match(text): self._capture_console('target reset condition detected') self._target_command('reset') if self.kill_on_end: if not ok_to_kill and '*** END OF TEST ' in text: self._capture_console('test end: %s' % (self.test_label)) if self.test_label is not None: ok_to_kill = '*** END OF TEST %s ***' % (self.test_label) in text text = [(self.console_prefix, l) for l in text.replace(chr(13), '').splitlines()] if self.output is not None: if self.realtime_trace: self._realtime_trace(text) self.output += text if reset_target: if self.index == self.total: self._target_command('off') else: self._target_command('reset') finally: self._unlock() if ok_to_kill: self._ok_kill() def capture_console(self, text): self._lock() try: self._capture_console(text) finally: self._unlock() def exe_trace(self, flag): dt = self.macros['exe_trace'] if dt: if flag in dt.split(','): return True return False def kill(self): if self.process: try: self.process.kill() except: pass def load(bsp, opts): mandatory = ['bsp', 'arch', 'tester'] cfg = configuration.configuration() path_ = opts.defaults.expand('%%{_configdir}/bsps/%s.ini' % (bsp)) ini_name = path.basename(path_) for p in path.dirname(path_).split(':'): if path.exists(path.join(p, ini_name)): cfg.load(path.join(p, ini_name)) if not cfg.has_section(bsp): raise error.general('bsp section not found in ini: [%s]' % (bsp)) item_names = cfg.get_item_names(bsp, err = False) for m in mandatory: if m not in item_names: raise error.general('mandatory item not found in bsp section: %s' % (m)) opts.defaults.set_write_map(bsp, add = True) for i in cfg.get_items(bsp, flatten = False): opts.defaults[i[0]] = i[1] if not opts.defaults.set_read_map(bsp): raise error.general('cannot set BSP read map: %s' % (bsp)) # Get a copy of the required fields we need requires = cfg.comma_list(bsp, 'requires', err = False) del cfg user_config = opts.find_arg('--user-config') if user_config is not None: user_config = path.expanduser(user_config[1]) if not path.exists(user_config): raise error.general('cannot find user configuration file: %s' % (user_config)) else: if 'HOME' in os.environ: user_config = path.join(os.environ['HOME'], '.rtemstesterrc') if user_config: if path.exists(user_config): cfg = configuration.configuration() cfg.load(user_config) if cfg.has_section(bsp): for i in cfg.get_items(bsp, flatten = False): opts.defaults[i[0]] = i[1] # Check for the required values. for r in requires: if opts.defaults.get(r) is None: raise error.general('user value missing, BSP %s requires \'%s\': missing: %s' % \ (bsp, ', '.join(requires), r)) return opts.defaults['bsp'] raise error.general('cannot find bsp configuration file: %s.ini' % (bsp))