# SPDX-License-Identifier: BSD-2-Clause
'''Executable test target.'''
# Copyright (C) 2013-2020 Chris Johns (chrisj@rtems.org)
#
# 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 OWNER 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.
from __future__ import print_function
import datetime
import os
import threading
import time
from rtemstoolkit import execute
class exe(object):
'''RTEMS Testing EXE base.'''
# pylint: disable=useless-object-inheritance
# pylint: disable=too-many-instance-attributes
def __init__(self, bsp_arch, bsp, trace=False):
self.trace = trace
self.lock_trace = False
self.lock_locked = None
self.lock = threading.RLock()
self.bsp = bsp
self.bsp_arch = bsp_arch
self.output = None
self.output_length = 0
self.process = None
self.ecode = None
self.output = None
self.output_buffer = ''
self.console = None
self.timeout = None
self.test_too_long = None
self.kill_good = True
def _lock(self, msg):
if self.lock_trace:
print('|[ LOCK:%s ]|' % (msg))
self.lock_locked = datetime.datetime.now()
self.lock.acquire()
def _unlock(self, msg):
if self.lock_trace:
period = datetime.datetime.now() - self.lock_locked
print('|] UNLOCK:%s [| : %s' % (msg, period))
self.lock.release()
def _capture(self, text):
self._lock('_capture')
self.output_length += len(text)
self._unlock('_capture')
if self.output is not None:
self.output(text)
def _timeout(self):
self._kill()
if self.timeout is not None:
self.timeout()
def _test_too_long(self):
self._kill()
if self.test_too_long is not None:
self.test_too_long()
def _kill(self):
self._lock('_kill')
self.kill_good = True
self._unlock('_kill')
if self.process:
# pylint: disable=bare-except
try:
self.process.kill()
except:
pass
self.process = None
def _execute(self, args):
'''Thread to execute the test and to wait for it to finish.'''
# pylint: disable=unused-variable
cmds = args
if self.console is not None:
self.console('exe: %s' % (' '.join(cmds)))
ecode, proc = self.process.open(cmds)
if self.trace:
print('gdb done', ecode)
self._lock('_execute')
self.ecode = ecode
self.process = None
self._unlock('_execute')
def _monitor(self, timeout):
output_length = self.output_length
step = 0.25
period = timeout[0]
seconds = timeout[1]
while self.process and period > 0 and seconds > 0:
current_length = self.output_length
if output_length != current_length:
period = timeout[0]
output_length = current_length
if seconds < step:
seconds = 0
else:
seconds -= step
if period < step:
step = period
period = 0
else:
period -= step
self._unlock('_monitor')
time.sleep(step)
self._lock('_monitor')
if self.process is not None:
if period == 0:
self._timeout()
elif seconds == 0:
self._test_too_long()
def open(self, command, ignore_exit_code, output, console, timeout):
'''Open the execute test run'''
# pylint: disable=too-many-arguments
self._lock('_open')
self.timeout = timeout[2]
self.test_too_long = timeout[3]
try:
cmds = execute.arg_list(command)
self.output = output
self.console = console
self.process = execute.execute(output=self._capture)
exec_thread = threading.Thread(target=self._execute, args=[cmds])
exec_thread.start()
self._monitor(timeout)
if self.ecode is not None and \
not (self.kill_good or ignore_exit_code) and self.ecode > 0:
if self.output:
self.output('*** TARGET ERROR %d %s ***' %
(self.ecode, os.strerror(self.ecode)))
finally:
self._unlock('_open')
def kill(self):
'''Kill the test run.'''
self._lock('_kill')
try:
self._kill()
finally:
self._unlock('_kill')
