util: uart stress tester using 'chargen' command

Uart stress tester runs a 'chargen' UART command on EC and/or AP, and
checks if any characters are lost from UART output.

BUG=b:131340067
BRANCH=None
TEST=ran on Bob and Octopus (Fleex)

$ ./util/uart_stress_tester.py -h
usage: uart_stress_tester.py [-h] [-c] [-d] [-t TIME] [port [port ...]]

uart_stress_tester.py repeats sending a uart console command
to each UART device for a given time, and check if output
has any missing characters.

Examples:
    uart_stress_tester.py /dev/ttyUSB2 --time 3600
    uart_stress_tester.py /dev/ttyUSB1 /dev/ttyUSB2 --debug
    uart_stress_tester.py /dev/ttyUSB1 /dev/ttyUSB2 --cr50

positional arguments:
  port                  UART device path to test

optional arguments:
  -h, --help            show this help message and exit
  -c, --cr50            generate TPM workload on cr50
  -d, --debug           enable debug messages
  -t TIME, --time TIME  Test duration in second

$ ./util/uart_stress_tester.py /dev/ttyUSB1 /dev/ttyUSB2 -t 120
INFO   | UartSerial| EC    |        0 char lost /    1382400 (0.0 %)
INFO   | UartSerial| AP    |        0 char lost /    1382400 (0.0 %)
INFO   | ChargenTest       | PASS: lost 0 character(s) from the test

$ ./util/uart_stress_tester.py /dev/ttyUSB1 /dev/ttyUSB2 -t 120 --cr50
INFO   | UartSerial| EC    |        0 char lost /    1382400 (0.0 %)
INFO   | UartSerial| AP    |        0 char lost /    1382400 (0.0 %)
INFO   | ChargenTest       | PASS: lost 0 character(s) from the test

Change-Id: I713fb0180db3ca5904bd7aae0dd26a4633733d2e
Signed-off-by: Namyoon Woo <namyoon@chromium.org>
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/ec/+/1683011
Reviewed-by: Mary Ruthven <mruthven@chromium.org>

3 files changed, 511 insertions, 353 deletions

diff --git a/util/presubmit_check.sh b/util/presubmit_check.sh
index ed1cbe6071..eeac33c863 100755
--- a/util/presubmit_check.sh
+++ b/util/presubmit_check.sh
@@ -36,8 +36,8 @@ for dir in $unittest_dirs; do
 done
 # Filter out flash_ec since it's not part of any unit tests.
 changed=$(echo "${changed}" | grep -v util/flash_ec)
-# Filter out uart_stress_tester.sh
-changed=$(echo "${changed}" | grep -v util/uart_stress_tester.sh)
+# Filter out uart_stress_tester
+changed=$(echo "${changed}" | grep -v util/uart_stress_tester.py)
 # Filter out this file itself.
 changed=$(echo "${changed}" | grep -v util/presubmit_check.sh)
 # Filter out the OWNERS file.
