diff --git a/docs/ota_wizard.md b/docs/ota_wizard.md index bb6530294..51794ffbc 100644 --- a/docs/ota_wizard.md +++ b/docs/ota_wizard.md @@ -48,123 +48,118 @@ Close other apps using the USB port before starting. ## Start -From the MeshCore checkout: +From the MeshCore checkout, run: ```bash python3 tools/lora_ota/ota_wizard.py ``` -On Windows, use `py` in place of `python3`. The first run installs USB and -verification dependencies in a private Python environment. On Raspberry Pi OS, -install `python3-venv` if Python reports that it is missing. +On Windows, use `py` in place of `python3`. No command-line options are needed. +The menu provides these actions: -The wizard selects the USB radio and destination automatically when there is -only one of each. Otherwise, choose from the displayed list. Enter the -destination admin password once, choose firmware, then review the plan and -press Enter to start. Password entry is hidden and passwords are never saved. +1. **Update a nearby radio** — the default. +2. **Check radios / save a preparation request** — inspect without installing. +3. **Prepare an update bundle on this computer** — no radios needed. +4. **Recover an interrupted update** — restore saved settings. +5. **Settings** — change transfer, bootloader, connection, or build preferences. +6. **Exit**. -Firmware choices are: +The first live connection installs USB and verification dependencies in a +private Python environment if needed. On Raspberry Pi OS, install +`python3-venv` if Python reports that it is missing. +Preparing a signed input file can also install signature-verification support +on first use, without installing the USB software. -- **Published GitHub firmware:** the default, matched to the exact board and - role. A running version that is already current is left in place. -- **Custom firmware file:** an application DFU ZIP, BIN, HEX, full mOTA, or a - prepared update bundle. -- **Build custom Full firmware from source:** offered on a computer. This - needs the firmware build tools and Bash; it is not offered by default on - a Pi. The generated version is newer than the detected installed version. +For an update, the wizard selects the USB radio and destination automatically +when there is only one of each. Otherwise, choose from the displayed lists. +Enter the destination admin password once, choose firmware, then review the +plan and press Enter to start. Password entry is hidden and passwords are +never saved. After an action, the menu is shown again. -The displayed plan includes package sizes, firmware version, and any -bootloader or trusted update signer being installed. After a reboot, the -wizard checks the running firmware hash and version, or the bootloader's -identity and version, before reporting success. +## Choose firmware from GitHub releases + +The firmware menu offers: + +- **Latest compatible GitHub release (recommended):** downloads the published + stable file for the detected board and role. +- **Choose another GitHub release:** lists compatible releases, including + clearly labelled previews, so you can choose a version. +- **Local firmware file or prepared bundle:** accepts an application DFU ZIP, + BIN, HEX, full mOTA, or a bundle prepared on another computer. +- **Build custom Full firmware from source:** offered on a computer with the + firmware build tools and Bash. It is not offered by default on a Pi. + +You do not need to copy release URLs or manually download published files. +The wizard checks the board and role, searches release files for an exact +running base when preparing a differential update, and checks for a compatible +signed bootloader on supported nRF52 destinations. Firmware that is already +installed or older is left in place. + +The final plan shows the package sizes, firmware version, and any bootloader +or trusted update signer being installed. After reboot, the wizard checks the +running firmware hash and version, or the bootloader identity and version, +before reporting success. ## Differential or full updates -The wizard looks for the **exact running firmware** in its cache and published -GitHub assets. It uses a differential package when it has a compatible builder, -can match that exact base, and the resulting package is smaller. A file with -the same version label but a different body hash is not a usable base. +**Automatic** is the default transfer setting. It uses a differential package +when a compatible builder and the **exact running firmware** are available +and the package is smaller. A file with the same version label but a different +body hash is not a usable base. Published bases are found through GitHub +releases; a local base can be selected when the running image was custom. On a computer, the differential builder can be prepared automatically with a -compatible Rust toolchain. On a Pi without a builder, the default is a full -update made by Python without compiling. This requires destination storage -that supports full updates, such as the Ikoka Stick's QSPI flash. +compatible Rust toolchain. On a Pi without a builder, a full update is made by +Python without compiling. The destination must support full updates, such as +an Ikoka Stick with working QSPI storage. + +To force a full application transfer, open **Settings**, choose **Full** for +the transfer method, then start the update. Settings also lets you turn the +bootloader check off for an application-only update, choose a USB port or +source type, or set a builder path, custom source folder/version, or cache folder. An internal-flash nRF52 destination that requires differential installation -cannot use that full fallback. Prepare its bundle on a computer instead: +cannot use the full fallback. Use the computer-to-Pi workflow below. -1. On the Pi, inspect the connected radios: +## Prepare on a computer, send from a Pi - ```bash - python3 tools/lora_ota/ota_wizard.py --check - ``` +All steps use the same menu; no command-line options are needed. - Copy the JSON file shown after **Preparation request:** to the computer. - It contains public board information and the exact running firmware hash, - without the admin password. +1. On the Pi, choose **Check radios / save a preparation request**. Connect to + the source and select the destination as prompted. Copy the JSON file + shown after **Preparation request:** to the computer. It contains public + board information and the exact running hash, without the password. +2. On the computer, choose **Prepare an update bundle on this computer**. + Select the saved request or enter its file path. Choose firmware from + GitHub releases, or select a local custom image. The wizard searches for + the exact base and offers a local base file if needed. Accept the proposed + output filename or choose another one. +3. Copy the resulting `.ota.zip` file to the Pi. Choose **Update a nearby + radio**, then **Local firmware file or prepared bundle**, and select that + ZIP. -2. On the computer, prepare the update using the request, new firmware, and - the exact firmware currently running on the destination: - - ```bash - python3 tools/lora_ota/ota_wizard.py --prepare-only \ - --request DEVICE.json --firmware NEW_FIRMWARE.zip \ - --base RUNNING_FIRMWARE.zip --output update.ota.zip - ``` - - No radio is opened. If the compatible differential builder and all input - files are already present, this also works without network access. Add - `--motatool /path/to/motatool` to use an existing builder. - -3. Copy `update.ota.zip` back to the Pi and run: - - ```bash - python3 tools/lora_ota/ota_wizard.py --firmware update.ota.zip - ``` - - The live destination is checked again before installation. A changed base - hash or a different board stops the differential update. - -For an offline full bundle on a destination that supports it, add `--full` -to the preparation command. This path needs neither the differential builder +The preparation step does not open a radio. When the builder and input files +are already present, local preparation also works without network access. +For a full bundle on a destination that supports it, choose **Full** in +**Settings** before preparation; this needs neither a differential compiler nor USB dependencies. -## Useful options - -| Option | Purpose | -| --- | --- | -| `--check` | Authenticate and inspect OTA support without downloading or installing firmware | -| `--port /dev/ttyUSB0` | Choose a USB port; Windows example: `--port COM5` | -| `--mode repeater` or `--mode companion` | Select the source type if automatic detection is unsuitable | -| `--target PUBLIC_KEY` | Choose a destination by its full 64-character key; it must still answer direct discovery | -| `--firmware FILE` | Use a local firmware image or prepared bundle | -| `--base FILE` | Supply the exact running firmware for differential preparation | -| `--full` | Choose a full application transfer when the destination supports it | -| `--build` | Build custom Full firmware from this checkout | -| `--version 1.18.0.3-custom` | Set the custom build version; it must be newer than the running version | -| `--no-bootloader` | Update application firmware only | -| `--cache DIRECTORY` | Choose where packages, public requests, and recovery records are saved | -| `--yes` | Accept the displayed plan; required choices and credentials must still be supplied | - -For unattended use, provide `MESHCORE_ADMIN_PASSWORD` in the process environment -and pass the port, source type, and target key. The wizard does not write that -password into its preferences, recovery records, or custom build environment. +The live destination is checked again before installation. A changed base +hash or a different board stops a differential update. ## If the connection is interrupted -Keep the same source radio connected and run the wizard again. It checks the -saved source and destination public keys before restoring settings from an -unfinished update. To restore settings and exit: +Keep the same source radio connected and run the wizard again. Choose +**Recover an interrupted update**, then select the saved session by its +public-key prefix and progress state. The wizard checks the complete source +and destination keys before restoring settings. A normal update also restores an unfinished +session for the selected destination before proceeding. -```bash -python3 tools/lora_ota/ota_wizard.py --recover --target PUBLIC_KEY -``` - -Keep the recovery record until the wizard reports that restoration is -complete. A lost install reply is followed by a running-image check; the -wizard does not blindly repeat an install command. Temporary radio windows -use the existing main-channel settings, so an interrupted process does not -leave the destination on a separate update frequency. +Keep the recovery record until restoration is complete. A lost install reply +is followed by a running-image check; the wizard does not blindly repeat an +install command. Temporary radio windows use the existing main-channel +settings. Existing temporary schedules block an update before settings are +changed, including on older firmware. The cache is normally `~/.cache/meshcore/ota` on Linux, `~/Library/Caches/MeshCore/ota` on macOS, or `%LOCALAPPDATA%\MeshCore\ota` on @@ -172,7 +167,33 @@ Windows. It contains firmware, public preparation records, selected board keys, and restoration settings. Trusted signer keys added for an update remain installed on the destination. -For the individual commands and a manual board-by-board example, see -[Station G2 to Ikoka Stick](ota_station_g2_to_ikoka_stick.md). For routed -updates and explicit multi-node setup, use the -[scripted LoRa OTA guide](lora_ota_automation.md). +
+Optional command-line use for automation + +Interactive users can use the menus for every step. Existing command-line +options remain available for scripts: + +| Option | Purpose | +| --- | --- | +| `--check` | Inspect without downloading or installing firmware | +| `--port /dev/ttyUSB0` | Choose a port; Windows example: `--port COM5` | +| `--mode repeater` or `--mode companion` | Select the source type | +| `--target PUBLIC_KEY` | Choose the full destination key; direct discovery is still required | +| `--firmware FILE` | Use a local firmware image or prepared bundle | +| `--base FILE` | Supply the exact running image for differential preparation | +| `--full` | Force a full transfer where supported | +| `--build` | Build custom Full firmware from source | +| `--no-bootloader` | Update application firmware only | +| `--prepare-only --request DEVICE.json --firmware FILE --output update.ota.zip` | Prepare without radios | +| `--recover --target PUBLIC_KEY` | Restore settings and exit | +| `--yes` | Accept the displayed plan; credentials and required choices must still be supplied | + +For unattended use, provide `MESHCORE_ADMIN_PASSWORD` in the process environment +and pass the port, source type, and target key. The password is excluded from +preferences, recovery records, and the custom build environment. + +
+ +For individual commands and a manual board example, see +[Station G2 to Ikoka Stick](ota_station_g2_to_ikoka_stick.md). For routed updates +and explicit multi-node setup, use the [scripted LoRa OTA guide](lora_ota_automation.md). diff --git a/tools/lora_ota/ota_wizard.py b/tools/lora_ota/ota_wizard.py index 7809be7c5..0a896104e 100644 --- a/tools/lora_ota/ota_wizard.py +++ b/tools/lora_ota/ota_wizard.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse import asyncio +from copy import copy import getpass import importlib.util import importlib.metadata @@ -73,6 +74,8 @@ def arguments(argv=None): parser.add_argument("--recover", action="store_true", help="restore settings from an interrupted update, then exit") parser.add_argument("--yes", action="store_true", help="accept the displayed update plan; does not guess missing credentials") parser.add_argument("--runtime-ready", action="store_true", help=argparse.SUPPRESS) + parser.add_argument("--menu-action", choices=("update", "check", "prepare", "recover"), + help=argparse.SUPPRESS) parser.add_argument("--check", action="store_true", help="inspect source and destination without preparing or installing an update") parser.add_argument("--prepare-only", action="store_true", help="prepare a portable update bundle on a computer without opening radios") parser.add_argument("--request", type=Path, help="destination preparation request saved by a previous live inspection") @@ -80,6 +83,131 @@ def arguments(argv=None): return parser.parse_args(argv) +class MenuCancelled(Exception): + """Return to the main menu without changing a radio.""" + + +def file_prompt(prompt: str, default: str | None = None) -> Path: + if default and Path(default).expanduser().is_file(): + selected = select(prompt, ["Use " + default, "Choose another file", "Back"]) + if selected == 0: + return Path(default).expanduser() + if selected == 2: + raise MenuCancelled + while True: + value = ask(prompt + " (blank to go back)") + if not value: + raise MenuCancelled + path = Path(value.strip().strip('"').strip("'")).expanduser() + if path.is_file(): + return path + print("That file was not found. Enter its path, or leave it blank to go back.") + + +def menu_settings_path() -> Path: + return cache_directory() / "menu-settings.json" + + +def load_menu_settings(args) -> None: + try: + values = json.loads(ota.read_bounded_file(menu_settings_path(), 8192, "menu settings")) + except (OSError, ValueError, ota.OtaError): + return + if not isinstance(values, dict): + return + for field in ("full", "no_bootloader"): + if type(values.get(field)) is bool: + setattr(args, field, values[field]) + if values.get("mode") in ("auto", "repeater", "companion"): + args.mode = values["mode"] + for field in ("motatool", "version", "port"): + value = values.get(field) + if value is None or isinstance(value, str) and len(value) <= 4096: + setattr(args, field, value) + for field in ("source_tree", "cache"): + value = values.get(field) + if isinstance(value, str) and value and len(value) <= 4096: + setattr(args, field, Path(value).expanduser()) + + +def settings_menu(args) -> None: + while True: + print("\nSettings") + options = [ + "Transfer: " + ("Full" if args.full else "Automatic (prefer differential)"), + "Check bootloader: " + ("Off" if args.no_bootloader else "On"), + "USB source type: " + args.mode.capitalize(), + "USB port: " + (args.port or "Automatic"), + "Differential builder: " + (args.motatool or "Automatic"), + "Custom source folder: " + str(args.source_tree), + "Custom version: " + (args.version or "Automatic"), + "Download/cache folder: " + str(args.cache), + "Back", + ] + choice = select("Setting", options, len(options)-1) + if choice == 8: + break + if choice == 0: + args.full = bool(select("Transfer", ["Automatic (prefer differential)", "Full"], int(args.full))) + elif choice == 1: + args.no_bootloader = bool(select("Check bootloader", ["On", "Off"], int(args.no_bootloader))) + elif choice == 2: + modes = ("auto", "repeater", "companion") + args.mode = modes[select("USB source type", ["Automatic", "Repeater", "Companion"], modes.index(args.mode))] + elif choice == 3: + args.port = ask("USB port (blank for automatic)") or None + elif choice == 4: + args.motatool = ask("Differential builder path (blank for automatic)") or None + elif choice == 5: + args.source_tree = Path(ask("MeshCore source folder", str(args.source_tree))).expanduser() + elif choice == 6: + args.version = ask("Custom version (blank for automatic)") or None + elif choice == 7: + args.cache = Path(ask("Download/cache folder", str(args.cache))).expanduser() + save_preferences(menu_settings_path(), **{ + field: str(getattr(args, field)) if field in ("cache", "source_tree") else getattr(args, field) + for field in ("full", "no_bootloader", "mode", "port", "motatool", "source_tree", "version", "cache")}) + + +def choose_request(args): + records = [] + for path in sorted((args.cache / "requests").glob("*.json"), key=lambda path: path.stat().st_mtime, reverse=True)[:50]: + try: + records.append((path, load_request(path))) + except (ota.OtaError, OSError): + continue + if not records: + return load_request(file_prompt("Preparation request JSON")) + choices = [f"{record.hw_id} — {record.public_key[:12].upper()} ({record.version_text})" + for _, record in records] + ["Open another preparation request file", "Back"] + chosen = select("Preparation request", choices) + if chosen == len(records) + 1: + raise MenuCancelled + return (records[chosen][1] if chosen < len(records) + else load_request(file_prompt("Preparation request JSON"))) + + +def choose_recovery(args) -> str: + records = [] + for path in sorted((args.cache / "sessions").glob("*.json"), key=lambda path: path.stat().st_mtime, reverse=True)[:50]: + try: + record = json.loads(ota.read_bounded_file(path, 64 * 1024, "recovery record")) + key = record.get("target_key") if isinstance(record, dict) else None + if (record.get("format") == 1 and isinstance(key, str) + and re.fullmatch(r"[0-9A-Fa-f]{64}", key) + and record.get("phase") not in ("finished", "prepared", "inspecting")): + records.append((key, record.get("phase", "unfinished"))) + except (ota.OtaError, OSError, ValueError, AttributeError): + continue + if not records: + print("No interrupted updates were found in this cache.") + raise MenuCancelled + chosen = select("Interrupted update", [f"{key[:12].upper()} — {phase}" for key, phase in records] + ["Back"]) + if chosen == len(records): + raise MenuCancelled + return records[chosen][0] + + def ensure_dependencies(args, argv) -> None: required = ("serial", "meshcore", "cryptography") missing = [name for name in required if importlib.util.find_spec(name) is None] @@ -231,6 +359,218 @@ def custom_build(args, destination, target_env: str) -> Path: raise ota.OtaError("The custom build did not produce a matching application image.") +async def latest_asset(catalog, destination): + print("Checking GitHub releases for this exact board and role…") + try: + return await asyncio.to_thread(catalog.choose_asset, target_id=destination.target_id) + except ota.OtaError as exc: + print(str(exc)) + return None + + +async def choose_release(catalog, destination): + pages = 3 + while True: + choices = await asyncio.to_thread(catalog.release_choices, target_id=destination.target_id, + include_prerelease=True, max_pages=pages) + if not choices: + print("No compatible firmware was found on these release pages.") + labels = [choice.label for choice in choices] + more = pages < 10 and catalog.has_more_releases + if more: + labels.append("Load older GitHub releases") + labels.append("Back") + selected = select("GitHub release", labels) + if selected < len(choices): + return choices[selected].asset + if more and selected == len(choices): + pages = min(10, pages + 3) + continue + raise MenuCancelled + + +async def choose_firmware(args, destination, catalog, asset, remembered): + if remembered.get("target") != destination.public_key.lower(): + remembered = {} + if args.firmware: + return args.firmware, "local" + preferred = remembered.get("firmware_source", "latest") + while True: + source = "build" if args.build else "latest" + if not args.build and not args.yes: + sources = ["latest", "release", "local"] + labels = ["Latest compatible GitHub release (recommended)", + "Choose another GitHub release", "Local firmware file or prepared bundle"] + if not small_host(): + sources.append("build") + labels.append("Build custom Full firmware from source") + default = sources.index(preferred) if preferred in sources else 0 + selected = select("Firmware", labels + ["Back"], default) + if selected == len(sources): + raise MenuCancelled + source = sources[selected] + if source == "local": + try: + return file_prompt("Firmware ZIP, BIN, HEX, full mOTA, or prepared bundle", + remembered.get("firmware_file")), source + except MenuCancelled: + continue + if source == "release": + try: + selected_asset = await choose_release(catalog, destination) + except MenuCancelled: + continue + else: + if asset is None: + asset = await latest_asset(catalog, destination) + if asset is None: + if args.yes or args.build: + raise ota.OtaError("No GitHub application matched this board. Choose another release or a local firmware file.") + preferred = "local" + continue + selected_asset = asset + if source in ("latest", "release"): + print(f"GitHub release: {selected_asset.release_tag}\n {selected_asset.name}") + return await asyncio.to_thread(download_asset, selected_asset, args.cache / "downloads"), source + if not args.version: + args.version = ota.format_version(max(destination.version, selected_asset.version) + 1) + "-custom" + return await asyncio.to_thread(custom_build, args, destination, selected_asset.target_env), source + + +async def choose_missing_base(args, destination, catalog): + """Offer the useful fallbacks only after the automatic exact-base search.""" + pages = 3 + while True: + choices, actions = [], [] + if pages < 10: + choices.append("Search older GitHub releases for the running firmware") + actions.append("github") + choices.append("Choose a local copy of the exact running firmware") + actions.append("local") + if destination.platform == "esp32" or destination.external_storage: + choices.append("Use a full update instead") + actions.append("full") + choices.append("Back") + actions.append("back") + action = actions[select("The exact running image was not found", choices)] + if action == "back": + raise MenuCancelled + if action == "full": + return None + if action == "local": + supplied = file_prompt("Exact running firmware ZIP, BIN, HEX, or full mOTA") + try: + return await asyncio.to_thread(find_running_image, destination, args.cache, catalog, supplied=supplied) + except ota.OtaError as exc: + print(str(exc)) + continue + pages = min(10, pages + 3) + try: + await asyncio.to_thread(catalog.release_choices, target_id=destination.target_id, + include_prerelease=True, max_pages=pages) + base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog, + candidate_limit=16 if pages < 10 else 32) + if base: + return base + if not catalog.has_more_releases: + pages = 10 + print("No downloaded GitHub candidate matched the exact running image hash.") + except ota.OtaError as exc: + print(str(exc)) + + +async def prepare_selected_firmware(args, destination, firmware, catalog): + prepared = await asyncio.to_thread(load_bundle, firmware, destination) + if prepared: + image, application = prepared + if args.full and not application.info.is_full: + application = await asyncio.to_thread(prepare_application, image, destination, args.cache, full=True) + else: + image = await asyncio.to_thread(load_application, firmware, + target_id=destination.target_id, hw_id=destination.hw_id) + application = await asyncio.to_thread(load_ready_application, firmware, image, destination) + ImageCache(args.cache).remember(image) + if image.fw_version <= destination.version: + print("The selected firmware is already installed or older; keeping the running firmware.") + return image, None + if application: + return image, application + builder = None if args.full else find_builder(args.cache, args.motatool) + if not args.full and not builder and not small_host(): + builder = await asyncio.to_thread(install_builder, args.cache) + base = None + if builder: + base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog, supplied=args.base) + if base is None and getattr(args, "interactive_menu", False): + base = await choose_missing_base(args, destination, catalog) + if not builder and not args.full: + print("Using a full update: no differential compiler setup is needed on this host.") + if destination.platform == "nrf52" and not destination.external_storage and not (base and builder): + raise ota.OtaError( + "This radio needs an exact-base differential update. On a computer, choose " + "Prepare an update bundle and select its saved preparation request. " + "A compatible differential builder must be available; its path can be set under Settings.") + application = await asyncio.to_thread(prepare_application, image, destination, args.cache, + base=base, builder=builder, full=args.full) + return image, application + + +async def prepare_interactively(args) -> int: + destination = choose_request(args) + print(f"Preparing for {destination.hw_id} — {destination.public_key[:12].upper()} ({destination.version_text})") + catalog = GitHubCatalog() + args.interactive_menu = True + firmware, _ = await choose_firmware(args, destination, catalog, None, {}) + verification_cache = ota.cryptography_repair_root() + if (verification_cache / "cryptography").is_dir(): + try: + ota.activate_cached_cryptography(verification_cache) + except ota.OtaError: + pass # A working system verifier may already be loaded. + try: + image, package = await prepare_selected_firmware(args, destination, firmware, catalog) + except ImportError: + # Unsigned full packaging stays dependency-free. Signed inputs request + # their verifier only when it is actually needed; no USB SDK is loaded. + await asyncio.to_thread(ensure_signature_support) + image, package = await prepare_selected_firmware(args, destination, firmware, catalog) + if package is None: + print("Choose a newer firmware release to prepare an update.") + return 0 + folder = args.cache / "bundles" + default = folder / f"{destination.hw_id}-{ota.format_version(image.fw_version)}.ota.zip" + index = 1 + while default.exists(): + default = folder / f"{destination.hw_id}-{ota.format_version(image.fw_version)}-{index}.ota.zip" + index += 1 + output = Path(ask("Save update bundle", str(default)).strip().strip('"').strip("'")).expanduser() + await asyncio.to_thread(save_bundle, output, image, package) + print(f"Prepared {package.info.kind}: {output} ({len(package.blob):,} transfer bytes)") + print("Copy this ZIP to the Pi. Choose Update a nearby radio, then Local firmware file or prepared bundle.") + return 0 + + +def ensure_signature_support() -> None: + root = ota.cryptography_repair_root() + if root.is_dir(): + try: + ota.activate_cached_cryptography(root) + return + except ota.OtaError: + pass + print("Preparing signature verification support. This is a one-time download.") + root.mkdir(parents=True, exist_ok=True) + log = root / "setup.log" + environment = {key: value for key, value in os.environ.items() + if key not in ("MESHCORE_ADMIN_PASSWORD", "MESHCORE_OTA_PASSWORD")} + with log.open("w", encoding="utf-8") as output: + completed = subprocess.run(ota.cryptography_repair_command(root), env=environment, + stdout=output, stderr=subprocess.STDOUT, timeout=600) + if completed.returncode: + raise ota.OtaError(f"Signature verification support could not be installed. Details: {log}") + ota.activate_cached_cryptography(root) + + async def run(args) -> int: args.cache.mkdir(parents=True, exist_ok=True, mode=0o700) if args.target and not re.fullmatch(r"[0-9A-Fa-f]{64}", args.target): @@ -314,65 +654,8 @@ async def run(args) -> int: return 0 await session.snapshot() catalog = GitHubCatalog() - print("Checking published firmware for this exact board and role…") - try: - asset = await asyncio.to_thread(catalog.choose_asset, target_id=destination.target_id) - except ota.OtaError as exc: - asset = None - if not args.firmware and not args.build: - print(str(exc)) - choice = 2 if args.build else 1 if args.firmware else 0 - if not args.yes and not args.firmware and not args.build: - options = ["Published GitHub firmware (recommended)", "Custom firmware file"] - if not small_host(): - options.append("Build custom Full firmware from source") - default_choice = remembered.get("firmware_choice", 0 if asset else 1) - if remembered.get("target") != key.lower() or default_choice not in range(len(options)): - default_choice = 0 if asset else 1 - choice = select("Firmware", options, default_choice) - if choice == 0: - if not asset: - raise ota.OtaError("No published firmware matched this board; supply --firmware or choose a custom build.") - firmware = await asyncio.to_thread(download_asset, asset, args.cache / "downloads") - elif choice == 1: - default_file = remembered.get("firmware_file") if remembered.get("target") == key.lower() else None - firmware = args.firmware or Path(ask("Custom application ZIP, BIN, HEX, full mOTA, or prepared bundle", default_file)) - else: - if not asset: - raise ota.OtaError("A custom build needs the exact target environment; use a local firmware file for this board.") - if not args.version: - args.version = ota.format_version(max(destination.version, asset.version) + 1) + "-custom" - firmware = await asyncio.to_thread(custom_build, args, destination, asset.target_env) - prepared = await asyncio.to_thread(load_bundle, firmware, destination) - if prepared: - image, application = prepared - if args.full and not application.info.is_full: - application = await asyncio.to_thread(prepare_application, image, destination, args.cache, full=True) - else: - image = await asyncio.to_thread(load_application, firmware, - target_id=destination.target_id, hw_id=destination.hw_id) - application = await asyncio.to_thread(load_ready_application, firmware, image, destination) - ImageCache(args.cache).remember(image) - if image.fw_version > destination.version and application is None: - builder = find_builder(args.cache, args.motatool) - if not args.full and not builder and not small_host(): - builder = await asyncio.to_thread(install_builder, args.cache) - base = None - if not args.full and builder: - base = await asyncio.to_thread(find_running_image, destination, args.cache, catalog, - supplied=args.base) - if not builder: - print("Using a full update: no differential compiler setup is needed on this host.") - if not builder and destination.platform == "nrf52" and not destination.external_storage: - raise ota.OtaError( - "This internal-flash radio requires a differential package. Prepare it on a computer with " - f"--prepare-only --request {request_file} --firmware NEW_FIRMWARE --base RUNNING_FIRMWARE " - "--output update.ota.zip, then select that ZIP on the Pi.") - application = await asyncio.to_thread(prepare_application, image, destination, args.cache, - base=base, builder=builder, full=args.full) - elif image.fw_version <= destination.version: - application = None - print("The selected published/application version is already installed or older; keeping the running firmware.") + firmware, choice = await choose_firmware(args, destination, catalog, None, remembered) + image, application = await prepare_selected_firmware(args, destination, firmware, catalog) bootloader = None if not args.no_bootloader and destination.platform == "nrf52": print("Checking the exact-board bootloader update…") @@ -398,8 +681,8 @@ async def run(args) -> int: destination = await session.install(package, destination) print(f"Verified {package.info.kind} installation.") save_preferences(args.cache / "choices.json", port=port, mode=mode, - target=key.lower(), hw=destination.hw_id, firmware_choice=choice, - firmware_file=str(firmware) if choice == 1 else None) + target=key.lower(), hw=destination.hw_id, firmware_source=choice, + firmware_file=str(firmware) if choice == "local" else None) print("Update verified successfully.") return 0 finally: @@ -437,18 +720,59 @@ def prepare_without_radios(args) -> int: return 0 +def menu_main(args) -> int: + load_menu_settings(args) + action = args.menu_action + while True: + if action is None: + print("\nMeshCore LoRa updater") + actions = ("update", "check", "prepare", "recover", "settings", "exit") + action = actions[select("Choose", [ + "Update a nearby radio", "Check radios / save a preparation request", + "Prepare an update bundle on this computer", "Recover an interrupted update", + "Settings", "Exit"])] + current = copy(args) + current.interactive_menu = True + current.check, current.recover = action == "check", action == "recover" + try: + if action == "exit": + return 0 + if action == "settings": + settings_menu(args) + elif action == "prepare": + asyncio.run(prepare_interactively(current)) + else: + # Preserve the chosen menu action if first-run dependency + # setup reexecutes inside the private Python environment. + ensure_dependencies(current, ["--menu-action", action, "--cache", str(args.cache)]) + if action == "recover": + current.target = choose_recovery(current) + asyncio.run(run(current)) + except MenuCancelled: + pass + except (ota.OtaError, RuntimeError, OSError, ValueError, ImportError, subprocess.TimeoutExpired) as exc: + print(f"Update stopped: {exc}", file=sys.stderr) + action = None + + def main(argv=None) -> int: argv = sys.argv[1:] if argv is None else argv args = arguments(argv) try: + if args.menu_action or not any(value != "--runtime-ready" for value in argv): + return menu_main(args) if args.prepare_only: return prepare_without_radios(args) ensure_dependencies(args, argv) return asyncio.run(run(args)) + except MenuCancelled: + return 0 + except EOFError: + return 0 except KeyboardInterrupt: print("\nStopped. Check the recovery record if the destination was unreachable during cleanup.", file=sys.stderr) return 130 - except (ota.OtaError, RuntimeError, OSError, ValueError, subprocess.TimeoutExpired) as exc: + except (ota.OtaError, RuntimeError, OSError, ValueError, ImportError, subprocess.TimeoutExpired) as exc: print(f"Update stopped: {exc}", file=sys.stderr) return 1 diff --git a/tools/lora_ota/ota_wizard_catalog.py b/tools/lora_ota/ota_wizard_catalog.py index d9e1d8c1d..1a9377a45 100644 --- a/tools/lora_ota/ota_wizard_catalog.py +++ b/tools/lora_ota/ota_wizard_catalog.py @@ -50,6 +50,30 @@ class ReleaseAsset: return ota.format_version(self.version) +@dataclass(frozen=True) +class ReleaseChoice: + """One exact-board application from a published release or split family.""" + release_tag: str + asset: ReleaseAsset + prerelease: bool + published_at: str + + @property + def label(self) -> str: + # GitHub tags are source data; do not echo control sequences to a terminal. + tag = "".join(character if character.isprintable() else "?" + for character in self.release_tag) + return tag + (" (prerelease)" if self.prerelease else "") + + +@dataclass(frozen=True) +class _CatalogRelease: + tag: str + prerelease: bool + published_at: str + assets: tuple[ReleaseAsset, ...] + + @dataclass(frozen=True) class UpdatePlan: kind: str @@ -199,61 +223,72 @@ class GitHubCatalog: self.opener = opener self._assets: list[ReleaseAsset] | None = None self._all_assets: list[ReleaseAsset] = [] + self._pages: dict[int, tuple[_CatalogRelease, ...]] = {} + self._last_page: int | None = None + + @property + def has_more_releases(self) -> bool: + return self._last_page is None and len(self._pages) < 10 + + def _releases(self, max_pages: int) -> list[_CatalogRelease]: + if type(max_pages) is not int or not 1 <= max_pages <= 10: + raise ota.OtaError("Catalog page limit must be between 1 and 10") + releases = [] + for page in range(1, max_pages + 1): + if self._last_page is not None and page > self._last_page: + break + if page not in self._pages: + # A split family contains thousands of assets. Cache only parsed + # application metadata, keeping each response bounded on a Pi. + url = f"https://api.github.com/repos/{self.repository}/releases?per_page=10&page={page}" + with _open(url, api=True, opener=self.opener) as response: + raw = response.read(MAX_CATALOG_BYTES + 1) + if len(raw) > MAX_CATALOG_BYTES: + raise ota.OtaError("GitHub release catalog is unexpectedly large") + try: + rows = json.loads(raw) + except (ValueError, UnicodeError) as exc: + raise ota.OtaError("GitHub returned an invalid release catalog") from exc + if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows): + raise ota.OtaError("GitHub returned an invalid release catalog") + parsed = [] + for release in rows: + tag = release.get("tag_name") + if release.get("draft") or not isinstance(tag, str) or _version(tag) is None: + continue + assets = release.get("assets", []) + if not isinstance(assets, list): + raise ota.OtaError("GitHub returned an invalid release asset list") + candidates = tuple(candidate for row in assets if isinstance(row, dict) + if (candidate := parse_release_asset(row, tag, self.repository)) is not None) + published = release.get("published_at") or "" + parsed.append(_CatalogRelease(tag, _is_prerelease(release), + published if isinstance(published, str) else "", candidates)) + self._pages[page] = tuple(parsed) + if len(rows) < 10: + self._last_page = page + releases.extend(self._pages[page]) + # Extra pages loaded through the release menu also become candidates for + # exact-base discovery. Their downloaded body hashes remain authoritative. + self._all_assets = [asset for page in sorted(self._pages) + for release in self._pages[page] for asset in release.assets] + return releases def fetch_assets(self, *, include_prerelease: bool = False, max_pages: int = 3) -> list[ReleaseAsset]: - if not 1 <= max_pages <= 10: - raise ota.OtaError("Catalog page limit must be between 1 and 10") - releases: list[dict] = [] - for page in range(1, max_pages + 1): - # Release families contain thousands of per-board assets. Small - # pages keep GitHub's embedded asset metadata bounded on a Pi. - url = f"https://api.github.com/repos/{self.repository}/releases?per_page=10&page={page}" - with _open(url, api=True, opener=self.opener) as response: - raw = response.read(MAX_CATALOG_BYTES + 1) - if len(raw) > MAX_CATALOG_BYTES: - raise ota.OtaError("GitHub release catalog is unexpectedly large") - try: - rows = json.loads(raw) - except (ValueError, UnicodeError) as exc: - raise ota.OtaError("GitHub returned an invalid release catalog") from exc - if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows): - raise ota.OtaError("GitHub returned an invalid release catalog") - releases.extend(row for row in rows if not row.get("draft") - and (include_prerelease or not _is_prerelease(row))) - if len(rows) < 10: - break - mains = [row for row in releases if re.match(r"^v\d", str(row.get("tag_name", "")), re.I) - and _version(row["tag_name"]) is not None] + releases = [release for release in self._releases(max_pages) + if include_prerelease or not release.prerelease] + mains = [release for release in releases if re.match(r"^v\d", release.tag, re.I)] if not mains: raise ota.OtaError("No published main firmware release was found") - main = max(mains, key=lambda row: (_version(row["tag_name"]), row.get("published_at") or "")) - family = main["tag_name"] - result: list[ReleaseAsset] = [] - historical: list[ReleaseAsset] = [] - for release in releases: - tag = str(release.get("tag_name", "")) - release_family = _release_family(tag) - if release_family is None: - continue - rows = release.get("assets", []) - if not isinstance(rows, list): - raise ota.OtaError("GitHub returned an invalid release asset list") - for row in rows: - if isinstance(row, dict): - candidate = parse_release_asset(row, tag, self.repository) - if candidate is not None: - historical.append(candidate) - if release_family == family: - result.append(candidate) - self._all_assets = historical - self._assets = result - return list(result) + main = max(mains, key=lambda release: (_version(release.tag), release.published_at)) + self._assets = [asset for release in releases if _release_family(release.tag) == main.tag + for asset in release.assets] + return list(self._assets) - def candidates(self, target_env: str | None = None, *, target_id: int | None = None, - profile: str = "full", include_embedded: bool = False) -> list[ReleaseAsset]: + @staticmethod + def _compatible(assets, target_env, target_id, profile, include_embedded=False): if target_env is None and target_id is None: raise ota.OtaError("Select an exact firmware target") - assets = self.fetch_assets() if self._assets is None else self._assets result = [row for row in assets if (target_env is None or row.target_env == target_env) and (target_id is None or row.target_id == target_id) and (include_embedded or not row.embedded_bootloader)] @@ -261,7 +296,12 @@ class GitHubCatalog: exact = [row for row in result if row.profile == profile] result = exact or [row for row in result if row.profile == "standard"] preference = {"mota": 0, "zip": 1, "bin": 2, "hex": 3} - return sorted(result, key=lambda row: (-row.version, preference[row.kind], row.name)) + return sorted(result, key=lambda row: (-row.version, preference[row.kind], row.name, row.release_tag)) + + def candidates(self, target_env: str | None = None, *, target_id: int | None = None, + profile: str = "full", include_embedded: bool = False) -> list[ReleaseAsset]: + assets = self.fetch_assets() if self._assets is None else self._assets + return self._compatible(assets, target_env, target_id, profile, include_embedded) def choose_asset(self, target_env: str | None = None, **kwargs) -> ReleaseAsset: choices = self.candidates(target_env, **kwargs) @@ -269,6 +309,35 @@ class GitHubCatalog: raise ota.OtaError("No published application matches this exact board, role, and profile") return choices[0] + def release_choices(self, target_env: str | None = None, *, target_id: int | None = None, + profile: str = "full", include_prerelease: bool = True, + max_pages: int = 3) -> list[ReleaseChoice]: + """Browse exact-board releases; loading older pages reuses parsed pages. + + Known split release shards are one family choice. Board-specific preview + releases are also selectable here, while choose_asset stays stable-only. + No file is downloaded until the caller selects a choice's asset. + """ + if target_env is None and target_id is None: + raise ota.OtaError("Select an exact firmware target") + families: dict[str, list[_CatalogRelease]] = {} + for release in self._releases(max_pages): + if not include_prerelease and release.prerelease: + continue + family = _release_family(release.tag) or release.tag + families.setdefault(family, []).append(release) + choices = [] + for family, releases in families.items(): + assets = self._compatible([asset for release in releases for asset in release.assets], + target_env, target_id, profile) + if assets: + choices.append(ReleaseChoice(family, assets[0], + any(release.prerelease for release in releases), + max(release.published_at for release in releases))) + choices.sort(key=lambda choice: (choice.asset.version, choice.published_at, choice.release_tag), reverse=True) + choices.sort(key=lambda choice: choice.prerelease) + return choices + def base_candidates(self, target_env: str | None = None, *, target_id: int | None = None, fw_version: int | None = None, limit: int = 8) -> list[ReleaseAsset]: """Find possible old images; their downloaded EndF body hash MUST be checked.""" @@ -276,8 +345,8 @@ class GitHubCatalog: raise ota.OtaError("Select an exact firmware target") if not 1 <= limit <= 32: raise ota.OtaError("Base candidate limit must be between 1 and 32") - if self._assets is None: - self.fetch_assets() + if not self._pages: + self._releases(3) preference = {"mota": 0, "zip": 1, "bin": 2, "hex": 3} matches = [row for row in self._all_assets if not row.embedded_bootloader diff --git a/tools/lora_ota/ota_wizard_packages.py b/tools/lora_ota/ota_wizard_packages.py index a7402bf5a..cd2b591f7 100644 --- a/tools/lora_ota/ota_wizard_packages.py +++ b/tools/lora_ota/ota_wizard_packages.py @@ -88,7 +88,8 @@ def install_builder(cache: Path, progress=print) -> str | None: def find_running_image(destination: Destination, cache: Path, catalog: GitHubCatalog | None, - *, supplied: Path | None = None, progress=print) -> Path | None: + *, supplied: Path | None = None, progress=print, + candidate_limit: int = 4) -> Path | None: images = ImageCache(cache) if supplied: identity = load_application(supplied, target_id=destination.target_id, hw_id=destination.hw_id) @@ -102,7 +103,7 @@ def find_running_image(destination: Destination, cache: Path, catalog: GitHubCat return found if catalog: for asset in catalog.base_candidates(target_id=destination.target_id, - fw_version=destination.version, limit=4): + fw_version=destination.version, limit=candidate_limit): try: candidate = download_asset(asset, cache / "downloads") identity = load_application(candidate, target_id=destination.target_id, diff --git a/tools/lora_ota/test_ota_wizard_catalog.py b/tools/lora_ota/test_ota_wizard_catalog.py index f0a5a9c8d..cddac3b09 100644 --- a/tools/lora_ota/test_ota_wizard_catalog.py +++ b/tools/lora_ota/test_ota_wizard_catalog.py @@ -92,6 +92,110 @@ class CatalogTest(unittest.TestCase): with self.assertRaises(ota.OtaError): source.choose_asset("another_repeater") + def test_release_menu_groups_shards_and_offers_exact_board_previews(self): + shard = f"lora-ota-10-{FAMILY}" + preview = "ikoka-v1.18.0.2-alpha2" + shard_asset = asset_row(tag=shard) + preview_asset = asset_row(f"{ENV}-full-ota-v1.18.0.2-alpha2.zip", preview) + old_asset = asset_row(f"{ENV}-full-ota-{OLD_FAMILY}.zip", OLD_FAMILY) + rows = [release(preview, [preview_asset], prerelease=True), + release(FAMILY, [asset_row(f"{ENV}-full-ota-{FAMILY}.bin")]), + release(shard, [shard_asset]), + release(f"full-profiles-{FAMILY}", [asset_row( + f"{ENV}-reduced-ota-{FAMILY}.zip", f"full-profiles-{FAMILY}")]), + release(OLD_FAMILY, [old_asset]), + release("v9.0.0-draft", [asset_row()], draft=True), + release("another-v8.0.0", [asset_row(f"another_repeater-full-ota-v8.0.0.zip", + "another-v8.0.0")])] + source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url)) + choices = source.release_choices(target_id=TARGET) + self.assertEqual([choice.release_tag for choice in choices], [FAMILY, OLD_FAMILY, preview]) + self.assertEqual(choices[0].asset.name, shard_asset["name"]) + self.assertEqual(choices[0].asset.release_tag, shard) + self.assertFalse(choices[0].prerelease) + self.assertTrue(choices[-1].prerelease) + self.assertIn("prerelease", choices[-1].label) + self.assertEqual(source.choose_asset(target_id=TARGET), choices[0].asset) + self.assertEqual([choice.release_tag for choice in source.release_choices( + target_id=TARGET, include_prerelease=False)], [FAMILY, OLD_FAMILY]) + self.assertFalse(source.has_more_releases) + + def test_menu_prefers_full_and_does_not_offer_embedded_or_wrong_role(self): + rows = [release(FAMILY, [ + asset_row(f"{ENV}-with-bootloader-2.4.12-full-ota-{FAMILY}.zip"), + asset_row(f"ikoka_stick_nrf_33dbm_room_server_lora_ota-full-ota-{FAMILY}.zip"), + asset_row(f"{ENV}-reduced-ota-{FAMILY}.zip"), + asset_row(f"{ENV}-ota-{FAMILY}.bin"), asset_row()])] + source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url)) + choices = source.release_choices(ENV) + self.assertEqual(len(choices), 1) + self.assertEqual(choices[0].asset.profile, "full") + self.assertEqual(source.release_choices(ENV, profile="reduced")[0].asset.profile, "reduced") + self.assertEqual(source.release_choices("no_matching_repeater"), []) + + def test_menu_loads_older_pages_once_and_adds_exact_base_candidates(self): + def page_rows(number): + if number == 1: + head = release(FAMILY, [asset_row()]) + elif number == 2: + head = release(OLD_FAMILY, [asset_row(f"{ENV}-full-ota-{OLD_FAMILY}.zip", OLD_FAMILY)]) + else: + return [] + return [head] + [release(f"v0.0.0-empty-{number}-{index}") for index in range(9)] + calls = [] + def opener(request, timeout): + number = int(request.full_url.rsplit("=", 1)[1]) + calls.append(number) + return Response(json.dumps(page_rows(number)).encode(), request.full_url) + source = catalog.GitHubCatalog(opener=opener) + self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=1)), 1) + self.assertTrue(source.has_more_releases) + self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=3)), 2) + self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=6)), 2) + self.assertEqual(calls, [1, 2, 3]) + self.assertFalse(source.has_more_releases) + self.assertEqual(source.base_candidates(target_id=TARGET, fw_version=0x01110109)[0].release_tag, OLD_FAMILY) + self.assertEqual(source.choose_asset(target_id=TARGET).release_tag, FAMILY) + self.assertEqual(calls, [1, 2, 3]) + + def test_preview_only_menu_still_supplies_an_exact_base_candidate(self): + tag = "ikoka-v1.18.0.2-alpha1" + rows = [release(tag, [asset_row(f"{ENV}-full-ota-v1.18.0.2-alpha1.zip", tag)], prerelease=True)] + source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(json.dumps(rows).encode(), req.full_url)) + choices = source.release_choices(target_id=TARGET) + self.assertEqual(len(choices), 1) + self.assertTrue(choices[0].prerelease) + self.assertEqual(source.base_candidates(target_id=TARGET, fw_version=0x01120002), [choices[0].asset]) + with self.assertRaisesRegex(ota.OtaError, "main firmware release"): + source.choose_asset(target_id=TARGET) + + def test_menu_page_errors_are_reported_without_caching_partial_results(self): + calls = [] + def opener(request, timeout): + page = int(request.full_url.rsplit("=", 1)[1]) + calls.append(page) + if page == 1: + rows = [release(FAMILY, [asset_row()])] + [release(f"v0.0.0-empty-{index}") for index in range(9)] + return Response(json.dumps(rows).encode(), request.full_url) + return Response(b"not a release catalog" if calls.count(2) == 1 else b"[]", request.full_url) + source = catalog.GitHubCatalog(opener=opener) + self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=1)), 1) + with self.assertRaisesRegex(ota.OtaError, "invalid release catalog"): + source.release_choices(target_id=TARGET, max_pages=3) + self.assertEqual(len(source.release_choices(target_id=TARGET, max_pages=3)), 1) + self.assertEqual(calls, [1, 2, 2]) + + def test_menu_validates_target_page_limits_and_terminal_labels(self): + source = catalog.GitHubCatalog(opener=lambda req, timeout: Response(b"[]", req.full_url)) + with self.assertRaisesRegex(ota.OtaError, "exact firmware target"): + source.release_choices() + for limit in (0, 11, True, "3"): + with self.subTest(limit=limit), self.assertRaisesRegex(ota.OtaError, "page limit"): + source.release_choices(target_id=TARGET, max_pages=limit) + item = catalog.ReleaseChoice("v1.18.0.2\x1b[1m\n", catalog.parse_release_asset(asset_row(), FAMILY), True, "") + self.assertNotIn("\x1b", item.label) + self.assertNotIn("\n", item.label) + def test_untrusted_asset_metadata_is_not_selectable(self): for mutation in ({"browser_download_url": "http://github.com/file.zip"}, {"browser_download_url": "https://example.com/file.zip"}, diff --git a/tools/lora_ota/test_ota_wizard_menu.py b/tools/lora_ota/test_ota_wizard_menu.py new file mode 100644 index 000000000..b54bd899e --- /dev/null +++ b/tools/lora_ota/test_ota_wizard_menu.py @@ -0,0 +1,335 @@ +"""Exercise the normal menu journeys without radios, compilers, or downloads.""" +import asyncio +from contextlib import ExitStack, redirect_stderr, redirect_stdout +import io +import json +import os +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest import mock + +import lora_ota as ota +import ota_wizard as wizard +import ota_wizard_bundle as bundles +import ota_wizard_packages as packages +from ota_wizard_catalog import ReleaseChoice +from test_ota_wizard_bundle import KEY, destination, image, signed_package + + +class WizardMenuTests(unittest.TestCase): + def setUp(self): + self.files = tempfile.TemporaryDirectory(prefix="meshcore-wizard-menu-") + self.addCleanup(self.files.cleanup) + self.cache = Path(self.files.name) + self.stack = ExitStack() + self.addCleanup(self.stack.close) + self.stack.enter_context(mock.patch.object(wizard, "cache_directory", return_value=self.cache)) + self.stack.enter_context(mock.patch.object(wizard.sys.stdin, "isatty", return_value=True)) + self.output = io.StringIO() + self.stack.enter_context(redirect_stdout(self.output)) + self.stack.enter_context(redirect_stderr(self.output)) + + def answers(self, values): + return mock.patch("builtins.input", side_effect=values) + + def test_plain_launch_can_exit_without_usb_or_dependency_setup(self): + with self.answers(["6"]), \ + mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \ + mock.patch.object(wizard, "run", side_effect=AssertionError("radio access")): + self.assertEqual(wizard.main([]), 0) + self.assertIn("Update a nearby radio", self.output.getvalue()) + self.assertIn("Prepare an update bundle on this computer", self.output.getvalue()) + + def test_update_and_check_need_only_menu_numbers(self): + for action, menu_number in (("update", "1"), ("check", "2")): + with self.subTest(action=action), self.answers([menu_number, "6"]), \ + mock.patch.object(wizard, "ensure_dependencies") as setup, \ + mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run: + self.assertEqual(wizard.main([]), 0) + run.assert_awaited_once() + options = run.await_args.args[0] + self.assertEqual(options.check, action == "check") + self.assertFalse(options.recover) + self.assertTrue(options.interactive_menu) + self.assertIsNone(options.port) + self.assertIsNone(options.target) + self.assertEqual(setup.call_args.args[1], + ["--menu-action", action, "--cache", str(self.cache)]) + + def test_runtime_resume_runs_selected_action_before_showing_main_menu(self): + with self.answers(["6"]) as input_call, \ + mock.patch.object(wizard, "ensure_dependencies"), \ + mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run: + self.assertEqual(wizard.main(["--runtime-ready", "--menu-action", "check"]), 0) + self.assertTrue(run.await_args.args[0].check) + self.assertEqual(input_call.call_count, 1) # Only the post-check Exit choice. + + def test_failed_update_returns_to_menu_without_a_traceback(self): + with self.answers(["1", "6"]), mock.patch.object(wizard, "ensure_dependencies"), \ + mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, + side_effect=ota.OtaError("Connect a source radio")): + self.assertEqual(wizard.main([]), 0) + self.assertIn("Update stopped: Connect a source radio", self.output.getvalue()) + self.assertNotIn("Traceback", self.output.getvalue()) + + def test_bad_menu_numbers_retry_and_back_cancels_before_usb_setup(self): + with self.answers(["nonsense", "0", "3", "", "6"]), \ + mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")): + self.assertEqual(wizard.main([]), 0) + self.assertEqual(self.output.getvalue().count("Enter one of the numbers above."), 2) + + def test_settings_round_trip_without_saving_environment_credentials(self): + source, download_cache = self.cache / "custom source", self.cache / "downloads elsewhere" + values = ["1", "2", "2", "2", "3", "2", "4", "COM12", "5", "/tools/motatool", + "6", str(source), "7", "v1.18.0.4-custom", "8", str(download_cache), "9"] + options = wizard.arguments([]) + with self.answers(values), mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "never save me"}): + wizard.settings_menu(options) + stored = json.loads(wizard.menu_settings_path().read_text()) + self.assertNotIn("never save me", json.dumps(stored)) + self.assertEqual(set(stored), {"full", "no_bootloader", "mode", "port", "motatool", + "source_tree", "version", "cache"}) + restored = wizard.arguments([]) + wizard.load_menu_settings(restored) + self.assertTrue(restored.full) + self.assertTrue(restored.no_bootloader) + self.assertEqual((restored.mode, restored.port), ("repeater", "COM12")) + self.assertEqual(restored.motatool, "/tools/motatool") + self.assertEqual(restored.source_tree, source) + self.assertEqual(restored.cache, download_cache) + self.assertEqual(restored.version, "v1.18.0.4-custom") + + def test_blank_settings_can_restore_automatic_port_builder_and_version(self): + options = wizard.arguments([]) + options.port, options.motatool, options.version = "COM12", "/tools/motatool", "v1.18.0.4-custom" + with self.answers(["4", "", "5", "", "7", "", "9"]): + wizard.settings_menu(options) + self.assertIsNone(options.port) + self.assertIsNone(options.motatool) + self.assertIsNone(options.version) + + def test_menu_prepares_real_full_bundle_from_github_and_cached_request(self): + request = self.cache / "requests" / (KEY + ".json") + bundles.save_request(request, destination()) + firmware = self.cache / "published.bin" + firmware.write_bytes(image().image) + asset = SimpleNamespace(release_tag="v1.18.0.2", name="ikoka-repeater-v1.18.0.2.zip") + catalog = SimpleNamespace(choose_asset=mock.Mock(return_value=asset)) + with self.answers(["3", "1", "1", "", "6"]), \ + mock.patch.object(wizard, "GitHubCatalog", return_value=catalog), \ + mock.patch.object(wizard, "download_asset", return_value=firmware) as download, \ + mock.patch.object(wizard, "small_host", return_value=True), \ + mock.patch.object(wizard, "find_builder", return_value=None), \ + mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \ + mock.patch.object(wizard, "choose_port", side_effect=AssertionError("USB access")), \ + mock.patch.object(wizard, "install_builder", side_effect=AssertionError("compiler setup")): + self.assertEqual(wizard.main([]), 0) + catalog.choose_asset.assert_called_once_with(target_id=destination().target_id) + download.assert_called_once_with(asset, self.cache / "downloads") + outputs = list((self.cache / "bundles").glob("*.ota.zip")) + self.assertEqual(len(outputs), 1) + prepared_image, package = bundles.load_bundle(outputs[0], destination()) + self.assertEqual(prepared_image.image, image().image) + self.assertTrue(package.info.is_full) + self.assertEqual(package.body_hash, image().body_hash) + self.assertIn("Local firmware file or prepared bundle", self.output.getvalue()) + + def test_recovery_menu_lists_only_unfinished_records_and_supplies_full_key(self): + folder = self.cache / "sessions" + folder.mkdir() + (folder / (KEY + ".json")).write_text(json.dumps( + {"format": 1, "target_key": KEY, "phase": "needs cleanup"})) + (folder / ("22" * 32 + ".json")).write_text(json.dumps( + {"format": 1, "target_key": "22" * 32, "phase": "finished"})) + (folder / "invalid.json").write_text("not json") + with self.answers(["4", "1", "6"]) as input_call, \ + mock.patch.object(wizard, "ensure_dependencies"), \ + mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run: + self.assertEqual(wizard.main([]), 0) + self.assertEqual(input_call.call_count, 3) + self.assertTrue(run.await_args.args[0].recover) + self.assertEqual(run.await_args.args[0].target, KEY) + self.assertIn(KEY[:12].upper(), self.output.getvalue()) + self.assertNotIn(KEY.upper(), self.output.getvalue()) + self.assertNotIn("finished", self.output.getvalue()) + + def test_release_browser_can_load_older_pages_and_marks_previews(self): + preview_asset, stable_asset = object(), object() + preview = ReleaseChoice("v1.18.0.3-alpha1", preview_asset, True, "2026-10-26") + stable = ReleaseChoice("v1.18.0.2", stable_asset, False, "2026-10-25") + catalog = SimpleNamespace(has_more_releases=True, release_choices=mock.Mock( + side_effect=[[preview], [stable, preview]])) + with self.answers(["2", "1"]): + chosen = asyncio.run(wizard.choose_release(catalog, destination())) + self.assertIs(chosen, stable_asset) + self.assertEqual([call.kwargs["max_pages"] for call in catalog.release_choices.call_args_list], [3, 6]) + self.assertTrue(all(call.kwargs["include_prerelease"] for call in catalog.release_choices.call_args_list)) + self.assertIn("v1.18.0.3-alpha1 (prerelease)", self.output.getvalue()) + + def test_firmware_back_does_not_download_or_open_a_file(self): + with self.answers(["5"]), mock.patch.object(wizard, "small_host", return_value=False), \ + mock.patch.object(wizard, "download_asset", side_effect=AssertionError("download")), \ + mock.patch.object(wizard, "file_prompt", side_effect=AssertionError("file")), \ + self.assertRaises(wizard.MenuCancelled): + asyncio.run(wizard.choose_firmware(wizard.arguments([]), destination(), object(), None, {})) + + def test_remembered_file_has_an_explicit_back_choice(self): + previous = self.cache / "previous.bin" + previous.write_bytes(image().image) + with self.answers(["3"]) as input_call, self.assertRaises(wizard.MenuCancelled): + wizard.file_prompt("Firmware file", str(previous)) + self.assertEqual(input_call.call_count, 1) + self.assertIn("Choose another file", self.output.getvalue()) + self.assertIn("Back", self.output.getvalue()) + + def test_missing_remembered_file_can_cancel_instead_of_retrying_stale_default(self): + missing = self.cache / "removed.bin" + with self.answers([""]) as input_call, self.assertRaises(wizard.MenuCancelled): + wizard.file_prompt("Firmware file", str(missing)) + self.assertEqual(input_call.call_count, 1) + self.assertNotIn("Use " + str(missing), self.output.getvalue()) + + def test_previous_boards_local_file_does_not_change_new_board_default(self): + previous = self.cache / "other-board.bin" + previous.write_bytes(b"other board") + published = self.cache / "matching.bin" + asset = SimpleNamespace(release_tag="v1.18.0.2", name="matching-board.zip") + remembered = {"target": "22" * 32, "firmware_source": "local", "firmware_file": str(previous)} + with self.answers([""]), mock.patch.object(wizard, "small_host", return_value=True), \ + mock.patch.object(wizard, "download_asset", return_value=published) as download, \ + mock.patch.object(wizard, "file_prompt", side_effect=AssertionError("other board's file")): + selected, source = asyncio.run(wizard.choose_firmware( + wizard.arguments([]), destination(), object(), asset, remembered)) + self.assertEqual((selected, source), (published, "latest")) + download.assert_called_once_with(asset, self.cache / "downloads") + + def test_local_firmware_selection_does_not_query_github(self): + local = self.cache / "local.bin" + local.write_bytes(image().image) + catalog = SimpleNamespace(choose_asset=mock.Mock(side_effect=AssertionError("GitHub query")), + release_choices=mock.Mock(side_effect=AssertionError("GitHub query"))) + with self.answers(["3", str(local)]), mock.patch.object(wizard, "small_host", return_value=True), \ + mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock, + side_effect=AssertionError("GitHub query")), \ + mock.patch.object(wizard, "download_asset", side_effect=AssertionError("GitHub download")): + selected, source = asyncio.run(wizard.choose_firmware( + wizard.arguments([]), destination(), catalog, None, {})) + self.assertEqual((selected, source), (local, "local")) + catalog.choose_asset.assert_not_called() + catalog.release_choices.assert_not_called() + + def test_release_back_returns_to_firmware_menu_then_accepts_local_file(self): + local = self.cache / "local.bin" + local.write_bytes(image().image) + preview = ReleaseChoice("v1.18.0.3-alpha1", object(), True, "2026-10-26") + catalog = SimpleNamespace(has_more_releases=False, + release_choices=mock.Mock(return_value=[preview])) + with self.answers(["2", "2", "3", str(local)]), \ + mock.patch.object(wizard, "small_host", return_value=True), \ + mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock, + side_effect=AssertionError("latest release query")), \ + mock.patch.object(wizard, "download_asset", side_effect=AssertionError("preview download")): + selected, source = asyncio.run(wizard.choose_firmware( + wizard.arguments([]), destination(), catalog, None, {})) + self.assertEqual((selected, source), (local, "local")) + catalog.release_choices.assert_called_once_with( + target_id=destination().target_id, include_prerelease=True, max_pages=3) + self.assertIn("v1.18.0.3-alpha1 (prerelease)", self.output.getvalue()) + + def test_signed_preparation_installs_verifier_then_retries_the_same_file(self): + options = wizard.arguments([]) + chosen = self.cache / "signed-full.mota" + new_image, package = image(), signed_package() + chosen.write_bytes(package.blob) + catalog = object() + with mock.patch.object(wizard, "choose_request", return_value=destination()), \ + mock.patch.object(wizard, "GitHubCatalog", return_value=catalog), \ + mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock, return_value=None), \ + mock.patch.object(wizard, "choose_firmware", new_callable=mock.AsyncMock, + return_value=(chosen, "local")) as choose, \ + mock.patch.object(wizard, "prepare_selected_firmware", new_callable=mock.AsyncMock, + side_effect=[ImportError("signature verifier missing"), + (new_image, package)]) as prepare, \ + mock.patch.object(wizard, "ensure_signature_support") as verifier, \ + mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \ + mock.patch.object(wizard, "choose_port", side_effect=AssertionError("USB access")), \ + mock.patch.object(wizard, "ask", side_effect=lambda prompt, default=None: default): + self.assertEqual(asyncio.run(wizard.prepare_interactively(options)), 0) + choose.assert_awaited_once() + verifier.assert_called_once_with() + self.assertEqual(prepare.await_args_list, [mock.call(options, destination(), chosen, catalog)] * 2) + outputs = list((self.cache / "bundles").glob("*.ota.zip")) + self.assertEqual(len(outputs), 1) + _, prepared = bundles.load_bundle(outputs[0], destination()) + self.assertEqual(prepared.blob, package.blob) + self.assertEqual(prepared.signer, package.signer) + + def test_failed_signature_setup_does_not_reselect_or_write_a_bundle(self): + options = wizard.arguments([]) + chosen = self.cache / "signed-full.mota" + with mock.patch.object(wizard, "choose_request", return_value=destination()), \ + mock.patch.object(wizard, "GitHubCatalog", return_value=object()), \ + mock.patch.object(wizard, "latest_asset", new_callable=mock.AsyncMock, return_value=None), \ + mock.patch.object(wizard, "choose_firmware", new_callable=mock.AsyncMock, + return_value=(chosen, "local")) as choose, \ + mock.patch.object(wizard, "prepare_selected_firmware", new_callable=mock.AsyncMock, + side_effect=ImportError("signature verifier missing")) as prepare, \ + mock.patch.object(wizard, "ensure_signature_support", + side_effect=ota.OtaError("Verifier download failed")) as verifier, \ + mock.patch.object(wizard, "ensure_dependencies", side_effect=AssertionError("USB setup")), \ + mock.patch.object(wizard, "save_bundle", side_effect=AssertionError("partial bundle")), \ + self.assertRaisesRegex(ota.OtaError, "Verifier download failed"): + asyncio.run(wizard.prepare_interactively(options)) + choose.assert_awaited_once() + prepare.assert_awaited_once() + verifier.assert_called_once_with() + self.assertFalse((self.cache / "bundles").exists()) + + def test_local_base_fallback_rejects_wrong_image_then_accepts_exact_running_image(self): + wrong, correct = self.cache / "wrong.bin", self.cache / "running.bin" + wrong.write_bytes(image().image) + expected = image(0x01120001, body=b"original application" * 16) + correct.write_bytes(expected.image) + with self.answers(["2", "2"]), \ + mock.patch.object(wizard, "file_prompt", side_effect=[wrong, correct]), \ + mock.patch.object(wizard, "GitHubCatalog", side_effect=AssertionError("network")): + base = asyncio.run(wizard.choose_missing_base(wizard.arguments([]), destination(), object())) + self.assertIsNotNone(base) + self.assertEqual(ota.parse_endf(base.read_bytes()).body_hash, destination().base_hash) + self.assertIn("not the exact running firmware", self.output.getvalue()) + + def test_older_github_base_search_checks_past_four_same_version_candidates(self): + candidates, paths = [object() for _ in range(5)], [] + for index in range(4): + path = self.cache / f"other-build-{index}.bin" + path.write_bytes(image(0x01120001, body=bytes([index])*64).image) + paths.append(path) + exact = self.cache / "running.bin" + exact.write_bytes(image(0x01120001, body=b"original application"*16).image) + paths.append(exact) + catalog = SimpleNamespace(has_more_releases=True, release_choices=mock.Mock(return_value=[]), + base_candidates=mock.Mock(side_effect=lambda **kwargs: candidates[:kwargs["limit"]])) + with self.answers(["1"]), mock.patch.object(packages, "download_asset", + side_effect=lambda asset, cache: paths[candidates.index(asset)]) as download: + base = asyncio.run(wizard.choose_missing_base(wizard.arguments([]), destination(), catalog)) + catalog.release_choices.assert_called_once_with( + target_id=destination().target_id, include_prerelease=True, max_pages=6) + catalog.base_candidates.assert_called_once_with( + target_id=destination().target_id, fw_version=destination().version, limit=16) + self.assertEqual(download.call_count, 5) + self.assertEqual(ota.parse_endf(base.read_bytes()).body_hash, destination().base_hash) + + def test_existing_automation_flags_do_not_enter_main_menu(self): + with mock.patch.object(wizard, "menu_main", side_effect=AssertionError("interactive menu")), \ + mock.patch.object(wizard, "ensure_dependencies") as setup, \ + mock.patch.object(wizard, "run", new_callable=mock.AsyncMock, return_value=0) as run: + self.assertEqual(wizard.main(["--check", "--port", "COM7"]), 0) + setup.assert_called_once() + self.assertTrue(run.await_args.args[0].check) + self.assertEqual(run.await_args.args[0].port, "COM7") + + +if __name__ == "__main__": + unittest.main()