test harness for qemu and pi via ssh
diff --git a/.bazelrc b/.bazelrc index 45cc761..dd92fd3 100644 --- a/.bazelrc +++ b/.bazelrc
@@ -55,18 +55,32 @@ build:k_common --build_tag_filters=-do_not_build,-kernel_doc_test test:k_common --test_tag_filters=-integration,-do_not_build,-do_not_run_test,-kernel_doc_test -# AST1060-EVB physical board (no QEMU, flash directly) -common:k_ast1060_evb --config=k_common +# AST1060-EVB physical board via Raspberry Pi SSH fixture. +# Requires key-based SSH auth: ssh-copy-id <user>@<pi-host> +# Usage: AST1060_EVB_PI_HOST=<pi-hostname> bazel test --config=k_ast1060_evb //... common:k_ast1060_evb --platforms=//target/ast10x0 -common:k_ast1060_evb --//target/ast10x0:uart_boot_header=true +build:k_ast1060_evb --build_tag_filters=-do_not_build,-kernel_doc_test +# Board tests share one physical board; run them one at a time. +test:k_ast1060_evb --local_test_jobs=1 +test:k_ast1060_evb --run_under="//target/ast10x0/harness:test_runner " +# Forward AST1060_EVB_PI_HOST from the user's shell into the test sandbox. +test:k_ast1060_evb --test_env=AST1060_EVB_PI_HOST +# Excludes semihosting tests (qemu_only) which HardFault on hardware without a debugger. +test:k_ast1060_evb --test_tag_filters=-integration,-do_not_build,-do_not_run_test,-kernel_doc_test,-qemu_only # `bazel {build,test,run} --config=virt_ast10x0 //target/ast10x0/...` -# launches the AST10x0 system images under QEMU's ast1030-evb machine -# with semihosting. The :qemu flag retunes SysTick to QEMU's 12 MHz clock. -common:virt_ast10x0 --config=k_common +# launches the AST10x0 system images under QEMU's ast1030-evb machine. +# The :qemu flag retunes SysTick to QEMU's 12 MHz clock. +# Pass/fail is signalled via UART sentinel (TEST_RESULT:PASS/FAIL) rather +# than semihosting, so the same firmware binary runs on physical hardware. common:virt_ast10x0 --platforms=//target/ast10x0 +build:virt_ast10x0 --build_tag_filters=-do_not_build,-kernel_doc_test common:virt_ast10x0 --//target/ast10x0:qemu=true -run:virt_ast10x0 --run_under="@pigweed//pw_kernel/tooling:qemu \ - --cpu cortex-m4 --machine ast1030-evb --semihosting --image " -test:virt_ast10x0 --run_under="@pigweed//pw_kernel/tooling:qemu \ - --cpu cortex-m4 --machine ast1030-evb --semihosting --image " \ No newline at end of file +run:virt_ast10x0 --run_under="//target/ast10x0/harness:qemu_runner \ + --cpu cortex-m4 --machine ast1030-evb --image " +test:virt_ast10x0 --run_under="//target/ast10x0/harness:qemu_runner \ + --cpu cortex-m4 --machine ast1030-evb --image " +# Mirrors k_common's test_tag_filters and adds -hardware (physical-board-only tests). +# Combined into one flag to avoid the Bazel 'expanded from multiple configs' warning +# that fires when --test_tag_filters is set by both k_common and this config. +test:virt_ast10x0 --test_tag_filters=-integration,-do_not_build,-do_not_run_test,-kernel_doc_test,-hardware \ No newline at end of file
diff --git a/.gitignore b/.gitignore index 563e893..e90698a 100644 --- a/.gitignore +++ b/.gitignore
@@ -3,6 +3,7 @@ user.bazelrc /bazel-* /out +target/rust-analyzer/ # Editors .pw_ide/
diff --git a/target/ast10x0/BUILD.bazel b/target/ast10x0/BUILD.bazel index bb7ccaf..0255c5b 100644 --- a/target/ast10x0/BUILD.bazel +++ b/target/ast10x0/BUILD.bazel
@@ -17,10 +17,10 @@ build_setting_default = False, ) -bool_flag( - name = "uart_boot_header", - build_setting_default = False, -) +# bool_flag( +# name = "uart_boot_header", +# build_setting_default = False, +# ) config_setting( name = "qemu_enabled",
diff --git a/target/ast10x0/README.md b/target/ast10x0/README.md index 83dc627..58fb1f8 100644 --- a/target/ast10x0/README.md +++ b/target/ast10x0/README.md
@@ -27,8 +27,10 @@ bazel test --config=virt_ast10x0 //target/ast10x0/... ``` -The `virt_ast10x0` config launches images with Pigweed's QEMU runner using the -`ast1030-evb` machine and semihosting. +The `virt_ast10x0` config launches images under QEMU (`ast1030-evb` machine) +using a local sentinel-based runner. Pass/fail is signalled by the firmware +writing `TEST_RESULT:PASS` or `TEST_RESULT:FAIL` to UART. See +`target/ast10x0/tests/README.md` for details. For more detailed failures:
diff --git a/target/ast10x0/harness/BUILD.bazel b/target/ast10x0/harness/BUILD.bazel index 788642f..5a4c437 100644 --- a/target/ast10x0/harness/BUILD.bazel +++ b/target/ast10x0/harness/BUILD.bazel
@@ -1,21 +1,50 @@ # Licensed under the Apache-2.0 license -# UART test execution tool -# -# This tool requires pyserial to be installed in the system Python: -# pip install pyserial -# -# Usage: -# bazel run //tools/uart_test:uart_test_exec -- [args] -# Or directly: python3 tools/uart_test/uart_test_exec.py [args] +load("@rules_platform//platform_data:defs.bzl", "platform_data") +load("@rules_python//python:py_binary.bzl", "py_binary") -sh_binary( - name = "uart_test_exec", - srcs = ["uart_test_exec_wrapper.sh"], - data = ["uart_test_exec.py"], - tags = ["manual"], +py_binary( + name = "qemu_runner_bin", + srcs = ["qemu_runner.py"], + main = "qemu_runner.py", + # @@pigweed++_repo_rules5+qemu is the canonical label for the qemu repo + # created by pigweed's cipd_repository use_repo_rule (not directly + # visible as @qemu from our module). If this breaks after a pigweed + # upgrade, run: ls $(bazel info output_base)/external/ | grep qemu + deps = [ + "@pigweed//pw_tokenizer/py:detokenize", + "@@pigweed++_repo_rules5+qemu//:qemu-system-arm-runfiles", + "@rules_python//python/runfiles", + ], +) + +platform_data( + name = "qemu_runner", + platform = "@bazel_tools//tools:host_platform", + target = ":qemu_runner_bin", visibility = ["//visibility:public"], ) -# Export the Python script for direct use -exports_files(["uart_test_exec.py"]) +py_binary( + name = "test_runner_bin", + srcs = [ + "test_runner.py", + "pi_test_runner.py", + ], + data = ["evb_config.toml"], + main = "test_runner.py", + deps = ["@pigweed//pw_tokenizer/py:detokenize"], +) + +platform_data( + name = "test_runner", + platform = "@bazel_tools//tools:host_platform", + target = ":test_runner_bin", + visibility = ["//visibility:public"], +) + +exports_files([ + "test_runner.py", + "pi_test_runner.py", + "evb_config.toml", +])
diff --git a/target/ast10x0/harness/evb_config.toml b/target/ast10x0/harness/evb_config.toml new file mode 100644 index 0000000..f0e86d6 --- /dev/null +++ b/target/ast10x0/harness/evb_config.toml
@@ -0,0 +1,7 @@ +[gpio] +srst_pin = 23 +fwspick_pin = 18 + +[uart] +serial_port = "/dev/ttyUSB0" +baudrate = 115200
diff --git a/target/ast10x0/harness/pi_test_runner.py b/target/ast10x0/harness/pi_test_runner.py new file mode 100644 index 0000000..8f96bfc --- /dev/null +++ b/target/ast10x0/harness/pi_test_runner.py
@@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +# Licensed under the Apache-2.0 license +# SPDX-License-Identifier: Apache-2.0 +""" +AST1060 EVB hardware interaction layer. + +Handles GPIO reset sequences, firmware upload via UART bootloader, and raw +UART byte streaming. All configuration is received as CLI arguments from +test_runner.py. Raw UART bytes are written to stdout; diagnostics go to +stderr. Runs locally or is SCP'd to the Pi for remote test execution. +""" + +import argparse +import subprocess +import sys +import time +from pathlib import Path + +try: + import serial +except ImportError: + print("Error: pyserial not installed. Install with: pip install pyserial", file=sys.stderr) + sys.exit(1) + + +def _gpio_set(pin: int, state: str) -> None: + subprocess.run(["pinctrl", "set", str(pin), "op"] + state.split(), check=True) + + +def _sequence_to_fwspick_mode(srst_pin: int, fwspick_pin: int, port: serial.Serial) -> None: + _gpio_set(srst_pin, "dl") + time.sleep(0.1) + port.timeout = 0.1 + port.read(4096) + _gpio_set(fwspick_pin, "pn dh") + time.sleep(1) + _gpio_set(srst_pin, "dh") + time.sleep(1) + + +def _wait_for_uart_ready(port: serial.Serial, timeout: int = 30) -> bool: + deadline = time.time() + timeout + buf = b"" + port.timeout = 0.1 + while time.time() < deadline: + data = port.read(1024) + if data: + buf += data + sys.stderr.buffer.write(data) + sys.stderr.buffer.flush() + if b"U" in buf: + return True + print("Timeout waiting for UART bootloader ready signal", file=sys.stderr) + return False + + +def _upload_firmware(port: serial.Serial, firmware_path: Path) -> None: + data = firmware_path.read_bytes() + size = len(data) + aligned = (size + 3) & ~3 + port.write(aligned.to_bytes(4, "little")) + chunk_size = 1024 + for i in range(0, size, chunk_size): + port.write(data[i : i + chunk_size]) + port.flush() + time.sleep(0.01) + padding = aligned - size + if padding: + port.write(bytes(padding)) + print(f"Uploaded {size} bytes ({padding} bytes padding)", file=sys.stderr) + + +_SUCCESS_SENTINEL = b"TEST_RESULT:PASS" +_FAILURE_SENTINELS = [b"TEST_RESULT:FAIL", b"panic"] + + +def _stream_uart(port: serial.Serial, timeout: int) -> bool: + port.timeout = 1.0 + deadline = time.time() + timeout if timeout else None + buf = b"" + while True: + if deadline and time.time() >= deadline: + print("Timeout waiting for test result sentinel", file=sys.stderr) + return False + data = port.read(1024) + if data: + try: + sys.stdout.buffer.write(data) + sys.stdout.buffer.flush() + except (BrokenPipeError, OSError): + return False + buf += data + if _SUCCESS_SENTINEL in buf: + return True + for s in _FAILURE_SENTINELS: + if s in buf: + return False + buf = buf[-256:] + + +def main() -> int: + parser = argparse.ArgumentParser( + description="AST1060 EVB hardware layer: GPIO, firmware upload, UART stream" + ) + parser.add_argument( + "uart_device", + help="Serial port device path (e.g. /dev/ttyUSB0)", + ) + parser.add_argument( + "firmware", + nargs="?", + help="Firmware binary to upload. Not required with --stream-only", + ) + parser.add_argument( + "--srst-pin", + type=int, + required=True, + help="BCM GPIO pin connected to the AST1060 SRST line", + ) + parser.add_argument( + "--fwspick-pin", + type=int, + required=True, + help="BCM GPIO pin connected to the AST1060 FWSPICK line", + ) + parser.add_argument( + "--baudrate", + type=int, + required=True, + help="Serial port baud rate, must match firmware UART initialisation", + ) + parser.add_argument( + "--timeout", + type=int, + default=600, + help="Seconds to wait for a result sentinel (0 = no timeout, default: 600)", + ) + parser.add_argument( + "--stream-only", + action="store_true", + help="Skip GPIO sequences and firmware upload; stream raw UART bytes only", + ) + args = parser.parse_args() + + if not args.stream_only: + if not args.firmware: + parser.error("firmware is required unless --stream-only is set") + firmware_path = Path(args.firmware) + if not firmware_path.exists(): + print(f"Error: firmware not found: {firmware_path}", file=sys.stderr) + return 1 + else: + firmware_path = None + + try: + port = serial.Serial( + args.uart_device, + baudrate=args.baudrate, + timeout=1.0, + write_timeout=1.0, + ) + except serial.SerialException as e: + print(f"Error: could not open {args.uart_device}: {e}", file=sys.stderr) + return 1 + + result = False + try: + if not args.stream_only: + _sequence_to_fwspick_mode(args.srst_pin, args.fwspick_pin, port) + if not _wait_for_uart_ready(port): + return 1 + _upload_firmware(port, firmware_path) + result = _stream_uart(port, args.timeout) + except KeyboardInterrupt: + pass + finally: + port.close() + + return 0 if result else 1 + + +if __name__ == "__main__": + sys.exit(main())
diff --git a/target/ast10x0/harness/qemu_runner.py b/target/ast10x0/harness/qemu_runner.py new file mode 100644 index 0000000..70fe998 --- /dev/null +++ b/target/ast10x0/harness/qemu_runner.py
@@ -0,0 +1,174 @@ +# Licensed under the Apache-2.0 license +# SPDX-License-Identifier: Apache-2.0 +"""AST10x0 QEMU test runner. + +Runs a firmware image under QEMU. Pass/fail is determined by whichever signal +arrives first: a TEST_RESULT:PASS/FAIL sentinel in UART output, or QEMU's own +exit code from a semihosting exit() call. Semihosting is always enabled in +QEMU (harmless when unused), so both signalling mechanisms work transparently. +""" + +import argparse +import logging +import subprocess +import sys +import tempfile +import threading +import time + +from pathlib import Path +from pw_tokenizer import detokenize + +_LOG = logging.getLogger(__name__) +_LOG.setLevel(logging.INFO) + +try: + # qemu-system-arm-runfiles is a pw_py_importable_runfile target from the + # qemu repo (canonical: @@pigweed++_repo_rules5+qemu). If this import + # breaks after a pigweed upgrade, run: + # ls $(bazel info output_base)/external/ | grep qemu + import qemu.qemu_system_arm # type: ignore + from python.runfiles import runfiles # type: ignore + + r = runfiles.Create() + assert r is not None + _QEMU_ARM = r.Rlocation(*qemu.qemu_system_arm.RLOCATION) +except ImportError as e: + print(f"Fatal: runfiles could not find qemu: {e}", file=sys.stderr) + sys.exit(1) + +assert _QEMU_ARM is not None + +PASS_SENTINEL = b"TEST_RESULT:PASS" +FAIL_SENTINEL = b"TEST_RESULT:FAIL" +TIMEOUT_SECONDS = 30 + + +def _parse_args(): + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument('--machine', type=str, help='qemu machine type') + parser.add_argument('--cpu', type=str, help='qemu cpu type') + parser.add_argument('--image', type=str, help='image file to run') + parser.add_argument('--qemu-args', nargs='*', help='Extra arguments to pass to qemu') + return parser.parse_args() + + +def _detokenizer(image: Path, tokenized_file: Path, qemu_finished: threading.Event): + try: + detokenizer = detokenize.Detokenizer(image) + line_buffer = "" + with open(tokenized_file, 'r', buffering=1) as f: + while not qemu_finished.is_set(): + try: + chunk = f.readline() + if chunk: + line_buffer += chunk + while '\n' in line_buffer: + newline_pos = line_buffer.find('\n') + 1 + complete_line = line_buffer[:newline_pos] + detokenizer.detokenize_text_to_file( + complete_line, sys.stdout.buffer + ) + sys.stdout.flush() + line_buffer = line_buffer[newline_pos:] + except BlockingIOError: + time.sleep(0.1) + if line_buffer: + detokenizer.detokenize_text_to_file(line_buffer, sys.stdout.buffer) + sys.stdout.flush() + except OSError as e: + print(f"Exception opening file {e}", file=sys.stderr) + + +def _sentinel_watcher( + tokenized_file: Path, + result: list, + qemu_finished: threading.Event, + proc: subprocess.Popen, +): + buf = b"" + try: + with open(tokenized_file, 'rb') as f: + while not qemu_finished.is_set(): + chunk = f.read(256) + if chunk: + buf += chunk + if PASS_SENTINEL in buf: + result[0] = 0 + proc.kill() + return + if FAIL_SENTINEL in buf: + result[0] = 1 + proc.kill() + return + else: + time.sleep(0.01) + except OSError as e: + print(f"Exception watching sentinel: {e}", file=sys.stderr) + + +def _main(args) -> None: + qemu_args = [ + _QEMU_ARM, + "-machine", args.machine, + "-cpu", args.cpu, + "-bios", "none", + "-nographic", + "-serial", "mon:stdio", + "-semihosting-config", "enable=on,target=native", + "-kernel", args.image, + ] + + if args.qemu_args: + qemu_args.extend(args.qemu_args) + + _LOG.info("Invoking QEMU: %s", qemu_args) + + result = [None] # 0 = pass, 1 = fail, None = no sentinel found + + with tempfile.NamedTemporaryFile() as f: + with subprocess.Popen(args=qemu_args, stdout=f) as proc: + qemu_finished = threading.Event() + sentinel_thread = threading.Thread( + target=_sentinel_watcher, + args=(Path(f.name), result, qemu_finished, proc), + daemon=True, + ) + stdout_thread = threading.Thread( + target=_detokenizer, + args=(Path(args.image), Path(f.name), qemu_finished), + daemon=True, + ) + sentinel_thread.start() + stdout_thread.start() + + try: + proc.wait(timeout=TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + _LOG.error( + "Test timed out after %ds — no sentinel detected", + TIMEOUT_SECONDS, + ) + proc.kill() + proc.wait() + + qemu_finished.set() + + stdout_thread.join(timeout=5) + + if result[0] is None: + # No UART sentinel — check if QEMU exited naturally via semihosting. + # Processes killed by timeout have a negative returncode (SIGKILL = -9). + if proc.returncode >= 0: + sys.exit(0 if proc.returncode == 0 else 1) + _LOG.error("No TEST_RESULT sentinel found in UART output") + sys.exit(1) + + sys.exit(result[0]) + + +if __name__ == '__main__': + _main(_parse_args())
diff --git a/target/ast10x0/harness/test_runner.py b/target/ast10x0/harness/test_runner.py new file mode 100644 index 0000000..fb359b6 --- /dev/null +++ b/target/ast10x0/harness/test_runner.py
@@ -0,0 +1,372 @@ +#!/usr/bin/env python3 +# Licensed under the Apache-2.0 license +# SPDX-License-Identifier: Apache-2.0 +""" +AST1060 EVB test orchestration layer. + +Loads evb_config.toml, resolves the ELF for detokenization, then either +spawns pi_test_runner.py locally (USB-attached board) or SCP+SSHes it to a +Raspberry Pi fixture. Raw UART bytes come back over a pipe or SSH stdout; +this script detokenizes them and pattern-matches for PASS/FAIL sentinels. +Diagnostics go to stderr; nothing is written to the Pi filesystem except the +firmware binary and pi_test_runner.py itself. +""" + +import argparse +import base64 +import binascii +import os +import signal +import subprocess +import sys +import threading +import time +import tomllib +from pathlib import Path + +AST1060_EVB_PI_HOST = "AST1060_EVB_PI_HOST" + +# ── pw_tokenizer discovery ──────────────────────────────────────────────────── +# When run as a Bazel py_binary, pw_tokenizer is already on sys.path via deps. +# When run outside Bazel, try PW_TOK_ROOT or the Bazel output base as fallbacks. + +def _extend_path_for_pw_tokenizer() -> None: + pw_tok_root = os.environ.get("PW_TOK_ROOT") + if pw_tok_root: + sys.path.insert(0, os.path.join(pw_tok_root, "pw_tokenizer", "py")) + return + try: + output_base = subprocess.check_output( + ["bazel", "info", "output_base"], text=True, stderr=subprocess.DEVNULL + ).strip() + candidate = os.path.join(output_base, "external", "pigweed+", "pw_tokenizer", "py") + if os.path.isdir(candidate): + sys.path.insert(0, candidate) + except (subprocess.CalledProcessError, FileNotFoundError): + pass + + +_extend_path_for_pw_tokenizer() + +try: + from pw_tokenizer import Detokenizer + from pw_tokenizer.detokenize import NestedMessageParser + _PW_TOKENIZER_AVAILABLE = True +except ImportError: + _PW_TOKENIZER_AVAILABLE = False + +# ── Remote lock constants ───────────────────────────────────────────────────── +# The Pi is a single-board computer; only one test session should hold the +# UART device and GPIO lines at a time. We use an atomic noclobber lock file +# on the Pi and touch it every _LOCK_TOUCH_INTERVAL seconds from a background +# thread so the stale-lock detector knows we're still alive. + +_LOCK_PATH = "/tmp/ast1060_evb.lock" +_LOCK_TOUCH_INTERVAL = 10 # seconds between lock touches +_LOCK_STALE_THRESHOLD = 60 # seconds since last touch → lock is stale +_LOCK_ACQUIRE_TIMEOUT = 120 # seconds to wait before giving up + + +# ── SSH helpers ─────────────────────────────────────────────────────────────── + +def _ssh(host: str, cmd: str, **kwargs) -> subprocess.CompletedProcess: + """Run a single SSH command synchronously.""" + return subprocess.run( + ["ssh", "-o", "BatchMode=yes", host, cmd], + **kwargs, + ) + + +def _ssh_stream(host: str, cmd: str) -> subprocess.Popen: + """Open a streaming SSH session; stdout is a pipe the caller reads.""" + return subprocess.Popen( + ["ssh", "-o", "BatchMode=yes", "-o", "ServerAliveInterval=5", host, cmd], + stdout=subprocess.PIPE, + stderr=sys.stderr, + ) + + +def _acquire_lock(host: str, timeout: int = _LOCK_ACQUIRE_TIMEOUT) -> bool: + """Atomically create lock file on Pi; retry until timeout.""" + deadline = time.time() + timeout + create = f"set -o noclobber && echo $$ > {_LOCK_PATH}" + stale_check = ( + f"mtime=$(stat -c %Y {_LOCK_PATH} 2>/dev/null) && " + f"now=$(date +%s) && " + f"[ $(( now - mtime )) -gt {_LOCK_STALE_THRESHOLD} ] && " + f"rm -f {_LOCK_PATH}" + ) + while time.time() < deadline: + r = _ssh(host, create, capture_output=True) + if r.returncode == 0: + return True + _ssh(host, stale_check, capture_output=True) + time.sleep(2) + print(f"Timeout acquiring Pi lock after {timeout}s", file=sys.stderr) + return False + + +def _release_lock(host: str) -> None: + _ssh(host, f"rm -f {_LOCK_PATH}", capture_output=True) + + +def _touch_lock_forever(host: str, stop: threading.Event) -> None: + """Background thread: touch the lock file every _LOCK_TOUCH_INTERVAL seconds.""" + while not stop.wait(_LOCK_TOUCH_INTERVAL): + _ssh(host, f"touch {_LOCK_PATH}", capture_output=True) + + +# ── UART monitor ────────────────────────────────────────────────────────────── + + +class UartMonitor: + """Detokenizes and displays raw UART bytes. Pass/fail is determined by pi_test_runner.py exit code.""" + + def __init__(self, args: argparse.Namespace, elf_path: Path) -> None: + # elf_path is always derived on the host from firmware path; caller + # validates existence before constructing this object. + self.args = args + self.log_file_handle = open(args.log_file, "w") if args.log_file else None + self.detokenizer = ( + Detokenizer(str(elf_path)) if _PW_TOKENIZER_AVAILABLE else None + ) + self._token_parser = NestedMessageParser() if _PW_TOKENIZER_AVAILABLE else None + + def _write_log(self, text: str) -> None: + if self.log_file_handle: + self.log_file_handle.write(text) + self.log_file_handle.flush() + + def print_uart_data(self, raw: bytes) -> None: + """Detokenize raw UART bytes and print them. + + pw_tokenizer embeds $<base64> token frames in the byte stream. + NestedMessageParser preserves state across calls so tokens split + across successive reads are reassembled correctly. + """ + if self.detokenizer and self._token_parser: + for is_token, span in self._token_parser.read_messages(raw): + if not is_token: + text = span.decode("utf-8", errors="replace") + print(text, end="", flush=True) + self._write_log(text) + continue + + # span is b'$<base64chars>' — strip '$' and add padding. + raw_text = span.decode("utf-8", errors="replace") + try: + b64 = span[1:] + b64 += b"=" * (-len(b64) % 4) + encoded = base64.b64decode(b64, validate=True) + result = self.detokenizer.detokenize(encoded) + except (binascii.Error, ValueError): + result = None + + if result is not None and result.ok(): + decoded_str = str(result) + print(f"\033[32m{decoded_str}\033[0m", end="", flush=True) + self._write_log(decoded_str) + else: + print(raw_text, end="", flush=True) + self._write_log(raw_text) + return + + text = raw.decode("utf-8", errors="replace") + print(text, end="", flush=True) + self._write_log(text) + + def display_stream(self, pipe) -> None: + """Read raw bytes from pipe and detokenize for display until EOF.""" + while True: + chunk = pipe.read(4096) + if not chunk: + break + self.print_uart_data(chunk) + + def cleanup(self) -> None: + if self.log_file_handle: + self.log_file_handle.close() + self.log_file_handle = None + + +# ── Local execution ─────────────────────────────────────────────────────────── + +def _run_local( + args: argparse.Namespace, + config: dict, + runner: Path, + monitor: UartMonitor, + uart_device: str, +) -> bool: + """Wired (non-SSH) connection to the Pi fixture — not yet implemented.""" + print( + "Error: wired Pi mode is not yet implemented. " + f"Set ${AST1060_EVB_PI_HOST} or pass --pi-host to use SSH.", + file=sys.stderr, + ) + return False + + +# ── Remote execution ────────────────────────────────────────────────────────── + +def _run_remote( + args: argparse.Namespace, + config: dict, + runner: Path, + monitor: UartMonitor, + uart_device: str, +) -> bool: + """SCP firmware and runner to Pi; stream UART back over SSH stdout.""" + host = args.pi_host + gpio = config["gpio"] + uart = config["uart"] + baudrate = args.baudrate if args.baudrate else uart["baudrate"] + + remote_dir = "/tmp/ast1060_test" + + if not _acquire_lock(host): + return False + + stop_touch = threading.Event() + touch_thread = threading.Thread( + target=_touch_lock_forever, args=(host, stop_touch), daemon=True + ) + touch_thread.start() + + _ssh(host, f"rm -rf {remote_dir} && mkdir -p {remote_dir}", check=True) + + subprocess.run( + ["scp", "-q", str(runner), f"{host}:{remote_dir}/pi_test_runner.py"], + check=True, + ) + + if not args.parse_only: + firmware_path = Path(args.firmware) + remote_fw = f"{remote_dir}/{firmware_path.name}" + subprocess.run( + ["scp", "-q", str(firmware_path), f"{host}:{remote_fw}"], + check=True, + ) + else: + remote_fw = None + + remote_cmd = f"python3 -u {remote_dir}/pi_test_runner.py {uart_device}" + if not args.parse_only: + remote_cmd += f" {remote_fw}" + remote_cmd += ( + f" --srst-pin {gpio['srst_pin']}" + f" --fwspick-pin {gpio['fwspick_pin']}" + f" --baudrate {baudrate}" + f" --timeout {args.timeout}" + ) + if args.parse_only: + remote_cmd += " --stream-only" + + proc = _ssh_stream(host, remote_cmd) + try: + monitor.display_stream(proc.stdout) + proc.wait() + finally: + if proc.returncode is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + stop_touch.set() + _release_lock(host) + + return proc.returncode == 0 + + +# ── Entry point ─────────────────────────────────────────────────────────────── + +def main() -> int: + with (Path(__file__).parent / "evb_config.toml").open("rb") as f: + config = tomllib.load(f) + + parser = argparse.ArgumentParser( + description="AST1060 EVB test orchestration: detokenize UART, match sentinels" + ) + parser.add_argument( + "firmware", + nargs="?", + help=( + "Firmware image (.elf or .bin). system_image emits both under the " + "same stem; pass either and the other is derived automatically." + ), + ) + parser.add_argument( + "--pi-host", + default=None, + help=f"Raspberry Pi hostname/IP. Falls back to ${AST1060_EVB_PI_HOST} env var. Omit both to use a locally attached board", + ) + parser.add_argument( + "--baudrate", + type=int, + default=None, + help=f"Override baud rate from evb_config.toml (default: {config['uart']['baudrate']})", + ) + parser.add_argument( + "--timeout", + type=int, + default=600, + help="Seconds to wait for a test result sentinel (0 = no timeout, default: 600)", + ) + parser.add_argument( + "--log-file", + help="Write detokenized UART output to this file", + ) + parser.add_argument( + "-q", "--quiet", + action="store_true", + help="Suppress diagnostic messages to stderr", + ) + parser.add_argument( + "--parse-only", + action="store_true", + help="Skip GPIO and firmware upload; stream and detokenize UART output only", + ) + args = parser.parse_args() + + uart_device = os.environ.get("UART_DEVICE") or config["uart"]["serial_port"] + + if not args.firmware: + parser.error("firmware is required") + + # system_image emits both .elf and .bin under the same stem; accept either. + # When invoked via --run_under on a system_image_test, Bazel passes the + # no-suffix symlink (e.g. threads_test → threads.elf); resolve it first. + image = Path(args.firmware) + if not image.suffix: + image = image.resolve() + elf_path = image.with_suffix(".elf") + args.firmware = str(image.with_suffix(".bin")) + if not elf_path.exists(): + print(f"Error: ELF not found at {elf_path}", file=sys.stderr) + return 1 + + args.pi_host = os.environ.get(AST1060_EVB_PI_HOST) or args.pi_host + + runner = Path(__file__).parent / "pi_test_runner.py" + monitor = UartMonitor(args, elf_path) + + signal.signal(signal.SIGINT, lambda s, f: sys.exit(130)) + + try: + if args.pi_host: + ok = _run_remote(args, config, runner, monitor, uart_device) + else: + ok = _run_local(args, config, runner, monitor, uart_device) + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + return 1 + finally: + monitor.cleanup() + + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main())
diff --git a/target/ast10x0/harness/uart_test_exec_wrapper.sh b/target/ast10x0/harness/uart_test_exec_wrapper.sh deleted file mode 100644 index e98a99b..0000000 --- a/target/ast10x0/harness/uart_test_exec_wrapper.sh +++ /dev/null
@@ -1,32 +0,0 @@ -#!/bin/bash -# Licensed under the Apache-2.0 license -# -# Wrapper script for uart_test_exec.py -# Invokes the Python script with system Python3 - -set -e - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -PYTHON_SCRIPT="$SCRIPT_DIR/uart_test_exec.py" - -# If running from runfiles, look there -if [[ -f "$PYTHON_SCRIPT" ]]; then - exec python3 "$PYTHON_SCRIPT" "$@" -fi - -# Try relative to workspace root -if [[ -f "tools/uart_test/uart_test_exec.py" ]]; then - exec python3 "tools/uart_test/uart_test_exec.py" "$@" -fi - -# Try runfiles directory -RUNFILES="${BASH_SOURCE[0]}.runfiles" -if [[ -d "$RUNFILES" ]]; then - SCRIPT="$RUNFILES/_main/tools/uart_test/uart_test_exec.py" - if [[ -f "$SCRIPT" ]]; then - exec python3 "$SCRIPT" "$@" - fi -fi - -echo "ERROR: Could not find uart_test_exec.py" >&2 -exit 1
diff --git a/target/ast10x0/harness/uart_upload_test.bzl b/target/ast10x0/harness/uart_upload_test.bzl deleted file mode 100644 index 9f14ed0..0000000 --- a/target/ast10x0/harness/uart_upload_test.bzl +++ /dev/null
@@ -1,368 +0,0 @@ -# Licensed under the Apache-2.0 license - -"""Bazel rules for UART test execution on AST1060 hardware. - -Provides test rules that upload firmware via UART and monitor test execution. -""" - -load( - "@bazel_skylib//rules:common_settings.bzl", - "BuildSettingInfo", -) - -load( - "@pigweed//pw_kernel/tooling:system_image.bzl", - "SystemImageInfo", -) - -def _firmware_bin(ctx): - if SystemImageInfo in ctx.attr.image: - return ctx.attr.image[SystemImageInfo].bin - return ctx.file.image - -def _declare_uart_boot_image(ctx, firmware_bin, name): - output = ctx.actions.declare_file(name + ".bin") - ctx.actions.run_shell( - inputs = [firmware_bin], - outputs = [output], - arguments = [firmware_bin.path, output.path], - mnemonic = "Ast10x0UartBootImage", - command = """set -eu -input=\"$1\" -output=\"$2\" -size=$(wc -c < \"$input\") -aligned=$(( (size + 3) & ~3 )) - -emit_byte() { - printf '%b' "$(printf '\\%03o' \"$1\")" -} - -{ - emit_byte $((aligned & 255)) - emit_byte $(((aligned >> 8) & 255)) - emit_byte $(((aligned >> 16) & 255)) - emit_byte $(((aligned >> 24) & 255)) - cat \"$input\" - padding=$((aligned - size)) - if [ \"$padding\" -gt 0 ]; then - dd if=/dev/zero bs=1 count=\"$padding\" status=none - fi -} > \"$output\" -""", - ) - return output - -def _uart_boot_image_impl(ctx): - firmware_bin = _firmware_bin(ctx) - if ctx.attr._uart_boot_header[BuildSettingInfo].value: - output = _declare_uart_boot_image(ctx, firmware_bin, ctx.label.name) - else: - output = firmware_bin - return [ - DefaultInfo(files = depset([output])), - ] - -uart_boot_image = rule( - implementation = _uart_boot_image_impl, - attrs = { - "image": attr.label( - mandatory = True, - doc = "system_image or binary target to wrap with the AST10x0 UART boot header", - ), - "_uart_boot_header": attr.label( - default = "//target/ast10x0:uart_boot_header", - providers = [BuildSettingInfo], - ), - }, - doc = "Generate an AST10x0 UART boot image by prepending the 4-byte size header when enabled by config.", -) - -def _uart_upload_test_impl(ctx): - """Implementation of uart_upload_test rule.""" - - # Get the firmware binary - firmware_bin = _firmware_bin(ctx) - - # Create test script - test_script = ctx.actions.declare_file(ctx.label.name + "_test.sh") - - # Build the command line arguments - args = [] - - if ctx.attr.baudrate: - args.extend(["--baudrate", str(ctx.attr.baudrate)]) - - if ctx.attr.test_timeout: - args.extend(["--test-timeout", str(ctx.attr.test_timeout)]) - - if ctx.attr.srst_pin: - args.extend(["--srst-pin", str(ctx.attr.srst_pin)]) - - if ctx.attr.fwspick_pin: - args.extend(["--fwspick-pin", str(ctx.attr.fwspick_pin)]) - - if ctx.attr.skip_gpio: - args.append("--skip-gpio") - - if ctx.attr.upload_only: - args.append("--upload-only") - - # Get path to the Python script - python_script = ctx.file._uart_test_exec - - script_content = """#!/bin/bash -set -e - -# Allow overriding UART device via environment -UART_DEVICE="${{UART_DEVICE:-{default_device}}}" - -# Check if device exists (unless skipping device check) -if [[ ! -e "$UART_DEVICE" && -z "$SKIP_DEVICE_CHECK" ]]; then - echo "ERROR: UART device not found: $UART_DEVICE" - echo "Set UART_DEVICE environment variable or connect hardware" - exit 1 -fi - -# Find the Python script in runfiles -SCRIPT_DIR="$(cd "$(dirname "${{BASH_SOURCE[0]}}")" && pwd)" -RUNFILES="${{SCRIPT_DIR}}/{test_name}_test.sh.runfiles/_main" - -if [[ -f "$RUNFILES/{script_path}" ]]; then - PYTHON_SCRIPT="$RUNFILES/{script_path}" -elif [[ -f "{script_path}" ]]; then - PYTHON_SCRIPT="{script_path}" -else - echo "ERROR: Could not find uart_test_exec.py" >&2 - exit 1 -fi - -# Find firmware in runfiles -if [[ -f "$RUNFILES/{firmware_path}" ]]; then - FIRMWARE="$RUNFILES/{firmware_path}" -elif [[ -f "{firmware_path}" ]]; then - FIRMWARE="{firmware_path}" -else - echo "ERROR: Could not find firmware: {firmware_path}" >&2 - exit 1 -fi - -# Run the UART test executor -exec python3 "$PYTHON_SCRIPT" "$UART_DEVICE" "$FIRMWARE" {args} -""".format( - default_device = ctx.attr.uart_device or "/dev/ttyUSB0", - test_name = ctx.label.name, - script_path = python_script.short_path, - firmware_path = firmware_bin.short_path, - args = " ".join(args), - ) - - ctx.actions.write( - output = test_script, - content = script_content, - is_executable = True, - ) - - runfiles = ctx.runfiles( - files = [firmware_bin, python_script], - ) - - return [ - DefaultInfo( - executable = test_script, - runfiles = runfiles, - ), - ] - -uart_upload_test = rule( - implementation = _uart_upload_test_impl, - test = True, - attrs = { - "image": attr.label( - mandatory = True, - allow_single_file = True, - doc = "system_image or uart_boot_image target to upload", - ), - "uart_device": attr.string( - default = "", - doc = "UART device path (default: /dev/ttyUSB0 or UART_DEVICE env var)", - ), - "baudrate": attr.int( - default = 115200, - doc = "UART baud rate", - ), - "test_timeout": attr.int( - default = 600, - doc = "Test execution timeout in seconds", - ), - "srst_pin": attr.int( - default = 23, - doc = "SRST GPIO pin number", - ), - "fwspick_pin": attr.int( - default = 18, - doc = "FWSPICK GPIO pin number", - ), - "skip_gpio": attr.bool( - default = False, - doc = "Skip GPIO operations (for pre-configured boards)", - ), - "upload_only": attr.bool( - default = False, - doc = "Upload firmware only, skip test monitoring", - ), - "_uart_test_exec": attr.label( - default = "//target/ast10x0/harness:uart_test_exec.py", - allow_single_file = [".py"], - ), - }, - doc = """Run a firmware test on AST1060 hardware via UART. - -This test rule: -1. Optionally enters FWSPICK mode via GPIO -2. Uploads firmware via UART bootloader -3. Monitors serial output for test pass/fail - -Environment variables: -- UART_DEVICE: Override the UART device path -- SKIP_DEVICE_CHECK: Skip device existence check (for CI) - -Usage: - load( - "//target/ast10x0/harness:uart_upload_test.bzl", - "uart_boot_image", - "uart_upload_test", - ) - - # AST1030-EVB UART boot expects the 4-byte size header. - uart_boot_image( - name = "threads_uart", - image = ":threads", - ) - - uart_upload_test( - name = "threads_uart_test", - image = ":threads_uart", - test_timeout = 300, - ) - -Run with: - bazel test //target/ast1060-evb/threads/kernel:threads_uart_test \\ - --test_env=UART_DEVICE=/dev/ttyUSB0 -""", -) - -def _uart_upload_impl(ctx): - """Implementation of uart_upload rule (non-test, just upload).""" - - # Get the firmware binary - firmware_bin = _firmware_bin(ctx) - - # Create upload script - upload_script = ctx.actions.declare_file(ctx.label.name + "_upload.sh") - - args = ["--upload-only"] - - if ctx.attr.baudrate: - args.extend(["--baudrate", str(ctx.attr.baudrate)]) - - if ctx.attr.skip_gpio: - args.append("--skip-gpio") - - # Get path to the Python script - python_script = ctx.file._uart_test_exec - - script_content = """#!/bin/bash -set -e - -UART_DEVICE="${{UART_DEVICE:-{default_device}}}" - -if [[ ! -e "$UART_DEVICE" ]]; then - echo "ERROR: UART device not found: $UART_DEVICE" - exit 1 -fi - -# Find the Python script in runfiles -SCRIPT_DIR="$(cd "$(dirname "${{BASH_SOURCE[0]}}")" && pwd)" -RUNFILES="${{SCRIPT_DIR}}/{script_name}_upload.sh.runfiles/_main" - -if [[ -f "$RUNFILES/{script_path}" ]]; then - PYTHON_SCRIPT="$RUNFILES/{script_path}" -elif [[ -f "{script_path}" ]]; then - PYTHON_SCRIPT="{script_path}" -else - echo "ERROR: Could not find uart_test_exec.py" >&2 - exit 1 -fi - -# Find firmware in runfiles -if [[ -f "$RUNFILES/{firmware_path}" ]]; then - FIRMWARE="$RUNFILES/{firmware_path}" -elif [[ -f "{firmware_path}" ]]; then - FIRMWARE="{firmware_path}" -else - echo "ERROR: Could not find firmware: {firmware_path}" >&2 - exit 1 -fi - -echo "Uploading firmware to $UART_DEVICE..." -exec python3 "$PYTHON_SCRIPT" "$UART_DEVICE" "$FIRMWARE" {args} -""".format( - default_device = ctx.attr.uart_device or "/dev/ttyUSB0", - script_name = ctx.label.name, - script_path = python_script.short_path, - firmware_path = firmware_bin.short_path, - args = " ".join(args), - ) - - ctx.actions.write( - output = upload_script, - content = script_content, - is_executable = True, - ) - - runfiles = ctx.runfiles( - files = [firmware_bin, python_script], - ) - - return [ - DefaultInfo( - executable = upload_script, - runfiles = runfiles, - ), - ] - -uart_upload = rule( - implementation = _uart_upload_impl, - executable = True, - attrs = { - "image": attr.label( - mandatory = True, - allow_single_file = True, - doc = "system_image or uart_boot_image target to upload", - ), - "uart_device": attr.string( - default = "", - doc = "UART device path (default: /dev/ttyUSB0 or UART_DEVICE env var)", - ), - "baudrate": attr.int( - default = 115200, - doc = "UART baud rate", - ), - "skip_gpio": attr.bool( - default = False, - doc = "Skip GPIO operations", - ), - "_uart_test_exec": attr.label( - default = "//target/ast10x0/harness:uart_test_exec.py", - allow_single_file = [".py"], - ), - }, - doc = """Upload firmware to AST10x0 hardware via UART. - -This is a non-test rule that just uploads firmware without monitoring. - -Usage: - bazel run //target/ast1060-evb/threads/kernel:upload_threads -- \\ - UART_DEVICE=/dev/ttyUSB0 -""", -)
diff --git a/target/ast10x0/peripherals/uart/mod.rs b/target/ast10x0/peripherals/uart/mod.rs index cec8ab1..0de7491 100644 --- a/target/ast10x0/peripherals/uart/mod.rs +++ b/target/ast10x0/peripherals/uart/mod.rs
@@ -133,23 +133,48 @@ Ok(()) } + #[inline(always)] fn write(&mut self, buf: &[u8]) -> Result<usize, Error> { - for (n, byte) in buf.iter().enumerate() { + let mut written = 0; + for byte in buf.iter() { if !self.is_tx_full() { // This is unsafe because we can transmit 7, 8 or 9 bits but the // interface can't know what it's been configured for. self.regs() .uartthr() .write(|w| unsafe { w.bits(*byte as u32) }); + written += 1; } else { - if n == 0 { - // spec demands to block until atleast one byte has been written - continue; + if written == 0 { + // spec demands to block until at least one byte has been written. + // `continue` would skip to the next byte rather than retrying + // this one, so we busy-wait inline instead. + while self.is_tx_full() {} + self.regs() + .uartthr() + .write(|w| unsafe { w.bits(*byte as u32) }); + written += 1; + } else { + break; } - return Ok(n); } } - Ok(buf.len()) + // Two invariants hold that LLVM cannot prove through value range analysis + // due to the busy-wait inner loop and early break: + // + // 1. n <= buf.len(): `written` is incremented at most once per element of + // `buf.iter()`. Without this, `write_all`'s `buf = &buf[n..]` retains + // a bounds-check panic. The `min` makes the assert mathematically sound. + // + // 2. n > 0 when buf is non-empty: the busy-wait guarantees at least one + // byte is written before returning. Without this, `write_all`'s + // `Ok(0) => panic!` branch is retained even though it is unreachable. + let n = written.min(buf.len()); + unsafe { + core::hint::assert_unchecked(n <= buf.len()); + core::hint::assert_unchecked(n > 0 || buf.is_empty()); + } + Ok(n) } }
diff --git a/target/ast10x0/tests/README.md b/target/ast10x0/tests/README.md new file mode 100644 index 0000000..9b4072b --- /dev/null +++ b/target/ast10x0/tests/README.md
@@ -0,0 +1,119 @@ +# AST10x0 Test Infrastructure + +## Overview + +Tests for the AST10x0 target are firmware images that run identically under +QEMU or on a physical board. Pass/fail is signalled by writing a sentinel +string to UART: + +``` +TEST_RESULT:PASS\n +TEST_RESULT:FAIL\n +``` + +The same `system_image_test` target is used for both execution environments — +no separate hardware-only test targets exist. + +## Running Tests + +### QEMU (no hardware required) + +``` +bazel test --config=virt_ast10x0 //target/ast10x0/tests/... +``` + +### Physical AST1060 EVB via Raspberry Pi SSH fixture + +``` +AST1060_EVB_PI_HOST=<pi-hostname> bazel test --config=k_ast1060_evb //target/ast10x0/tests/... +``` + +or inline without modifying the shell environment: + +``` +bazel test --config=k_ast1060_evb --test_env=AST1060_EVB_PI_HOST=<pi-hostname> //target/ast10x0/tests/... +``` + +Key-based SSH auth is required (`ssh-copy-id <user>@<pi-host>`). The Pi runs +`pi_test_runner.py`, which handles GPIO reset sequencing, firmware upload over +the UART bootloader, and sentinel detection. UART output is streamed back to +the host for detokenization and display. Tests no longer use cortex_m_semihosting. + +### Physical AST1060 EVB wired (not yet implemented) + +A wired mode — where the host connects to the Pi fixture over a local serial +port rather than SSH — is not yet implemented. The physical connection type +between the host and the Pi has not been defined (e.g. Pi serial console over +UART, USB serial gadget, or USB networking), so the host-side protocol cannot +be specified. Omitting `AST1060_EVB_PI_HOST` should default to using a wired +connection, but currently logs an unimplemented error. + +## Test Results (2026-05-12) + +| Test | QEMU | Physical board | +|------|------|----------------| +| `interrupts/kernel:interrupts_test` | PASSED | TIMEOUT — see note below | +| `interrupts/user:interrupts_test` | PASSED | PASSED | +| `ipc/user:ipc_test` | FAILED (hangs at `object_set_peer_user_signal`) | PASSED | +| `threads/kernel:threads_test` | PASSED | PASSED | +| `unittest_runner:unittest_runner` | PASSED | SKIPPED (qemu_only) | +| `usart:usart_test` | PASSED | SKIPPED (qemu_only) | +| `*/no_panics_test` (×5) | SKIPPED (host-only) | SKIPPED | + +### `interrupts/kernel:interrupts_test` — times out on physical board + +The firmware produces no UART output after upload, indicating a crash before +UART initialisation. The same binary passes in QEMU. The `interrupts/user` +variant (which manages IRQ 42 through the kernel IPC abstraction rather than +raw NVIC manipulation) passes on both. + +The suspected cause is the `interrupt_table` entry in `system.json5`: the +codegen for that entry installs an NVIC handler via `early_init()`, which runs +before UART is initialised. If `early_init()` faults (e.g. invalid vector table +layout, bad IRQ number on hardware), the firmware crashes with no UART output +and no way to signal failure. This cannot be verified without a hardware +debugger (GDB via OpenOCD or J-Link) attached to the board. + +## How Pass/Fail Signalling Works + +Firmware writes the sentinel via `console_backend_write_all`, which calls +`Usart::write_all` directly, bypassing `pw_log` and the tokenizer. This means +the sentinel is always plain ASCII regardless of whether the rest of the log +output is tokenized, and it can be detected without an ELF for detokenization. + +### QEMU + +`qemu_runner.py` starts QEMU with a PTY for serial I/O and a named pipe for +the raw byte stream. A sentinel watcher thread scans the raw stream; when a +sentinel is found QEMU is killed and the runner exits 0 or 1. A 30-second +watchdog kills QEMU if no sentinel arrives. + +### Physical board + +`pi_test_runner.py` (running on the Raspberry Pi) sequences the GPIO reset +lines to enter UART bootloader mode, uploads the firmware binary, then streams +raw UART bytes to stdout while scanning for the sentinel. It exits 0 (PASS) or +1 (FAIL/timeout). `test_runner.py` on the host SCP's the script to the Pi, +streams the output back for detokenization and display, and reports the Pi's +exit code to Bazel. + +Because the Pi is a shared fixture, `test_runner.py` holds an atomic noclobber +lock file at `/tmp/ast1060_evb.lock` on the Pi for the duration of each test, +preventing multiple users from driving the board over SSH simultaneously. The +lock is touched every 10 seconds by a background thread and considered stale +after 60 seconds of inactivity (e.g. after a crash). If the lock cannot be +acquired within 120 seconds the run is aborted. + +## Semihosting Migration + +This infrastructure previously used ARM semihosting to signal pass/fail. On +real hardware with no attached debugger, a semihosting trap causes a HardFault, +so hardware testing was impossible. Replacing semihosting with UART sentinels +removed that constraint and enabled the Pi SSH test fixture. + +## `uart_upload_test` Targets (removed) + +The five `*_uart_upload_test` targets that previously existed in these BUILD +files have been removed. They used an earlier harness (`uart_upload_test.bzl`) +that predated the `run_under` approach. The `system_image_test` targets cover +both QEMU and hardware execution; no separate hardware-only test rule is needed.
diff --git a/target/ast10x0/tests/interrupts/kernel/BUILD.bazel b/target/ast10x0/tests/interrupts/kernel/BUILD.bazel index 2b3dd08..417a008 100644 --- a/target/ast10x0/tests/interrupts/kernel/BUILD.bazel +++ b/target/ast10x0/tests/interrupts/kernel/BUILD.bazel
@@ -7,7 +7,6 @@ load("@pigweed//pw_kernel/tooling/panic_detector:rust_binary_no_panics_test.bzl", "rust_binary_no_panics_test") load("@rules_rust//rust:defs.bzl", "rust_binary") load("//target/ast10x0:defs.bzl", "TARGET_COMPATIBLE_WITH") -load("//target/ast10x0/harness:uart_upload_test.bzl", "uart_boot_image", "uart_upload_test") system_image( name = "interrupts", @@ -18,17 +17,6 @@ userspace = False, ) -uart_boot_image( - name = "interrupts_uart_image", - image = ":interrupts", -) - -uart_upload_test( - name = "interrupts_uart_upload_test", - image = ":interrupts_uart_image", - test_timeout = 300, -) - system_image_test( name = "interrupts_test", image = ":interrupts", @@ -79,6 +67,5 @@ "@pigweed//pw_kernel/tests/interrupts/kernel:test_interrupts", "@pigweed//pw_log/rust:pw_log", "@pigweed//pw_status/rust:pw_status", - "@rust_crates//:cortex-m-semihosting", ], )
diff --git a/target/ast10x0/tests/interrupts/kernel/target.rs b/target/ast10x0/tests/interrupts/kernel/target.rs index 2eb0944..6d16108 100644 --- a/target/ast10x0/tests/interrupts/kernel/target.rs +++ b/target/ast10x0/tests/interrupts/kernel/target.rs
@@ -6,8 +6,7 @@ use arch_arm_cortex_m::Arch; use codegen as _; -use console_backend as _; -use cortex_m_semihosting::debug::{EXIT_FAILURE, EXIT_SUCCESS, exit}; +use console_backend::console_backend_write_all; use entry as _; use target_common::{TargetInterface, declare_target}; @@ -20,11 +19,11 @@ const NAME: &'static str = "AST10x0 Kernel Interrupts"; fn main() -> ! { - let exit_status = match test_interrupts::main::<Arch>(TEST_IRQ) { - Ok(()) => EXIT_SUCCESS, - Err(_e) => EXIT_FAILURE, + let sentinel: &[u8] = match test_interrupts::main::<Arch>(TEST_IRQ) { + Ok(()) => b"TEST_RESULT:PASS\n", + Err(_e) => b"TEST_RESULT:FAIL\n", }; - exit(exit_status); + let _ = console_backend_write_all(sentinel); #[expect(clippy::empty_loop)] loop {} }
diff --git a/target/ast10x0/tests/interrupts/user/BUILD.bazel b/target/ast10x0/tests/interrupts/user/BUILD.bazel index 0517efb..653e68c 100644 --- a/target/ast10x0/tests/interrupts/user/BUILD.bazel +++ b/target/ast10x0/tests/interrupts/user/BUILD.bazel
@@ -7,8 +7,6 @@ load("@pigweed//pw_kernel/tooling/panic_detector:rust_binary_no_panics_test.bzl", "rust_binary_no_panics_test") load("@rules_rust//rust:defs.bzl", "rust_binary") load("//target/ast10x0:defs.bzl", "TARGET_COMPATIBLE_WITH") -load("//target/ast10x0/harness:uart_upload_test.bzl", "uart_boot_image", "uart_upload_test") - system_image( name = "interrupts", apps = [ @@ -21,17 +19,6 @@ tags = ["kernel"], ) -uart_boot_image( - name = "interrupts_uart_image", - image = ":interrupts", -) - -uart_upload_test( - name = "interrupts_uart_upload_test", - image = ":interrupts_uart_image", - test_timeout = 300, -) - system_image_test( name = "interrupts_test", image = ":interrupts", @@ -80,6 +67,5 @@ "@pigweed//pw_kernel/target:target_common", "@pigweed//pw_kernel/userspace", "@pigweed//pw_log/rust:pw_log", - "@rust_crates//:cortex-m-semihosting", ], )
diff --git a/target/ast10x0/tests/interrupts/user/target.rs b/target/ast10x0/tests/interrupts/user/target.rs index 46d032c..92cb3fc 100644 --- a/target/ast10x0/tests/interrupts/user/target.rs +++ b/target/ast10x0/tests/interrupts/user/target.rs
@@ -4,8 +4,7 @@ #![no_std] #![no_main] -use console_backend as _; -use cortex_m_semihosting::debug::{EXIT_FAILURE, EXIT_SUCCESS, exit}; +use console_backend::console_backend_write_all; use entry as _; use target_common::{TargetInterface, declare_target}; @@ -22,11 +21,8 @@ fn shutdown(code: u32) -> ! { pw_log::info!("Shutting down with code {}", code as u32); - let status = match code { - 0 => EXIT_SUCCESS, - _ => EXIT_FAILURE, - }; - exit(status); + let sentinel: &[u8] = if code == 0 { b"TEST_RESULT:PASS\n" } else { b"TEST_RESULT:FAIL\n" }; + let _ = console_backend_write_all(sentinel); #[expect(clippy::empty_loop)] loop {} }
diff --git a/target/ast10x0/tests/ipc/user/BUILD.bazel b/target/ast10x0/tests/ipc/user/BUILD.bazel index 6aac1fd..553dfde 100644 --- a/target/ast10x0/tests/ipc/user/BUILD.bazel +++ b/target/ast10x0/tests/ipc/user/BUILD.bazel
@@ -7,8 +7,6 @@ load("@pigweed//pw_kernel/tooling/panic_detector:rust_binary_no_panics_test.bzl", "rust_binary_no_panics_test") load("@rules_rust//rust:defs.bzl", "rust_binary") load("//target/ast10x0:defs.bzl", "TARGET_COMPATIBLE_WITH") -load("//target/ast10x0/harness:uart_upload_test.bzl", "uart_boot_image", "uart_upload_test") - system_image( name = "ipc", apps = ["@pigweed//pw_kernel/tests/ipc/user:ipc"], @@ -19,17 +17,6 @@ target_compatible_with = TARGET_COMPATIBLE_WITH, ) -uart_boot_image( - name = "ipc_uart_image", - image = ":ipc", -) - -uart_upload_test( - name = "ipc_uart_upload_test", - image = ":ipc_uart_image", - test_timeout = 300, -) - system_image_test( name = "ipc_test", image = ":ipc", @@ -78,6 +65,5 @@ "@pigweed//pw_kernel/target:target_common", "@pigweed//pw_kernel/userspace", "@pigweed//pw_log/rust:pw_log", - "@rust_crates//:cortex-m-semihosting", ], )
diff --git a/target/ast10x0/tests/ipc/user/target.rs b/target/ast10x0/tests/ipc/user/target.rs index abfb96b..d386f54 100644 --- a/target/ast10x0/tests/ipc/user/target.rs +++ b/target/ast10x0/tests/ipc/user/target.rs
@@ -4,9 +4,9 @@ #![no_std] #![no_main] -use cortex_m_semihosting::debug::{EXIT_FAILURE, EXIT_SUCCESS, exit}; +use console_backend::console_backend_write_all; use target_common::{TargetInterface, declare_target}; -use {console_backend as _, entry as _}; +use entry as _; pub struct Target {} @@ -21,11 +21,8 @@ fn shutdown(code: u32) -> ! { pw_log::info!("Shutting down with code {}", code as u32); - let status = match code { - 0 => EXIT_SUCCESS, - _ => EXIT_FAILURE, - }; - exit(status); + let sentinel: &[u8] = if code == 0 { b"TEST_RESULT:PASS\n" } else { b"TEST_RESULT:FAIL\n" }; + let _ = console_backend_write_all(sentinel); #[expect(clippy::empty_loop)] loop {} }
diff --git a/target/ast10x0/tests/threads/kernel/BUILD.bazel b/target/ast10x0/tests/threads/kernel/BUILD.bazel index 4504994..aad4c2c 100644 --- a/target/ast10x0/tests/threads/kernel/BUILD.bazel +++ b/target/ast10x0/tests/threads/kernel/BUILD.bazel
@@ -6,8 +6,6 @@ load("@pigweed//pw_kernel/tooling/panic_detector:rust_binary_no_panics_test.bzl", "rust_binary_no_panics_test") load("@rules_rust//rust:defs.bzl", "rust_binary") load("//target/ast10x0:defs.bzl", "TARGET_COMPATIBLE_WITH") -load("//target/ast10x0/harness:uart_upload_test.bzl", "uart_boot_image", "uart_upload_test") - system_image( name = "threads", kernel = ":target", @@ -23,17 +21,6 @@ target_compatible_with = TARGET_COMPATIBLE_WITH, ) -uart_boot_image( - name = "threads_uart_image", - image = ":threads", -) - -uart_upload_test( - name = "threads_uart_upload_test", - image = ":threads_uart_image", - test_timeout = 300, -) - rust_binary_no_panics_test( name = "no_panics_test", binary = ":threads", @@ -68,6 +55,5 @@ "@pigweed//pw_kernel/target:target_common", "@pigweed//pw_kernel/tests/threads/kernel:threads", "@pigweed//pw_log/rust:pw_log", - "@rust_crates//:cortex-m-semihosting", ], )
diff --git a/target/ast10x0/tests/threads/kernel/target.rs b/target/ast10x0/tests/threads/kernel/target.rs index 2c2e7fb..d9376d9 100644 --- a/target/ast10x0/tests/threads/kernel/target.rs +++ b/target/ast10x0/tests/threads/kernel/target.rs
@@ -7,9 +7,9 @@ #![no_main] use arch_arm_cortex_m::Arch; -use cortex_m_semihosting::debug::{EXIT_FAILURE, EXIT_SUCCESS, exit}; +use console_backend::console_backend_write_all; use target_common::{TargetInterface, declare_target}; -use {console_backend as _, entry as _}; +use entry as _; pub struct Target {} @@ -21,11 +21,11 @@ // SAFETY: `main` is only executed once, so we never generate more // than one `&mut` reference to `APP_STATE`. #[expect(static_mut_refs)] - let exit_status = match threads::main(Arch, unsafe { &mut APP_STATE }) { - Ok(()) => EXIT_SUCCESS, - Err(_e) => EXIT_FAILURE, + let sentinel: &[u8] = match threads::main(Arch, unsafe { &mut APP_STATE }) { + Ok(()) => b"TEST_RESULT:PASS\n", + Err(_e) => b"TEST_RESULT:FAIL\n", }; - exit(exit_status); + let _ = console_backend_write_all(sentinel); #[expect(clippy::empty_loop)] loop {} }
diff --git a/target/ast10x0/tests/unittest_runner/BUILD.bazel b/target/ast10x0/tests/unittest_runner/BUILD.bazel index 64f799c..abc253f 100644 --- a/target/ast10x0/tests/unittest_runner/BUILD.bazel +++ b/target/ast10x0/tests/unittest_runner/BUILD.bazel
@@ -21,6 +21,7 @@ "kernel", # Clippy lints do not work with `use_libtest_harness = False` "no_clippy", + "qemu_only", ], target_compatible_with = TARGET_COMPATIBLE_WITH, use_libtest_harness = False, @@ -33,6 +34,5 @@ "@pigweed//pw_kernel/subsys/console:console_backend", "@pigweed//pw_kernel/target:target_common", "@pigweed//pw_log/rust:pw_log", - "@rust_crates//:cortex-m-semihosting", ], )
diff --git a/target/ast10x0/tests/unittest_runner/target.rs b/target/ast10x0/tests/unittest_runner/target.rs index fb7ca9d..176eeca 100644 --- a/target/ast10x0/tests/unittest_runner/target.rs +++ b/target/ast10x0/tests/unittest_runner/target.rs
@@ -4,10 +4,10 @@ #![no_std] #![no_main] -use cortex_m_semihosting::debug::{EXIT_FAILURE, EXIT_SUCCESS, exit}; +use console_backend::console_backend_write_all; use target_common::{TargetInterface, declare_target}; use unittest_core::TestsResult; -use {console_backend as _, entry as _, integration_tests as _}; +use {entry as _, integration_tests as _}; pub struct Target {} @@ -20,11 +20,11 @@ // calling `run_all_tests` below. unsafe { target_common::run_ctors() }; - let exit_status = match unittest_core::run_all_tests!() { - TestsResult::AllPassed => EXIT_SUCCESS, - TestsResult::SomeFailed => EXIT_FAILURE, + let sentinel: &[u8] = match unittest_core::run_all_tests!() { + TestsResult::AllPassed => b"TEST_RESULT:PASS\n", + TestsResult::SomeFailed => b"TEST_RESULT:FAIL\n", }; - exit(exit_status); + let _ = console_backend_write_all(sentinel); loop {} } }
diff --git a/target/ast10x0/tests/usart/BUILD.bazel b/target/ast10x0/tests/usart/BUILD.bazel index 20ab68e..cac8839 100644 --- a/target/ast10x0/tests/usart/BUILD.bazel +++ b/target/ast10x0/tests/usart/BUILD.bazel
@@ -8,8 +8,6 @@ load("@pigweed//pw_kernel/tooling/panic_detector:rust_binary_no_panics_test.bzl", "rust_binary_no_panics_test") load("@rules_rust//rust:defs.bzl", "rust_binary") load("//target/ast10x0:defs.bzl", "TARGET_COMPATIBLE_WITH") -load("//target/ast10x0/harness:uart_upload_test.bzl", "uart_boot_image", "uart_upload_test") - filegroup( name = "system_config", srcs = ["system.json5"], @@ -46,7 +44,6 @@ "@pigweed//pw_kernel/subsys/console:console_backend", "@pigweed//pw_kernel/target:target_common", "@pigweed//pw_kernel/userspace", - "@pigweed//pw_log/rust:pw_log", "@rust_crates//:cortex-m-semihosting", ], ) @@ -111,20 +108,10 @@ visibility = ["//visibility:public"], ) -uart_boot_image( - name = "usart_uart_image", - image = ":usart", -) - -uart_upload_test( - name = "usart_uart_upload_test", - image = ":usart_uart_image", - test_timeout = 300, -) - system_image_test( name = "usart_test", image = ":usart", + tags = ["qemu_only"], target_compatible_with = TARGET_COMPATIBLE_WITH, )
diff --git a/workflows.json b/workflows.json index 8ea4a2f..d1f1919 100644 --- a/workflows.json +++ b/workflows.json
@@ -129,6 +129,7 @@ "args": [ "--keep_going", "--config=virt_ast10x0", + "--test_tag_filters=-hardware", "--test_output=streamed" ], "driver_options": {