diff --git a/util/uart_stress_tester.py b/util/uart_stress_tester.py
new file mode 100755
index 0000000000..2dd5e12515
--- /dev/null
+++ b/util/uart_stress_tester.py
@@ -0,0 +1,509 @@
+#!/usr/bin/env python2
+# -*- 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.
+
+"""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 print_function
+from chromite.lib import cros_logging as logging
+
+import argparse
+import atexit
+import serial
+import sys
+import threading
+import time
+
+
+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
+  """
+  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):
+    """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
+
+    Attributes:
+      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
+      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
+    """
+
+    # 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.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)
+      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())
+
+  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']
+
+      # Check whether the command 'chargen' is available in the device.
+      # 'chargen 1 4' is supposed to print '0000'
+      self.get_output()      # drain data
+      self.run_command(['chargen 1 4'])
+      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)
+      self.test_cli = ['chargen %d %d' % (CHARGEN_TXT_LEN, self.num_ch_exp)]
+
+      self.logger.info('Ready to test')
+    finally:
+      self.serial.close()
+
+  def stress_test_thread(self):
+    """Test thread"""
+    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.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(err_msg, 'Broken char captured',
+                                ch_exp, hex(ord(ch_cap)), self.num_ch_cap)
+              self.logger.error('Whole captured characters: %r', captured)
+            # 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:
+    cr50_workload: True if cr50 should be stressed, or False otherwise
+    duration: Time to keep testing in seconds
+    logger: logging object
+    ports: List of Uart device filename
+    serials: Dictionary where key is filename of UART device, and the value is
+             UartSerial object
+  """
+
+  def __init__(self, ports, duration, cr50_workload=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
+    """
+
+    # Save the arguments
+    self.ports = ports
+
+    if duration <= 0:
+      raise ChargenTestError('Input error: duration is not positive.')
+    self.duration = duration
+
+    self.cr50_workload = cr50_workload
+
+    # Initialize logging object
+    self.logger = logging.getLogger(type(self).__name__)
+
+    # Create an UartSerial object per UART port
+    self.serials = {}     # UartSerial objects
+    for port in self.ports:
+      self.serials[port] = UartSerial(port=port, duration=self.duration,
+                                      cr50_workload=self.cr50_workload)
+
+  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"""
+    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)
+
+  def run(self):
+    """Run the stress test on UART port(s)"""
+
+    # 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.
+    self.print_result()
+    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)
+  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, options.cr50)
+    utest.run()    # Run
+
+  except KeyboardInterrupt:
+    sys.exit(0)
+
+  except ChargenTestError as e:
+    print('Error: ', str(e))
+    sys.exit(1)
+
+if __name__ == '__main__':
+  main()
diff --git a/util/uart_stress_tester.sh b/util/uart_stress_tester.sh
deleted file mode 100755
index f22c90c291..0000000000
--- a/util/uart_stress_tester.sh
+++ /dev/null
@@ -1,351 +0,0 @@
-#!/bin/bash
-#
-# 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.
-
-# ChromeOS UART Stress Test
-#
-# This script compares the UART output on a console command against the expected
-# output, and checks if there are any lost characters.
-#
-# Use: Run uart_stress_tester.sh --help.
-#
-# Output: At the end of test, character loss rates are displayed on screen.
-#
-# How it works:
-#          1. Run a console command on each UART, and capture the output for
-#             a base text in comparison.
-#          2. Run a console command on a UART or more multiple times,
-#             and capture the output.
-#          3. Compare the captured output against the base text, and check
-#             if the base pattern is repeated or if the character is lost.
-#          4. Print the result
-#
-# Prerequisite:
-#     Turn off CR50 and EC uart output channels with the console command
-#     'chan 0'
-#     If servod is running, turn uart_timestamp off before running this test.
-#     e.g. dut-control cr50_uart_timestamp:off
-#
-
-SCRIPT_NAME="$(basename "$0")"
-
-# Load chromeOS common bash library.
-. "/mnt/host/source/src/scripts/common.sh" || exit 1
-
-# Loads script libraries.
-. "/usr/share/misc/shflags" || exit 1
-
-# Flags
-DEFINE_string pty "" "List of UART device path(s) to test"
-DEFINE_integer min_char  "40000" "Minimum number of characters to generate."
-
-FLAGS_HELP="usage: ${SCRIPT_NAME} [flags]
-example:
-  ${SCRIPT_NAME} --pty /dev/ttyUSB0 --min_char 100000
-  ${SCRIPT_NAME} --pty=\"/dev/ttyUSB0 /dev/ttyUSB2\""
-
-FLAGS "$@" || exit 1
-eval set -- "${FLAGS_ARGV}"
-
-if [[ $# -gt 0 ]]; then
-	die "invalid arguments: \"$*\""
-fi
-
-set -e
-
-PIDS=()
-TEST_PASS=false
-
-# Trap function on EXIT
-cleanup() {
-	local LINK_LATEST="/tmp/${SCRIPT_NAME}_latest"
-
-	[[ -e ${DIR_TMP} ]] && ln -snf "${DIR_TMP}" "${LINK_LATEST}"
-
-	info "Test files are in ${LINK_LATEST}"
-	info "and also in ${DIR_TMP}."
-	${TEST_PASS} && info "PASS" || error "FAIL"
-
-	# Kill any background processes.
-	[[ ${#PIDS[@]} -gt 0 ]] && kill -KILL ${PIDS[@]} 2> /dev/null
-	wait
-}
-trap cleanup EXIT
-
-#######################################
-# Calculate the number of characters.
-# Arguments:
-#   $1: Input text file
-# Returns:
-#   The number of characters from the input file
-#######################################
-get_num_char() {
-	wc -c < "$1"
-}
-
-#######################################
-# Calculate the percentage.
-# Arguments:
-#   $1: Numerator
-#   $2: Denominator
-# Returns:
-#   The percentage $1 over $2
-#######################################
-calc_percent() {
-	bc <<< "scale=1;100.0 * $1 / $2"
-}
-
-#######################################
-# Get time of the day in millisecond.
-#######################################
-get_msecond() {
-	date +%s%3N
-}
-
-#######################################
-# Calculate the character loss rate based on the given test files.
-# Arguments:
-#   $1: Device path
-#   $2: Base sample text file to compare with
-#   $3: Test text file to compare against
-#   $4: Count that a console command repeated to get a test file
-#######################################
-calc_char_loss_rate() {
-	[[ $# -eq 4 ]] || die "${FUNCNAME[0]}: argument error: $*"
-
-	local PTY="$1"
-	local FILE_SPL="$2"
-	local FILE_COMP="$3"
-	local REPEATS=$4
-	local FILE_BASE="${FILE_SPL%.*}_${REPEATS}.cap"
-
-	# Create a base text with the sample output, if not exists.
-	if [[ ! -e ${FILE_BASE} ]]; then
-		local i
-
-		for (( i=1; i<=${REPEATS}; i++ )) do
-			cat "${FILE_SPL}"
-		done > "${FILE_BASE}"
-	fi
-
-	# Count the characters in captured data files, and get the difference
-	# between them.
-	local CH_EXPC=$( get_num_char "${FILE_BASE}" )
-	local CH_RESL=$( get_num_char "${FILE_COMP}" )
-	local CH_LOST=$(( ${CH_EXPC} - ${CH_RESL} ))
-	local STR="${PTY}: ${CH_LOST} lost / ${CH_EXPC}"
-
-	TOTAL_CH_EXPC=$(( ${TOTAL_CH_EXPC} + ${CH_EXPC} ))
-	TOTAL_CH_LOST=$(( ${TOTAL_CH_LOST} + ${CH_LOST} ))
-
-	# Check if test output is not smaller than expected.
-	# If so, it must contain some other pattern.
-	if [[ ${CH_LOST} -eq 0 ]]; then
-		# If the sizes are same each other, then compare the text.
-		if diff --brief "${FILE_BASE}" "${FILE_COMP}"; then
-			info "${STR} : 0 %"
-			return
-		else
-			die "${FILE_COMP} does not match to ${FILE_BASE}"
-		fi
-	elif [[ ${CH_LOST} -gt 0 ]]; then
-		# Calculate the character loss rate.
-		local RATE
-
-		RATE=$( calc_percent ${CH_LOST} ${CH_EXPC} )
-
-		error "${STR} : ${RATE} %"
-	else
-		error "Check console output channels are turned off."
-		error "Check uart_timestamp is off if servod is running."
-		echo
-		die "${FILE_COMP} corrupted: $(( -${CH_LOST} )) more found."
-	fi
-}
-
-#######################################
-# Start to capture UART output. Call this function in background.
-# Arguments:
-#   $1: Device path.
-#   $2: File path to save the capture
-# Returns:
-#   Process ID capturing the UART output in background
-#######################################
-start_capture() {
-	[[ $# -eq 2 ]] || die "${FUNCNAME[0]}: argument error: $*"
-
-	local PTY="$1"
-	local FILE_CAP="$2"
-	local STTY_ARGS=( "cs8" "ignbrk" "igncr" "noflsh" "-brkint" "-clocal"
-			"-echo" "-echoe" "-echok" "-echoctl" "-echoke"
-			"-icanon" "-icrnl" "-iexten" "-imaxbel" "-isig" "-ixon"
-			"-onlcr" "-opost" )
-
-	stty -F "${PTY}" "${STTY_ARGS[@]}" || die "stty failed: ${STTY_ARGS[*]}"
-
-	# Drain output
-	cat "${PTY}" &>/dev/null &
-
-	local PID=$!
-
-	echo "" > "${PTY}"
-	sleep 2
-	kill ${PID} &>/dev/null
-	wait
-
-	# Start to capture
-	cat "${PTY}" > "${FILE_CAP}" 2>/dev/null &
-	echo $!
-}
-
-#######################################
-# Run a UART stress test on target device(s).
-# Arguments:
-#   $1: Number of times to run a console command
-#######################################
-stress_test() {
-	# Check the number of arguments.
-	[[ $# -gt 1 ]] || die "${FUNCNAME[0]}: wrong number of arguments: $*"
-
-	local ITER=$1
-	shift
-	local TEST_PTYS=( "$@" )
-	local pd
-	local i
-
-	# Start to capture.
-	for pd in "${TEST_PTYS[@]}"; do
-		FILE_RES["${pd}"]="${DIR_TMP}/$(basename ${pd})_res_${ITER}.cap"
-		PIDS+=( $( start_capture "${pd}" "${FILE_RES["${pd}"]}" ) )
-	done
-
-	TS_START=$( get_msecond )
-
-	# Generate traffic.
-	for (( i=1; i<=${ITER}; i++ )) do
-		for pd in "${TEST_PTYS[@]}"; do
-			echo "${CONSOLE_CMDS["${pd}"]}" > "${pd}"
-		done
-
-		(( i % 10 == 0 )) || continue
-
-		echo -n "."
-		sleep 2
-	done
-	DURATION=$(( $(get_msecond) - TS_START ))
-	echo
-
-	# Stop capturing.
-	sleep 5
-	kill ${PIDS[@]} &>/dev/null
-	wait
-	PIDS=()
-}
-
-MIN_CHAR_SMPL=99999999
-#######################################
-# Choose a console command for sampling, and get a sample output.
-# Global Variables:
-#       FILE_SAMPLE
-#       CONSOLE_CMDS
-# Arguments:
-#       @: Device paths
-#######################################
-get_sample_txt() {
-	local FILE_CAP
-	local PID
-	local NUM_CH
-	local CMDS=( "" "help" )
-	local pd
-	local cmd
-
-	for pd in "$@"; do
-		FILE_CAP="${DIR_TMP}/$(basename ${pd})_sample.cap"
-
-		for cmd in "${CMDS[@]}"; do
-			# Start to capture
-			PID=$( start_capture "${pd}" "${FILE_CAP}" )
-			if [[ -n ${cmd} ]]; then
-				# Since it just started to capture, it might
-				# lose a few beginning bytes from echoed input
-				# command. Let's put some space to prevent it.
-				# Exception: AP uart regards "" with space as
-				# an attempt to login.
-				echo -n "    " > "${pd}"
-			fi
-
-			echo "${cmd}" > "${pd}"
-
-			# Stop capturing
-			sleep 1
-			kill ${PID} &>/dev/null
-			wait
-
-			if [[ -n ${cmd} ]]; then
-				# Remove any spaces that were attached
-				# at the beginning of this for loop statement.
-				# It should apply to the first line only.
-				sed "1s/^ *${cmd}$/${cmd}/" ${FILE_CAP} -i
-			fi
-
-			# Calculate the number of characters from the captured.
-			NUM_CH=$( get_num_char "${FILE_CAP}" )
-
-			if [[ ${NUM_CH} -gt 50 ]]; then
-				CONSOLE_CMDS["${pd}"]="${cmd}"
-				break
-			fi
-		done
-
-		[[ ${NUM_CH} -gt 50 ]] || die "${pd} does not seem to respond"
-
-		if [[ ${NUM_CH} -lt ${MIN_CHAR_SMPL} ]]; then
-			MIN_CHAR_SMPL=${NUM_CH}
-		fi
-
-		FILE_SAMPLE["${pd}"]="${FILE_CAP}"
-	done
-}
-
-info "ChromeOS UART stress test starts."
-
-declare -A CONSOLE_CMDS
-declare -A FILE_SAMPLE
-declare -A FILE_RES
-TOTAL_CH_LOST=0
-TOTAL_CH_EXPC=0
-
-# Check whether the given devices are available.
-read -a PTYS <<< "${FLAGS_pty}"
-
-[[ ${#PTYS[@]} -gt 0 ]] || \
-	die "Flag '--pty' value, ${FLAGS_pty} is not correct."
-
-for pd in "${PTYS[@]}"; do
-	[[ -e ${pd} ]] || die "Device '${pd}' does not exist."
-	if lsof -V "${pd}" &>/dev/null; then
-		die "${pd} is already opened"
-	fi
-done
-
-DIR_TMP="$( mktemp -d --suffix=.${SCRIPT_NAME} )"
-
-# Get sample output as base for comparison.
-get_sample_txt "${PTYS[@]}"
-
-# Calculate the iteration to run console command for traffic.
-REPEATS=$(( (${FLAGS_min_char} + ${MIN_CHAR_SMPL} - 1) / ${MIN_CHAR_SMPL} ))
-
-# Start the stress test
-info "UART devices: ${PTYS[*]}"
-stress_test ${REPEATS} "${PTYS[@]}"
-
-# Calculate average rate calculation.
-for pd in "${PTYS[@]}"; do
-	calc_char_loss_rate "${pd}" "${FILE_SAMPLE["${pd}"]}" \
-			"${FILE_RES["${pd}"]}" ${REPEATS}
-done
-
-[[ ${TOTAL_CH_LOST} -eq 0 ]] && TEST_PASS=true