Merge pull request #10857 from ARMmbed/feature-watchdog

Add Watchdog and ResetReason

Author: 0xc0170

TESTS/host_tests/reset_reason.py

@@ -0,0 +1,163 @@
+"""
+Copyright (c) 2018-2019 Arm Limited and affiliates.
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+import time
+from mbed_host_tests import BaseHostTest
+from mbed_host_tests.host_tests_runner.host_test_default import DefaultTestSelector
+
+DEFAULT_SYNC_DELAY = 4.0
+
+MSG_VALUE_WATCHDOG_PRESENT = 'wdg_present'
+MSG_VALUE_DUMMY = '0'
+MSG_VALUE_RESET_REASON_GET = 'get'
+MSG_VALUE_RESET_REASON_CLEAR = 'clear'
+MSG_VALUE_DEVICE_RESET_NVIC = 'nvic'
+MSG_VALUE_DEVICE_RESET_WATCHDOG = 'watchdog'
+
+MSG_KEY_DEVICE_READY = 'ready'
+MSG_KEY_RESET_REASON_RAW = 'reason_raw'
+MSG_KEY_RESET_REASON = 'reason'
+MSG_KEY_DEVICE_RESET = 'reset'
+MSG_KEY_SYNC = '__sync'
+
+RESET_REASONS = {
+    'POWER_ON': '0',
+    'PIN_RESET': '1',
+    'BROWN_OUT': '2',
+    'SOFTWARE': '3',
+    'WATCHDOG': '4',
+    'LOCKUP': '5',
+    'WAKE_LOW_POWER': '6',
+    'ACCESS_ERROR': '7',
+    'BOOT_ERROR': '8',
+    'MULTIPLE': '9',
+    'PLATFORM': '10',
+    'UNKNOWN': '11'
+}
+
+
+def raise_if_different(expected, actual, text=''):
+    """Raise a RuntimeError if actual is different than expected."""
+    if expected != actual:
+        raise RuntimeError('{}Got {!r}, expected {!r}'
+                           .format(text, actual, expected))
+
+
+class ResetReasonTest(BaseHostTest):
+    """Test for the Reset Reason HAL API.
+
+    Given a device supporting a Reset Reason API.
+    When the device is restarted using various methods.
+    Then the device returns a correct reset reason for every restart.
+    """
+
+    def __init__(self):
+        super(ResetReasonTest, self).__init__()
+        self.device_has_watchdog = None
+        self.raw_reset_reasons = set()
+        self.sync_delay = DEFAULT_SYNC_DELAY
+        self.test_steps_sequence = self.test_steps()
+        # Advance the coroutine to it's first yield statement.
+        self.test_steps_sequence.send(None)
+
+    def setup(self):
+        sync_delay = self.get_config_item('forced_reset_timeout')
+        self.sync_delay = sync_delay if sync_delay is not None else DEFAULT_SYNC_DELAY
+        self.register_callback(MSG_KEY_DEVICE_READY, self.cb_device_ready)
+        self.register_callback(MSG_KEY_RESET_REASON_RAW, self.cb_reset_reason_raw)
+        self.register_callback(MSG_KEY_RESET_REASON, self.cb_reset_reason)
+        self.register_callback(MSG_KEY_DEVICE_RESET, self.cb_reset_reason)
+
+    def cb_device_ready(self, key, value, timestamp):
+        """Request a raw value of the reset_reason register.
+
+        Additionally, save the device's watchdog status on the first call.
+        """
+        if self.device_has_watchdog is None:
+            self.device_has_watchdog = (value == MSG_VALUE_WATCHDOG_PRESENT)
+        self.send_kv(MSG_KEY_RESET_REASON_RAW, MSG_VALUE_RESET_REASON_GET)
+
+    def cb_reset_reason_raw(self, key, value, timestamp):
+        """Verify that the raw reset_reason register value is unique.
+
+        Fail the test suite if the raw reset_reason value is not unique.
+        Request a platform independent reset_reason otherwise.
+        """
+        if value in self.raw_reset_reasons:
+            self.log('TEST FAILED: The raw reset reason is not unique. '
+                     '{!r} is already present in {!r}.'
+                     .format(value, self.raw_reset_reasons))
+            self.notify_complete(False)
+        else:
+            self.raw_reset_reasons.add(value)
+        self.send_kv(MSG_KEY_RESET_REASON, MSG_VALUE_RESET_REASON_GET)
+
+    def cb_reset_reason(self, key, value, timestamp):
+        """Feed the test_steps coroutine with reset_reason value.
+
+        Pass the test suite if the coroutine yields True.
+        Fail the test suite if the iterator stops or raises a RuntimeError.
+        """
+        try:
+            if self.test_steps_sequence.send(value):
+                self.notify_complete(True)
+        except (StopIteration, RuntimeError) as exc:
+            self.log('TEST FAILED: {}'.format(exc))
+            self.notify_complete(False)
+
+    def test_steps(self):
+        """Generate a sequence of test steps.
+
+        This coroutine calls yield to wait for the input from the device
+        (the reset_reason). If the device gives the wrong response, the
+        generator raises a RuntimeError exception and fails the test.
+        """
+        # Ignore the first reason.
+        __ignored_reset_reason = yield
+        self.raw_reset_reasons.clear()
+        self.send_kv(MSG_KEY_RESET_REASON, MSG_VALUE_RESET_REASON_CLEAR)
+        __ignored_clear_ack = yield
+
+        # Request a NVIC_SystemReset() call.
+        self.send_kv(MSG_KEY_DEVICE_RESET, MSG_VALUE_DEVICE_RESET_NVIC)
+        __ignored_reset_ack = yield
+        time.sleep(self.sync_delay)
+        self.send_kv(MSG_KEY_SYNC, MSG_VALUE_DUMMY)
+        reset_reason = yield
+        raise_if_different(RESET_REASONS['SOFTWARE'], reset_reason, 'Wrong reset reason. ')
+        self.send_kv(MSG_KEY_RESET_REASON, MSG_VALUE_RESET_REASON_CLEAR)
+        __ignored_clear_ack = yield
+
+        # Reset the device using DAP.
+        self.reset_dut(DefaultTestSelector.RESET_TYPE_SW_RST)
+        reset_reason = yield
+        raise_if_different(RESET_REASONS['PIN_RESET'], reset_reason, 'Wrong reset reason. ')
+        self.send_kv(MSG_KEY_RESET_REASON, MSG_VALUE_RESET_REASON_CLEAR)
+        __ignored_clear_ack = yield
+
+        # Start a watchdog timer and wait for it to reset the device.
+        if not self.device_has_watchdog:
+            self.log('DUT does not have a watchdog. Skipping this reset reason.')
+        else:
+            self.send_kv(MSG_KEY_DEVICE_RESET, MSG_VALUE_DEVICE_RESET_WATCHDOG)
+            __ignored_reset_ack = yield
+            time.sleep(self.sync_delay)
+            self.send_kv(MSG_KEY_SYNC, MSG_VALUE_DUMMY)
+            reset_reason = yield
+            raise_if_different(RESET_REASONS['WATCHDOG'], reset_reason, 'Wrong reset reason. ')
+
+        # The sequence is correct -- test passed.
+        yield True

TESTS/host_tests/sync_on_reset.py

@@ -0,0 +1,74 @@
+"""
+Copyright (c) 2018-2019 Arm Limited and affiliates.
+
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+import time
+from mbed_host_tests import BaseHostTest
+
+DEFAULT_SYNC_DELAY = 4.0
+
+MSG_VALUE_DUMMY = '0'
+MSG_KEY_DEVICE_READY = 'ready'
+MSG_KEY_START_CASE = 'start_case'
+MSG_KEY_DEVICE_RESET = 'reset_on_case_teardown'
+MSG_KEY_SYNC = '__sync'
+
+
+class SyncOnReset(BaseHostTest):
+    """Host side test that handles device reset during case teardown.
+
+    Given a device that performs a reset during a test case teardown.
+    When the device notifies the host about the reset.
+    Then the host:
+    * keeps track of the test case index of the current test suite,
+    * performs a dev-host handshake,
+    * advances the test suite to next test case.
+
+    Note:
+    Developed for a watchdog test, so that it can be run on devices that
+    do not support watchdog timeout updates after the initial setup.
+    As a solution, after testing watchdog with one set of settings, the
+    device performs a reset and notifies the host with the test case number,
+    so that the test suite may be advanced once the device boots again.
+    """
+
+    def __init__(self):
+        super(SyncOnReset, self).__init__()
+        self.test_case_num = 0
+        self.sync_delay = DEFAULT_SYNC_DELAY
+
+    def setup(self):
+        sync_delay = self.get_config_item('forced_reset_timeout')
+        self.sync_delay = sync_delay if sync_delay is not None else DEFAULT_SYNC_DELAY
+        self.register_callback(MSG_KEY_DEVICE_READY, self.cb_device_ready)
+        self.register_callback(MSG_KEY_DEVICE_RESET, self.cb_device_reset)
+
+    def cb_device_ready(self, key, value, timestamp):
+        """Advance the device test suite to the next test case."""
+        self.send_kv(MSG_KEY_START_CASE, self.test_case_num)
+
+    def cb_device_reset(self, key, value, timestamp):
+        """Wait for the device to boot and perform a handshake.
+
+        Additionally, keep track of the last test case number.
+        """
+        try:
+            self.test_case_num = int(value)
+        except ValueError:
+            pass
+        self.test_case_num += 1
+        time.sleep(self.sync_delay)
+        self.send_kv(MSG_KEY_SYNC, MSG_VALUE_DUMMY)

