diff --git a/pureboot/pureboot.py b/pureboot/pureboot.py index 99893d4..4f061b1 100644 --- a/pureboot/pureboot.py +++ b/pureboot/pureboot.py @@ -484,8 +484,13 @@ class Loader: self.port.write(knock) knocks += 1 if PROMPT in self.port.read_available(0.4): + # Settle: absorb a real loader's trailing bytes before asking + # for the identity. Bounded by the deadline so a target that + # never falls quiet — a board stuck in a reset loop, whose + # garbage carries a stray prompt — cannot spin here forever. while self.port.read_available(0.3): - pass + if time.monotonic() > deadline: + break self.port.write(b"b") try: # A version the tool cannot speak is the loader's own diff --git a/pureboot/test/test_handshake.py b/pureboot/test/test_handshake.py new file mode 100644 index 0000000..7fdfbb3 --- /dev/null +++ b/pureboot/test/test_handshake.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Host-tool activation handshake: it must not hang on a flooding target. + +`_handshake` drains the line after it sees a prompt, to absorb a real loader's +trailing bytes before it asks for the identity. That drain must be bounded: a +target that never falls quiet — a board stuck in a reset loop presents exactly +this, ~60 reboots/s of UART-reset garbage in which a stray 0x2b reads as a +prompt — otherwise spins the tool forever. Regression for that hang, plus a +control that a well-behaved loader still connects. + +Stdlib only; run with `python test/test_handshake.py`. +""" +import importlib.util +import pathlib +import threading +import time + +PB = pathlib.Path(__file__).resolve().parents[1] / "pureboot.py" +_spec = importlib.util.spec_from_file_location("pureboot", PB) +pb = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(pb) + +P = F = 0 + + +def check(name, ok): + global P, F + P, F = P + (1 if ok else 0), F + (0 if ok else 1) + print(f" [{'PASS' if ok else 'FAIL'}] {name}") + + +class FloodPort: + """A line that never falls quiet: read_available always returns bytes, and + they contain a prompt. No identity ever completes.""" + + def flush_input(self): + pass + + def write(self, data): + pass + + def read_available(self, wait): + time.sleep(0.01) # a real read waits; keep the busy loop off a core + return b"+\x00\xff" + + def read_exact(self, count, timeout): + raise pb.Error("no identity") + + +class LoaderPort: + """A well-behaved pureboot 5: one prompt to the knock, then quiet, then the + slim identity (version 5 + m328p signature) and a closing prompt.""" + + def __init__(self): + self.reads = self.exacts = 0 + + def flush_input(self): + pass + + def write(self, data): + pass + + def read_available(self, wait): + self.reads += 1 + return b"+" if self.reads == 1 else b"" # prompt once, then settle quiet + + def read_exact(self, count, timeout): + self.exacts += 1 + return b"\x05\x1e\x95\x0f" if self.exacts == 1 else b"+" # identity, then prompt + + +def terminates(port, wait, budget): + """Run connect_autobaud in a thread; True if it returns/raises within + `budget` seconds rather than hanging.""" + done = threading.Event() + + def run(): + try: + pb.Loader(port).connect_autobaud(wait) + except Exception: + pass + finally: + done.set() + + threading.Thread(target=run, daemon=True).start() + return done.wait(budget) + + +def main(): + # the hang: a flooding target must not spin the drain forever. With wait=0.5 + # the whole handshake has to give up well inside a few seconds. + check("flooding target: handshake terminates, drain is bounded", + terminates(FloodPort(), wait=0.5, budget=4.0)) + + # the control: a real loader still connects and reads identity. + info = pb.Loader(LoaderPort()).connect_autobaud(2.0) + check("well-behaved loader still connects (version 5)", info.version == 5) + + print(f"\n {P} passed, {F} failed") + return 1 if F else 0 + + +if __name__ == "__main__": + raise SystemExit(main())