rtems-tools/tester/rt/config.py

#
# 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

    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 _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 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):
        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 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:
            self.process.kill()