add new test for serial

Testing serial is quite useful especially since we've had some issues
with RS232 and RS485 already, so let's add a Python script that is
nicely configurable to test many different scenarios.

Signed-off-by: Quentin Schulz <quentin.schulz@theobroma-systems.com>

3 files changed, 198 insertions, 0 deletions

diff --git a/testing/serial/README.md b/testing/serial/README.md
new file mode 100644
index 0000000..50d67c0
--- /dev/null
+++ b/testing/serial/README.md
@@ -0,0 +1,10 @@
+# Test UART controllers, transceivers, adapters and connectors
+
+This is a small autonomous test script written in Python to test UART.
+
+One needs to setup a "loop" on the same device. In other words, both ends of the
+UART communication needs to be connected to the same device under test.
+
+## Dependencies
+
+- pyserial (python3-pyserial on Fedora; python3-serial on Debian-based systems)
diff --git a/testing/serial/requirements.txt b/testing/serial/requirements.txt
new file mode 100644
index 0000000..24b4c2f
--- /dev/null
+++ b/testing/serial/requirements.txt
@@ -0,0 +1 @@
+pyserial>=3.0
diff --git a/testing/serial/test-serial.py b/testing/serial/test-serial.py
new file mode 100755
index 0000000..5f0f66b
--- /dev/null
+++ b/testing/serial/test-serial.py
@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+# Test UART controllers, transceivers, adapters and connectors
+#
+# This is a small autonomous test script to test UART.
+# Both ends of the UART communication needs to be connected to the same device.
+
+import argparse
+import math
+import random
+import serial
+import serial.rs485
+import string
+import sys
+
+
+def randomword(length):
+    letters = string.ascii_lowercase
+    return "".join(random.choice(letters) for i in range(length))
+
+
+def transfer(tx, rx, size):
+    # tx.reset_output_buffer()
+    # rx.reset_input_buffer()
+    rand = randomword(size)
+    tx.write(rand.encode("ascii"))
+    # tx.flush()
+
+    recv = rx.read(size).decode("ascii", "ignore")
+
+    if rand == recv:
+        return True
+
+    print()
+    print(f"ERROR: sent from {tx.name}")
+    print(f"{rand}")
+    print(f"and received on {rx.name}")
+    print(f"{recv}")
+
+    return False
+
+
+def update_serial_settings(tx, rx, baudrate, size):
+    rx.baudrate = tx.baudrate = baudrate
+    # Increase timeout based on baudrate and transfer size while giving room for
+    # "life" (TM) to happen by adding a second to it.
+    timeout = math.ceil(size / baudrate) + 1
+    rx.timeout = tx.timeout = timeout
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-b", "--baudrate", type=int, default=115200)
+    parser.add_argument("--rs485", action="store_true")
+    parser.add_argument(
+        "-s",
+        "--size",
+        type=int,
+        default=256,
+        help="number of characters to send for one test",
+    )
+    parser.add_argument(
+        "-F",
+        "--fuzz",
+        action="append",
+        choices=["size", "baudrate"],
+        help="randomly select baudrate and/or size",
+    )
+    parser.add_argument(
+        "--min-baudrate",
+        type=int,
+        default=serial.Serial.BAUDRATES[0],
+        help="if --fuzz is passed, limits the tested baudrates to anything above that baudrate (included)",
+    )
+    parser.add_argument(
+        "--max-baudrate",
+        type=int,
+        default=serial.Serial.BAUDRATES[-1],
+        help="if --fuzz is passed, limits the tested baudrates to anything below that baudrate (included)",
+    )
+    parser.add_argument(
+        "--min-size",
+        type=int,
+        default=1,
+        help="if --fuzz is passed, limits the tested sizes to anything above that size (included) up to --max-size",
+    )
+    parser.add_argument(
+        "--max-size",
+        type=int,
+        default=1024 * 1024,
+        help="if --fuzz is passed, limits the tested sizes to anything below that size (included) down to --min-size",
+    )
+    parser.add_argument(
+        "-n",
+        "--loops",
+        type=int,
+        default=1,
+        help="number of tests/loops to run for; -1 for infinite",
+    )
+    parser.add_argument(
+        "-e", "--exit-on-error", action="store_true", help="exit on first error"
+    )
+    direction = parser.add_mutually_exclusive_group()
+    direction.add_argument(
+        "-R",
+        "--reverse",
+        default=False,
+        action="store_true",
+        help="swap RX and TX roles",
+    )
+    direction.add_argument(
+        "-B",
+        "--bidirectional",
+        default=False,
+        action="store_true",
+        help="each test has RX assume RX and TX roles while TX assumes TX and RX roles respectively",
+    )
+    parser.add_argument("TX", type=str, help="/dev path to TX UART device")
+    parser.add_argument("RX", type=str, help="/dev path to RX UART device")
+
+    args = parser.parse_args()
+
+    if args.rs485:
+        rx = serial.rs485.RS485(args.TX if args.reverse else args.RX)
+        tx = serial.rs485.RS485(args.RX if args.reverse else args.TX)
+        tx.rs485_mode = rx.rs485_mode = serial.rs485.RS485Settings()
+    else:
+        rx = serial.Serial(args.TX if args.reverse else args.RX)
+        tx = serial.Serial(args.RX if args.reverse else args.TX)
+
+    size = args.size
+    baudrate = args.baudrate
+    baudrate_choices = list(
+        filter(
+            lambda baud: args.min_baudrate <= baud <= args.max_baudrate, tx.BAUDRATES
+        )
+    )
+    update_serial_settings(tx, rx, baudrate, size)
+
+    tx_errors = 0
+    rx_errors = 0
+    loop = 0
+
+    try:
+        while loop != args.loops:
+            loop += 1
+            if args.fuzz:
+                if "baudrate" in args.fuzz:
+                    baudrate = random.choice(baudrate_choices)
+                if "size" in args.fuzz:
+                    size = random.randint(args.min_size, args.max_size)
+                update_serial_settings(tx, rx, baudrate, size)
+
+            print(
+                f"Loop {loop} {tx.name}->{rx.name} {baudrate}bauds size {size}...",
+                end="",
+            )
+            sys.stdout.flush()
+            ret = transfer(tx, rx, args.size)
+            print("success" if ret else "fail")
+            tx_errors += 0 if ret else 1
+            if args.exit_on_error and not ret:
+                sys.exit(1)
+            if args.bidirectional:
+                print(
+                    f"Loop {loop} {rx.name}->{tx.name} {baudrate}bauds size {size}...",
+                    end="",
+                )
+                sys.stdout.flush()
+                ret = transfer(rx, tx, args.size)
+                print("success" if ret else "fail")
+                rx_errors += 0 if ret else 1
+                if args.exit_on_error and not ret:
+                    sys.exit(1)
+
+            if tx_errors:
+                print(f"{tx.name}->{rx.name}: {tx_errors}/{loop} errors")
+            if rx_errors:
+                print(f"{rx.name}->{tx.name}: {rx_errors}/{loop} errors")
+    except KeyboardInterrupt:
+        print("")
+        print("Normal operation interrupted, stopping...")
+
+    print("Summary:")
+    print(f"{tx.name}->{rx.name}: {tx_errors}/{loop} errors")
+    if args.bidirectional:
+        print(f"{rx.name}->{tx.name}: {rx_errors}/{loop} errors")
+    sys.exit(tx_errors or rx_errors)