diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index 5f8d0c54..1e25b749 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -139,8 +139,38 @@ command plus the `--motatool` argument to use afterward. The motatool repair builds the [mikecarper/motatool](https://github.com/mikecarper/motatool) fork at pinned revision `8c38369e7d35ad50cf74261869676d52dd24adf7`, with Cargo's `--locked` dependencies, in a separate per-user cache directory. It needs -Rust/Cargo and a native linker already installed; otherwise it points to +matching stable Cargo and rustc **1.86.0 or newer** and a native linker already +installed; otherwise it points to [Rust installation](https://rustup.rs/) and stops without installing anything. +The pinned dependency set needs more than the older `rust-version` in motatool's +own manifest: `zeroize 1.9.0` needs edition 2024, and ICU/idna dependencies need +Rust 1.86. Cargo 1.75 is a version problem, not a root-permission problem; no +nightly toolchain is necessary. + +Before asking to build or downloading crates, the runner checks both executable +versions. It keeps a compatible default pair; otherwise it looks for versioned +executables on PATH (such as Ubuntu's `cargo-1.91` and `rustc-1.91`) and already +installed stable rustup toolchains. It displays the selected paths and pins both +for the build process, so a newer Cargo cannot silently invoke an old rustc. +Rustup proxies are resolved to their actual toolchain binaries before changing +to the build directory. No default toolchain, symlink, system PATH, or package +installation is changed, and probes disable rustup's automatic toolchain install. +Tool version probes have ten-second timeouts. A working cached motatool does not +need Cargo or Rust checks at all. + +To choose a pair explicitly, append: + +```text +--cargo cargo-1.91 --rustc rustc-1.91 +``` + +These options also accept executable paths, not shell aliases or a command with +embedded arguments. Explicit selections and `RUSTC`/`CARGO_BUILD_RUSTC` compiler +overrides are honored; if incompatible, they fail instead of silently selecting +another compiler. `--rustc` takes precedence over those environment variables. +The manual build command includes the chosen compiler as well as Cargo. Native +linker, SDK, and system-library availability are still checked by the real build; +a passed version check is not a guarantee that every host prerequisite is present. Source/dependency downloads, build time and disk use are disclosed before confirmation. Cargo progress is displayed, with a one-hour build timeout. The radio admin-password environment variable is not passed to the build. @@ -186,6 +216,17 @@ redirect the run. Duplicate names, ambiguous prefixes, or the same radio under different participant aliases stop with an actionable error. Missing contacts must first be imported or discovered; a key alone does not create a contact. +For a separate source with a managed USB/TCP console, automatic source selection +reads `get public.key` from the physical source and matches that full key in the +controller's contacts. It does not require the source's current name to match an +old saved advert, and duplicate/emoji names cannot redirect this lookup. An +explicit `--source-contact` must identify that same physical source. If its key +really is missing, import the source contact or advertise/discover it on the +normal channel first; the normal/TempRadio on-air ACK checks still require the +contact. The script does not silently create contacts or skip those checks. +When the controller and source are the same TCP Full Companion, use the verified +`--source-shares-controller` topology; it does not need a contact for itself. + ## Destination requirements | Destination | Package installed | One-time prerequisite | Raw ZIP handling | diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 7b4f3196..c280c95f 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -147,6 +147,10 @@ MIN_MESHCLI_VERSION = (1, 6, 0) MOTATOOL_REPAIR_REPOSITORY = "https://github.com/mikecarper/motatool.git" MOTATOOL_REPAIR_REVISION = "8c38369e7d35ad50cf74261869676d52dd24adf7" MOTATOOL_REPAIR_TIMEOUT_SECONDS = 60 * 60 +# The pinned lockfile, not motatool's older package-level rust-version: zeroize +# needs edition 2024 (1.85), and idna_adapter/ICU 2.2 require Rust 1.86. +MOTATOOL_MIN_RUST_VERSION = (1, 86, 0) +RUST_TOOL_PROBE_TIMEOUT_SECONDS = 10 CRYPTOGRAPHY_REPAIR_REQUIREMENT = "cryptography==50.0.1" CRYPTOGRAPHY_REPAIR_TIMEOUT_SECONDS = 10 * 60 # v1.17.1.5 is the first release-version contract in which every packet, @@ -2558,11 +2562,34 @@ def bind_contact_selectors(controller: Controller, args: argparse.Namespace) -> relays = [(resolve(name), password) for name, password in args.relay_values] source_selector = args.source_contact_value if ( - not source_selector and not args.source_shares_controller + not args.source_shares_controller and has_managed_source_cli(args) ): - source_selector = read_source_name_bounded(args) - source = resolve(source_selector) if source_selector else None + # USB/TCP identifies the physical source. Its current display name may + # differ from the controller's saved advert (renames, truncation, emoji). + # Never bind a same-named but different radio, even with --source-contact. + source_key = read_source_public_key_bounded(args).lower() + if source_selector: + source = resolve(source_selector) + if source != source_key: + raise OtaError( + f"--source-contact identifies {source}, but the connected OTA source " + f"reports {source_key}; select the contact for the physical source" + ) + else: + if source_key not in contacts: + raise OtaError( + f"the connected OTA source public key {source_key} is not in the " + "controller's contacts; import its contact or advertise/discover it " + "on the normal channel first. Renaming a different contact or passing " + "--source-contact cannot replace this identity check. If source and " + "controller are the same TCP Full Companion, use the verified " + "--source-shares-controller topology instead" + ) + source = source_key + print(f"[contact] OTA source -> {contacts[source].get('adv_name', '?')} [{source}]") + else: + source = resolve(source_selector) if source_selector else None keys = [target, *(key for key, _password in relays)] if source: keys.append(source) @@ -6938,6 +6965,12 @@ def build_parser() -> argparse.ArgumentParser: help=argparse.SUPPRESS, ) parser.add_argument("--meshcli", default="meshcli") + parser.add_argument( + "--cargo", help="Cargo executable for motatool repair (default: find a compatible installed toolchain)", + ) + parser.add_argument( + "--rustc", help="Rust compiler for motatool repair (default: pair with selected Cargo)", + ) parser.add_argument( "--motatool", default="motatool", help=( @@ -7480,6 +7513,195 @@ def offer_cryptography_repair(failure: BootloaderCryptoError) -> None: print("[repair] Ed25519 self-test passed; retrying full package/signature verification.") +@dataclass(frozen=True) +class RustBuildTools: + cargo: str + rustc: str + cargo_version: tuple[int, int, int] + rustc_version: tuple[int, int, int] + + def environment(self) -> dict[str, str]: + env = rust_probe_environment() + # Pin both executables for this child process. In particular, a versioned + # distro Cargo must not fall back to the old unversioned rustc on PATH. + env["CARGO"] = self.cargo + env["RUSTC"] = self.rustc + return env + + +def rust_probe_environment() -> dict[str, str]: + env = os.environ.copy() + env.pop("MESHCORE_ADMIN_PASSWORD", None) + env["RUSTUP_AUTO_INSTALL"] = "0" + return env + + +def rust_host_output(command: list[str]) -> str: + """Bounded, read-only probes; never invoke rustup install/update/default.""" + try: + result = subprocess.run( + command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, check=False, + timeout=RUST_TOOL_PROBE_TIMEOUT_SECONDS, env=rust_probe_environment(), + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise OtaError(f"cannot probe {command[0]}: {exc}") from exc + if result.returncode != 0: + detail = (result.stderr or result.stdout).strip()[:1000] + raise OtaError(f"cannot probe {command[0]} (exit {result.returncode}): {detail}") + return result.stdout.strip() + + +def rust_tool_version(command: str, tool: str) -> tuple[int, int, int]: + output = rust_host_output([command, "--version"]) + match = re.match(rf"^{tool} (\d+)\.(\d+)\.(\d+)(?=\s|$)", output) + if match is None: + raise OtaError(f"cannot determine a stable {tool} version from {command}: {output[:200]}") + version = tuple(int(part) for part in match.groups()) + if version < MOTATOOL_MIN_RUST_VERSION: + needed = ".".join(map(str, MOTATOOL_MIN_RUST_VERSION)) + raise OtaError(f"{command} reports {tool} {'.'.join(map(str, version))}; need {needed} or newer") + return version + + +def rust_executable(command: str) -> str: + path = shutil.which(command) + if path is None: + raise OtaError(f"Rust tool executable not found: {command}") + # Keep argv[0] intact for symlink/hardlink rustup proxies. + return str(Path(path).absolute()) + + +def resolve_rustup_proxy(command: str, tool: str) -> str: + rustup = shutil.which("rustup") + if rustup is not None: + try: + is_proxy = os.path.samefile(command, rustup) + except OSError: + is_proxy = False + if is_proxy: + # Resolve in the caller's context, then retain the real binary path: + # changing to the build cache must not change the active toolchain. + return rust_executable(rust_host_output([rustup, "which", tool])) + return command + + +def versioned_cargo_candidates() -> list[str]: + found: dict[str, tuple[int, ...]] = {} + for directory in dict.fromkeys(os.get_exec_path()): + if not directory: + continue + try: + for entry in Path(directory).iterdir(): + match = re.fullmatch( + r"cargo-(\d+\.\d+(?:\.\d+)?)(?:\.exe)?", entry.name, + flags=re.IGNORECASE if os.name == "nt" else 0, + ) + if match and entry.is_file() and os.access(entry, os.X_OK): + found[str(entry.absolute())] = tuple(map(int, match[1].split("."))) + except OSError: + continue + return sorted(found, key=lambda path: found[path], reverse=True)[:16] + + +def installed_rustup_cargos() -> list[str]: + rustup = shutil.which("rustup") + if rustup is None: + return [] + result = [] + for line in rust_host_output([rustup, "toolchain", "list"]).splitlines(): + name = line.split()[0] if line.strip() else "" + if not re.fullmatch(r"(?:stable|\d+\.\d+\.\d+)(?:-[A-Za-z0-9_-]+)?", name): + continue # Do not select nightly/beta or arbitrary custom toolchains. + try: + result.append(rust_host_output([rustup, "which", "--toolchain", name, "cargo"])) + except OtaError: + continue # An incompletely installed toolchain is not usable. + if len(result) == 16: + break + return result + + +def select_motatool_build_tools(args: argparse.Namespace) -> RustBuildTools: + explicit_cargo = getattr(args, "cargo", None) + explicit_rustc = ( + getattr(args, "rustc", None) or os.environ.get("RUSTC") + or os.environ.get("CARGO_BUILD_RUSTC") + ) + checked: set[str] = set() + failures: list[str] = [] + + def check(candidate: str) -> RustBuildTools | None: + try: + cargo = resolve_rustup_proxy(rust_executable(candidate), "cargo") + if cargo in checked: + return None + checked.add(cargo) + cargo_version = rust_tool_version(cargo, "cargo") + compiler_candidates = [explicit_rustc] if explicit_rustc else [] + if not explicit_rustc: + name = re.sub(r"^cargo", "rustc", Path(cargo).name, count=1, flags=re.IGNORECASE) + compiler_candidates = [str(Path(cargo).with_name(name)), name, + f"rustc-{cargo_version[0]}.{cargo_version[1]}", "rustc"] + compiler_errors = [] + seen_compilers: set[str] = set() + for compiler in dict.fromkeys(compiler_candidates): + try: + rustc = resolve_rustup_proxy(rust_executable(compiler), "rustc") + if rustc in seen_compilers: + continue + seen_compilers.add(rustc) + rustc_version = rust_tool_version(rustc, "rustc") + if cargo_version[:2] != rustc_version[:2]: + raise OtaError( + f"toolchain mismatch: Cargo {'.'.join(map(str, cargo_version))} at {cargo}, " + f"rustc {'.'.join(map(str, rustc_version))} at {rustc}; select matching releases" + ) + break + except OtaError as exc: + compiler_errors.append(str(exc)) + else: + raise OtaError(" | ".join(compiler_errors)) + tools = RustBuildTools(cargo, rustc, cargo_version, rustc_version) + print(f"[host] build Cargo {'.'.join(map(str, cargo_version))}: {cargo}") + print(f"[host] build rustc {'.'.join(map(str, rustc_version))}: {rustc}") + return tools + except OtaError as exc: + failures.append(str(exc)) + print(f"[host] {exc}") + return None + + selected = check(explicit_cargo or "cargo") + if selected is not None: + return selected + if not explicit_cargo: + for candidate in versioned_cargo_candidates(): + selected = check(candidate) + if selected is not None: + return selected + try: + installed = installed_rustup_cargos() + except OtaError as exc: + failures.append(str(exc)) + installed = [] + for candidate in installed: + selected = check(candidate) + if selected is not None: + return selected + needed = ".".join(map(str, MOTATOOL_MIN_RUST_VERSION)) + raise OtaError( + f"motatool repair needs matching stable Cargo and rustc {needed} or newer " + "for its pinned dependencies, plus a native linker. " + "Select an installed pair with --cargo cargo-1.91 --rustc rustc-1.91, " + "or install a current stable user toolchain from https://rustup.rs " + "(with rustup already installed: rustup toolchain install stable --profile minimal). " + "Root access and nightly are not required for this version issue. " + "Shell aliases do not affect this script; explicit --cargo/--rustc and compiler " + "environment overrides are not silently replaced. No build or radio changes " + "were made. " + " | ".join(failures[-4:]) + ) + + def motatool_repair_command(root: Path, cargo: str = "cargo") -> list[str]: return [ cargo, "install", "--git", MOTATOOL_REPAIR_REPOSITORY, @@ -7520,9 +7742,14 @@ def offer_motatool_repair( print(f"[host] using previously installed, rechecked motatool: {binary}") return - cargo = shutil.which("cargo") - command = motatool_repair_command(root, cargo or "cargo") + build_tools = None if cached else select_motatool_build_tools(args) + command = motatool_repair_command(root, build_tools.cargo if build_tools else "cargo") manual = display_host_command(command) + if build_tools is not None: + if os.name == "nt": + manual = "$env:RUSTC = '" + build_tools.rustc.replace("'", "''") + "'; " + manual + else: + manual = "RUSTC=" + shlex.quote(build_tools.rustc) + " " + manual selection = display_host_command(["--motatool", str(binary)]) # Display an argument, not a PowerShell call operator for an option. if os.name == "nt": @@ -7533,12 +7760,6 @@ def offer_motatool_repair( print("[repair] Existing motatool installations and PATH will not be changed.") print(f"[repair] Manual install command: {manual}") print(f"[repair] To select this build explicitly, append: {selection}") - if not cached and cargo is None: - raise OtaError( - "automatic motatool repair requires Rust/Cargo and a native linker; " - "install the toolchain from https://rustup.rs and rerun, or select a " - "bootloader-capable build with --motatool. No installation was attempted" - ) if not sys.stdin.isatty(): raise OtaError( "motatool repair needs separate interactive approval (--yes does not approve " @@ -7565,8 +7786,8 @@ def offer_motatool_repair( # No cache directories, downloads or builds before affirmative consent. root.mkdir(parents=True, exist_ok=True) print("[repair] Building motatool (Cargo output follows)...", flush=True) - build_env = os.environ.copy() - build_env.pop("MESHCORE_ADMIN_PASSWORD", None) + assert build_tools is not None + build_env = build_tools.environment() try: result = subprocess.run( command, cwd=str(root), stdin=subprocess.DEVNULL, diff --git a/tools/lora_ota/test_lora_ota_bootloader.py b/tools/lora_ota/test_lora_ota_bootloader.py index 4aed554a..da8100a3 100644 --- a/tools/lora_ota/test_lora_ota_bootloader.py +++ b/tools/lora_ota/test_lora_ota_bootloader.py @@ -304,6 +304,9 @@ class BootloaderStagingTests(unittest.TestCase): mock.patch.object(ota, "run_checked", side_effect=ota.OtaError("unsupported format_ver 3")), mock.patch.object(ota, "motatool_repair_root", return_value=Path(directory) / "cache"), mock.patch.object(ota.shutil, "which", return_value="cargo"), + mock.patch.object(ota, "select_motatool_build_tools", return_value=ota.RustBuildTools( + "cargo", "rustc", (1, 91, 0), (1, 91, 0), + )), mock.patch.object(ota.sys.stdin, "isatty", return_value=False), self.assertRaisesRegex(ota.OtaError, "separate interactive approval"), ): @@ -350,7 +353,7 @@ class BootloaderStagingTests(unittest.TestCase): "switch_controller_to_temp_radio", "source_cli_command", ): stack.enter_context(mock.patch.object(ota, name)) - stack.enter_context(mock.patch.object(ota, "read_source_name_bounded", return_value="source")) + stack.enter_context(mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key)) stack.enter_context(mock.patch.object(ota, "read_source_rxps", return_value=ota.RxpsSettings(False, 0, 0))) stack.enter_context(mock.patch.object(ota, "read_remote_rxps", return_value=ota.RxpsSettings(False, 0, 0))) stack.enter_context(mock.patch.object( @@ -428,6 +431,84 @@ class StationSelectionTests(unittest.TestCase): with contextlib.redirect_stdout(io.StringIO()), self.assertRaisesRegex(ota.OtaError, "different radios"): ota.bind_contact_selectors(self.controller(), args) + def test_source_binds_by_local_key_despite_stale_or_emoji_name(self): + source_key = "b2" * 32 + controller = self.controller() + controller._run.return_value[0][source_key] = { + "public_key": source_key, "adv_name": "Old source name \U0001f4e1", + } + args = self.args(self.name) + args.source_shares_controller = False + args.source_cli_tcp = "source:5002" + with ( + mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key.upper()) as read_key, + mock.patch.object(ota, "read_source_name_bounded") as read_name, + contextlib.redirect_stdout(io.StringIO()), + ): + ota.bind_contact_selectors(controller, args) + self.assertEqual(args.source_contact_value, source_key) + read_key.assert_called_once_with(args) + read_name.assert_not_called() + controller.remote_command.assert_not_called() + + def test_missing_source_key_does_not_bind_same_named_other_radio(self): + args = self.args(self.name) + args.source_shares_controller = False + args.source_serial = "source" + source_key = "b2" * 32 + controller = self.controller() + controller._run.return_value[0]["c3" * 32] = { + "public_key": "c3" * 32, "adv_name": "W4JEC MC OTA2", + } + with ( + mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key), + contextlib.redirect_stdout(io.StringIO()), + self.assertRaisesRegex(ota.OtaError, "connected OTA source public key.*not in") as error, + ): + ota.bind_contact_selectors(controller, args) + self.assertIn(source_key, str(error.exception)) + self.assertIn("normal channel", str(error.exception)) + controller.remote_command.assert_not_called() + + def test_explicit_source_contact_must_match_connected_source_key(self): + args = self.args(self.name) + args.source_shares_controller = False + args.source_serial = "source" + args.source_contact_value = "Saved source" + controller = self.controller() + controller._run.return_value[0]["c3" * 32] = { + "public_key": "c3" * 32, "adv_name": "Saved source", + } + with ( + mock.patch.object(ota, "read_source_public_key_bounded", return_value="b2" * 32), + contextlib.redirect_stdout(io.StringIO()), + self.assertRaisesRegex(ota.OtaError, "--source-contact identifies.*connected OTA source"), + ): + ota.bind_contact_selectors(controller, args) + controller.remote_command.assert_not_called() + + def test_auto_source_cannot_also_be_destination(self): + args = self.args(self.name) + args.source_shares_controller = False + args.source_serial = "source" + with ( + mock.patch.object(ota, "read_source_public_key_bounded", return_value=self.key), + contextlib.redirect_stdout(io.StringIO()), + self.assertRaisesRegex(ota.OtaError, "different radios"), + ): + ota.bind_contact_selectors(self.controller(), args) + + def test_shared_source_does_not_require_its_own_contact_or_repeater_cli(self): + args = self.args(self.name) + args.source_cli_tcp = "shared:5002" + with ( + mock.patch.object(ota, "read_source_public_key_bounded") as read_key, + contextlib.redirect_stdout(io.StringIO()), + ): + ota.bind_contact_selectors(self.controller(), args) + read_key.assert_not_called() + self.assertIsNone(args.source_contact_value) + def test_remote_reply_accepts_key_selector_with_emoji_name(self): for selector in (self.name, self.key, self.key[:12]): controller = object.__new__(ota.Controller) diff --git a/tools/lora_ota/test_lora_ota_rust_tools.py b/tools/lora_ota/test_lora_ota_rust_tools.py new file mode 100644 index 00000000..98e0ebf2 --- /dev/null +++ b/tools/lora_ota/test_lora_ota_rust_tools.py @@ -0,0 +1,257 @@ +"""Toolchain preflight regressions; fake executables are never actually run.""" + +import argparse +import contextlib +import io +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest import mock + +import lora_ota as ota + + +class RustToolSelectionTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.outputs = {} + self.args = argparse.Namespace(cargo=None, rustc=None) + stack = contextlib.ExitStack() + self.addCleanup(stack.close) + stack.enter_context(mock.patch.dict(os.environ, { + "PATH": str(self.root), "PATHEXT": ".com;.exe;.bat;.cmd", + }, clear=True)) + self.probe = stack.enter_context(mock.patch.object(ota.subprocess, "run", side_effect=self.run_probe)) + self.rustup = stack.enter_context(mock.patch.object(ota, "installed_rustup_cargos", return_value=[])) + stack.enter_context(mock.patch.object(ota, "resolve_rustup_proxy", side_effect=lambda path, _: path)) + self.output = stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + + def executable(self, name, version): + suffix = ".exe" if os.name == "nt" else "" + path = self.root / (name + suffix) + path.write_bytes(b"test placeholder: must never execute") + path.chmod(0o755) + tool = "cargo" if name.startswith("cargo") else "rustc" + self.outputs[str(path)] = f"{tool} {version} (test-only build)" + return str(path) + + def pair(self, version, suffix=""): + return self.executable("cargo" + suffix, version), self.executable("rustc" + suffix, version) + + def run_probe(self, command, **kwargs): + self.assertEqual(command[1:], ["--version"]) + self.assertIn(command[0], self.outputs) + return subprocess.CompletedProcess(command, 0, self.outputs[command[0]], "") + + def select(self): + return ota.select_motatool_build_tools(self.args) + + def test_compatible_default_pair_is_preserved(self): + cargo, rustc = self.pair("1.91.0") + self.pair("1.98.0", "-1.98") + selected = self.select() + self.assertEqual((selected.cargo, selected.rustc), (cargo, rustc)) + self.rustup.assert_not_called() + self.assertEqual(self.probe.call_count, 2) + + def test_reported_ubuntu_old_default_selects_versioned_pair(self): + old_cargo, _ = self.pair("1.75.0") + cargo, rustc = self.pair("1.91.0", "-1.91") + selected = self.select() + self.assertEqual((selected.cargo, selected.rustc), (cargo, rustc)) + self.assertEqual(self.probe.call_args_list[0].args[0][0], old_cargo) + self.assertIn("1.75.0", self.output.getvalue()) + self.assertEqual(selected.environment()["RUSTC"], rustc) + self.assertEqual(selected.environment()["CARGO"], cargo) + + def test_new_cargo_with_old_rustc_is_not_accepted(self): + self.executable("cargo", "1.91.0") + self.executable("rustc", "1.75.0") + with self.assertRaisesRegex(ota.OtaError, "rustc 1.75.0"): + self.select() + + def test_edition_2024_alone_is_not_enough_for_icu_dependencies(self): + self.pair("1.85.0") + with self.assertRaisesRegex(ota.OtaError, "1.86.0 or newer"): + self.select() + + def test_minimum_supported_pair_is_accepted(self): + self.pair("1.86.0") + self.assertEqual(self.select().rustc_version, (1, 86, 0)) + + def test_incompatible_major_minor_pair_is_rejected(self): + self.executable("cargo", "1.91.0") + self.executable("rustc", "1.90.0") + with self.assertRaisesRegex(ota.OtaError, "toolchain mismatch"): + self.select() + + def test_patch_release_difference_is_allowed(self): + self.executable("cargo", "1.91.1") + self.executable("rustc", "1.91.0") + self.assertEqual(self.select().cargo_version, (1, 91, 1)) + + def test_missing_unversioned_cargo_can_use_versioned_pair(self): + cargo, _ = self.pair("1.91.0", "-1.91") + self.assertEqual(self.select().cargo, cargo) + + def test_versioned_cargo_does_not_fall_back_to_old_unversioned_rustc(self): + self.executable("cargo-1.91", "1.91.0") + self.executable("rustc", "1.75.0") + with self.assertRaisesRegex(ota.OtaError, "rustc-1.91"): + self.select() + + def test_versioned_candidates_are_ranked_numerically(self): + self.pair("1.9.0", "-1.9") + self.pair("1.91.0", "-1.91") + cargo, _ = self.pair("1.100.0", "-1.100") + self.assertEqual(self.select().cargo, cargo) + + def test_versioned_cargo_can_use_compatible_unversioned_compiler(self): + self.executable("cargo-1.91", "1.91.0") + rustc = self.executable("rustc", "1.91.0") + self.assertEqual(self.select().rustc, rustc) + + def test_new_default_cargo_can_find_matching_versioned_compiler(self): + self.executable("cargo", "1.91.0") + self.executable("rustc", "1.75.0") + rustc = self.executable("rustc-1.91", "1.91.0") + self.assertEqual(self.select().rustc, rustc) + + def test_explicit_cargo_is_not_replaced(self): + self.args.cargo, _ = self.pair("1.75.0") + self.pair("1.91.0", "-1.91") + with self.assertRaisesRegex(ota.OtaError, "not silently replaced"): + self.select() + self.assertEqual(self.probe.call_count, 1) + self.rustup.assert_not_called() + + def test_explicit_versioned_pair_is_used(self): + cargo, rustc = self.pair("1.91.0", "-1.91") + self.args.cargo = "cargo-1.91" + self.args.rustc = "rustc-1.91" + self.assertEqual(self.select().rustc, rustc) + self.assertEqual(self.select().cargo, cargo) + + def test_explicit_missing_cargo_never_falls_back(self): + self.pair("1.91.0") + self.args.cargo = "missing-cargo" + with self.assertRaisesRegex(ota.OtaError, "missing-cargo"): + self.select() + self.probe.assert_not_called() + + def test_explicit_old_compiler_environment_is_not_silently_overridden(self): + self.pair("1.91.0") + compiler = self.executable("rustc-1.75", "1.75.0") + for variable in ("RUSTC", "CARGO_BUILD_RUSTC"): + with self.subTest(variable=variable), mock.patch.dict(os.environ, {variable: compiler}): + with self.assertRaisesRegex(ota.OtaError, "rustc 1.75.0"): + self.select() + + def test_cli_compiler_takes_precedence_over_environment(self): + self.pair("1.91.0") + old = self.executable("rustc-1.75", "1.75.0") + self.args.rustc = "rustc" + with mock.patch.dict(os.environ, {"RUSTC": old}): + self.assertEqual(self.select().rustc_version, (1, 91, 0)) + + def test_installed_rustup_toolchain_is_fallback(self): + self.pair("1.75.0") + tools = self.root / "installed-stable" / "bin" + tools.mkdir(parents=True) + cargo, rustc = self.pair("1.91.0", "-1.91") + actual_cargo = tools / Path(cargo).name.replace("-1.91", "") + actual_rustc = tools / Path(rustc).name.replace("-1.91", "") + Path(cargo).rename(actual_cargo) + Path(rustc).rename(actual_rustc) + self.outputs[str(actual_cargo)] = self.outputs[cargo] + self.outputs[str(actual_rustc)] = self.outputs[rustc] + self.rustup.return_value = [str(actual_cargo)] + self.assertEqual(self.select().rustc, str(actual_rustc)) + + def test_malformed_version_fails_closed(self): + cargo, _ = self.pair("unknown") + with self.assertRaisesRegex(ota.OtaError, "cannot determine a stable cargo version"): + self.select() + self.assertIn(cargo, self.output.getvalue()) + + def test_nightly_is_not_automatically_selected(self): + self.pair("1.91.0-nightly") + with self.assertRaisesRegex(ota.OtaError, "stable cargo version"): + self.select() + + def test_no_toolchain_explains_non_root_stable_fix(self): + with self.assertRaises(ota.OtaError) as error: + self.select() + message = str(error.exception) + self.assertIn("--cargo cargo-1.91 --rustc rustc-1.91", message) + self.assertIn("Root access and nightly are not required", message) + self.assertIn("No build or radio changes", message) + + +class RustProbeTests(unittest.TestCase): + def test_probe_is_bounded_noninteractive_and_strips_admin_secret(self): + result = subprocess.CompletedProcess([], 0, "cargo 1.91.0\n", "") + with ( + mock.patch.object(ota.subprocess, "run", return_value=result) as run, + mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "secret"}), + ): + self.assertEqual(ota.rust_host_output(["cargo", "--version"]), "cargo 1.91.0") + options = run.call_args.kwargs + self.assertEqual(options["timeout"], ota.RUST_TOOL_PROBE_TIMEOUT_SECONDS) + self.assertEqual(options["stdin"], subprocess.DEVNULL) + self.assertEqual(options["env"]["RUSTUP_AUTO_INSTALL"], "0") + self.assertNotIn("MESHCORE_ADMIN_PASSWORD", options["env"]) + + def test_probe_timeout_and_missing_executable_are_actionable(self): + for error in (FileNotFoundError("not installed"), subprocess.TimeoutExpired("cargo", 10)): + with mock.patch.object(ota.subprocess, "run", side_effect=error): + with self.assertRaisesRegex(ota.OtaError, "cannot probe cargo"): + ota.rust_host_output(["cargo", "--version"]) + + def test_failed_probe_preserves_diagnostic(self): + result = subprocess.CompletedProcess([], 1, "", "toolchain not installed") + with mock.patch.object(ota.subprocess, "run", return_value=result): + with self.assertRaisesRegex(ota.OtaError, "toolchain not installed"): + ota.rust_host_output(["cargo", "--version"]) + + def test_rustup_discovery_only_uses_installed_stable_toolchains(self): + def output(command): + if command[1:] == ["toolchain", "list"]: + return "stable-host (default)\n1.91.0-host\nnightly-host\nbeta-host\ncustom\n" + return "/installed/" + command[3] + "/cargo" + with ( + mock.patch.object(ota.shutil, "which", return_value="rustup"), + mock.patch.object(ota, "rust_host_output", side_effect=output) as probe, + ): + self.assertEqual(ota.installed_rustup_cargos(), [ + "/installed/stable-host/cargo", "/installed/1.91.0-host/cargo", + ]) + self.assertEqual(probe.call_count, 3) + self.assertTrue(all(call.args[0][1] in ("toolchain", "which") for call in probe.call_args_list)) + + def test_proxy_is_resolved_to_real_binary_before_build_cwd_changes(self): + with ( + mock.patch.object(ota.shutil, "which", return_value="rustup"), + mock.patch.object(ota.os.path, "samefile", return_value=True), + mock.patch.object(ota, "rust_host_output", return_value="/installed/stable/bin/cargo") as probe, + mock.patch.object(ota, "rust_executable", side_effect=lambda path: path), + ): + self.assertEqual(ota.resolve_rustup_proxy("proxy/cargo", "cargo"), "/installed/stable/bin/cargo") + probe.assert_called_once_with(["rustup", "which", "cargo"]) + + def test_build_environment_is_private_and_pins_both_binaries(self): + pair = ota.RustBuildTools("/new/cargo", "/new/rustc", (1, 91, 0), (1, 91, 0)) + with mock.patch.dict(os.environ, {"CARGO": "old-cargo", "RUSTC": "old-rustc"}): + env = pair.environment() + self.assertEqual(env["CARGO"], "/new/cargo") + self.assertEqual(env["RUSTC"], "/new/rustc") + self.assertEqual(os.environ["CARGO"], "old-cargo") + self.assertEqual(os.environ["RUSTC"], "old-rustc") + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/lora_ota/test_lora_ota_tool_repair.py b/tools/lora_ota/test_lora_ota_tool_repair.py index edc13b0d..e32d6a8c 100644 --- a/tools/lora_ota/test_lora_ota_tool_repair.py +++ b/tools/lora_ota/test_lora_ota_tool_repair.py @@ -29,6 +29,10 @@ class MotatoolRepairTests(unittest.TestCase): self.addCleanup(stack.close) stack.enter_context(mock.patch.object(ota, "motatool_repair_root", return_value=self.root)) stack.enter_context(mock.patch.object(ota.shutil, "which", return_value="cargo")) + self.build_tools = ota.RustBuildTools("cargo", "rustc", (1, 91, 0), (1, 91, 0)) + self.select_tools = stack.enter_context(mock.patch.object( + ota, "select_motatool_build_tools", return_value=self.build_tools, + )) stack.enter_context(mock.patch.object(ota.sys.stdin, "isatty", return_value=True)) self.prompt = stack.enter_context(mock.patch("builtins.input", return_value="yes")) self.check = stack.enter_context(mock.patch.object(ota, "check_bootloader_tool")) @@ -54,6 +58,9 @@ class MotatoolRepairTests(unittest.TestCase): self.assertIn(ota.MOTATOOL_REPAIR_REVISION, command) self.assertIn("--locked", command) self.assertNotIn("MESHCORE_ADMIN_PASSWORD", self.install.call_args.kwargs["env"]) + self.assertEqual(self.install.call_args.kwargs["env"]["RUSTC"], "rustc") + self.assertEqual(self.install.call_args.kwargs["env"]["CARGO"], "cargo") + self.assertEqual(self.install.call_args.kwargs["env"]["RUSTUP_AUTO_INSTALL"], "0") self.assertEqual(self.install.call_args.kwargs["stdin"], subprocess.DEVNULL) self.assertEqual(self.install.call_args.kwargs["timeout"], ota.MOTATOOL_REPAIR_TIMEOUT_SECONDS) self.check.assert_called_once_with(str(self.binary), self.probe) @@ -89,6 +96,7 @@ class MotatoolRepairTests(unittest.TestCase): self.assertFalse(self.root.exists()) def test_missing_cargo_gives_prerequisite_without_installing_anything(self): + self.select_tools.side_effect = ota.OtaError("install a matching toolchain from https://rustup.rs") with mock.patch.object(ota.shutil, "which", return_value=None): with self.assertRaisesRegex(ota.OtaError, "https://rustup.rs"): self.repair() @@ -96,6 +104,14 @@ class MotatoolRepairTests(unittest.TestCase): self.prompt.assert_not_called() self.assertFalse(self.root.exists()) + def test_old_compiler_stops_before_prompt_build_or_cache_creation(self): + self.select_tools.side_effect = ota.OtaError("rustc 1.75.0; need 1.86.0 or newer") + with self.assertRaisesRegex(ota.OtaError, "rustc 1.75.0"): + self.repair() + self.prompt.assert_not_called() + self.install.assert_not_called() + self.assertFalse(self.root.exists()) + def test_cargo_failure_stops_and_preserves_original_selection(self): self.install.side_effect = None self.install.return_value = subprocess.CompletedProcess([], 101) @@ -134,6 +150,7 @@ class MotatoolRepairTests(unittest.TestCase): self.repair() self.prompt.assert_not_called() self.install.assert_not_called() + self.select_tools.assert_not_called() self.check.assert_called_once_with(str(self.binary), self.probe) self.assertEqual(self.args.motatool, str(self.binary)) @@ -151,6 +168,7 @@ class MotatoolRepairTests(unittest.TestCase): with mock.patch.object(ota.shutil, "which", return_value=None): self.repair() self.install.assert_not_called() + self.select_tools.assert_not_called() self.assertEqual(self.check.call_count, 2) self.assertEqual(self.args.motatool, str(self.binary))