#!/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())