test: Add script for running unit tests on device

This is still a prototype, but is functional and useful for verifying
all the unit tests pass on device.

BRANCH=none
BUG=b:151105339
TEST=With dragonclaw v0.2 connected to Segger J-Trace and servo micro:
     ./test/run_device_tests.py
TEST=With dragonclaw v0.2 connected to Segger J-Trace and servo micro:
     ./test/run_device_tests.py -t mpu

Signed-off-by: Tom Hughes <tomhughes@chromium.org>
Change-Id: Iab39c092b6637544ac37ca32a0b0c94274f05868
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/ec/+/2212623
Commit-Queue: Yicheng Li <yichengli@chromium.org>
Reviewed-by: Yicheng Li <yichengli@chromium.org>

1 files changed, 346 insertions, 0 deletions

diff --git a/test/run_device_tests.py b/test/run_device_tests.py
new file mode 100755
index 0000000000..5045009521
--- /dev/null
+++ b/test/run_device_tests.py
@@ -0,0 +1,346 @@
+#!/usr/bin/env python
+
+# Copyright 2020 The Chromium OS Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Runs unit tests on device and displays the results.
+
+This script assumes you have a ~/.servodrc config file with a line that
+corresponds to the board being tested.
+
+See https://chromium.googlesource.com/chromiumos/third_party/hdctools/+/HEAD/docs/servo.md#servodrc
+"""
+import argparse
+import concurrent
+import io
+import logging
+import os
+import re
+import subprocess
+import sys
+import time
+from concurrent.futures.thread import ThreadPoolExecutor
+from enum import Enum
+from pathlib import Path
+import colorama
+
+EC_DIR = Path(os.path.dirname(os.path.realpath(__file__))).parent
+FLASH_SCRIPT = os.path.join(EC_DIR, 'util/flash_jlink.py')
+
+ALL_TESTS_PASSED_REGEX = re.compile(r'Pass!\r\n')
+ALL_TESTS_FAILED_REGEX = re.compile(r'Fail! \(\d+ tests\)\r\n')
+
+SINGLE_CHECK_PASSED_REGEX = re.compile(r'Pass: .*')
+SINGLE_CHECK_FAILED_REGEX = re.compile(r'.*failed:.*')
+
+DATA_ACCESS_VIOLATION_8020000_REGEX = re.compile(
+    r'Data access violation, mfar = 8020000\r\n')
+DATA_ACCESS_VIOLATION_20000000_REGEX = re.compile(
+    r'Data access violation, mfar = 20000000\r\n')
+
+
+class ImageType(Enum):
+    """EC Image type to use for the test."""
+    RO = 1
+    RW = 2
+
+
+class BoardConfig:
+    """Board-specific configuration."""
+    def __init__(self, test_list, servo_uart_name, servo_power_enable):
+        self.test_list = test_list
+        self.servo_uart_name = servo_uart_name
+        self.servo_power_enable = servo_power_enable
+
+
+class TestConfig:
+    """Configuration for a given test."""
+
+    def __init__(self, name, image_to_use=ImageType.RW, finish_regexes=None,
+                 toggle_power=False):
+        if finish_regexes is None:
+            finish_regexes = [ALL_TESTS_PASSED_REGEX, ALL_TESTS_FAILED_REGEX]
+
+        self.name = name
+        self.image_to_use = image_to_use
+        self.finish_regexes = finish_regexes
+        self.toggle_power = toggle_power
+        self.logs = []
+        self.passed = False
+        self.num_fails = 0
+        self.num_passes = 0
+
+
+# All possible tests.
+ALL_TESTS = {
+    'aes':
+        TestConfig(name='aes'),
+    'crc32':
+        TestConfig(name='crc32'),
+    'flash_physical':
+        TestConfig(name='flash_physical', image_to_use=ImageType.RO,
+                   toggle_power=True),
+    'flash_write_protect':
+        TestConfig(name='flash_write_protect', image_to_use=ImageType.RO,
+                   toggle_power=True),
+    'mpu':
+        TestConfig(name='mpu',
+                   finish_regexes=[DATA_ACCESS_VIOLATION_20000000_REGEX]),
+    'mutex':
+        TestConfig(name='mutex'),
+    'pingpong':
+        TestConfig(name='pingpong'),
+    'rollback':
+        TestConfig(name='rollback', finish_regexes=[
+            DATA_ACCESS_VIOLATION_8020000_REGEX]),
+    'rollback_entropy':
+        TestConfig(name='rollback_entropy', image_to_use=ImageType.RO),
+    'rtc':
+        TestConfig(name='rtc'),
+    'sha256':
+        TestConfig(name='sha256'),
+    'sha256_unrolled':
+        TestConfig(name='sha256_unrolled'),
+    'stm32f_rtc':
+        TestConfig(name='stm32f_rtc'),
+}
+
+BLOONCHIPPER_CONFIG = BoardConfig(
+    test_list=ALL_TESTS.values(),
+    servo_uart_name='raw_fpmcu_uart_pty',
+    servo_power_enable='spi1_vref'
+)
+DARTMONKEY_CONFIG = BLOONCHIPPER_CONFIG
+
+BOARD_CONFIGS = {
+    'bloonchipper': BLOONCHIPPER_CONFIG,
+    'dartmonkey': DARTMONKEY_CONFIG,
+}
+
+
+def get_console(board_name, board_config):
+    """Get the name of the console for a given board."""
+    cmd = [
+        'dut-control',
+        '-n', board_name,
+        board_config.servo_uart_name,
+    ]
+    logging.debug('Running command: "%s"', ' '.join(cmd))
+
+    with subprocess.Popen(cmd, stdout=subprocess.PIPE) as proc:
+        for line in io.TextIOWrapper(proc.stdout):
+            logging.debug(line)
+            pty = line.split(':')
+            if len(pty) == 2 and pty[0] == board_config.servo_uart_name:
+                return pty[1].strip()
+
+    return None
+
+
+def power(board_name, board_config, on):
+    """Turn power to board on/off."""
+    if on:
+        state = 'pp3300'
+    else:
+        state = 'off'
+
+    cmd = [
+        'dut-control',
+        '-n', board_name,
+        board_config.servo_power_enable + ':' + state,
+    ]
+    logging.debug('Running command: "%s"', ' '.join(cmd))
+    subprocess.run(cmd).check_returncode()
+
+
+def build(test_name, board_name):
+    """Build specified test for specified board."""
+    cmd = [
+        'make',
+        'BOARD=' + board_name,
+        'test-' + test_name,
+        '-j',
+    ]
+
+    logging.debug('Running command: "%s"', ' '.join(cmd))
+    subprocess.run(cmd).check_returncode()
+
+
+def flash(test_name, board):
+    """Flash specified test to specified board."""
+    logging.info("Flashing test")
+
+    # TODO(b/151105339): Support ./util/flash_ec as well. It's slower, but only
+    # requires servo micro.
+    cmd = [
+        FLASH_SCRIPT,
+        '--board', board,
+        '--image', os.path.join(EC_DIR, 'build', board, test_name,
+                                test_name + '.bin'),
+    ]
+    logging.debug('Running command: "%s"', ' '.join(cmd))
+    subprocess.run(cmd).check_returncode()
+
+
+def readline(executor, f, timeout_secs):
+    """Read a line with timeout."""
+    a = executor.submit(f.readline)
+    try:
+        return a.result(timeout_secs)
+    except concurrent.futures.TimeoutError:
+        return None
+
+
+def readlines_until_timeout(executor, f, timeout_secs):
+    """Continuously read lines for timeout_secs."""
+    lines = []
+    while True:
+        line = readline(executor, f, timeout_secs)
+        if not line:
+            return lines
+        lines.append(line)
+
+
+def run_test(test, console, executor, timeout_secs=10):
+    """Run specified test."""
+    start = time.time()
+    with open(console, "wb+", buffering=0) as c:
+        # Wait for boot to finish
+        time.sleep(1)
+        c.write('\n'.encode())
+        if test.image_to_use == ImageType.RO:
+            c.write('reboot ro\n'.encode())
+            time.sleep(1)
+        c.write('runtest\n'.encode())
+
+        while True:
+            c.flush()
+            line = readline(executor, c, 1)
+            if not line:
+                now = time.time()
+                if now - start > timeout_secs:
+                    logging.debug("Test timed out")
+                    return False
+                continue
+
+            logging.debug(line)
+            test.logs.append(line)
+            # Look for test_print_result() output (success or failure)
+            try:
+                line_str = line.decode()
+
+                if SINGLE_CHECK_PASSED_REGEX.match(line_str):
+                    test.num_passes += 1
+
+                if SINGLE_CHECK_FAILED_REGEX.match(line_str):
+                    test.num_fails += 1
+
+                if ALL_TESTS_FAILED_REGEX.match(line_str):
+                    test.num_fails += 1
+
+                for r in test.finish_regexes:
+                    if r.match(line_str):
+                        # flush read the remaining
+                        lines = readlines_until_timeout(executor, c, 1)
+                        logging.debug(lines)
+                        test.logs.append(lines)
+                        return test.num_fails == 0
+
+            except UnicodeDecodeError:
+                # Sometimes we get non-unicode from the console (e.g., when the
+                # board reboots.) Not much we can do in this case, so we'll just
+                # ignore it.
+                pass
+
+
+def get_test_list(config, test_args):
+    """Get a list of tests to run."""
+    if test_args == 'all':
+        return config.test_list
+
+    test_list = []
+    for t in test_args:
+        logging.debug('test: %s', t)
+        config = ALL_TESTS.get(t)
+        if config is None:
+            logging.error('Unable to find test config for "%s"', t)
+            sys.exit(1)
+        test_list.append(config)
+
+    return test_list
+
+
+def main():
+    parser = argparse.ArgumentParser()
+
+    default_board = 'bloonchipper'
+    parser.add_argument(
+        '--board', '-b',
+        help='Board (default: ' + default_board + ')',
+        default=default_board)
+
+    default_tests = 'all'
+    parser.add_argument(
+        '--tests', '-t',
+        nargs='+',
+        help='Tests (default: ' + default_tests + ')',
+        default=default_tests)
+
+    log_level_choices = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
+    parser.add_argument(
+        '--log_level', '-l',
+        choices=log_level_choices,
+        default='DEBUG'
+    )
+
+    args = parser.parse_args()
+    logging.basicConfig(level=args.log_level)
+
+    if args.board not in BOARD_CONFIGS:
+        logging.error('Unable to find a config for board: "%s"', args.board)
+        sys.exit(1)
+
+    board_config = BOARD_CONFIGS[args.board]
+
+    e = ThreadPoolExecutor(max_workers=1)
+
+    test_list = get_test_list(board_config, args.tests)
+    logging.debug('Running tests: %s', [t.name for t in test_list])
+
+    for test in test_list:
+        # build test binary
+        build(test.name, args.board)
+
+        # flash test binary
+        flash(test.name, args.board)
+
+        if test.toggle_power:
+            power(args.board, board_config, on=False)
+            time.sleep(1)
+            power(args.board, board_config, on=True)
+
+        # run the test
+        logging.info('Running test: "%s"', test.name)
+        console = get_console(args.board, board_config)
+        test.passed = run_test(test, console, executor=e)
+
+    colorama.init()
+    exit_code = 0
+    for test in test_list:
+        # print results
+        print('Test "' + test.name + '": ', end='')
+        if test.passed:
+            print(colorama.Fore.GREEN + 'PASSED')
+        else:
+            print(colorama.Fore.RED + 'FAILED')
+            exit_code = 1
+
+        print(colorama.Style.RESET_ALL)
+
+    e.shutdown(wait=False)
+    sys.exit(exit_code)
+
+
+if __name__ == '__main__':
+  sys.exit(main())