TESTS/host_tests/watchdog_reset.py

@@ -0,0 +1,145 @@
+"""
+Copyright (c) 2018-2019 Arm Limited and affiliates.
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+import collections
+import threading
+from mbed_host_tests import BaseHostTest
+
+TestCaseData = collections.namedtuple('TestCaseData', ['index', 'data_to_send'])
+
+DEFAULT_SYNC_DELAY = 4.0
+MAX_HB_PERIOD = 2.5  # [s] Max expected heartbeat period.
+
+MSG_VALUE_DUMMY = '0'
+CASE_DATA_INVALID = 0xffffffff
+CASE_DATA_PHASE2_OK = 0xfffffffe
+CASE_DATA_INSUFF_HB = 0x0
+
+MSG_KEY_SYNC = '__sync'
+MSG_KEY_DEVICE_READY = 'ready'
+MSG_KEY_START_CASE = 'start_case'
+MSG_KEY_DEVICE_RESET = 'dev_reset'
+MSG_KEY_HEARTBEAT = 'hb'
+
+
+class WatchdogReset(BaseHostTest):
+    """Host side test that handles device reset.
+
+    Given a device with a watchdog timer started.
+    When the device notifies the host about an incoming reset.
+    Then the host:
+    * keeps track of the test case index of the current test suite,
+    * performs a dev-host handshake.
+    """
+
+    def __init__(self):
+        super(WatchdogReset, self).__init__()
+        self.current_case = TestCaseData(0, CASE_DATA_INVALID)
+        self.__handshake_timer = None
+        self.sync_delay = DEFAULT_SYNC_DELAY
+        self.drop_heartbeat_messages = True
+        self.hb_timestamps_us = []
+
+    def handshake_timer_start(self, seconds=1.0, pre_sync_fun=None):
+        """Start a new handshake timer."""
+
+        def timer_handler():
+            """Perform a dev-host handshake by sending a sync message."""
+            if pre_sync_fun is not None:
+                pre_sync_fun()
+            self.send_kv(MSG_KEY_SYNC, MSG_VALUE_DUMMY)
+
+        self.__handshake_timer = threading.Timer(seconds, timer_handler)
+        self.__handshake_timer.start()
+
+    def handshake_timer_cancel(self):
+        """Cancel the current handshake timer."""
+        try:
+            self.__handshake_timer.cancel()
+        except AttributeError:
+            pass
+        finally:
+            self.__handshake_timer = None
+
+    def heartbeat_timeout_handler(self):
+        """Handler for the heartbeat timeout.
+
+        Compute the time span of the last heartbeat sequence.
+        Set self.current_case.data_to_send to CASE_DATA_INVALID if no heartbeat was received.
+        Set self.current_case.data_to_send to CASE_DATA_INSUFF_HB if only one heartbeat was
+        received.
+        """
+        self.drop_heartbeat_messages = True
+        dev_data = CASE_DATA_INVALID
+        if len(self.hb_timestamps_us) == 1:
+            dev_data = CASE_DATA_INSUFF_HB
+            self.log('Not enough heartbeats received.')
+        elif len(self.hb_timestamps_us) >= 2:
+            dev_data = int(round(0.001 * (self.hb_timestamps_us[-1] - self.hb_timestamps_us[0])))
+            self.log('Heartbeat time span was {} ms.'.format(dev_data))
+        self.current_case = TestCaseData(self.current_case.index, dev_data)
+
+    def setup(self):
+        sync_delay = self.get_config_item('forced_reset_timeout')
+        self.sync_delay = sync_delay if sync_delay is not None else DEFAULT_SYNC_DELAY
+        self.register_callback(MSG_KEY_DEVICE_READY, self.cb_device_ready)
+        self.register_callback(MSG_KEY_DEVICE_RESET, self.cb_device_reset)
+        self.register_callback(MSG_KEY_HEARTBEAT, self.cb_heartbeat)
+
+    def teardown(self):
+        self.handshake_timer_cancel()
+
+    def cb_device_ready(self, key, value, timestamp):
+        """Advance the device test suite to a proper test case.
+
+        Additionally, send test case data to the device.
+        """
+        self.handshake_timer_cancel()
+        msg_value = '{0.index:02x},{0.data_to_send:08x}'.format(self.current_case)
+        self.send_kv(MSG_KEY_START_CASE, msg_value)
+        self.drop_heartbeat_messages = False
+        self.hb_timestamps_us = []
+
+    def cb_device_reset(self, key, value, timestamp):
+        """Keep track of the test case number.
+
+        Also set a new handshake timeout, so when the device gets
+        restarted by the watchdog, the communication will be restored
+        by the __handshake_timer.
+        """
+        self.handshake_timer_cancel()
+        case_num, dev_reset_delay_ms = (int(i, base=16) for i in value.split(','))
+        self.current_case = TestCaseData(case_num, CASE_DATA_PHASE2_OK)
+        self.handshake_timer_start(self.sync_delay + dev_reset_delay_ms / 1000.0)
+
+    def cb_heartbeat(self, key, value, timestamp):
+        """Save the timestamp of a heartbeat message.
+
+        Additionally, keep track of the test case number.
+
+        Also each heartbeat sets a new timeout, so when the device gets
+        restarted by the watchdog, the communication will be restored
+        by the __handshake_timer.
+        """
+        if self.drop_heartbeat_messages:
+            return
+        self.handshake_timer_cancel()
+        case_num, timestamp_us = (int(i, base=16) for i in value.split(','))
+        self.current_case = TestCaseData(case_num, CASE_DATA_INVALID)
+        self.hb_timestamps_us.append(timestamp_us)
+        self.handshake_timer_start(
+            seconds=(MAX_HB_PERIOD + self.sync_delay),
+            pre_sync_fun=self.heartbeat_timeout_handler)