rtems-tools/tester/rt/wait.py

# SPDX-License-Identifier: BSD-2-Clause
'''Wait test target.'''

# Copyright (C) 2013-2021 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


class wait(object):
    '''RTEMS Testing WAIT 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._init()

    def _init(self):
        self.output_length = None
        self.console = None
        self.server = None
        self.timeout = None
        self.test_too_long = None
        self.timer = None
        self.waiting = False
        self.seconds = 0

    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 _trace(self, text):
        if log.tracing:
            log.trace('] wait: ' + text)

    def _console(self, text):
        if self.console:
            self.console(text)

    def _kill(self):
        self._lock('_kill')
        self._console('wait: kill')
        self.waiting = False
        self._unlock('_kill')

    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 _monitor(self, timeout):
        output_length = self.output_length
        period = timeout[0]
        seconds = timeout[1]
        self._console('wait: start: period=%d seconds=%d' % (period, seconds))
        step = 0.5
        period *= step
        seconds *= step
        self.waiting = True
        while self.waiting and period > 0 and seconds > 0:
            self._unlock('_monitor')
            current_length = self.output_length()
            if output_length != current_length:
                period = timeout[0] * step
            output_length = current_length
            if seconds < step:
                seconds = 0
            else:
                seconds -= step
            if period < step:
                step = period
                period = 0
            else:
                period -= step
            time.sleep(step)
            self._lock('_monitor')
        if self.waiting:
            if period == 0:
                self._timeout()
            elif seconds == 0:
                self._test_too_long()
        self._console('wait: finished: period=%d seconds=%d' % (period, seconds))

    def open(self, output_length, console, timeout):
        '''Start to wait'''
        # pylint: disable=too-many-arguments
        self._lock('_open')
        self._init()
        self.output_length = output_length
        self.console = console
        self.timeout = timeout[2]
        self.test_too_long = timeout[3]
        self._monitor(timeout)

    def kill(self):
        '''Kill the test run.'''
        self._kill()

    def target_restart(self, started):
        pass

    def target_reset(self, started):
        pass

    def target_start(self):
        pass

    def target_end(self):
        pass