summaryrefslogtreecommitdiff
path: root/util/uart_stress_tester.py
diff options
context:
space:
mode:
authorJack Rosenthal <jrosenth@chromium.org>2021-11-04 12:11:58 -0600
committerCommit Bot <commit-bot@chromium.org>2021-11-05 04:22:34 +0000
commit252457d4b21f46889eebad61d4c0a65331919cec (patch)
tree01856c4d31d710b20e85a74c8d7b5836e35c3b98 /util/uart_stress_tester.py
parent08f5a1e6fc2c9467230444ac9b582dcf4d9f0068 (diff)
downloadchrome-ec-release-R101-14588.B-ish.tar.gz
In the interest of making long-term branch maintenance incur as little technical debt on us as possible, we should not maintain any files on the branch we are not actually using. This has the added effect of making it extremely clear when merging CLs from the main branch when changes have the possibility to affect us. The follow-on CL adds a convenience script to actually pull updates from the main branch and generate a CL for the update. BUG=b:204206272 BRANCH=ish TEST=make BOARD=arcada_ish && make BOARD=drallion_ish Signed-off-by: Jack Rosenthal <jrosenth@chromium.org> Change-Id: I17e4694c38219b5a0823e0a3e55a28d1348f4b18 Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/ec/+/3262038 Reviewed-by: Jett Rink <jettrink@chromium.org> Reviewed-by: Tom Hughes <tomhughes@chromium.org>
Diffstat (limited to 'util/uart_stress_tester.py')
-rwxr-xr-xutil/uart_stress_tester.py562
1 files changed, 0 insertions, 562 deletions
diff --git a/util/uart_stress_tester.py b/util/uart_stress_tester.py
deleted file mode 100755
index b3db60060e..0000000000
--- a/util/uart_stress_tester.py
+++ /dev/null
@@ -1,562 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-# Copyright 2019 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.
-#
-# Ignore indention messages, since legacy scripts use 2 spaces instead of 4.
-# pylint: disable=bad-indentation,docstring-section-indent
-# pylint: disable=docstring-trailing-quotes
-
-"""ChromeOS Uart Stress Test
-
-This tester runs the command 'chargen' on EC and/or AP, captures the
-output, and compares it against the expected output to check any characters
-lost.
-
-Prerequisite:
- (1) This test needs PySerial. Please check if it is available before test.
- Can be installed by 'pip install pyserial'
- (2) If servod is running, turn uart_timestamp off before running this test.
- e.g. dut-control cr50_uart_timestamp:off
-"""
-
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
-import argparse
-import atexit
-import logging
-import os
-import stat
-import sys
-import threading
-import time
-
-import serial
-
-BAUDRATE = 115200 # Default baudrate setting for UART port
-CROS_USERNAME = 'root' # Account name to login to ChromeOS
-CROS_PASSWORD = 'test0000' # Password to login to ChromeOS
-CHARGEN_TXT = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
- # The result of 'chargen 62 62'
-CHARGEN_TXT_LEN = len(CHARGEN_TXT)
-CR = '\r' # Carriage Return
-LF = '\n' # Line Feed
-CRLF = CR + LF
-FLAG_FILENAME = '/tmp/chargen_testing'
-TPM_CMD = ('trunks_client --key_create --rsa=2048 --usage=sign'
- ' --key_blob=/tmp/blob &> /dev/null')
- # A ChromeOS TPM command for the cr50 stress
- # purpose.
-CR50_LOAD_GEN_CMD = ('while [[ -f %s ]]; do %s; done &'
- % (FLAG_FILENAME, TPM_CMD))
- # A command line to run TPM_CMD in background
- # infinitely.
-
-
-class ChargenTestError(Exception):
- """Exception for Uart Stress Test Error"""
- pass
-
-
-class UartSerial(object):
- """Test Object for a single UART serial device
-
- Attributes:
- UART_DEV_PROFILES
- char_loss_occurrences: Number that character loss happens
- cleanup_cli: Command list to perform before the test exits
- cr50_workload: True if cr50 should be stressed, or False otherwise
- usb_output: True if output should be generated to USB channel
- dev_prof: Dictionary of device profile
- duration: Time to keep chargen running
- eol: Characters to add at the end of input
- logger: object that store the log
- num_ch_exp: Expected number of characters in output
- num_ch_cap: Number of captured characters in output
- test_cli: Command list to run for chargen test
- test_thread: Thread object that captures the UART output
- serial: serial.Serial object
- """
- UART_DEV_PROFILES = (
- # Kernel
- {
- 'prompt':'localhost login:',
- 'device_type':'AP',
- 'prepare_cmd':[
- CROS_USERNAME, # Login
- CROS_PASSWORD, # Password
- 'dmesg -D', # Disable console message
- 'touch ' + FLAG_FILENAME, # Create a temp file
- ],
- 'cleanup_cmd':[
- 'rm -f ' + FLAG_FILENAME, # Remove the temp file
- 'dmesg -E', # Enable console message
- 'logout', # Logout
- ],
- 'end_of_input':LF,
- },
- # EC
- {
- 'prompt':'> ',
- 'device_type':'EC',
- 'prepare_cmd':[
- 'chan save',
- 'chan 0' # Disable console message
- ],
- 'cleanup_cmd':['', 'chan restore'],
- 'end_of_input':CRLF,
- },
- )
-
- def __init__(self, port, duration, timeout=1,
- baudrate=BAUDRATE, cr50_workload=False,
- usb_output=False):
- """Initialize UartSerial
-
- Args:
- port: UART device path. e.g. /dev/ttyUSB0
- duration: Time to test, in seconds
- timeout: Read timeout value.
- baudrate: Baud rate such as 9600 or 115200.
- cr50_workload: True if a workload should be generated on cr50
- usb_output: True if a workload should be generated to USB channel
- """
-
- # Initialize serial object
- self.serial = serial.Serial()
- self.serial.port = port
- self.serial.timeout = timeout
- self.serial.baudrate = baudrate
-
- self.duration = duration
- self.cr50_workload = cr50_workload
- self.usb_output = usb_output
-
- self.logger = logging.getLogger(type(self).__name__ + '| ' + port)
- self.test_thread = threading.Thread(target=self.stress_test_thread)
-
- self.dev_prof = {}
- self.cleanup_cli = []
- self.test_cli = []
- self.eol = CRLF
- self.num_ch_exp = 0
- self.num_ch_cap = 0
- self.char_loss_occurrences = 0
- atexit.register(self.cleanup)
-
- def run_command(self, command_lines, delay=0):
- """Run command(s) at UART prompt
-
- Args:
- command_lines: list of commands to run.
- delay: delay after a command in second
- """
- for cli in command_lines:
- self.logger.debug('run %r', cli)
-
- self.serial.write((cli + self.eol).encode())
- self.serial.flush()
- if delay:
- time.sleep(delay)
-
- def cleanup(self):
- """Before termination, clean up the UART device."""
- self.logger.debug('Closing...')
-
- self.serial.open()
- self.run_command(self.cleanup_cli) # Run cleanup commands
- self.serial.close()
-
- self.logger.debug('Cleanup done')
-
- def get_output(self):
- """Capture the UART output
-
- Args:
- stop_char: Read output buffer until it reads stop_char.
-
- Returns:
- text from UART output.
- """
- if self.serial.inWaiting() == 0:
- time.sleep(1)
-
- return self.serial.read(self.serial.inWaiting()).decode()
-
- def prepare(self):
- """Prepare the test:
-
- Identify the type of UART device (EC or Kernel?), then
- decide what kind of commands to use to generate stress loads.
-
- Raises:
- ChargenTestError if UART source can't be identified.
- """
- try:
- self.logger.info('Preparing...')
-
- self.serial.open()
-
- # Prepare the device for test
- self.serial.flushInput()
- self.serial.flushOutput()
-
- self.get_output() # drain data
-
- # Give a couple of line feeds, and capture the prompt text
- self.run_command(['', ''])
- prompt_txt = self.get_output()
-
- # Detect the device source: EC or AP?
- # Detect if the device is AP or EC console based on the captured.
- for dev_prof in self.UART_DEV_PROFILES:
- if dev_prof['prompt'] in prompt_txt:
- self.dev_prof = dev_prof
- break
- else:
- # No prompt patterns were found. UART seems not responding or in
- # an undesirable status.
- if prompt_txt:
- raise ChargenTestError('%s: Got an unknown prompt text: %s\n'
- 'Check manually whether %s is available.' %
- (self.serial.port, prompt_txt,
- self.serial.port))
- else:
- raise ChargenTestError('%s: Got no input. Close any other connections'
- ' to this port, and try it again.' %
- self.serial.port)
-
- self.logger.info('Detected as %s UART', self.dev_prof['device_type'])
- # Log displays the UART type (AP|EC) instead of device filename.
- self.logger = logging.getLogger(type(self).__name__ + '| ' +
- self.dev_prof['device_type'])
-
- # Either login to AP or run some commands to prepare the device
- # for test
- self.eol = self.dev_prof['end_of_input']
- self.run_command(self.dev_prof['prepare_cmd'], delay=2)
- self.cleanup_cli += self.dev_prof['cleanup_cmd']
-
- # 'chargen' of AP does not have option for USB output.
- # Force it work on UART.
- if self.dev_prof['device_type'] == 'AP':
- self.usb_output = False
-
- # Check whether the command 'chargen' is available in the device.
- # 'chargen 1 4' is supposed to print '0000'
- self.get_output() # drain data
-
- chargen_cmd = 'chargen 1 4'
- if self.usb_output:
- chargen_cmd += ' usb'
- self.run_command([chargen_cmd])
- tmp_txt = self.get_output()
-
- # Check whether chargen command is available.
- if '0000' not in tmp_txt:
- raise ChargenTestError('%s: Chargen got an unexpected result: %s' %
- (self.dev_prof['device_type'], tmp_txt))
-
- self.num_ch_exp = int(self.serial.baudrate * self.duration / 10)
- chargen_cmd = 'chargen ' + str(CHARGEN_TXT_LEN) + ' ' + \
- str(self.num_ch_exp)
- if self.usb_output:
- chargen_cmd += ' usb'
- self.test_cli = [chargen_cmd]
-
- self.logger.info('Ready to test')
- finally:
- self.serial.close()
-
- def stress_test_thread(self):
- """Test thread
-
- Raises:
- ChargenTestError: if broken character is found.
- """
- try:
- self.serial.open()
- self.serial.flushInput()
- self.serial.flushOutput()
-
- # Run TPM command in background to burden cr50.
- if self.dev_prof['device_type'] == 'AP' and self.cr50_workload:
- self.run_command([CR50_LOAD_GEN_CMD])
- self.logger.debug('run TPM job while %s exists', FLAG_FILENAME)
-
- # Run the command 'chargen', one time
- self.run_command(['']) # Give a line feed
- self.get_output() # Drain the output
- self.run_command(self.test_cli)
- self.serial.readline() # Drain the echoed command line.
-
- err_msg = '%s: Expected %r but got %s after %d char received'
-
- # Keep capturing the output until the test timer is expired.
- self.num_ch_cap = 0
- self.char_loss_occurrences = 0
- data_starve_count = 0
-
- total_num_ch = self.num_ch_exp # Expected number of characters in total
- ch_exp = CHARGEN_TXT[0]
- ch_cap = 'z' # any character value is ok for loop initial condition.
- while self.num_ch_cap < total_num_ch:
- captured = self.get_output()
-
- if captured:
- # There is some output data. Reset the data starvation count.
- data_starve_count = 0
- else:
- data_starve_count += 1
- if data_starve_count > 1:
- # If nothing was captured more than once, then terminate the test.
- self.logger.debug('No more output')
- break
-
- for ch_cap in captured:
- if ch_cap not in CHARGEN_TXT:
- # If it is not alpha-numeric, terminate the test.
- if ch_cap not in CRLF:
- # If it is neither a CR nor LF, then it is an error case.
- self.logger.error('Whole captured characters: %r', captured)
- raise ChargenTestError(err_msg % ('Broken char captured', ch_exp,
- hex(ord(ch_cap)),
- self.num_ch_cap))
-
- # Set the loop termination condition true.
- total_num_ch = self.num_ch_cap
-
- if self.num_ch_cap >= total_num_ch:
- break
-
- if ch_exp != ch_cap:
- # If it is alpha-numeric but not continuous, then some characters
- # are lost.
- self.logger.error(err_msg, 'Char loss detected',
- ch_exp, repr(ch_cap), self.num_ch_cap)
- self.char_loss_occurrences += 1
-
- # Recalculate the expected number of characters to adjust
- # termination condition. The loss might be bigger than this
- # adjustment, but it is okay since it will terminates by either
- # CR/LF detection or by data starvation.
- idx_ch_exp = CHARGEN_TXT.find(ch_exp)
- idx_ch_cap = CHARGEN_TXT.find(ch_cap)
- if idx_ch_cap < idx_ch_exp:
- idx_ch_cap += len(CHARGEN_TXT)
- total_num_ch -= (idx_ch_cap - idx_ch_exp)
-
- self.num_ch_cap += 1
-
- # Determine What character is expected next?
- ch_exp = CHARGEN_TXT[(CHARGEN_TXT.find(ch_cap) + 1) % CHARGEN_TXT_LEN]
-
- finally:
- self.serial.close()
-
- def start_test(self):
- """Start the test thread"""
- self.logger.info('Test thread starts')
- self.test_thread.start()
-
- def wait_test_done(self):
- """Wait until the test thread get done and join"""
- self.test_thread.join()
- self.logger.info('Test thread is done')
-
- def get_result(self):
- """Display the result
-
- Returns:
- Integer = the number of lost character
-
- Raises:
- ChargenTestError: if the capture is corrupted.
- """
- # If more characters than expected are captured, it means some messages
- # from other than chargen are mixed. Stop processing further.
- if self.num_ch_exp < self.num_ch_cap:
- raise ChargenTestError('%s: UART output is corrupted.' %
- self.dev_prof['device_type'])
-
- # Get the count difference between the expected to the captured
- # as the number of lost character.
- char_lost = self.num_ch_exp - self.num_ch_cap
- self.logger.info('%8d char lost / %10d (%.1f %%)',
- char_lost, self.num_ch_exp,
- char_lost * 100.0 / self.num_ch_exp)
-
- return char_lost, self.num_ch_exp, self.char_loss_occurrences
-
-
-class ChargenTest(object):
- """UART stress tester
-
- Attributes:
- logger: logging object
- serials: Dictionary where key is filename of UART device, and the value is
- UartSerial object
- """
-
- def __init__(self, ports, duration, cr50_workload=False,
- usb_output=False):
- """Initialize UART stress tester
-
- Args:
- ports: List of UART ports to test.
- duration: Time to keep testing in seconds.
- cr50_workload: True if a workload should be generated on cr50
- usb_output: True if a workload should be generated to USB channel
-
- Raises:
- ChargenTestError: if any of ports is not a valid character device.
- """
-
- # Save the arguments
- for port in ports:
- try:
- mode = os.stat(port).st_mode
- except OSError as e:
- raise ChargenTestError(e)
- if not stat.S_ISCHR(mode):
- raise ChargenTestError('%s is not a character device.' % port)
-
- if duration <= 0:
- raise ChargenTestError('Input error: duration is not positive.')
-
- # Initialize logging object
- self.logger = logging.getLogger(type(self).__name__)
-
- # Create an UartSerial object per UART port
- self.serials = {} # UartSerial objects
- for port in ports:
- self.serials[port] = UartSerial(port=port, duration=duration,
- cr50_workload=cr50_workload,
- usb_output=usb_output)
-
- def prepare(self):
- """Prepare the test for each UART port"""
- self.logger.info('Prepare ports for test')
- for _, ser in self.serials.items():
- ser.prepare()
- self.logger.info('Ports are ready to test')
-
- def print_result(self):
- """Display the test result for each UART port
-
- Returns:
- char_lost: Total number of characters lost
- """
- char_lost = 0
- for _, ser in self.serials.items():
- (tmp_lost, _, _) = ser.get_result()
- char_lost += tmp_lost
-
- # If any characters are lost, then test fails.
- msg = 'lost %d character(s) from the test' % char_lost
- if char_lost > 0:
- self.logger.error('FAIL: %s', msg)
- else:
- self.logger.info('PASS: %s', msg)
-
- return char_lost
-
- def run(self):
- """Run the stress test on UART port(s)
-
- Raises:
- ChargenTestError: If any characters are lost.
- """
-
- # Detect UART source type, and decide which command to test.
- self.prepare()
-
- # Run the test on each UART port in thread.
- self.logger.info('Test starts')
- for _, ser in self.serials.items():
- ser.start_test()
-
- # Wait all tests to finish.
- for _, ser in self.serials.items():
- ser.wait_test_done()
-
- # Print the result.
- char_lost = self.print_result()
- if char_lost:
- raise ChargenTestError('Test failed: lost %d character(s)' %
- char_lost)
-
- self.logger.info('Test is done')
-
-def parse_args(cmdline):
- """Parse command line arguments.
-
- Args:
- cmdline: list to be parsed
-
- Returns:
- tuple (options, args) where args is a list of cmdline arguments that the
- parser was unable to match i.e. they're servod controls, not options.
- """
- description = """%(prog)s repeats sending a uart console command
-to each UART device for a given time, and check if output
-has any missing characters.
-
-Examples:
- %(prog)s /dev/ttyUSB2 --time 3600
- %(prog)s /dev/ttyUSB1 /dev/ttyUSB2 --debug
- %(prog)s /dev/ttyUSB1 /dev/ttyUSB2 --cr50
-"""
-
- parser = argparse.ArgumentParser(description=description,
- formatter_class=argparse.RawTextHelpFormatter
- )
- parser.add_argument('port', type=str, nargs='*',
- help='UART device path to test')
- parser.add_argument('-c', '--cr50', action='store_true', default=False,
- help='generate TPM workload on cr50')
- parser.add_argument('-d', '--debug', action='store_true', default=False,
- help='enable debug messages')
- parser.add_argument('-t', '--time', type=int,
- help='Test duration in second', default=300)
- parser.add_argument('-u', '--usb', action='store_true', default=False,
- help='Generate output to USB channel instead')
- return parser.parse_known_args(cmdline)
-
-
-def main():
- """Main function wrapper"""
- try:
- (options, _) = parse_args(sys.argv[1:])
-
- # Set Log format
- log_format = '%(asctime)s %(levelname)-6s | %(name)-25s'
- date_format = '%Y-%m-%d %H:%M:%S'
- if options.debug:
- log_format += ' | %(filename)s:%(lineno)4d:%(funcName)-18s'
- loglevel = logging.DEBUG
- else:
- loglevel = logging.INFO
- log_format += ' | %(message)s'
-
- logging.basicConfig(level=loglevel, format=log_format,
- datefmt=date_format)
-
- # Create a ChargenTest object
- utest = ChargenTest(options.port, options.time,
- cr50_workload=options.cr50,
- usb_output=options.usb)
- utest.run() # Run
-
- except KeyboardInterrupt:
- sys.exit(0)
-
- except ChargenTestError as e:
- logging.error(str(e))
- sys.exit(1)
-
-if __name__ == '__main__':
- main()