fpga robots: Add a library to run robots on FPGA
This includes adding a shell script to make running of robots on
FPGA possible as well.
Unfortunately, the only way to run these robots against the FPGAs
is to use this shell script, because Robot does not allow for
dynamic changes to test suites in the Settings section of a suite.
Change-Id: I96acb0ae1b475e1ecc204c96627aba95efa7acc8
diff --git a/FPGALibrary.py b/FPGALibrary.py
new file mode 100644
index 0000000..841b250
--- /dev/null
+++ b/FPGALibrary.py
@@ -0,0 +1,318 @@
+#!/usr/bin/env python3
+#
+# Copyright 2023 Google LLC
+#
+# 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
+#
+# https://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 robot.api.logger import info
+
+import serial
+
+
+# Ensure this class gets re-instantiated for every test.
+ROBOT_LIBRARY_SCOPE = "TEST"
+
+
+def OpenSerialPort(device, baudrate, timeout, write_timeout) -> serial.Serial:
+ """
+ Helper function to open a pyserial device using the given settings.
+
+ Throws an AssertionError if it could not open the given device.
+ """
+ ser = serial.Serial(port=device,
+ timeout=timeout,
+ write_timeout=write_timeout
+ baudrate=baudrate)
+ if not ser.is_open:
+ raise AssertionError(f"Could not open {device}!")
+ return ser
+
+
+class FPGALibrary:
+ """
+ Class to extend the Robot test framework
+
+ Adds Renode-compatible faked keywords as well as serial input and
+ output for the various tests.
+ """
+
+ # Ensure this class gets re-instantiated for every test.
+ ROBOT_LIBRARY_SCOPE = "TEST"
+
+ def __init__(self, *args, **kwargs) -> None:
+ """
+ Initialize the class library.
+
+ On construction, opens the serial ports for the given FPGA
+ board ID. Throws an AssertionError if it fails.
+
+ Arguments expected in kwargs:
+ board_id: string. Required. The FPGA board ID to test. Used to
+ construct path names to the UARTs.
+ timeout: string. The amount of time in seconds to wait in a read
+ or write to the serial ports.
+ """
+ info(f"FPGALibrary installed ({args}, {kwargs})")
+
+ self.board_id = kwargs['board_id']
+
+ self.timeout = None
+ if kwargs.get('timeout'):
+ self.timeout = float(kwargs['timeout'])
+
+ self.smc_uart = f"/dev/Nexus-CP210-FPGA-UART-{self.board_id}"
+ self.sc_uart = f"/dev/Nexus-FTDI-{self.board_id}-FPGA-UART"
+
+ self.uarts = {}
+ self.default_uart = None
+
+ self.open_serial_ports()
+
+ def _quiesce_input(self, port) -> None:
+ """
+ Drains a given port of all incoming data.
+
+ Because serial has no actual "end state", it's possible for
+ a device to continually send data. The best we can do to sync
+ up with the output is delay a second each time we drain to
+ really wait for the output to stop.
+
+ Essentially, if no character is sent for at least one second,
+ we consider the port to be "quiesced". It's a bad heuristic,
+ but since there is no actual command and control protocol,
+ this is the best we get for synchronizing I/O to a command
+ line.
+
+ Arguments:
+ port: serial.Serial instance. The port to drain of data.
+ """
+ # Loop while there is data available on the port, then delay
+ # for a second to make sure we've actually caught all of the
+ # incoming data.
+ while port.in_waiting > 0:
+ info(f"_quiesce_input: port.read({port.in_waiting})")
+ result += port.read(port.in_waiting)
+ info("_quiesce_input: sleep(1)")
+ time.sleep(1)
+ info(f"_quiesce_input read: [{result.decode()}]")
+
+ def _write_string_to_uart(self, port, s, wait_for_echo=True) -> None:
+ """
+ Writes a given string to the given port.
+
+ Throws AssertionError if the echoed back data doesn't match, or a
+ read timeout occurs.
+
+ Arguments:
+ port: serial.Serial instance. The port to write data to.
+ s: string. The string to write to the port.
+ wait_for_echo: boolean. Whether or not to wait for echoed
+ data back from the port.
+ """
+ # Flush the write buffers and drain inputs -- might not be
+ # needed, but do it anyway to ensure we're synced up.
+ port.flush()
+ self._quiesce_input(port)
+
+ port.write(s.encode('utf-8'))
+ port.flush()
+ info(f"wrote [{s}]")
+
+ if wait_for_echo:
+ result = port.read(len(s)).decode()
+ info(f"read [{result}]")
+
+ # We didn't get the same length string back -- likely caused by a timeout.
+ if len(result) < len(s):
+ raise AssertionError(
+ "Timeout when reading from UART -- did not read " +
+ f"{s}]({len(result)})! " +
+ f"Got: [{result}]({len(result)})")
+
+ if s != result:
+ raise AssertionError(
+ "Write String to UART: Echo back: Did not read " +
+ f"[{s}]({len(result)}) from port! " +
+ f"Got: [{result}]({len(result)})")
+
+ def _write_line_to_uart(self, port, line, wait_for_echo=True) -> None:
+ """
+ Writes the given string with a newline to the given port.
+
+ Throws AssertionError if the echoed back data doesn't match, or
+ a read timeout occurs.
+
+ Arguments:
+ port: serial.Serial instance. The port to write data to.
+ s: string. The string to write to the port.
+ wait_for_echo: boolean. Whether or not to wait for echoed
+ data back from the port.
+ """
+ self._write_string_to_uart(port, line + '\n', wait_for_echo)
+
+ def _wait_for_string_on_uart(self, port, s) -> None:
+ """
+ Waits until the given string is found in the port buffer.
+
+ Throws AssertionError if a timeout occurs, or if the buffer
+ doesn't match up with the string to wait for.
+
+ Arguments:
+ port: serial.Serial instance. The port to wait for data from.
+ s: string. The string to look for on the port.
+ """
+ result = port.read_until(expected=s.encode('utf-8')).decode()
+ info(f"_wait_for_string_on_uart read: [{result}]")
+
+ # Short read likely resulting from a timeout.
+ if len(result) < len(s):
+ raise AssertionError(
+ f"Timeout while reading on UART: Did not find string [{s}]!")
+
+ if not result.endswith(s):
+ raise AssertionError(
+ "Wait for String on UART: " +
+ f"Did not get string [{s}] from port! Got [{result}]")
+
+ def set_timeout(self, timeout) -> None:
+ """
+ Sets the read/write timeouts for serial operations.
+
+ Robot keyword. Can be called as Set Timeout.
+
+ Arguments:
+ timeout: float. The amount of time in seconds to wait for a
+ read or write to complete.
+ """
+ self.timeout = float(timeout)
+ if self.uarts['sc']:
+ self.uarts['sc'].timeout = self.timeout
+ self.uarts['sc'].write_timeout = self.timeout
+ if self.uarts['smc']:
+ self.uarts['smc'].timeout = self.timeout
+ self.uarts['smc'].write_timeout = self.timeout
+
+ def open_serial_ports(self) -> None:
+ """
+ Opens the UART ports to the FPGA and sets the default UART.
+
+ For now, sets the default UART to the SMC side only.
+
+ Robot keyword. Can be called as Open Serial Ports.
+ """
+ self.uarts['sc'] = OpenSerialPort(self.sc_uart,
+ read_timeout=self.timeout,
+ write_timeout=self.timeout,
+ baudrate=115200)
+ self.uarts['smc'] = OpenSerialPort(self.smc_uart,
+ read_timeout=self.timeout,
+ write_timeout=self.timeout,
+ baudrate=115200)
+ self.default_uart = self.uarts['smc']
+
+ def close_serial_ports(self) -> None:
+ """
+ Closes previously opened UARTs to the FPGA.
+
+ Robot keyword. Can be called as Close Serial Ports.
+
+ Throws AssertionError if it was unable to close a port.
+ """
+ for name, port in self.uarts.items():
+ port.close()
+ if port.is_open:
+ raise AssertionError(f"Port {name} did not close.")
+
+ def write_line_to_uart(self, line, **kwargs) -> None:
+ """
+ Writes a given string to the default UART.
+
+ Throws AssertionError if a timeout occurs, or if the echoed back
+ characters don't match the given string.
+
+ Robot keyword. Can be called as Write Line To UART.
+
+ Arguments expected in kwargs:
+ waitForEcho: boolean. Deafults to True. Whether or not to check
+ echoed back characters against the given string.
+ """
+ if kwargs.get('waitForEcho'):
+ self._write_line_to_uart(self.default_uart,
+ line,
+ wait_for_echo=kwargs['waitForEcho'])
+ else:
+ self._write_line_to_uart(self.default_uart, line)
+
+ def wait_for_line_on_uart(self, s, **kwargs) -> None:
+ """
+ Waits for a given string on the default UART.
+
+ This does not actually look for a newline in the output. The
+ name is a misnomer and holdover from Renode terms.
+
+ Throws AssertionError if a timeout occurs, or if the read
+ characters don't match the given string.
+
+ Robot keyword. Used as Wait For Line On UART.
+ """
+ self._wait_for_string_on_uart(self.default_uart, s)
+
+ def wait_for_prompt_on_uart(self, prompt, *args) -> None:
+ """
+ Waits for the given prompt on the default UART.
+
+ Robot keyword. Can be called as Wait For Prompt On UART.
+ """
+ self._wait_for_string_on_uart(self.default_uart, prompt)
+
+ def execute_command(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding command `{args}`")
+
+ def execute_script(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding script `{args}`")
+
+ def set_default_uart_timeout(self, timeout, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding set default uart timeout `{args}`")
+
+ def create_log_tester(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding log tester `{args}`")
+
+ def run_process(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding run process `{args}`")
+
+ def requires(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding requires `{args}`")
+
+ def provides(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding provides `{args}`")
+
+ def start_emulation(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding start emulation `{args}`")
+
+ def create_terminal_tester(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding create terminal tester `{args}`")
+
+ def wait_for_logentry(self, *args) -> None:
+ """Renode-compatible do nothing keyword."""
+ info(f"Eliding wait for logentry `{args}`")
diff --git a/fpga_header.robot b/fpga_header.robot
new file mode 100644
index 0000000..ab1c2c4
--- /dev/null
+++ b/fpga_header.robot
@@ -0,0 +1,20 @@
+# Copyright 2023 Google LLC
+#
+# 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
+#
+# https://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.
+
+*** Settings ***
+# These variables are defined in the individual robot tests this script is
+# prepended to, and otherwise via command line arguments to the robot test
+# framework.
+Library FPGALibrary.py board_id=${FPGA_BOARD_ID} timeout=${FPGA_UART_TIMEOUT}
+
diff --git a/fpga_test.sh b/fpga_test.sh
new file mode 100755
index 0000000..3a3b3ef
--- /dev/null
+++ b/fpga_test.sh
@@ -0,0 +1,106 @@
+#!/bin/bash
+#
+# Copyright 2023 Google LLC
+#
+# 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
+#
+# https://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.
+
+if [[ -z "${ROOTDIR}" ]]; then
+ echo "Source build/setup.sh first"
+ exit 1
+fi
+
+function die {
+ echo "$@" >/dev/stderr
+ exit 1
+}
+
+function usage {
+ cat <<EOF
+Usage: fpga_test.sh [options] <fpga-board-number> <test-suite-filename>
+
+Where [options] is one of:
+
+ --no-echo-check Disables the checks for echoed back characters
+ from the FPGA shell prompt.
+ --timeout NN Sets the length of time to wait for a response
+ on read or write to the UART. Defaults to 60
+ seconds.
+ --robot PATH Overrides the path to the robot test script
+ runner. Defaults to searching the PATH.
+ --help | -h Show this usage information.
+
+fpga-board-number must be a two digit number, prefixed with a leading
+zero if necessary.
+
+test-suite-filename is the path to the test suite to run.
+Ie: sim/tests/shodan_boot.robot.
+
+Note: options are positional -- fpga-board-number and test-suite-filename
+must be the last arguments on the command line, in that order.
+EOF
+ exit 1
+}
+
+function cleanup {
+ rm -f "${TMP_TEST_SUITE}"
+}
+
+ROBOT=$(which robot)
+FPGA_HEADER="${ROOTDIR}/sim/tests/fpga_header.robot"
+
+ARGS=()
+
+if [[ "$1" == "--no-echo-check" ]]; then
+ shift
+ ARGS+=(--variable "WAIT_ECHO:false")
+fi
+
+if [[ "$1" == "--timeout" ]]; then
+ shift
+ ARGS+=(--variable "FPGA_UART_TIMEOUT:$1")
+ shift
+fi
+
+if [[ "$1" == "--robot" ]]; then
+ shift
+ ROBOT="$1"
+ shift
+fi
+
+if [[ "$1" == "--help" ]] || [[ "$1" == "-h" ]]; then
+ usage
+fi
+
+FPGA_BOARD_ID="$(printf '%02d' $1)"
+ARGS+=(--variable "FPGA_BOARD_ID:${FPGA_BOARD_ID}")
+shift
+TEST_SUITE="$1"
+shift
+
+if [[ -z "${TEST_SUITE}" ]]; then
+ echo "No FPGA number or test suite specified." >/dev/stderr
+ echo >/dev/stderr
+ usage
+fi
+
+if [[ ! -z "$1" ]]; then
+ die "Unknown argument $1"
+fi
+
+trap cleanup EXIT
+TMP_TEST_SUITE=$(mktemp /tmp/test_suite.robot.XXXXXX)
+
+cat "${FPGA_HEADER}" "${TEST_SUITE}" > "${TMP_TEST_SUITE}"
+echo "${ROBOT}" "${ARGS[@]}" "${TMP_TEST_SUITE}"
+"${ROBOT}" "${ARGS[@]}" "${TMP_TEST_SUITE}" || exit 1
+
diff --git a/shodan_boot.robot b/shodan_boot.robot
index 31a0d92..af34c45 100644
--- a/shodan_boot.robot
+++ b/shodan_boot.robot
@@ -6,13 +6,15 @@
# sim/tests/test.sh --debug sim/tests/shodan_boot.robot
${RUN_DEBUG} 0
-${WAIT_ECHO} true
-IF ${NO_UART_ECHO} == 1
- ${WAIT_ECHO} false
-END
+# This variable is set to be 0 for renode tests, and should be
+# overridden on CLI to test on the FPGA. Ie:
+# sim/tests/test.sh --fpga 02 sim/tests/shodan_boot.robot
+${FPGA_BOARD_ID} 0
+${WAIT_ECHO} true
${LOG_TIMEOUT} 2
${DEBUG_LOG_TIMEOUT} 10
+${FPGA_UART_TIMEOUT} 60
${ROOTDIR} ${CURDIR}/../..
${SCRIPT} sim/config/shodan.resc
${PROMPT} CANTRIP>
diff --git a/shodan_stress.robot b/shodan_stress.robot
index d42ae08..c91bb2a 100644
--- a/shodan_stress.robot
+++ b/shodan_stress.robot
@@ -5,6 +5,7 @@
${MAX_ITER} 100
${LOG_TIMEOUT} 2
+${FPGA_UART_TIMEOUT} 60
${ROOTDIR} ${CURDIR}/../..
${SCRIPT} sim/config/shodan.resc
${PROMPT} CANTRIP>
@@ -59,7 +60,7 @@
FOR ${iter} IN RANGE ${MAX_ITER}
IF ${{random.randint(0, 2)}} == 0
Write Line to Uart start hello
- Wait For Line On Uart Done, sleeping in WFI loop
+ Wait For Line On Uart Done
Stop App hello
END
diff --git a/test.sh b/test.sh
index df115f6..718f54a 100755
--- a/test.sh
+++ b/test.sh
@@ -1,4 +1,18 @@
#!/bin/bash
+#
+# Copyright 2023 Google LLC
+#
+# 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
+#
+# https://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.
if [[ -z "${ROOTDIR}" ]]; then
echo "Source build/setup.sh first"
@@ -38,7 +52,7 @@
echo "Disable UART input echo check"
shift
ARGS+=(
- --variable "NO_UART_ECHO:1"
+ --variable "WAIT_ECHO:false"
)
fi
