bootloader/test/pbtest.py

#!/usr/bin/env python3
"""End-to-end pureboot protocol test: spawn the simavr device, then drive it
with the real host tool (pureboot.py, as a subprocess over the device's pty)
through flash + EEPROM + fuse + hand-over scenarios, and cross-check
the tool's view against the simulator's ground-truth memory dumps.

Usage: pbtest.py <device_bin> <pureboot_elf> <mcu> <hz> <base_hex> <page>
                 <baud> <eeprom_size> <app_bin> <tool_py> <workdir>
Exits 0 if every scenario passes.
"""

import os
import signal
import subprocess
import sys
import time


def fail(message):
    print(f"FAIL: {message}")
    sys.exit(1)


def rjmp_decode(word, at, flash_words):
    """Where an rjmp word at word-address `at` lands — deliberately written
    against the instruction-set definition (12-bit signed offset), not with
    the host tool's encoder, so an encoding bug cannot verify itself."""
    if word & 0xF000 != 0xC000:
        fail(f"word at {at * 2:#06x} is {word:#06x}, not an rjmp")
    offset = word & 0x0FFF
    if offset >= 0x800:
        offset -= 0x1000
    return (at + 1 + offset) % flash_words


class Device:
    def __init__(self, binary, elf, mcu, hz, base, page, baud, dump):
        self.proc = subprocess.Popen(
            [binary, elf, mcu, hz, base, str(page), str(baud), dump],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
        )
        self.dump = dump
        self.pty = None
        deadline = time.time() + 5
        while time.time() < deadline:
            line = self.proc.stdout.readline()
            if not line:
                break
            if line.startswith("PB_PTY"):
                self.pty = line.split()[1]
                break
        if not self.pty:
            self.stop()
            raise RuntimeError("device did not report a pty")

    def stop(self):
        self.proc.terminate()
        try:
            self.proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            self.proc.kill()


def run_tool(tool, pty, baud, *args):
    result = subprocess.run(
        [sys.executable, tool, "--port", pty, "--baud", str(baud), "--wait", "20", *args],
        capture_output=True,
        text=True,
        timeout=120,
    )
    print(result.stdout, end="")
    if result.returncode != 0:
        fail(f"tool exited {result.returncode}: {result.stderr.strip()}")
    return result.stdout


def main():
    (device_bin, elf, mcu, hz, base_hex, page, baud, eeprom_size, app_bin, tool, workdir) = sys.argv[1:]
    base, page, baud, eeprom_size = int(base_hex, 0), int(page), int(baud), int(eeprom_size)
    sys.path.insert(0, os.path.dirname(os.path.abspath(tool)))
    import pureboot as pb

    os.makedirs(workdir, exist_ok=True)
    ee_image = bytes(range(0xA0, 0xB0))
    ee_path = os.path.join(workdir, "ee.bin")
    open(ee_path, "wb").write(ee_image)
    dump = os.path.join(workdir, "flash_dump.bin")
    read_flash = os.path.join(workdir, "readback_flash.bin")
    read_eeprom = os.path.join(workdir, "readback_eeprom.bin")

    # The geometry the host will discover, for computing the expected image.
    info = pb.Info(
        bytes([ord("P"), ord("B"), 1, 0, 0, 0, page])
        + bytes([base & 0xFF, base >> 8, eeprom_size & 0xFF, eeprom_size >> 8])
        + bytes([0 if mcu == "atmega328p" else 1])
    )

    device = Device(device_bin, elf, mcu, hz, base_hex, page, baud, dump)
    try:
        # Session 1: knock from reset, identify, program everything, stay.
        out = run_tool(tool, device.pty, baud, "--info", "--fuses", "--flash", app_bin,
                       "--eeprom", ee_path, "--stay")
        for needed in ("device: signature", "fuses:", "verify:", "stays"):
            if needed not in out:
                fail(f"session 1 output lacks {needed!r}")

        # Session 2: reconnect into the live session, verify, dump, hand over
        # is deferred — the pty must be reopened for the APP banner first.
        out = run_tool(tool, device.pty, baud, "--verify-flash", app_bin, "--verify-eeprom", ee_path,
                       "--read-flash", read_flash, "--read-eeprom", read_eeprom, "--stay")
        if out.count("verify:") != 2:
            fail("session 2 did not verify both memories")

        eeprom_back = open(read_eeprom, "rb").read()
        if eeprom_back[: len(ee_image)] != ee_image:
            fail("EEPROM read-back mismatch")

        # The expected post-surgery flash, straight from the tool's planner.
        pages = pb.plan_flash(open(app_bin, "rb").read(), info)
        flash_back = open(read_flash, "rb").read()
        for address, data in pages.items():
            if flash_back[address : address + page] != data:
                fail(f"flash read-back mismatch in page {address:#06x}")

        # An external reset re-enters through the patched word 0 (tinies; the
        # runner resets them to address 0 like silicon) or BOOTRST (mega).
        # The loader must answer a fresh knock, and the 'J' hand-over must
        # land in the application, which banners on the same link.
        device.proc.send_signal(signal.SIGUSR1)
        port = pb.Port(device.pty, baud)
        try:
            loader = pb.Loader(port)
            loader.connect(15)
            loader.run_application()
            banner = port.read_exact(3, 5.0)
            if banner != b"APP":
                fail(f"application banner was {banner!r}")
        finally:
            port.close()
    finally:
        device.stop()

    # Ground truth: the simulator's own memories, against the host's view.
    flash_true = open(dump, "rb").read()
    if flash_true[:base] != flash_back:
        fail("host flash read-back differs from the simulator's flash")
    if flash_true[base] == 0xFF and flash_true[base + 1] == 0xFF:
        fail("loader region looks erased in the ground-truth dump")

    # The surgery, decoded independently: the patched vector must land on the
    # loader, the trampoline on the application's own entry.
    if mcu != "atmega328p":
        flash_words = (base + 512) // 2
        app = open(app_bin, "rb").read()
        word0 = flash_true[0] | (flash_true[1] << 8)
        if rjmp_decode(word0, 0, flash_words) != base // 2:
            fail("patched reset vector does not land on the loader base")
        trampoline = flash_true[base - 2] | (flash_true[base - 1] << 8)
        original = app[0] | (app[1] << 8)
        if rjmp_decode(trampoline, (base - 2) // 2, flash_words) != rjmp_decode(original, 0, flash_words):
            fail("trampoline does not land on the application's own entry")
    ee_true_path = dump + ".eeprom"
    if os.path.exists(ee_true_path):
        ee_true = open(ee_true_path, "rb").read()
        if ee_true[: len(ee_image)] != ee_image:
            fail("ground-truth EEPROM does not match what was programmed")

    print("pbtest: all scenarios pass")


if __name__ == "__main__":
    main()