diff options
| author | HiFiPhile <[email protected]> | 2026-06-23 23:12:51 +0200 |
|---|---|---|
| committer | HiFiPhile <[email protected]> | 2026-06-23 23:12:51 +0200 |
| commit | 033dc6bb77e8a9e1c1172d74d47707848b5c5bf5 (patch) | |
| tree | e7672f9bda54aa1a243b2a1c0f74a48055569c63 /test | |
| parent | b5e63ca44c846813ece9ad9df684cae0d1d5b542 (diff) | |
| parent | cd3561bf158afd5a5718904b8139a338d1e3b67c (diff) | |
Merge remote-tracking branch 'tinyusb/master' into fix-stm32-usbc
Diffstat (limited to 'test')
| -rw-r--r-- | test/fuzz/rules.mk | 1 | ||||
| -rw-r--r-- | test/hil/hfp.json | 4 | ||||
| -rw-r--r-- | test/hil/hil_ci.sh | 114 | ||||
| -rw-r--r-- | test/hil/hil_ci_set_matrix.py | 83 | ||||
| -rwxr-xr-x | test/hil/hil_test.py | 964 | ||||
| -rw-r--r-- | test/hil/requirements.txt | 11 | ||||
| -rw-r--r-- | test/hil/tinyusb.json | 343 | ||||
| -rw-r--r-- | test/unit-test/CMakeLists.txt | 4 | ||||
| -rw-r--r-- | test/unit-test/test/device/midi2/test_midi2_device.c | 266 | ||||
| -rw-r--r-- | test/unit-test/test/device/usbd/test_usbd.c | 52 | ||||
| -rw-r--r-- | test/unit-test/test/host/midi2/test_midi2_host.c | 101 |
11 files changed, 1624 insertions, 319 deletions
diff --git a/test/fuzz/rules.mk b/test/fuzz/rules.mk index 329dcce11..c14330312 100644 --- a/test/fuzz/rules.mk +++ b/test/fuzz/rules.mk @@ -23,7 +23,6 @@ SRC_C += \ src/tusb.c \ src/common/tusb_fifo.c \ src/device/usbd.c \ - src/device/usbd_control.c \ src/class/audio/audio_device.c \ src/class/cdc/cdc_device.c \ src/class/dfu/dfu_device.c \ diff --git a/test/hil/hfp.json b/test/hil/hfp.json index 8ba7a8f44..bb146d2fc 100644 --- a/test/hil/hfp.json +++ b/test/hil/hfp.json @@ -15,6 +15,10 @@ { "name": "stm32f746disco", "uid": "210041000C51343237303334", + "variant": [ + { "name": "stm32f746disco", "flags": "" }, + { "name": "stm32f746disco-DMA", "flags": "-DCFG_TUD_DWC2_DMA_ENABLE=1 -DCFG_TUH_DWC2_DMA_ENABLE=1" } + ], "tests": { "device": true, "host": false, "dual": false }, diff --git a/test/hil/hil_ci.sh b/test/hil/hil_ci.sh index fa8bb0245..3ec907979 100644 --- a/test/hil/hil_ci.sh +++ b/test/hil/hil_ci.sh @@ -1,15 +1,24 @@ -#!/bin/bash +#!/usr/bin/env bash # Run HIL test remotely on ci.lan # Usage: test/hil/hil_ci.sh [-b BOARD] [-t TEST] [extra hil_test.py args...] # Example: # test/hil/hil_ci.sh -b stm32f723disco # test/hil/hil_ci.sh -b stm32f723disco -t host/cdc_msc_hid -r 1 +# +# Env overrides: REMOTE, REMOTE_DIR, CONFIG (path to HIL config json), +# ROOT_DIR (tinyusb checkout to test; defaults to the script's own checkout). -set -e +set -euo pipefail + +REMOTE=${REMOTE:-ci.lan} +REMOTE_DIR=${REMOTE_DIR:-/tmp/tinyusb-hil} +ROOT_DIR=${ROOT_DIR:-$(cd "$(dirname "$0")/../.." && pwd)} +CONFIG=${CONFIG:-$ROOT_DIR/test/hil/tinyusb.json} -REMOTE=ci.lan -REMOTE_DIR=/tmp/tinyusb-hil -SCRIPT_DIR="$(cd "$(dirname "$0")/../.." && pwd)" +[[ -f "$ROOT_DIR/test/hil/hil_test.py" && -d "$ROOT_DIR/examples" ]] || { + echo "error: $ROOT_DIR does not look like a tinyusb checkout" >&2 + exit 1 +} # Parse -b BOARD from arguments to know which build to copy BOARD="" @@ -17,6 +26,7 @@ ARGS=() while [[ $# -gt 0 ]]; do case "$1" in -b) + [[ $# -ge 2 ]] || { echo "error: -b requires a BOARD argument" >&2; exit 1; } BOARD="$2" ARGS+=("$1" "$2") shift 2 @@ -28,42 +38,98 @@ while [[ $# -gt 0 ]]; do esac done -# Setup remote directory +# Setup remote directory. Use `bash -s` + heredoc so REMOTE_DIR (user-overridable) +# is passed as a positional parameter and never reinterpreted by the remote shell. echo "==> Setting up remote $REMOTE:$REMOTE_DIR" -ssh "$REMOTE" "rm -rf $REMOTE_DIR && mkdir -p $REMOTE_DIR/test/hil $REMOTE_DIR/examples" +ssh "$REMOTE" bash -s -- "$REMOTE_DIR" <<'REMOTE' +set -e +rm -rf -- "$1" +mkdir -p -- "$1/test/hil" "$1/examples" +REMOTE # Copy HIL test script and config echo "==> Copying test scripts" -scp -q "$SCRIPT_DIR/test/hil/hil_test.py" \ - "$SCRIPT_DIR/test/hil/pymtp.py" \ - "$SCRIPT_DIR/test/hil/tinyusb.json" \ +scp -q "$ROOT_DIR/test/hil/hil_test.py" \ + "$ROOT_DIR/test/hil/pymtp.py" \ + "$CONFIG" \ "$REMOTE:$REMOTE_DIR/test/hil/" -# Copy only firmware binaries (elf/bin/hex), preserving directory structure +# Copy only firmware binaries (elf/bin/hex) plus esptool metadata +# (config.env + flash_args needed by the esptool flasher), preserving structure copy_board_binaries() { local src="$1" - local board_name - board_name=$(basename "$src") - rsync -a --include='*/' --include='*.elf' --include='*.bin' --include='*.hex' --exclude='*' \ + rsync -a --prune-empty-dirs \ + --include='*/' --include='*.elf' --include='*.bin' --include='*.hex' \ + --include='config.env' --include='flash_args' \ + --exclude='*' \ "$src" "$REMOTE:$REMOTE_DIR/examples/" } if [ -n "$BOARD" ]; then - BUILD_DIR="$SCRIPT_DIR/examples/cmake-build-$BOARD" - if [ ! -d "$BUILD_DIR" ]; then - echo "Error: build directory not found: $BUILD_DIR" - echo "Build first with: cd examples && cmake -DBOARD=$BOARD -G Ninja -B cmake-build-$BOARD .. && cmake --build cmake-build-$BOARD" + # Copy the board's build dir plus its variant dirs. Variant names come from + # $CONFIG (they are not required to be prefixed with the board name); the + # cmake-build-<BOARD>-* glob is kept as a fallback for ad-hoc local builds. + # Collect only dirs that actually exist, deduplicated. + declare -A SEEN_DIRS=() + BUILD_DIRS=() + add_build_dir() { + [[ -d "$1" && -z "${SEEN_DIRS[$1]:-}" ]] || return 0 + SEEN_DIRS[$1]=1 + BUILD_DIRS+=("$1") + } + shopt -s nullglob + for d in "$ROOT_DIR"/examples/cmake-build-"$BOARD" "$ROOT_DIR"/examples/cmake-build-"$BOARD"-*; do + add_build_dir "$d" + done + shopt -u nullglob + while IFS= read -r v; do + add_build_dir "$ROOT_DIR/examples/cmake-build-$v" + done < <(python3 -c ' +import json, sys +cfg = json.load(open(sys.argv[1])) +for b in cfg.get("boards", []): + if b["name"] == sys.argv[2]: + for v in b.get("variant") or []: + print(v["name"]) +' "$CONFIG" "$BOARD") + if [ ${#BUILD_DIRS[@]} -eq 0 ]; then + echo "Error: no build directory found for $BOARD under $ROOT_DIR/examples/" + echo "Build first with: cd examples && cmake --preset $BOARD && cmake --build --preset $BOARD" exit 1 fi - echo "==> Copying binaries for $BOARD" - copy_board_binaries "$BUILD_DIR" + echo "==> Copying binaries for $BOARD (${#BUILD_DIRS[@]} build dir(s))" + for d in "${BUILD_DIRS[@]}"; do + copy_board_binaries "$d" + done else echo "==> Copying all built binaries" - for dir in "$SCRIPT_DIR"/examples/cmake-build-*/; do - [ -d "$dir" ] && copy_board_binaries "$dir" + # Use `%/` parameter expansion to strip the trailing slash from the glob — + # rsync needs the bare dir name so the per-board cmake-build-<BOARD>/ subdir + # is preserved on the remote (hil_test.py looks up binaries by that path). + for dir in "$ROOT_DIR"/examples/cmake-build-*/; do + [ -d "$dir" ] && copy_board_binaries "${dir%/}" done fi -# Run test +# Run test. Use `bash -s` so REMOTE_DIR + ARGS reach the remote shell as positional +# parameters; quoting and metacharacters in args are preserved. +CONFIG_BASENAME="$(basename "$CONFIG")" echo "==> Running HIL test on $REMOTE" -ssh -t "$REMOTE" "cd $REMOTE_DIR && python3 -u test/hil/hil_test.py -B examples ${ARGS[*]} tinyusb.json" +rc=0 +ssh "$REMOTE" bash -s -- "$REMOTE_DIR" "${ARGS[@]}" "test/hil/$CONFIG_BASENAME" <<'REMOTE' || rc=$? +cd -- "$1" +shift +# Flasher CLIs live in the user bin dirs on ci.lan (esptool/idf in ~/.local/bin, +# STM32CubeProgrammer's STM32_Programmer_CLI in ~/bin); the non-interactive shell +# subprocess used for flashing doesn't source profile/rc, so add them explicitly. +export PATH="$HOME/.local/bin:$HOME/bin:$PATH" +python3 -u test/hil/hil_test.py -B examples "$@" +REMOTE + +# Copy the generated report back to the local checkout (best-effort; the run's +# exit code is preserved regardless of whether a report was produced). +scp -q "$REMOTE:$REMOTE_DIR/hil_report.md" "$ROOT_DIR/hil_report.md" \ + && echo "==> Report copied to $ROOT_DIR/hil_report.md" \ + || echo "==> warning: no hil_report.md copied back" >&2 + +exit $rc diff --git a/test/hil/hil_ci_set_matrix.py b/test/hil/hil_ci_set_matrix.py index ecd964d87..13f7f1882 100644 --- a/test/hil/hil_ci_set_matrix.py +++ b/test/hil/hil_ci_set_matrix.py @@ -3,45 +3,70 @@ import json import os +def _resolve_config_path(config_file): + if os.path.exists(config_file): + return config_file + + script_relative = os.path.join(os.path.dirname(__file__), config_file) + if os.path.exists(script_relative): + return script_relative + + raise FileNotFoundError(f'Config file not found: {config_file}') + + def main(): parser = argparse.ArgumentParser() - parser.add_argument('config_file', help='Configuration JSON file') + parser.add_argument('config_files', nargs='+', help='Configuration JSON file(s)') args = parser.parse_args() - config_file = args.config_file - - # if config file is not found, try to find it in the same directory as this script - if not os.path.exists(config_file): - config_file = os.path.join(os.path.dirname(__file__), config_file) - with open(config_file) as f: - config = json.load(f) - + # Toolchain buckets must match the toolchains instantiated by the hil-build + # job in .github/workflows/build.yml. Keep all keys present (even if empty) + # so `fromJSON(hil_json)[toolchain]` always resolves to a list. matrix = { 'arm-gcc': [], + 'riscv-gcc': [], 'esp-idf': [] } - for board in config['boards']: - name = board['name'] - flasher = board['flasher'] - if flasher['name'] == 'esptool': - toolchain = 'esp-idf' - else: - toolchain = 'arm-gcc' - build_board = f'-b {name}' - if 'build' in board: - if 'args' in board['build']: - build_board += ' ' + ' '.join(f'-D{a}' for a in board['build']['args']) - if 'flags_on' in board['build']: - for f in board['build']['flags_on']: - if f == '': - matrix[toolchain].append(build_board) - else: - matrix[toolchain].append(f'{build_board} -f1 {f.replace(" ", " -f1 ")}') + seen = {toolchain: set() for toolchain in matrix} + + def append_build_arg(toolchain, build_arg): + if build_arg not in seen[toolchain]: + seen[toolchain].add(build_arg) + matrix[toolchain].append(build_arg) + + for config_file in args.config_files: + with open(_resolve_config_path(config_file)) as f: + config = json.load(f) + + for board in config['boards']: + name = board['name'] + flasher = board['flasher'] + # esptool boards must build under esp-idf; others default to arm-gcc + # but may opt into another bucket via an explicit "toolchain" field + # (e.g. RISC-V boards like ch32v20x need "riscv-gcc"). + if flasher['name'] == 'esptool': + toolchain = 'esp-idf' else: - matrix[toolchain].append(build_board) - else: - matrix[toolchain].append(build_board) + toolchain = board.get('toolchain', 'arm-gcc') + + build_board = f'-b {name}' + if 'build' in board and 'args' in board['build']: + build_board += ' ' + ' '.join(f'-D{a}' for a in board['build']['args']) + + # Each variant builds into cmake-build-<variant.name> with its own cmake + # -D defines and raw CFLAGS. No 'variant' -> a single build named after + # the board. + variants = board.get('variant') or [{'name': name, 'flags': ''}] + for v in variants: + arg = build_board + if v['name'] != name: + arg += f' --build-name {v["name"]}' + for d in v.get('defines', []): + arg += f' -D{d}' + for tok in v.get('flags', '').split(): + arg += f' --cflag={tok}' + append_build_arg(toolchain, arg) print(json.dumps(matrix)) diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py index d50a60894..07375c1ad 100755 --- a/test/hil/hil_test.py +++ b/test/hil/hil_test.py @@ -22,29 +22,37 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. +# Host setup (required: a missing tool fails its test rather than skipping it): +# - System packages: sudo apt install mtools libmtp9 alsa-utils iperf +# mtools - read_disk_file (device/cdc_msc, device/msc_dual_lun) +# libmtp9 - pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64 +# alsa-utils - arecord (device/audio_test_freertos) +# iperf - throughput tests (device/net_lwip_*) +# - Python packages: pip install -r requirements.txt +# # udev rules : # ACTION=="add", SUBSYSTEM=="tty", SUBSYSTEMS=="usb", MODE="0666", PROGRAM="/bin/sh -c 'echo $$ID_SERIAL_SHORT | rev | cut -c -8 | rev'", SYMLINK+="ttyUSB_%c.%s{bInterfaceNumber}" # ACTION=="add", SUBSYSTEM=="block", SUBSYSTEMS=="usb", ENV{ID_FS_USAGE}=="filesystem", MODE="0666", PROGRAM="/bin/sh -c 'echo $$ID_SERIAL_SHORT | rev | cut -c -8 | rev'", RUN{program}+="/usr/bin/systemd-mount --no-block --automount=yes --collect $devnode /media/blkUSB_%c.%s{bInterfaceNumber}" import argparse +import io import os import random import re +import select import sys import time -import warnings - -# Suppress pkg_resources deprecation warning from fs module -warnings.filterwarnings("ignore", message="pkg_resources is deprecated") -# Suppress pyfatfs unclean unmount warning -warnings.filterwarnings("ignore", message="Filesystem was not cleanly unmounted") +import signal +from contextlib import redirect_stdout +from pathlib import Path +from typing import Any, TypedDict, NotRequired, cast import serial import subprocess import json import glob -from multiprocessing import Pool -import fs +from multiprocessing import Pool, Lock +from multiprocessing import TimeoutError as MpTimeoutError import hashlib import ctypes from pymtp import MTP @@ -56,9 +64,97 @@ STATUS_OK = "\033[32mOK\033[0m" STATUS_FAILED = "\033[31mFailed\033[0m" STATUS_SKIPPED = "\033[33mSkipped\033[0m" +# Plain (non-ANSI) cell symbols for the markdown matrix report (hil_report.md). +# A missing binary is reported as skipped too. +REPORT_CELL = {'pass': '✅', 'fail': '❌', 'skip': '⚪'} + verbose = False test_only = [] +board_test = {} build_dir = 'cmake-build' +skip_flash = False +print_lock = None + + +def init_worker(lock): + global print_lock + print_lock = lock + + +def log_line(msg: str) -> None: + out = sys.__stdout__ if sys.__stdout__ is not None else sys.stdout + if print_lock is not None: + with print_lock: + print(msg, file=out, flush=True) + else: + print(msg, file=out, flush=True) + + +def compact_output(raw: str) -> str: + if not raw: + return '' + lines = [ln.strip() for ln in raw.replace('\r', '\n').split('\n') if ln.strip()] + return ' | '.join(lines) + +class FlasherCfg(TypedDict): + name: str + uid: str + args: str + + +class AttachedDevCfg(TypedDict, total=False): + vid_pid: str + serial: str + is_cdc: bool + is_msc: bool + block_count: int + block_size: int + + +class TestsCfg(TypedDict, total=False): + device: bool + dual: bool + host: bool + only: list[str] + skip: list[str] + dev_attached: list[AttachedDevCfg] + + +class BuildCfg(TypedDict, total=False): + args: list[str] + + +class VariantCfg(TypedDict, total=False): + name: str # build dir (cmake-build-<name>) and HIL report row + flags: str # raw CFLAGS, e.g. "-DCFG_TUD_DWC2_DMA_ENABLE=1" + defines: list[str] # cmake -D defines, e.g. ["RHPORT_DEVICE=1"] (vs flags which are compiler-only) + + +class Board(TypedDict): + name: str + uid: str + tests: TestsCfg + flasher: FlasherCfg + build: NotRequired[BuildCfg] + variant: NotRequired[list[VariantCfg]] + toolchain: NotRequired[str] # CI build bucket override, e.g. "riscv-gcc" (consumed by hil_ci_set_matrix.py) + + +class HilConfig(TypedDict): + boards: list[Board] + +CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180')) +POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '3000')) +SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5')) +SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10')) + + +def cmd_stdout_text(out: Any) -> str: + if out is None: + return '' + if isinstance(out, bytes): + return out.decode('utf-8', errors='ignore') + return str(out) WCH_RISCV_CONTENT = """ adapter driver wlinke @@ -88,8 +184,8 @@ issue at github.com/hathach/tinyusb" # ------------------------------------------------------------- # Path # ------------------------------------------------------------- -OPENCOD_ADI_PATH = f'{os.getenv("HOME")}/app/openocd_adi' -TINYUSB_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi' +TINYUSB_ROOT = Path(__file__).resolve().parents[2] # get usb serial by id def get_serial_dev(id, vendor_str, product_str, ifnum): @@ -116,13 +212,29 @@ def get_hid_dev(id, vendor_str, product_str, event): return f'/dev/input/by-id/usb-{vendor_str}_{product_str}_{id}-{event}' -def open_serial_dev(port): +def get_alsa_capture_dev(id): + pattern = f'/dev/snd/by-id/usb-*_{id}-*' + for dev in glob.glob(pattern): + try: + link = os.path.basename(os.path.realpath(dev)) + except OSError: + continue + m = re.match(r'controlC(\d+)', link) + if m: + return f'hw:{m.group(1)},0' + return None + + +def open_serial_dev(port: str): timeout = ENUM_TIMEOUT ser = None while timeout > 0: if os.path.exists(port): try: - ser = serial.Serial(port, baudrate=115200, timeout=5) + # write_timeout: a wedged device otherwise blocks ser.write() forever, + # hanging the worker until the pool/job timeout kills the whole run + ser = serial.Serial(port, baudrate=115200, timeout=SERIAL_READ_TIMEOUT, + write_timeout=SERIAL_WRITE_TIMEOUT) break except serial.SerialException: print(f'serial {port} not reaady {timeout} sec') @@ -131,28 +243,39 @@ def open_serial_dev(port): timeout -= 0.1 assert timeout > 0, f'Cannot open port f{port}' if os.path.exists(port) else f'Port {port} not existed' + assert ser is not None return ser -def read_disk_file(uid, lun, fname): - # open_fs("fat://{dev}) require 'pip install pyfatfs' +def serial_write_all(ser: serial.Serial, data: bytes): + # write_timeout is a total deadline for the whole call (pyserial keeps partial progress + # internally). A timeout means the device stopped draining — treat it as fatal: pyserial + # loses the partial-write count on raise, so retrying would duplicate bytes on the wire. + try: + ser.write(data) + except serial.SerialTimeoutException: + raise AssertionError(f'Serial write timeout after {SERIAL_WRITE_TIMEOUT:.1f}s') + + +def read_disk_file(uid: str, lun: int, fname: str) -> bytes: + # Reads a file from a FAT volume on a block device without mounting it. + # Requires mtools: `apt install mtools` (no pip dependency). dev = get_disk_dev(uid, 'TinyUSB', lun) timeout = ENUM_TIMEOUT + last_err = None while timeout > 0: if os.path.exists(dev): - fat = fs.open_fs(f'fat://{dev}?read_only=true') try: - with fat.open(fname, 'rb') as f: - data = f.read() - finally: - fat.close() - assert data, f'Cannot read file {fname} from {dev}' - return data + data = subprocess.check_output( + ['mtype', '-i', dev, f'::/{fname}'], stderr=subprocess.PIPE) + assert data, f'Cannot read file {fname} from {dev}' + return data + except subprocess.CalledProcessError as e: + last_err = e.stderr.decode(errors='replace').strip() time.sleep(1) timeout -= 1 - assert timeout > 0, f'Storage {dev} not existed' - return None + raise AssertionError(f'mtype failed on {dev}: {last_err}' if last_err else f'Storage {dev} not existed') def open_mtp_dev(uid): @@ -174,7 +297,7 @@ def open_mtp_dev(uid): return None -def get_printer_dev(id, vendor_str, product_str, ifnum): +def get_printer_dev(id: str, vendor_str, product_str, ifnum: int): """Find /dev/usb/lpX by matching USB serial, vendor, product, and interface number via sysfs""" vendor_str = vendor_str.replace(' ', '_') if vendor_str else '' product_str = product_str.replace(' ', '_') if product_str else '' @@ -188,7 +311,7 @@ def get_printer_dev(id, vendor_str, product_str, ifnum): return None -def open_printer_dev(id, vendor_str, product_str, ifnum): +def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str: """Wait for printer device to enumerate and return its path""" timeout = ENUM_TIMEOUT while timeout > 0: @@ -203,41 +326,77 @@ def open_printer_dev(id, vendor_str, product_str, ifnum): # ------------------------------------------------------------- # Flashing firmware # ------------------------------------------------------------- -def run_cmd(cmd, cwd=None): - r = subprocess.run(cmd, cwd=cwd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) +def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess: + popen_kwargs = { + 'cwd': cwd, + 'shell': True, + 'stdout': subprocess.PIPE, + 'stderr': subprocess.STDOUT, + 'text': True, + 'encoding': 'utf-8', + 'errors': 'replace', + } + if os.name != 'nt': + popen_kwargs['preexec_fn'] = os.setsid + + p = subprocess.Popen(cmd, **popen_kwargs) + try: + out, _ = p.communicate(timeout=timeout) + r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out) + except subprocess.TimeoutExpired as ex: + if os.name != 'nt': + try: + os.killpg(p.pid, signal.SIGKILL) + except ProcessLookupError: + pass + else: + p.kill() + out, _ = p.communicate() + timeout_out = ex.stdout or out or b'' + title = f'COMMAND TIMEOUT ({timeout}s): {cmd}' + print() + if os.getenv('CI'): + print(f"::group::{title}") + print(cmd_stdout_text(timeout_out)) + print(f"::endgroup::") + else: + print(title) + print(cmd_stdout_text(timeout_out)) + return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out) + if r.returncode != 0: title = f'COMMAND FAILED: {cmd}' print() if os.getenv('CI'): print(f"::group::{title}") - print(r.stdout.decode("utf-8")) + print(cmd_stdout_text(r.stdout)) print(f"::endgroup::") else: print(title) - print(r.stdout.decode("utf-8")) + print(cmd_stdout_text(r.stdout)) elif verbose: print(cmd) - print(r.stdout.decode("utf-8")) + print(cmd_stdout_text(r.stdout)) return r -def flash_jlink(board, firmware): +def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess: flasher = board['flasher'] script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit'] - f_jlink = f'{board["name"]}_{os.path.basename(firmware)}.jlink' - with open(f_jlink, 'w') as f: + f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink') + with f_jlink.open('w') as f: f.writelines(f'{s}\n' for s in script) ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}') - os.remove(f_jlink) + f_jlink.unlink(missing_ok=True) return ret -def reset_jlink(board): +def reset_jlink(board: Board) -> subprocess.CompletedProcess: flasher = board['flasher'] script = ['halt', 'r', 'go', 'exit'] - f_jlink = f'{board["name"]}_reset.jlink' - if not os.path.exists(f_jlink): - with open(f_jlink, 'w') as f: + f_jlink = Path(f'{board["name"]}_reset.jlink') + if not f_jlink.exists(): + with f_jlink.open('w') as f: f.writelines(f'{s}\n' for s in script) ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}') return ret @@ -300,16 +459,20 @@ def reset_openocd_wch(board): return ret -def flash_openocd_adi(board, firmware): +def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess: flasher = board['flasher'] - ret = run_cmd(f'{OPENCOD_ADI_PATH}/src/openocd -c "adapter serial {flasher["uid"]}" -s {OPENCOD_ADI_PATH}/tcl ' + openocd = OPENCOD_ADI_PATH / 'src' / 'openocd' + tcl_dir = OPENCOD_ADI_PATH / 'tcl' + ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} ' f'{flasher["args"]} -c "program {firmware}.elf reset exit"') return ret -def reset_openocd_adi(board): +def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess: flasher = board['flasher'] - ret = run_cmd(f'{OPENCOD_ADI_PATH}/src/openocd -c "adapter serial {flasher["uid"]}" -s {OPENCOD_ADI_PATH}/tcl ' + openocd = OPENCOD_ADI_PATH / 'src' / 'openocd' + tcl_dir = OPENCOD_ADI_PATH / 'tcl' + ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} ' f'{flasher["args"]} -c "program reset exit"') return ret @@ -328,17 +491,17 @@ def reset_wlink_rs(board): return ret -def flash_esptool(board, firmware): +def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess: flasher = board['flasher'] port = get_serial_dev(flasher["uid"], None, None, 0) - fw_dir = os.path.dirname(f'{firmware}.bin') - with open(f'{fw_dir}/config.env') as f: + fw_dir = Path(f'{firmware}.bin').parent + with (fw_dir / 'config.env').open() as f: idf_target = json.load(f)['IDF_TARGET'] - with open(f'{fw_dir}/flash_args') as f: + with (fw_dir / 'flash_args').open() as f: flash_args = f.read().strip().replace('\n', ' ') command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} ' f'--before=default_reset --after=hard_reset write_flash {flash_args}') - ret = run_cmd(command, cwd=fw_dir) + ret = run_cmd(command, cwd=str(fw_dir)) return ret @@ -358,6 +521,19 @@ def reset_uniflash(board): return subprocess.CompletedProcess(args=['dummy'], returncode=0) +def flash_lm4flash(board, firmware): + # TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write + flasher = board['flasher'] + ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin') + return ret + + +def reset_lm4flash(board): + # lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op + flasher = board['flasher'] + return subprocess.CompletedProcess(args=['dummy'], returncode=0) + + # ------------------------------------------------------------- # Tests: dual # ------------------------------------------------------------- @@ -562,8 +738,7 @@ def test_host_cdc_msc_hid(board): offset = 0 while offset < echo_len: chunk_size = min(random.randint(1, packet_size), echo_len - offset) - ser.write(echo_data[offset:offset + chunk_size]) - ser.flush() + serial_write_all(ser, echo_data[offset:offset + chunk_size]) # wait until this chunk is echoed back echo = b'' t_end = time.monotonic() + 1.0 @@ -612,8 +787,7 @@ def test_host_msc_file_explorer(board): time.sleep(1) ser.reset_input_buffer() for ch in 'cat README.TXT\r': - ser.write(ch.encode()) - ser.flush() + serial_write_all(ser, ch.encode()) time.sleep(0.002) resp = b'' @@ -635,8 +809,7 @@ def test_host_msc_file_explorer(board): time.sleep(0.5) ser.reset_input_buffer() for ch in 'dd 1024\r': - ser.write(ch.encode()) - ser.flush() + serial_write_all(ser, ch.encode()) time.sleep(0.002) # Read dd output until prompt @@ -652,12 +825,21 @@ def test_host_msc_file_explorer(board): t -= 0.05 resp_text = resp.decode('utf-8', errors='ignore') + speed = None for line in resp_text.splitlines(): if 'KB/s' in line: print(f'{line.strip()} ', end='') + m = re.search(r'([\d.]+\s*[KMG]B/s)', line) # MSC read speed for the report cell + if m: + speed = 'rd ' + m.group(1).replace(' ', '') break ser.close() + return speed + + +def test_host_msc_file_explorer_freertos(board): + return test_host_msc_file_explorer(board) # ------------------------------------------------------------- @@ -681,7 +863,7 @@ def test_device_cdc_dual_ports(board): sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)] - def write_and_check(writer, payload): + def write_and_check(writer, payload : bytes): payload_len = len(payload) for s in ser: s.reset_input_buffer() @@ -691,8 +873,7 @@ def test_device_cdc_dual_ports(board): # Write in chunks of random 1-64 bytes (device has 64-byte buffer) while offset < payload_len: chunk_size = min(random.randint(1, 64), payload_len - offset) - ser[writer].write(payload[offset:offset + chunk_size]) - ser[writer].flush() + serial_write_all(ser[writer], payload[offset:offset + chunk_size]) rd0 += ser[0].read(chunk_size) rd1 += ser[1].read(chunk_size) offset += chunk_size @@ -726,8 +907,7 @@ def test_device_cdc_msc(board): # Write in chunks of random 1-64 bytes (device has 64-byte buffer) while offset < size: chunk_size = min(random.randint(1, 64), size - offset) - ser.write(test_str[offset:offset + chunk_size]) - ser.flush() + serial_write_all(ser, test_str[offset:offset + chunk_size]) rd_str += ser.read(chunk_size) offset += chunk_size assert rd_str == test_str, f'CDC wrong data ({size} bytes):\n expected: {test_str}\n received: {rd_str}' @@ -737,56 +917,84 @@ def test_device_cdc_msc(board): data = read_disk_file(uid, 0, 'README.TXT') assert data == MSC_README_TXT, f'MSC wrong data in README.TXT\n expected: {MSC_README_TXT.decode()}\n received: {data.decode()}' - # MSC dd throughput test: read all sectors then write back same data + +def test_device_cdc_msc_freertos(board): + test_device_cdc_msc(board) + + +def test_device_cdc_msc_throughput(board): + uid = board['uid'] + + def parse_speed(dd_output): + for line in dd_output.splitlines(): + m = re.search(r'([\d.]+)\s+([kMG]?B)/s', line) + if m: + return f'{float(m.group(1)):.1f} {m.group(2)}ps' + return '?' + + # Wait for MSC disk enumeration dev = get_disk_dev(uid, 'TinyUSB', 0) timeout = ENUM_TIMEOUT while timeout > 0: if os.path.exists(dev): break - time.sleep(1) - timeout -= 1 - assert timeout > 0, f'Disk {dev} not found for dd test' + time.sleep(0.1); timeout -= 0.1 + assert timeout > 0, f'Disk {dev} not found' - block_count = 16 - block_size = 512 - tmp_file = f'/tmp/msc_dd_{uid}.bin' + # Wait for CDC tty enumeration + tty = get_serial_dev(uid, 'TinyUSB', 'Throughput', 0) + timeout = ENUM_TIMEOUT + while timeout > 0: + if os.path.exists(tty): + break + time.sleep(0.1); timeout -= 0.1 + assert timeout > 0, f'CDC tty {tty} not found' - # dd reports speed based on payload only. Each block also transfers 31-byte CBW + 13-byte CSW on USB. - scsi_ratio = (block_size + 31 + 13) / block_size + # Detect speed (12 Mbps FS / 480 Mbps HS) for payload scaling + is_fs = False + for f in glob.glob('/sys/bus/usb/devices/*/serial'): + try: + if open(f).read().strip() == uid: + is_fs = (open(os.path.join(os.path.dirname(f), 'speed')).read().strip() == '12') + break + except (OSError, ValueError): + pass - def parse_dd_speed(dd_output): - """Parse dd output, return USB-adjusted speed string""" - for line in dd_output.splitlines(): - m = re.search(r'([\d.]+)\s+([kMG]?B/s)', line) - if m: - speed_val = float(m.group(1)) * scsi_ratio - return f'{speed_val:.1f} {m.group(2)}' - return '' + # Put tty in raw mode so dd sees pure binary throughput. + rs = run_cmd(f'timeout 30 stty -F {tty} raw -echo') + assert rs.returncode == 0, f'stty failed: {cmd_stdout_text(rs.stdout)}' - # Read: dd from device to file - ret = run_cmd(f'dd if={dev} of={tmp_file} bs={block_size} count={block_count} iflag=direct 2>&1') - assert ret.returncode == 0, f'dd read failed: {ret.stdout.decode()}' - read_speed = parse_dd_speed(ret.stdout.decode()) + # Payload aim: ~5 s per direction at FS (~830 kB/s), much less at HS. + msc_count = 2 if is_fs else 16 # bs=1M + cdc_count = 16 if is_fs else 128 # bs=64K - # Write back the same data to avoid corrupting the disk (skip if read-only) - ret = run_cmd(f'dd if={tmp_file} of={dev} bs={block_size} count={block_count} oflag=direct 2>&1') - if ret.returncode != 0 and 'Read-only' in ret.stdout.decode(): - write_speed = 'skip (read-only)' - else: - assert ret.returncode == 0, f'dd write failed: {ret.stdout.decode()}' - write_speed = parse_dd_speed(ret.stdout.decode()) + tmp_file = f'/tmp/cdc_msc_tp_{uid}.bin' + + rw = run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1') + assert rw.returncode == 0, f'CDC dd write failed: {cmd_stdout_text(rw.stdout)}' + cdc_w = parse_speed(cmd_stdout_text(rw.stdout)) + + rr = run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1') + assert rr.returncode == 0, f'CDC dd read failed: {cmd_stdout_text(rr.stdout)}' + cdc_r = parse_speed(cmd_stdout_text(rr.stdout)) + + rmr = run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1') + assert rmr.returncode == 0, f'MSC dd read failed: {cmd_stdout_text(rmr.stdout)}' + msc_r = parse_speed(cmd_stdout_text(rmr.stdout)) + + rmw = run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1') + assert rmw.returncode == 0, f'MSC dd write failed: {cmd_stdout_text(rmw.stdout)}' + msc_w = parse_speed(cmd_stdout_text(rmw.stdout)) try: os.remove(tmp_file) except OSError: pass - if read_speed and write_speed: - print(f' dd read: {read_speed}, write: {write_speed}', end='') - - -def test_device_cdc_msc_freertos(board): - test_device_cdc_msc(board) + print(f' CDC read {cdc_r} write {cdc_w}, MSC read {msc_r} write {msc_w} ', end='') + # compact read/write speed for the report cell, e.g. "✅ CDC 652k/422k MSC 1.1M/783k" + short = lambda s: (s.split()[0].rstrip('0').rstrip('.') + s.split()[-1][0]) if ' ' in s else s + return f'{REPORT_CELL["pass"]} CDC {short(cdc_r)}/{short(cdc_w)} MSC {short(msc_r)}/{short(msc_w)}' def test_device_dfu(board): @@ -796,7 +1004,7 @@ def test_device_dfu(board): timeout = ENUM_TIMEOUT while timeout > 0: ret = run_cmd(f'dfu-util -l') - stdout = ret.stdout.decode() + stdout = cmd_stdout_text(ret.stdout) if f'serial="{uid}"' in stdout and 'Found DFU: [cafe:4000]' in stdout: break time.sleep(1) @@ -836,7 +1044,7 @@ def test_device_dfu_runtime(board): timeout = ENUM_TIMEOUT while timeout > 0: ret = run_cmd(f'dfu-util -l') - stdout = ret.stdout.decode() + stdout = cmd_stdout_text(ret.stdout) if f'serial="{uid}"' in stdout and 'Found Runtime: [cafe:4000]' in stdout: break time.sleep(1) @@ -896,18 +1104,27 @@ def test_device_printer_to_cdc(board): ser.reset_input_buffer() # Test 1: Printer -> CDC with multiple sizes, write in random 1-64 byte chunks + LP_WRITE_TIMEOUT = 5.0 # seconds; firmware may stall draining the printer OUT endpoint for size in sizes: test_data = rand_ascii(size) ser.reset_input_buffer() rd = b'' offset = 0 - with open(lp_dev, 'wb') as lp: + lp_fd = os.open(lp_dev, os.O_WRONLY | os.O_NONBLOCK) + try: while offset < size: chunk_size = min(random.randint(1, 64), size - offset) - lp.write(test_data[offset:offset + chunk_size]) - lp.flush() + buf = test_data[offset:offset + chunk_size] + written = 0 + while written < len(buf): + _, wr, _ = select.select([], [lp_fd], [], LP_WRITE_TIMEOUT) + assert wr, f'Printer write timeout after {LP_WRITE_TIMEOUT}s (firmware not draining OUT endpoint)' + n = os.write(lp_fd, buf[written:]) + written += n rd += ser.read(chunk_size) offset += chunk_size + finally: + os.close(lp_fd) # read any remaining bytes (fullspeed devices may need extra time) while len(rd) < size: remaining = ser.read(size - len(rd)) @@ -954,8 +1171,7 @@ def test_device_printer_to_cdc(board): offset = 0 while offset < size: chunk_size = min(random.randint(1, 64), size - offset) - ser.write(test_data[offset:offset + chunk_size]) - ser.flush() + serial_write_all(ser, test_data[offset:offset + chunk_size]) time.sleep(0.01) offset += chunk_size @@ -1028,6 +1244,65 @@ def test_device_mtp(board): mtp.disconnect() +def test_device_net_lwip_webserver(board): + # MAC hard-coded in examples/device/net_lwip_webserver/src/main.c; Linux names the + # USB network interface enx<MAC_lowercase_no_colons>. Device IP is 192.168.7.1 and + # the example runs an iperf2 TCP server on port 5001 (INCLUDE_IPERF). + import socket + mac_no_colons = '0202846a9600' + iface = 'enx' + mac_no_colons + device_ip = '192.168.7.1' + iperf_port = 5001 + + # Wait for the host to get an IPv4 address in the device's subnet (DHCP served by the device). + # USB enum + DHCP serve can take longer on the CI HIL hardware than on local — give it 30s. + iface_timeout = 30 + deadline = time.time() + iface_timeout + host_ip = None + while time.time() < deadline: + ret = subprocess.run(['ip', '-o', '-4', 'addr', 'show', iface], + capture_output=True, text=True, timeout=2) + m = re.search(r'inet (192\.168\.7\.\d+)/', ret.stdout) if ret.returncode == 0 else None + if m: + host_ip = m.group(1) + break + time.sleep(0.5) + assert host_ip, f'USB net iface {iface} did not come up with 192.168.7.x within {iface_timeout}s' + + # Poll the iperf TCP port until the device is accepting. The net stack comes up a bit + # after DHCP completes; iperf server binding isn't instantaneous after reflash. + deadline = time.time() + ENUM_TIMEOUT + last_err = None + while time.time() < deadline: + try: + with socket.create_connection((device_ip, iperf_port), timeout=1): + last_err = None + break + except OSError as e: + last_err = e + time.sleep(0.3) + assert last_err is None, f'iperf TCP {device_ip}:{iperf_port} not accepting within {ENUM_TIMEOUT}s: {last_err}' + + # Throughput: 5-second iperf2 TCP test, CSV output for stable parsing. + # iperf2 CSV final summary line: timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps + ret = subprocess.run(['iperf', '-c', device_ip, '-t', '5', '-y', 'C'], + capture_output=True, text=True, timeout=30) + stderr = ret.stderr.strip() + stdout = ret.stdout.strip() + assert ret.returncode == 0, f'iperf rc={ret.returncode}: stderr={stderr!r} stdout={stdout!r}' + lines = [l for l in stdout.splitlines() if l] + assert lines, f'iperf produced no output (rc={ret.returncode}, stderr={stderr!r})' + try: + bps = int(lines[-1].split(',')[-1]) + except (ValueError, IndexError) as e: + raise AssertionError(f'could not parse iperf output: {lines[-1]!r} ({e})') + mbps = bps / 1e6 + print(f' iperf {mbps:5.1f} Mbps', end='') + + # Reject implausibly low throughput - a working USB-net link should clear this easily. + assert mbps >= 1.0, f'iperf throughput too low: {mbps:.2f} Mbps' + + def test_device_msc_dual_lun(board): uid = board['uid'] @@ -1098,9 +1373,78 @@ def test_device_midi_test(board): assert n in note_sequence, f'Unexpected MIDI note {n}' +def test_device_audio_test_freertos(board): + uid = board['uid'] + + if os.name == 'nt': + return 'skipped' + + pcm = None + timeout = ENUM_TIMEOUT + while timeout > 0: + pcm = get_alsa_capture_dev(uid) + if pcm: + break + time.sleep(1) + timeout -= 1 + + assert pcm is not None, f'ALSA capture device not found for {uid}' + + raw_path = f'/tmp/tinyusb_audio_{uid}.raw' + cmd = [ + 'arecord', + '-D', pcm, + '-q', + '-f', 'S16_LE', + '-c', '1', + '-r', '48000', + '-d', '2', + '-t', 'raw', + raw_path, + ] + + ret = subprocess.run(cmd, capture_output=True, text=True, timeout=20) + assert ret.returncode == 0, f'arecord failed: {ret.stderr.strip() or ret.stdout.strip()}' + + try: + with open(raw_path, 'rb') as f: + raw = f.read() + finally: + try: + os.remove(raw_path) + except OSError: + pass + + assert len(raw) >= 48000, f'Captured too little audio: {len(raw)} bytes' + assert (len(raw) % 2) == 0, f'Invalid 16-bit audio length: {len(raw)}' + + sample_count = len(raw) // 2 + samples = [int.from_bytes(raw[i:i + 2], 'little', signed=False) for i in range(0, len(raw), 2)] + assert sample_count > 1024, f'Not enough samples captured: {sample_count}' + + # The firmware sends a continuous uint16 ramp. Using ALSA hw: capture bypasses + # PulseAudio processing, so most adjacent samples should differ by exactly 1. + total_diffs = sample_count - 1 + one_step = 0 + near_step = 0 + for i in range(total_diffs): + d = (samples[i + 1] - samples[i]) & 0xFFFF + if d == 1: + one_step += 1 + if d in (0, 1, 2, 47, 48, 49): + near_step += 1 + + one_ratio = one_step / total_diffs + near_ratio = near_step / total_diffs + assert one_ratio >= 0.85, f'Unexpected audio pattern (strict ratio={one_ratio:.3f})' + assert near_ratio >= 0.98, f'Unexpected audio pattern (relaxed ratio={near_ratio:.3f})' + + print(f' ALSA {pcm} strict={one_ratio:.3f} relaxed={near_ratio:.3f}', end='') + + def test_device_hid_generic_inout(board): uid = board['uid'] - import hid + import hid # cython-hidapi (pip: hidapi, apt: python3-hid) # Find HID device by UID (VID=0xCafe) timeout = ENUM_TIMEOUT @@ -1116,22 +1460,23 @@ def test_device_hid_generic_inout(board): timeout -= 1 assert dev is not None, f'HID device not found for {uid}' - h = hid.Device(vid=dev['vendor_id'], pid=dev['product_id'], serial=uid) - - # Echo test: send random data and verify echo - for size in [8, 32, 63]: - # Report ID (0) + payload, padded to 64 bytes - payload = bytes([random.randint(1, 255) for _ in range(size)]) - report = bytes([0]) + payload + bytes(64 - size) - h.write(report) - echo = h.read(64, timeout=2000) - assert echo is not None and len(echo) >= size, ( - f'HID echo timeout or short read ({size} bytes)') - assert bytes(echo[:size]) == payload, ( - f'HID echo wrong data ({size} bytes):\n' - f' expected: {payload.hex()}\n received: {bytes(echo[:size]).hex()}') - - h.close() + h = hid.device() + h.open(dev['vendor_id'], dev['product_id'], uid) + try: + # Echo test: send random data and verify echo + for size in [8, 32, 63]: + # Report ID (0) + payload, padded to 64 bytes + payload = bytes([random.randint(1, 255) for _ in range(size)]) + report = bytes([0]) + payload + bytes(64 - size) + h.write(report) + echo = h.read(64, 2000) + assert echo and len(echo) >= size, ( + f'HID echo timeout or short read ({size} bytes)') + assert bytes(echo[:size]) == payload, ( + f'HID echo wrong data ({size} bytes):\n' + f' expected: {payload.hex()}\n received: {bytes(echo[:size]).hex()}') + finally: + h.close() # ------------------------------------------------------------- @@ -1140,9 +1485,15 @@ def test_device_hid_generic_inout(board): # device tests # note don't test 2 examples with cdc or 2 msc next to each other device_tests = [ + # Order matters: cdc_msc and cdc_msc_throughput share the same VID:PID (cafe:4003), so keep a + # differently-PID'd example (dfu, cafe:4000) between them. Boards whose CPU-reset does not drop + # D+ (e.g. WCH CH58x via openocd) only re-enumerate when the PID changes; back-to-back same-PID + # firmware would otherwise leave the host on the previous example's cached descriptors. 'device/cdc_dual_ports', - 'device/dfu', 'device/cdc_msc', + 'device/dfu', + 'device/cdc_msc_throughput', + 'device/audio_test_freertos', 'device/dfu_runtime', 'device/cdc_msc_freertos', 'device/hid_boot_interface', @@ -1150,7 +1501,8 @@ device_tests = [ 'device/hid_generic_inout', 'device/printer_to_cdc', 'device/midi_test', - 'device/mtp' + 'device/mtp', + # 'device/net_lwip_webserver', # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host ] dual_tests = [ @@ -1160,85 +1512,179 @@ dual_tests = [ host_test = [ 'host/cdc_msc_hid', 'host/msc_file_explorer', + 'host/msc_file_explorer_freertos', 'host/device_info', ] -def test_example(board, f1, example): +def find_firmware(variant: str, example: str): + """Locate a built example's firmware base path (no extension) under + cmake-build-<variant>/<example>/. Accepts the single-config layout (firmware + directly in the example dir) or Ninja Multi-Config (a per-config subdir like + RelWithDebInfo/). Returns the base Path, or None if not built.""" + fw_dir = TINYUSB_ROOT / build_dir / f'cmake-build-{variant}' / example + base = Path(example).name + if fw_dir.is_dir(): + for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base, + *(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]: + if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists(): + return cand + return None + + +def test_example(board: Board, variant: str, example: str) -> tuple[int, str]: """ Test example firmware :param board: board dict - :param f1: flags on + :param variant: build variant name = build dir (cmake-build-<variant>) and report row :param example: example name - :return: 0 if success/skip, 1 if failed + :return: (err_count, status, metric) where err_count is 0 on success/skip or + 1 on failure, status is one of 'pass'/'fail'/'skip' (a missing binary + counts as 'skip'), and metric is an optional string a test returns to + show in its report cell instead of the pass symbol (e.g. speed) """ - name = board['name'] err_count = 0 + result_status = 'fail' + metric = None - f1_str = "" - if f1 != "": - f1_str = '-f1_' + f1.replace(' ', '_') + test_name = f'{variant:40} {example:30} ...' - fw_dir = f'{TINYUSB_ROOT}/{build_dir}/cmake-build-{name}{f1_str}/{example}' - fw_name = f'{fw_dir}/{os.path.basename(example)}' - print(f'{name+f1_str:40} {example:30} ...', end='') - - if not os.path.exists(fw_dir) or not (os.path.exists(f'{fw_name}.elf') or os.path.exists(f'{fw_name}.bin')): - print('Skip (no binary)') - return 0 + fw_name = find_firmware(variant, example) + if fw_name is None: + log_line(f'{test_name} Skip (no binary)') + return 0, 'skip', None if verbose: - print(f'Flashing {fw_name}.elf') + log_line(f'Flashing {fw_name}.elf') - # flash firmware. It may fail randomly, retry a few times + # flash firmware (unless --skip-flash), then run the test. Both may fail randomly, + # retry a few times. start_s = time.time() + flash_ok = True + last_err = '' + last_detail = '' for i in range(max_retry): - ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, fw_name) - if ret.returncode == 0: - try: - tret = globals()[f'test_{example.replace("/", "_")}'](board) - if tret == 'skipped': - print(f' {STATUS_SKIPPED}', end='') - else: - print(' OK', end='') - break - except Exception as e: - if i == max_retry - 1: - err_count += 1 - print(f'{STATUS_FAILED}: {e}') - else: - print(f'\n Test failed: {e}, retry {i+2}/{max_retry}', end='') - time.sleep(0.5) - else: - print(f'\n Flash failed, retry {i+2}/{max_retry}', end='') - time.sleep(0.5) + attempt_out = io.StringIO() + with redirect_stdout(attempt_out): + if not skip_flash: + ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, str(fw_name)) + flash_ok = (ret.returncode == 0) + if flash_ok: + try: + tret = globals()[f'test_{example.replace("/", "_")}'](board) + last_detail = compact_output(attempt_out.getvalue()) + if tret == 'skipped': + status = STATUS_SKIPPED + result_status = 'skip' + else: + status = STATUS_OK + result_status = 'pass' + # a test may return a string to show in its report cell (e.g. speed) + metric = tret if isinstance(tret, str) else None + msg = f'{test_name} {status}' + if last_detail: + msg += f' {last_detail}' + msg += f' in {time.time() - start_s:.1f}s' + log_line(msg) + break + except Exception as e: + last_err = str(e) + last_detail = compact_output(attempt_out.getvalue()) + if i == max_retry - 1: + err_count += 1 + msg = f'{test_name} {STATUS_FAILED}: {e}' + if last_detail: + msg += f' {last_detail}' + msg += f' in {time.time() - start_s:.1f}s' + log_line(msg) + else: + msg = f'{test_name} retry {i+2}/{max_retry}: test failed: {e}' + if last_detail: + msg += f' {last_detail}' + log_line(msg) + time.sleep(0.5) + else: + last_err = 'Flash failed' + last_detail = compact_output(attempt_out.getvalue()) + if i < max_retry - 1: + msg = f'{test_name} retry {i+2}/{max_retry}: flash failed' + if last_detail: + msg += f' {last_detail}' + log_line(msg) + time.sleep(0.5) - if ret.returncode != 0: + if not flash_ok: err_count += 1 - print(f' Flash {STATUS_FAILED}', end='') + msg = f'{test_name} Flash {STATUS_FAILED}' + if last_err: + msg += f': {last_err}' + if last_detail: + msg += f' {last_detail}' + msg += f' in {time.time() - start_s:.1f}s' + log_line(msg) - print(f' in {time.time() - start_s:.1f}s') + return err_count, result_status, metric - return err_count +def build_board(board: Board) -> tuple[str, int]: + """Build firmware for this board via tools/build.py. + Honors board config's variant list and build.args defines. + Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout).""" + name = board['name'] + bcfg = cast(BuildCfg, board.get('build', {})) + extra_defs = bcfg.get('args', []) + variants = board.get('variant') or [{'name': name, 'flags': ''}] -def test_board(board): + failed = 0 + for v in variants: + cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name] + for d in extra_defs: + cmd += ['-D', d] + if v['name'] != name: + cmd += ['--build-name', v['name']] + for d in v.get('defines', []): + cmd += ['-D', d] + for tok in v.get('flags', '').split(): + cmd += [f'--cflag={tok}'] + if verbose: + cmd.append('-v') + print(f' + {" ".join(cmd)}') + r = subprocess.run(cmd, cwd=TINYUSB_ROOT) + if r.returncode != 0: + failed += 1 + return name, failed + + +def test_board(board: Board) -> tuple[str, int, list[str], list]: name = board['name'] flasher = board['flasher'] # default to all tests test_list = [] - if len(test_only) > 0: - test_list = test_only + if name in board_test: + test_list = board_test[name] + elif len(test_only) > 0: + # Explicit -t: filter against the board's capabilities so a device-only + # board doesn't try to run host/dual tests (the test functions need a + # `dev_attached` entry in the board config that won't exist). + board_tests = board.get('tests', {}) + if 'only' in board_tests: + allowed = set(board_tests['only']) + test_list = [t for t in test_only if t in allowed] + else: + for t in test_only: + category = t.split('/', 1)[0] + if board_tests.get(category) is True: + test_list.append(t) else: if 'tests' in board: board_tests = board['tests'] - if 'device' in board_tests and board_tests['device'] == True: + if board_tests.get('device') is True: test_list += list(device_tests) - if 'dual' in board_tests and board_tests['dual'] == True: + if board_tests.get('dual') is True: test_list += dual_tests - if 'host' in board_tests and board_tests['host'] == True: + if board_tests.get('host') is True: test_list += host_test if 'only' in board_tests: test_list = board_tests['only'] @@ -1246,75 +1692,207 @@ def test_board(board): for skip in board_tests['skip']: if skip in test_list: test_list.remove(skip) - print(f'{name:25} {skip:30} ... Skip') + log_line(f'{name:25} {skip:30} ... Skip') err_count = 0 - flags_on_list = [""] - if 'build' in board and 'flags_on' in board['build']: - flags_on_list = board['build']['flags_on'] + failed_tests = [] + rows = [] # list of (row_label, {example: status}) — one row per build variant + variants = board.get('variant') or [{'name': name, 'flags': ''}] - for f1 in flags_on_list: + for v in variants: + vname = v['name'] + cells = {} for test in test_list: - err_count += test_example(board, f1, test) + ec, status, metric = test_example(board, vname, test) + err_count += ec + cells[test] = metric if metric else status + if ec > 0: + failed_tests.append(test) + rows.append((vname, cells)) + + # flash board_test last to disable board's usb (skipped when --skip-flash is set); + # this is teardown/park, not a test — not recorded in the report + if not skip_flash: + test_example(board, variants[0]['name'], 'device/board_test') + + return name, err_count, sorted(set(failed_tests)), rows - # flash board_test last to disable board's usb - test_example(board, flags_on_list[0], 'device/board_test') - return name, err_count +REPORT_MD = 'hil_report.md' +REPORT_JSON = 'hil_report.json' -def main(): +def render_matrix(rows_all: list) -> str: + """Render rows (list of (row_label, {example: status})) as an aligned markdown + matrix: columns = tests (bare names) centered, boards left-aligned.""" + canonical = device_tests + dual_tests + host_test + seen = set() + for _, cells in rows_all: + seen.update(cells) + if not seen: + return 'No tests were run.' + + # columns: canonical order first, then any extras (e.g. from -t) alphabetically + columns = [t for t in canonical if t in seen] + columns += [t for t in sorted(seen) if t not in canonical] + headers = [c.rsplit('/', 1)[-1] for c in columns] # bare example name + + def cell(cells, col): + v = cells.get(col) + if v is None: + return '' + return REPORT_CELL.get(v, v) # status symbol, or a metric string (e.g. speed) verbatim + + board_hdr = 'Board' + board_w = max([len(board_hdr)] + [len(lbl) for lbl, _ in rows_all]) + col_w = [max([len(h)] + [len(cell(cells, c)) for _, cells in rows_all]) + for h, c in zip(headers, columns)] + + def line(label, values): + padded = [label.ljust(board_w)] + [v.center(w) for v, w in zip(values, col_w)] + return '| ' + ' | '.join(padded) + ' |' + + header = line(board_hdr, headers) + sep = '| ' + '-' * board_w + ' | ' + ' | '.join(':' + '-' * (w - 2) + ':' for w in col_w) + ' |' + body = [line(lbl, [cell(cells, c) for c in columns]) for lbl, cells in rows_all] + + legend = 'Legend: ✅ pass · ❌ fail · ⚪ skipped · blank not run' + return '\n'.join([header, sep] + body) + '\n\n' + legend + + +def accumulate_report(mret: list, report_dir: Path, fresh: bool) -> str: + """Merge this run's results into hil_report.json in report_dir, then (re)write + the markdown matrix to hil_report.md. `fresh` (a full run, no --skip-board/-bt) + starts a new report; otherwise a re-run accumulates so boards/tests that + already passed are preserved while re-run cells are updated. Returns the md.""" + acc = {} # ordered {row_label: {example: status}} + jpath = report_dir / REPORT_JSON + if not fresh and jpath.is_file(): + try: + for entry in json.loads(jpath.read_text()).get('rows', []): + acc[entry['board']] = dict(entry['cells']) + except (ValueError, KeyError, TypeError): + pass # corrupt/old sidecar: start fresh + + # merge this run: current cells override prior for boards/tests that ran + for _, _, _, rows in mret: + for row_label, cells in rows: + acc.setdefault(row_label, {}).update(cells) + + report_dir.mkdir(parents=True, exist_ok=True) + jpath.write_text(json.dumps({'rows': [{'board': k, 'cells': v} for k, v in acc.items()]}, + indent=2) + '\n') + + md = render_matrix(list(acc.items())) + (report_dir / REPORT_MD).write_text(md + '\n', encoding='utf-8') + return md + + +def main() -> None: """ Hardware test on specified boards """ global verbose global test_only + global board_test global build_dir global max_retry + global skip_flash duration = time.time() parser = argparse.ArgumentParser() parser.add_argument('config_file', help='Configuration JSON file') parser.add_argument('-b', '--board', action='append', default=[], help='Boards to test, all if not specified') - parser.add_argument('-s', '--skip', action='append', default=[], help='Skip boards from test') + parser.add_argument('-s', '--skip-board', action='append', default=[], help='Skip boards from test') + parser.add_argument('-sf', '--skip-flash', action='store_true', help='Run tests without flashing firmware (use whatever is already on the board)') parser.add_argument('-t', '--test-only', action='append', default=[], help='Tests to run, all if not specified') - parser.add_argument('-B', '--build', default='cmake-build', help='Build folder name (default: cmake-build)') + parser.add_argument('-bt', '--board-test', action='append', default=[], + help='Per-board test list as BOARD:test1,test2 (overrides -t for that board); repeat for multiple boards') + parser.add_argument('-B', '--build-dir', default='cmake-build', help='Build folder name (default: cmake-build)') + parser.add_argument('--build', action='store_true', help='Build firmware for selected boards with cmake before running tests') parser.add_argument('-r', '--retry', type=int, default=3, help='Retry count for failed tests (default: 3)') parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output') args = parser.parse_args() - config_file = args.config_file + config_file = Path(args.config_file) boards = args.board - skip_boards = args.skip + skip_boards = args.skip_board verbose = args.verbose test_only = args.test_only - build_dir = args.build + for entry in args.board_test: + bname, _, tnames = entry.partition(':') + if not bname or not tnames: + parser.error(f'invalid --board-test value: {entry!r} (expected BOARD:test1,test2)') + board_test[bname] = [t for t in tnames.split(',') if t] + build_dir = args.build_dir max_retry = args.retry + skip_flash = args.skip_flash # if config file is not found, try to find it in the same directory as this script - if not os.path.exists(config_file): - config_file = os.path.join(os.path.dirname(__file__), config_file) - with open(config_file) as f: - config = json.load(f) + if not config_file.exists(): + config_file = Path(__file__).resolve().parent / config_file + with config_file.open() as f: + config = cast(HilConfig, json.load(f)) if len(boards) == 0: config_boards = [e for e in config['boards'] if e['name'] not in skip_boards] else: config_boards = [e for e in config['boards'] if e['name'] in boards] - err_count = 0 - with Pool(processes=os.cpu_count()) as pool: - mret = pool.map(test_board, config_boards) - err_count = sum(e[1] for e in mret) - # generate skip list for next re-run if failed - skip_fname = f'{config_file}.skip' + build_err = 0 + if args.build: + if build_dir != 'cmake-build': + print(f'warning: --build writes into cmake-build/, but -B is {build_dir!r}; ' + f'tests will not find the freshly built firmware') + print('-' * 30) + print(f'Build phase: {len(config_boards)} board(s)') + print('-' * 30) + for board in config_boards: + _, nfail = build_board(board) + build_err += nfail + print('-' * 30) + print(f'Build phase done: {build_err} failed') + print('-' * 30) + + # HIL report sidecar (hil_report.json/.md). A full run starts fresh; a re-run + # (--skip-board / -bt, i.e. the .skip file) accumulates so already-passed + # boards/tests are preserved. Clear any prior report up front on a fresh run so + # a crash mid-run can't leave stale results to be merged by a retry or posted. + report_dir = Path(os.environ.get('HIL_REPORT_DIR', '.')) + fresh = not (args.skip_board or args.board_test) + if fresh: + report_dir.mkdir(parents=True, exist_ok=True) + for f in (REPORT_JSON, REPORT_MD): + (report_dir / f).unlink(missing_ok=True) + + with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=(Lock(),)) as pool: + async_ret = pool.map_async(test_board, config_boards) + try: + mret = async_ret.get(timeout=POOL_TIMEOUT) + except MpTimeoutError: + pool.terminate() + pool.join() + raise RuntimeError(f'HIL worker pool timed out after {POOL_TIMEOUT}s') + + err_count = build_err + sum(e[1] for e in mret) + # generate skip list for next re-run if failed: skip boards that fully passed, + # and emit -bt BOARD:t1,t2 so each failed board only re-runs its own failed tests. + skip_fname = config_file.with_suffix(config_file.suffix + '.skip') if err_count > 0: - skip_boards += [name for name, err in mret if err == 0] - with open(skip_fname, 'w') as f: - f.write(' '.join(f'-s {i}' for i in skip_boards)) - elif os.path.exists(skip_fname): - os.remove(skip_fname) + skip_boards += [name for name, err, _, _ in mret if err == 0] + parts = [f'--skip-board {i}' for i in skip_boards] + parts += [f'-bt {name}:{",".join(fts)}' for name, err, fts, _ in mret if err > 0 and fts] + with skip_fname.open('w') as f: + f.write(' '.join(parts)) + elif skip_fname.exists(): + skip_fname.unlink() + + # board x test result matrix -> hil_report.md (accumulates across re-runs) + stdout + report = accumulate_report(mret, report_dir, fresh) + print() + print(report) + print(f'\nReport written to {(report_dir / REPORT_MD).resolve()}') duration = time.time() - duration print() diff --git a/test/hil/requirements.txt b/test/hil/requirements.txt index ef2fecebe..ef1cf575b 100644 --- a/test/hil/requirements.txt +++ b/test/hil/requirements.txt @@ -1,4 +1,9 @@ -fs -hid -pyfatfs +# System packages (install separately): +# sudo apt install mtools libmtp9 alsa-utils iperf +# mtools - read_disk_file (device/cdc_msc, device/msc_dual_lun) +# libmtp9 - pymtp ctypes load (device/mtp); Debian 13 uses libmtp9t64 +# alsa-utils - arecord (device/audio_test_freertos) +# iperf - throughput tests (device/net_lwip_*) +hidapi pyserial +esptool diff --git a/test/hil/tinyusb.json b/test/hil/tinyusb.json index 92b7b21b0..2ea910c4c 100644 --- a/test/hil/tinyusb.json +++ b/test/hil/tinyusb.json @@ -1,14 +1,49 @@ { "boards": [ { + "name": "ek_tm4c123gxl", + "uid": "010105186C60A110", + "tests": { + "device": true, + "host": false, + "dual": false + }, + "flasher": { + "name": "lm4flash", + "uid": "0E205D19", + "args": "-v" + } + }, + { "name": "espressif_p4_function_ev", "uid": "6055F9F98715", - "build" : { - "flags_on": ["", "CFG_TUD_DWC2_DMA_ENABLE CFG_TUH_DWC2_DMA_ENABLE"] - }, + "variant": [ + { "name": "espressif_p4_function_ev", "flags": "" }, + { "name": "espressif_p4_function_ev-DMA", "flags": "-DCFG_TUD_DWC2_DMA_ENABLE=1 -DCFG_TUH_DWC2_DMA_ENABLE=1" } + ], "tests": { - "only": ["device/cdc_msc_freertos", "device/hid_composite_freertos", "host/device_info"], - "dev_attached": [{"vid_pid": "1a86_55d4", "serial": "52D2002427", "is_cdc": true}] + "only": [ + "device/cdc_msc_freertos", + "device/hid_composite_freertos", + "device/audio_test_freertos", + "host/device_info", + "host/msc_file_explorer_freertos" + ], + "dev_attached": [ + { + "vid_pid": "1a86_55d4", + "serial": "52D2002427", + "is_cdc": true + }, + { + "vid_pid": "21c4_0cc7", + "serial": "900058944CB80A53", + "is_msc": true, + "block_size": 512, + "block_count": 60620800, + "msc_inquiry": "Lexar USB Flash Drive PMAP" + } + ] }, "flasher": { "name": "esptool", @@ -21,12 +56,34 @@ { "name": "espressif_s3_devkitm", "uid": "84F703C084E4", - "build" : { - "flags_on": ["", "CFG_TUD_DWC2_DMA_ENABLE CFG_TUH_DWC2_DMA_ENABLE"] - }, + "variant": [ + { "name": "espressif_s3_devkitm", "flags": "" }, + { "name": "espressif_s3_devkitm-DMA", "flags": "-DCFG_TUD_DWC2_DMA_ENABLE=1 -DCFG_TUH_DWC2_DMA_ENABLE=1" } + ], "tests": { - "only": ["device/cdc_msc_freertos", "device/hid_composite_freertos", "host/device_info"], - "dev_attached": [{"vid_pid": "1a86_55d4", "serial": "52D2005402", "is_cdc": true}] + "only": [ + "device/cdc_msc_freertos", + "device/hid_composite_freertos", + "device/audio_test_freertos", + "host/device_info", + "host/msc_file_explorer_freertos" + ], + "dev_attached": [ + { + "vid_pid": "1a86_55d4", + "serial": "52D2005402", + "is_cdc": true + }, + { + "vid_pid": "048d_04d2", + "serial": "\u0409", + "is_msc": true, + "block_size": 512, + "block_count": 30720000, + "msc_inquiry": "General UDisk 5.00", + "comment": "General UDisk reports iSerialNumber=U+0409" + } + ] }, "flasher": { "name": "esptool", @@ -39,11 +96,13 @@ "name": "feather_nrf52840_express", "uid": "1F0479CD0F764471", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "jlink", - "uid": "000682804350", + "uid": "681295394", "args": "-device nrf52840_xxaa" } }, @@ -51,7 +110,9 @@ "name": "max32666fthr", "uid": "0C81464124010B20FF0A08CC2C", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "openocd_adi", @@ -71,8 +132,15 @@ "device": true, "host": false, "dual": true, - "dev_attached": [{"vid_pid": "067b_2303", "serial": "0", "is_cdc": true}], - "comment": "pl23x" + "skip": ["device/audio_test_freertos"], + "dev_attached": [ + { + "vid_pid": "067b_2303", + "serial": "0", + "is_cdc": true + } + ], + "comment": "pl23x; audio_test_freertos skipped: samd51 iso-IN capture fails (arecord EIO)" }, "flasher": { "name": "jlink", @@ -84,7 +152,9 @@ "name": "mimxrt1015_evk", "uid": "DC28F865D2111D228D00B0543A70463C", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "jlink", @@ -96,9 +166,25 @@ "name": "mimxrt1064_evk", "uid": "BAE96FB95AFA6DBB8F00005002001200", "tests": { - "device": true, "host": true, "dual": true, - "dev_attached": [{"vid_pid": "10c4_ea60", "serial": "0001", "is_cdc": true}], - "comment": "cp2102" + "device": true, + "host": true, + "dual": true, + "dev_attached": [ + { + "vid_pid": "10c4_ea60", + "serial": "0001", + "is_cdc": true, + "comment": "cp2102" + }, + { + "vid_pid": "21c4_0cc7", + "serial": "900058874D871F66", + "is_msc": true, + "block_size": 512, + "block_count": 60620800, + "msc_inquiry": "Lexar USB Flash Drive PMAP" + } + ] }, "flasher": { "name": "jlink", @@ -110,7 +196,9 @@ "name": "lpcxpresso11u37", "uid": "17121919", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "jlink", @@ -135,13 +223,30 @@ { "name": "raspberry_pi_pico", "uid": "E6614C311B764A37", - "build" : { - "flags_on": ["CFG_TUH_RPI_PIO_USB"] - }, + "variant": [ + { "name": "raspberry_pi_pico", "flags": "-DCFG_TUH_RPI_PIO_USB=1" } + ], "tests": { - "device": true, "host": true, "dual": true, - "dev_attached": [{"vid_pid": "1a86_7523", "serial": "0", "is_cdc": true}], - "comment": "ch34x" + "device": true, + "host": true, + "dual": true, + "dev_attached": [ + { + "vid_pid": "1a86_7523", + "serial": "0", + "is_cdc": true, + "comment": "ch34x" + }, + { + "vid_pid": "048d_04d2", + "serial": "\u0409", + "is_msc": true, + "block_size": 512, + "block_count": 30720000, + "msc_inquiry": "General UDisk 5.00", + "comment": "General UDisk reports iSerialNumber=U+0409" + } + ] }, "flasher": { "name": "openocd", @@ -153,11 +258,13 @@ "name": "raspberry_pi_pico_w", "uid": "E6614864D35DAE36", "tests": { - "device": false, "host": true, "dual": false, + "device": false, + "host": true, + "dual": false, "dev_attached": [ { "vid_pid": "1a86_55d4", - "serial": "52D2023934", + "serial": "52D2002694", "is_cdc": true }, { @@ -181,20 +288,17 @@ "name": "raspberry_pi_pico2", "uid": "560AE75E1C7152C9", "tests": { - "device": false, "host": true, "dual": false, + "device": false, + "host": true, + "dual": false, "dev_attached": [ { - "vid_pid": "1a86_55d4", - "serial": "52D2002694", - "is_cdc": true - }, - { "vid_pid": "0951_1603", "serial": "820000000000000045B46338", "is_msc": true, "block_size": 512, "block_count": 3987456, - "msc_inquiry": "Kingston DataTraveler 2.0 1.0" + "msc_inquiry": "Kingston DataTraveler 2.0 1.00" } ] }, @@ -212,20 +316,35 @@ "host": true, "dual": true, "dev_attached": [ - {"vid_pid": "0403_6001", "serial": "0", "is_cdc": true}, - {"vid_pid": "058f_6387", "serial": "A8BEE062633D", "is_msc": true, - "block_size": 512, "block_count": 7639040, "msc_inquiry": "Generic Flash Disk 8.07"} + { + "vid_pid": "0403_6001", + "serial": "0", + "is_cdc": true + }, + { + "vid_pid": "058f_6387", + "serial": "A8BEE062633D", + "is_msc": true, + "block_size": 512, + "block_count": 7639040, + "msc_inquiry": "Generic Flash Disk 8.07" + } ] }, "flasher": { "name": "openocd", - "uid": "E6614103E78E8324", + "uid": "E663AC91D3359B38", "args": "-f interface/cmsis-dap.cfg -f target/rp2350.cfg -c \"adapter speed 5000\"" } }, { "name": "stm32f072disco", "uid": "3A001A001357364230353532", + "tests": { + "device": true, + "host": false, + "dual": false + }, "flasher": { "name": "jlink", "uid": "779541626", @@ -234,14 +353,45 @@ "comment": "2x16 access scheme with 1KB USB SRAM" }, { + "name": "stm32f407disco", + "uid": "30001A000647313332353735", + "tests": { + "device": true, + "host": false, + "dual": false + }, + "flasher": { + "name": "jlink", + "uid": "000773661813", + "args": "-device stm32f407vg" + } + }, + { "name": "stm32f723disco", "uid": "460029001951373031313335", - "build" : { - "flags_on": ["", "CFG_TUH_DWC2_DMA_ENABLE"] - }, + "variant": [ + { "name": "stm32f723disco", "flags": "" }, + { "name": "stm32f723disco-DMA", "flags": "-DCFG_TUH_DWC2_DMA_ENABLE=1" } + ], "tests": { - "device": true, "host": true, "dual": false, - "dev_attached": [{"vid_pid": "1a86_55d4", "serial": "52D2003414", "is_cdc": true}] + "device": true, + "host": true, + "dual": false, + "dev_attached": [ + { + "vid_pid": "1a86_55d4", + "serial": "52D2003414", + "is_cdc": true + }, + { + "vid_pid": "21c4_0cc7", + "serial": "90005893730A1A63", + "is_msc": true, + "block_size": 512, + "block_count": 60620800, + "msc_inquiry": "Lexar USB Flash Drive PMAP" + } + ] }, "flasher": { "name": "jlink", @@ -253,11 +403,14 @@ { "name": "stm32h743nucleo", "uid": "110018000951383432343236", - "build" : { - "flags_on": ["", "CFG_TUD_DWC2_DMA_ENABLE"] - }, + "variant": [ + { "name": "stm32h743nucleo", "flags": "" }, + { "name": "stm32h743nucleo-DMA", "flags": "-DCFG_TUD_DWC2_DMA_ENABLE=1 -DCFG_TUH_DWC2_DMA_ENABLE=1" } + ], "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "openocd", @@ -269,7 +422,9 @@ "name": "stm32g0b1nucleo", "uid": "4D0038000450434E37343120", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "openocd", @@ -277,29 +432,46 @@ "args": "-f interface/stlink.cfg -f target/stm32g0x.cfg" }, "comment": "32-bit scheme, 2KB USB SRAM" - } - ], - "boards-skip": [ + }, { - "name": "stm32f769disco", - "uid": "21002F000F51363531383437", - "build" : { - "flags_on": ["", "CFG_TUD_DWC2_DMA_ENABLE"] - }, + "name": "stm32l476disco", + "uid": "3C0050001150334258343920", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "jlink", - "uid": "000778170924", - "args": "-device stm32f769ni" + "uid": "777632258", + "args": "-device STM32L476VG" + } + }, + { + "name": "stm32u083nucleo", + "uid": "300044000D5036394E373620", + "tests": { + "device": true, + "host": false, + "dual": false + }, + "flasher": { + "name": "stlink", + "uid": "0668FF575457657187061314" } }, { "name": "nanoch32v203", "uid": "CDAB277B0FBC03E339E339E3", + "toolchain": "riscv-gcc", + "variant": [ + {"name": "nanoch32v203-fsdev", "defines": ["RHPORT_DEVICE=0"]}, + {"name": "nanoch32v203-usbfs", "defines": ["RHPORT_DEVICE=1"]} + ], "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "openocd_wch", @@ -308,15 +480,54 @@ } }, { - "name": "stm32f407disco", - "uid": "30001A000647313332353735", + "name": "ch32v103r_r1_1v0", + "uid": "CDAB3E8749BC54EF0F410025", + "toolchain": "riscv-gcc", "tests": { - "device": true, "host": false, "dual": false + "device": true, + "host": false, + "dual": false, + "skip": ["device/cdc_msc_throughput"] + }, + "flasher": { + "name": "openocd_wch", + "uid": "BC4954081051", + "args": "" + } + }, + { + "name": "ch582m_evt", + "uid": "D443627B5450", + "toolchain": "riscv-gcc", + "tests": { + "device": true, + "host": false, + "dual": false + }, + "flasher": { + "name": "openocd_wch", + "uid": "7FD88F0604B5", + "args": "" + } + } + ], + "boards-skip": [ + { + "name": "stm32f769disco", + "uid": "21002F000F51363531383437", + "variant": [ + { "name": "stm32f769disco", "flags": "" }, + { "name": "stm32f769disco-DMA", "flags": "-DCFG_TUD_DWC2_DMA_ENABLE=1 -DCFG_TUH_DWC2_DMA_ENABLE=1" } + ], + "tests": { + "device": true, + "host": false, + "dual": false }, "flasher": { "name": "jlink", - "uid": "000773661813", - "args": "-device stm32f407vg" + "uid": "000778170924", + "args": "-device stm32f769ni" } } ] diff --git a/test/unit-test/CMakeLists.txt b/test/unit-test/CMakeLists.txt index b44a91d57..a33af4563 100644 --- a/test/unit-test/CMakeLists.txt +++ b/test/unit-test/CMakeLists.txt @@ -117,14 +117,14 @@ add_ceedling_test( add_ceedling_test( test_usbd ${CEEDLING_WORKDIR}/test/device/usbd/test_usbd.c - "${CEEDLING_WORKDIR}/../../src/tusb.c;${CEEDLING_WORKDIR}/../../src/device/usbd.c;${CEEDLING_WORKDIR}/../../src/device/usbd_control.c;${CEEDLING_WORKDIR}/../../src/common/tusb_fifo.c" + "${CEEDLING_WORKDIR}/../../src/tusb.c;${CEEDLING_WORKDIR}/../../src/device/usbd.c;${CEEDLING_WORKDIR}/../../src/common/tusb_fifo.c" "${CEEDLING_BUILD_DIR}/test/mocks/test_usbd/mock_dcd.c;${CEEDLING_BUILD_DIR}/test/mocks/test_usbd/mock_msc_device.c" ) add_ceedling_test( test_msc_device ${CEEDLING_WORKDIR}/test/device/msc/test_msc_device.c - "${CEEDLING_WORKDIR}/../../src/tusb.c;${CEEDLING_WORKDIR}/../../src/device/usbd.c;${CEEDLING_WORKDIR}/../../src/device/usbd_control.c;${CEEDLING_WORKDIR}/../../src/class/msc/msc_device.c;${CEEDLING_WORKDIR}/../../src/common/tusb_fifo.c" + "${CEEDLING_WORKDIR}/../../src/tusb.c;${CEEDLING_WORKDIR}/../../src/device/usbd.c;${CEEDLING_WORKDIR}/../../src/class/msc/msc_device.c;${CEEDLING_WORKDIR}/../../src/common/tusb_fifo.c" "${CEEDLING_BUILD_DIR}/test/mocks/test_msc_device/mock_dcd.c" ) diff --git a/test/unit-test/test/device/midi2/test_midi2_device.c b/test/unit-test/test/device/midi2/test_midi2_device.c new file mode 100644 index 000000000..1314c2585 --- /dev/null +++ b/test/unit-test/test/device/midi2/test_midi2_device.c @@ -0,0 +1,266 @@ +/* + * The MIT License (MIT) + * + * Copyright (c) 2026 Saulo Verissimo + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +#include "unity.h" +#include "tusb_types.h" +#include "class/audio/audio.h" +#include "class/midi/midi.h" +#include "device/usbd.h" + +void setUp(void) {} +void tearDown(void) {} + +//--------------------------------------------------------------------+ +// UMP Word Count: all 16 message types +//--------------------------------------------------------------------+ + +void test_ump_word_count_1word_types(void) { + uint8_t types[] = {0x0, 0x1, 0x2, 0x6, 0x7}; + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL(1, midi2_ump_word_count(types[i])); + } +} + +void test_ump_word_count_2word_types(void) { + uint8_t types[] = {0x3, 0x4, 0x8, 0x9, 0xA}; + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL(2, midi2_ump_word_count(types[i])); + } +} + +void test_ump_word_count_3word_types(void) { + TEST_ASSERT_EQUAL(3, midi2_ump_word_count(0xB)); + TEST_ASSERT_EQUAL(3, midi2_ump_word_count(0xC)); +} + +void test_ump_word_count_4word_types(void) { + uint8_t types[] = {0x5, 0xD, 0xE, 0xF}; + for (int i = 0; i < 4; i++) { + TEST_ASSERT_EQUAL(4, midi2_ump_word_count(types[i])); + } +} + +void test_ump_word_count_covers_all_16(void) { + for (uint8_t mt = 0; mt <= 0xF; mt++) { + uint8_t wc = midi2_ump_word_count(mt); + TEST_ASSERT_TRUE(wc >= 1 && wc <= 4); + } +} + +//--------------------------------------------------------------------+ +// CS Endpoint subtypes (defined in midi.h) +//--------------------------------------------------------------------+ + +void test_cs_endpoint_subtypes(void) { + TEST_ASSERT_EQUAL(0x01, MIDI_CS_ENDPOINT_GENERAL); + TEST_ASSERT_EQUAL(0x02, MIDI_CS_ENDPOINT_GENERAL_2_0); +} + +//--------------------------------------------------------------------+ +// Descriptor macro length calculations +//--------------------------------------------------------------------+ + +void test_midi1_desc_len(void) { + TEST_ASSERT_EQUAL(TUD_MIDI_DESC_HEAD_LEN + TUD_MIDI_DESC_JACK_LEN + TUD_MIDI_DESC_EP_LEN(1) * 2, + TUD_MIDI_DESC_LEN); +} + +void test_midi2_alt1_head_len(void) { + TEST_ASSERT_EQUAL(16, TUD_MIDI2_DESC_ALT1_HEAD_LEN); +} + +void test_midi2_alt1_ep_len(void) { + // EP(7) + CS base(4) + numgtbs + TEST_ASSERT_EQUAL(12, TUD_MIDI2_DESC_ALT1_EP_LEN(1)); + TEST_ASSERT_EQUAL(13, TUD_MIDI2_DESC_ALT1_EP_LEN(2)); + TEST_ASSERT_EQUAL(18, TUD_MIDI2_DESC_ALT1_EP_LEN(7)); +} + +void test_midi2_desc_len(void) { + int expected = TUD_MIDI_DESC_LEN + TUD_MIDI2_DESC_ALT1_HEAD_LEN + TUD_MIDI2_DESC_ALT1_EP_LEN(1) * 2; + TEST_ASSERT_EQUAL(expected, TUD_MIDI2_DESC_LEN); +} + +void test_midi2_desc_len_greater_than_midi1(void) { + TEST_ASSERT_TRUE(TUD_MIDI2_DESC_LEN > TUD_MIDI_DESC_LEN); +} + +//--------------------------------------------------------------------+ +// Descriptor macro byte validation +//--------------------------------------------------------------------+ + +void test_midi2_descriptor_bytes(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(0, 0, 0x01, 0x81, 64) }; + + TEST_ASSERT_EQUAL(TUD_MIDI2_DESC_LEN, sizeof(desc)); + + // First byte: Audio Control Interface descriptor length = 9 + TEST_ASSERT_EQUAL(9, desc[0]); + TEST_ASSERT_EQUAL(TUSB_DESC_INTERFACE, desc[1]); + TEST_ASSERT_EQUAL(0, desc[2]); + + // Find Alt Setting 1 by scanning + int alt1_offset = -1; + int pos = 0; + while (pos < (int)sizeof(desc)) { + if (desc[pos + 1] == TUSB_DESC_INTERFACE && desc[pos + 3] == 1) { + alt1_offset = pos; + break; + } + pos += desc[pos]; + } + + TEST_ASSERT_TRUE_MESSAGE(alt1_offset >= 0, "Alt Setting 1 interface not found"); + + TEST_ASSERT_EQUAL(9, desc[alt1_offset]); + TEST_ASSERT_EQUAL(TUSB_DESC_INTERFACE, desc[alt1_offset + 1]); + TEST_ASSERT_EQUAL(1, desc[alt1_offset + 2]); // bInterfaceNumber + TEST_ASSERT_EQUAL(1, desc[alt1_offset + 3]); // bAlternateSetting + TEST_ASSERT_EQUAL(2, desc[alt1_offset + 4]); // bNumEndpoints + TEST_ASSERT_EQUAL(TUSB_CLASS_AUDIO, desc[alt1_offset + 5]); + + // MS Header after Alt Setting 1 interface: bcdMSC = 0x0200 + int ms2_offset = alt1_offset + 9; + TEST_ASSERT_EQUAL(7, desc[ms2_offset]); + TEST_ASSERT_EQUAL(TUSB_DESC_CS_INTERFACE, desc[ms2_offset + 1]); + TEST_ASSERT_EQUAL(MIDI_CS_INTERFACE_HEADER, desc[ms2_offset + 2]); + TEST_ASSERT_EQUAL(0x00, desc[ms2_offset + 3]); + TEST_ASSERT_EQUAL(0x02, desc[ms2_offset + 4]); + // USB-MIDI 2.0 Table 5-2: wTotalLength shall match bLength (= 0x0007) + TEST_ASSERT_EQUAL(0x07, desc[ms2_offset + 5]); + TEST_ASSERT_EQUAL(0x00, desc[ms2_offset + 6]); +} + +void test_midi2_descriptor_alt1_cs_endpoint_subtype(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(0, 0, 0x01, 0x81, 64) }; + + int cs_ep_count = 0; + int pos = 0; + while (pos < (int)sizeof(desc)) { + if (desc[pos + 1] == TUSB_DESC_CS_ENDPOINT && + desc[pos + 2] == MIDI_CS_ENDPOINT_GENERAL_2_0) { + cs_ep_count++; + TEST_ASSERT_EQUAL(1, desc[pos + 3]); + } + pos += desc[pos]; + } + TEST_ASSERT_EQUAL(2, cs_ep_count); +} + +void test_midi2_descriptor_has_both_alt_settings(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(0, 0, 0x01, 0x81, 64) }; + + int alt0_count = 0; + int alt1_count = 0; + int pos = 0; + while (pos < (int)sizeof(desc)) { + if (desc[pos + 1] == TUSB_DESC_INTERFACE) { + if (desc[pos + 3] == 0) alt0_count++; + if (desc[pos + 3] == 1) alt1_count++; + } + pos += desc[pos]; + } + TEST_ASSERT_TRUE(alt0_count >= 2); + TEST_ASSERT_EQUAL(1, alt1_count); +} + +void test_midi2_descriptor_endpoint_addresses(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(0, 0, 0x02, 0x82, 64) }; + + int ep_out_count = 0; + int ep_in_count = 0; + int pos = 0; + while (pos < (int)sizeof(desc)) { + if (desc[pos + 1] == TUSB_DESC_ENDPOINT) { + uint8_t ep_addr = desc[pos + 2]; + if (ep_addr == 0x02) ep_out_count++; + if (ep_addr == 0x82) ep_in_count++; + TEST_ASSERT_EQUAL(TUSB_XFER_BULK, desc[pos + 3]); + TEST_ASSERT_EQUAL(64, desc[pos + 4]); + TEST_ASSERT_EQUAL(0, desc[pos + 5]); + } + pos += desc[pos]; + } + TEST_ASSERT_EQUAL(2, ep_out_count); + TEST_ASSERT_EQUAL(2, ep_in_count); +} + +void test_midi2_descriptor_nonzero_itfnum(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(2, 0, 0x03, 0x83, 64) }; + + TEST_ASSERT_EQUAL(2, desc[2]); + + int pos = desc[0]; + while (pos < (int)sizeof(desc)) { + if (desc[pos + 1] == TUSB_DESC_INTERFACE) { + TEST_ASSERT_EQUAL(3, desc[pos + 2]); + break; + } + pos += desc[pos]; + } +} + +//--------------------------------------------------------------------+ +// Descriptor traversal integrity +//--------------------------------------------------------------------+ + +void test_midi2_descriptor_no_zero_length(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(0, 0, 0x01, 0x81, 64) }; + + int pos = 0; + int desc_count = 0; + while (pos < (int)sizeof(desc)) { + TEST_ASSERT_TRUE_MESSAGE(desc[pos] > 0, "Zero-length descriptor found"); + TEST_ASSERT_TRUE_MESSAGE(desc[pos] <= (int)sizeof(desc) - pos, + "Descriptor length exceeds remaining bytes"); + pos += desc[pos]; + desc_count++; + } + TEST_ASSERT_EQUAL((int)sizeof(desc), pos); + TEST_ASSERT_TRUE(desc_count > 5); +} + +void test_midi2_descriptor_valid_types(void) { + uint8_t desc[] = { TUD_MIDI2_DESCRIPTOR(0, 0, 0x01, 0x81, 64) }; + + int pos = 0; + while (pos < (int)sizeof(desc)) { + uint8_t dtype = desc[pos + 1]; + bool valid = (dtype == TUSB_DESC_INTERFACE || + dtype == TUSB_DESC_ENDPOINT || + dtype == TUSB_DESC_CS_INTERFACE || + dtype == TUSB_DESC_CS_ENDPOINT); + TEST_ASSERT_TRUE_MESSAGE(valid, "Invalid descriptor type found"); + pos += desc[pos]; + } +} + +//--------------------------------------------------------------------+ +// Edge cases +//--------------------------------------------------------------------+ + +void test_ump_word_count_with_values_beyond_0xf(void) { + TEST_ASSERT_EQUAL(4, midi2_ump_word_count(0x10)); + TEST_ASSERT_EQUAL(4, midi2_ump_word_count(0xFF)); +} diff --git a/test/unit-test/test/device/usbd/test_usbd.c b/test/unit-test/test/device/usbd/test_usbd.c index 3a2cf3217..7f3c3f5b2 100644 --- a/test/unit-test/test/device/usbd/test_usbd.c +++ b/test/unit-test/test/device/usbd/test_usbd.c @@ -29,7 +29,7 @@ #include "tusb_fifo.h" #include "tusb.h" #include "usbd.h" -TEST_SOURCE_FILE("usbd_control.c") +TEST_SOURCE_FILE("usbd.c") // Mock File #include "mock_dcd.h" @@ -100,6 +100,16 @@ tusb_control_request_t const req_get_desc_configuration = .wLength = 256 }; +// Vendor OUT control request (direction OUT, type Vendor, recipient Device), 8-byte data stage +tusb_control_request_t const req_vendor_out = +{ + .bmRequestType = 0x40, + .bRequest = 0x01, + .wValue = 0x0000, + .wIndex = 0x0000, + .wLength = 8 +}; + uint8_t const* desc_device; uint8_t const* desc_configuration; @@ -120,6 +130,19 @@ uint16_t const* tud_descriptor_string_cb(uint8_t index, uint16_t langid) { return NULL; } +// Backing buffer for the vendor OUT data stage. Sized to EP0 max packet so an (untested) regression +// that drops the clamp can't corrupt memory here; the regression is caught by the expectation below. +static uint8_t vendor_out_buf[CFG_TUD_ENDPOINT0_SIZE]; + +bool tud_vendor_control_xfer_cb(uint8_t rhport_, uint8_t stage, tusb_control_request_t const* request) { + (void) request; + if (stage == CONTROL_STAGE_SETUP) { + // Offer only an 8-byte capacity even though the data stage may receive a larger packet + return tud_control_xfer(rhport_, request, vendor_out_buf, 8); + } + return true; +} + void setUp(void) { dcd_int_disable_Ignore(); dcd_int_enable_Ignore(); @@ -246,3 +269,30 @@ void test_usbd_control_in_zlp(void) tud_task(); } + +//--------------------------------------------------------------------+ +// Control OUT data stage host overrun +//--------------------------------------------------------------------+ + +// A non-compliant host sends an OUT data packet larger than the buffer the class offered: +// wLength = 8, but the DCD reports a full CFG_TUD_ENDPOINT0_SIZE packet. usbd must clamp the +// copy/accounting to the 8-byte capacity so total_xferred reaches wLength, ends the data stage, +// and queues the IN status stage. Without the clamp total_xferred overshoots wLength and usbd +// re-arms an OUT data packet (EDPT_CTRL_OUT) instead, failing the EDPT_CTRL_IN expectation below. +void test_usbd_control_out_overrun_clamp(void) +{ + dcd_event_setup_received(rhport, (uint8_t*) &req_vendor_out, false); + + // Data stage: usbd arms an 8-byte OUT into its internal bounce buffer (buffer ptr is internal) + dcd_edpt_xfer_ExpectAndReturn(rhport, EDPT_CTRL_OUT, NULL, 8, false, true); + dcd_edpt_xfer_IgnoreArg_buffer(); + // Host overrun: DCD reports a full max packet, larger than the 8-byte capacity + dcd_event_xfer_complete(rhport, EDPT_CTRL_OUT, CFG_TUD_ENDPOINT0_SIZE, XFER_RESULT_SUCCESS, false); + + // Clamp -> total_xferred == wLength -> data stage done -> IN status stage queued + dcd_edpt_xfer_ExpectAndReturn(rhport, EDPT_CTRL_IN, NULL, 0, false, true); + dcd_event_xfer_complete(rhport, EDPT_CTRL_IN, 0, 0, false); + dcd_edpt0_status_complete_ExpectWithArray(rhport, &req_vendor_out, 1); + + tud_task(); +} diff --git a/test/unit-test/test/host/midi2/test_midi2_host.c b/test/unit-test/test/host/midi2/test_midi2_host.c new file mode 100644 index 000000000..8ad77c14e --- /dev/null +++ b/test/unit-test/test/host/midi2/test_midi2_host.c @@ -0,0 +1,101 @@ +/* + * The MIT License (MIT) + * + * Copyright (c) 2026 Saulo Verissimo + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ + +#include "unity.h" +#include "tusb_option.h" +#include "class/midi/midi.h" +#include "class/midi/midi2_host.h" + +void setUp(void) {} +void tearDown(void) {} + +//--------------------------------------------------------------------+ +// UMP Word Count (shared helper, defined in midi.h) +//--------------------------------------------------------------------+ + +void test_midi2_host_ump_word_count_1word(void) { + uint8_t types[] = {0x0, 0x1, 0x2, 0x6, 0x7}; + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL(1, midi2_ump_word_count(types[i])); + } +} + +void test_midi2_host_ump_word_count_2word(void) { + uint8_t types[] = {0x3, 0x4, 0x8, 0x9, 0xA}; + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL(2, midi2_ump_word_count(types[i])); + } +} + +void test_midi2_host_ump_word_count_4word(void) { + uint8_t types[] = {0x5, 0xD, 0xE, 0xF}; + for (int i = 0; i < 4; i++) { + TEST_ASSERT_EQUAL(4, midi2_ump_word_count(types[i])); + } +} + +//--------------------------------------------------------------------+ +// Callback struct field validation +//--------------------------------------------------------------------+ + +void test_midi2_descriptor_cb_struct_fields(void) { + tuh_midi2_descriptor_cb_t desc = { + .protocol_version = 1, + .bcdMSC_hi = 0x02, + .bcdMSC_lo = 0x00, + .rx_cable_count = 1, + .tx_cable_count = 1 + }; + TEST_ASSERT_EQUAL(1, desc.protocol_version); + TEST_ASSERT_EQUAL(0x02, desc.bcdMSC_hi); + TEST_ASSERT_EQUAL(0x00, desc.bcdMSC_lo); + TEST_ASSERT_EQUAL(1, desc.rx_cable_count); + TEST_ASSERT_EQUAL(1, desc.tx_cable_count); +} + +void test_midi2_mount_cb_struct_fields(void) { + tuh_midi2_mount_cb_t mount = { + .daddr = 1, + .bInterfaceNumber = 0, + .protocol_version = 1, + .alt_setting_active = 1, + .rx_cable_count = 2, + .tx_cable_count = 2 + }; + TEST_ASSERT_EQUAL(1, mount.daddr); + TEST_ASSERT_EQUAL(0, mount.bInterfaceNumber); + TEST_ASSERT_EQUAL(1, mount.protocol_version); + TEST_ASSERT_EQUAL(1, mount.alt_setting_active); + TEST_ASSERT_EQUAL(2, mount.rx_cable_count); + TEST_ASSERT_EQUAL(2, mount.tx_cable_count); +} + +//--------------------------------------------------------------------+ +// CS Endpoint subtypes +//--------------------------------------------------------------------+ + +void test_midi2_host_cs_endpoint_subtypes(void) { + TEST_ASSERT_EQUAL(0x01, MIDI_CS_ENDPOINT_GENERAL); + TEST_ASSERT_EQUAL(0x02, MIDI_CS_ENDPOINT_GENERAL_2_0); +} |
