#
# RTEMS Tools Project (http://www.rtems.org/)
# Copyright 2013-2014 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 threading
from rtemstoolkit import config
from rtemstoolkit import error
from rtemstoolkit import execute
from rtemstoolkit import log
from rtemstoolkit import path
from . import console
from . import gdb
timeout = 15
class file(config.file):
"""RTEMS Testing configuration."""
_directives = ['%execute',
'%gdb',
'%console']
def __init__(self, report, name, opts, _directives = _directives):
super(file, self).__init__(name, opts, directives = _directives)
self.lock = threading.Lock()
self.realtime_trace = self.debug_trace('output')
self.process = None
self.console = None
self.output = None
self.report = report
self.name = name
self.timedout = False
self.kill_good = False
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 _ok_kill(self):
self.kill_good = True
try:
self.process.kill()
except:
pass
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.debug_trace('console')
if data[0] == 'stdio':
self.console = 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 = 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, exe, bsp_arch, bsp):
self.process = execute.execute(output = self.capture)
if not self.in_error:
if self.console:
self.console.open()
self.capture_console('run: %s' % (' '.join(data)))
ec, proc = self.process.open(data,
timeout = (int(self.expand('%{timeout}')),
self._timeout))
self._lock()
if not self.kill_good and ec > 0:
self._error('execute failed: %s: exit-code:%d' % (' '.join(data), ec))
elif self.timedout:
self.process.kill()
self._unlock()
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 = gdb.gdb(bsp_arch, bsp,
trace = self.debug_trace('gdb'),
mi_trace = self.debug_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()
self.process.open(data[0], data[1],
script = script,
output = self.capture,
gdb_console = self.capture_console,
timeout = int(self.expand('%{timeout}')))
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:
total = int(self.expand('%{test_total}'))
index = int(self.expand('%{test_index}'))
exe = self.expand('%{test_executable}')
bsp_arch = self.expand('%{bsp_arch}')
bsp = self.expand('%{bsp}')
self.report.start(index, total, exe, exe, bsp_arch, bsp)
self.output = []
finally:
self._unlock()
if _directive == '%execute':
self._dir_execute(ds, total, index, exe, bsp_arch, bsp)
elif _directive == '%gdb':
self._dir_gdb(ds, total, index, exe, bsp_arch, bsp)
else:
raise error.general(self._name_line_msg('invalid directive'))
self._lock()
try:
self.report.end(exe, self.output)
self.process = None
self.output = None
finally:
self._unlock()
return None, None, None
def _realtime_trace(self, text):
if self.realtime_trace:
for l in text:
print(' '.join(l))
def run(self):
self.load(self.name)
def capture(self, text):
ok_to_kill = '*** TEST STATE: USER_INPUT' in text or '*** TEST STATE: BENCHMARK' in text
text = [(']', l) for l in text.replace(chr(13), '').splitlines()]
self._lock()
if self.output is not None:
self._realtime_trace(text)
self.output += text
if ok_to_kill:
self._ok_kill()
self._unlock()
def capture_console(self, text):
text = [('>', l) for l in text.replace(chr(13), '').splitlines()]
self._lock()
if self.output is not None:
self._realtime_trace(text)
self.output += text
self._unlock()
def debug_trace(self, flag):
dt = self.macros['debug_trace']
if dt:
if flag in dt.split(','):
return True
return False
def kill(self):
if self.process:
try:
self.process.kill()
except:
pass
