From 38f5d8a61adb624ee7a57356ca44ad972551cef6 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 20 Oct 2026 21:55:37 -0700 Subject: [PATCH] Fix adaptive OTA capability reporting and application preflight --- src/helpers/ota/OtaCli.cpp | 10 +- test/test_ota_self_serve_policy.py | 127 ++++++++++++ tools/lora_ota/README.md | 18 ++ tools/lora_ota/lora_ota.py | 143 +++++++++++-- tools/lora_ota/test_lora_ota_adaptive_caps.py | 193 ++++++++++++++++++ 5 files changed, 474 insertions(+), 17 deletions(-) create mode 100644 tools/lora_ota/test_lora_ota_adaptive_caps.py diff --git a/src/helpers/ota/OtaCli.cpp b/src/helpers/ota/OtaCli.cpp index c254bf366..81a2292cb 100644 --- a/src/helpers/ota/OtaCli.cpp +++ b/src/helpers/ota/OtaCli.cpp @@ -938,11 +938,16 @@ static __attribute__((noinline)) bool handle_control_command( #if defined(OTA_SEEDER_ONLY) (void)bl; snprintf(reply + n, 160 - n, " | source-only; local install disabled"); -#elif defined(OTA_TOWER_AUTO_STORE) +#elif defined(OTA_TOWER_AUTO_STORE) || defined(OTA_RAK_AUTO_STORE) + // These adaptive stores choose their backend at runtime. Keep the actual + // application capability fields ahead of storage text so remote preflight + // can verify them even when a diagnostic tail reaches the reply limit. + n += snprintf(reply + n, 160 - n, " abi=%u codecs=0x%x", bl.apply_abi, bl.codec_mask); +#if defined(OTA_TOWER_AUTO_STORE) snprintf(reply + n, 160 - n, " | store:%s | bootloader:%s", c.fetch_store.usesExternal() ? "SD" : c.fetch_store.usesInternal() ? "internal" : "unsafe", c.fetch_store.compatible() ? "apply OK" : "unsupported"); -#elif defined(OTA_RAK_AUTO_STORE) +#else #if defined(OTA_RAK15002_SD_STORE) if (c.fetch_store.usesSd()) { snprintf(reply + n, 160 - n, " | SD store | bootloader: %s", @@ -963,6 +968,7 @@ static __attribute__((noinline)) bool handle_control_command( snprintf(reply + n, 160 - n, " | storage unsafe: %s", c.fetch_store.selectionReason()); } +#endif #elif defined(OTA_QSPI_STORE) uint32_t qspi_capacity = c.fetch_store.capacity(); n += snprintf(reply + n, 160 - n, " | QSPI store:%s%uK", diff --git a/test/test_ota_self_serve_policy.py b/test/test_ota_self_serve_policy.py index e4c900978..aacac225f 100644 --- a/test/test_ota_self_serve_policy.py +++ b/test/test_ota_self_serve_policy.py @@ -13,6 +13,133 @@ ROOT = Path(__file__).resolve().parents[1] class OtaSelfServePolicyTest(unittest.TestCase): + def test_adaptive_self_reports_actual_capabilities_inside_remote_reply(self): + cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text(encoding="ascii") + branch = method(cli, 'else if (is_cmd(a, "self|id", &rest))').removeprefix("else ") + harness = r''' +#include +#include +#include +#include +#include +#include +struct SelfFwInfo { + bool valid = true; + uint32_t body_len = UINT32_MAX, image_len = UINT32_MAX; + uint8_t body_hash[8] = {}; +}; +static SelfFwInfo running; +static bool ota_self_firmware(SelfFwInfo& info) { info = running; return true; } +namespace mesh { struct Utils { + static void toHex(char* out, const uint8_t*, unsigned count) { + assert(count == 8); memcpy(out, "0011223344556677", 17); + } +}; } +static unsigned ota_max_block_capability() { return 2048; } +struct OtaBlCaps { + bool present = true; + uint16_t apply_abi = 3, codec_mask = 5; + uint8_t storage_flags = 4, optional_sd_app_storage = 1; +}; +enum { OTA_BL_STORAGE_SD = 1, OTA_BL_STORAGE_QSPI = 4 }; +struct Store { + enum Kind { Internal, Qspi, Sd, Unsafe } kind = Internal; + uint32_t bytes = 252 * 1024; + bool supported = true; + bool usesInternal() const { return kind == Internal; } + bool usesQspi() const { return kind == Qspi; } + bool usesSd() const { return kind == Sd; } + bool usesExternal() const { return usesSd() || usesQspi(); } + bool compatible() const { return supported; } + uint32_t capacity() const { return bytes; } + uint32_t qspiCapacity() const { return bytes; } + const char* selectionReason() const { + return "unsafe storage diagnostic intentionally longer than the remote reply " + "budget: capabilities must remain complete before this explanatory tail"; + } +}; +struct Context { + OtaBlCaps caps; + Store fetch_store; + unsigned caps_reads = 0; + const OtaBlCaps& bootloaderAppCaps() { ++caps_reads; return caps; } +}; +static bool is_cmd(const char*, const char*, const char**) { return true; } +static bool command(char* reply, Context& c) { + const char* a = "self", *rest = nullptr; +@BRANCH@ + return true; +} +int main() { + for (auto kind : {Store::Internal, Store::Qspi, Store::Sd, Store::Unsafe}) + for (bool supported : {false, true}) + for (uint16_t abi : {uint16_t(3), uint16_t(65535)}) { + Context c; + c.fetch_store.kind = kind; + c.fetch_store.bytes = supported ? UINT32_MAX : 0; + c.fetch_store.supported = supported; + c.caps.apply_abi = abi; + c.caps.codec_mask = abi == 3 ? 5 : 65535; + c.caps.present = supported; + c.caps.storage_flags = supported ? OTA_BL_STORAGE_QSPI : 0; + c.caps.optional_sd_app_storage = supported ? OTA_BL_STORAGE_SD : 0; + struct { char before[8], reply[160], after[8]; } out; + memset(&out, 0x5A, sizeof(out)); + assert(command(out.reply, c)); + for (char byte : out.before) assert(byte == 0x5A); + for (char byte : out.after) assert(byte == 0x5A); + assert(memchr(out.reply, 0, sizeof(out.reply))); + assert(c.caps_reads == 1); + assert(strstr(out.reply, "base_hash=0011223344556677")); + assert(strstr(out.reply, "maxblk:2048")); +#if defined(OTA_SEEDER_ONLY) + assert(strstr(out.reply, "source-only; local install disabled")); + assert(!strstr(out.reply, "abi=")); +#else + char expected[40]; + snprintf(expected, sizeof(expected), "abi=%u codecs=0x%x", abi, c.caps.codec_mask); + const char* caps = strstr(out.reply, expected); + assert(caps && caps < out.reply + 110); +#if defined(OTA_TOWER_AUTO_STORE) + assert(strstr(out.reply, supported ? "bootloader:apply OK" : "bootloader:unsupported")); +#else + if (kind == Store::Qspi) { + assert(strstr(out.reply, supported ? "QSPI apply OK" : "NO QSPI apply")); + assert(strstr(out.reply, supported ? "QSPI store:4194303K" : "QSPI store:ERR 0K")); + } else if (kind == Store::Internal) { + assert(strstr(out.reply, supported ? "delta apply OK" : "NO delta apply")); + assert(strstr(out.reply, "internal store:")); + } +#if defined(OTA_RAK15002_SD_STORE) + else if (kind == Store::Sd) { + assert(strstr(out.reply, supported ? "authenticated SD apply OK" : "NO SD apply")); + assert(strstr(out.reply, "SD store")); + } +#endif +#endif +#endif + } + running.valid = false; + Context c; + char reply[160]; + assert(command(reply, c)); + assert(!strncmp(reply, "ERR no EndF", 11) && c.caps_reads == 0); +} +'''.replace("@BRANCH@", branch) + with tempfile.TemporaryDirectory(prefix="ota-self-caps-") as directory: + path = Path(directory) + source = path / "test.cpp" + source.write_text(harness, encoding="ascii") + variants = ( + ("NRF52_PLATFORM=1", "OTA_RAK_AUTO_STORE=1"), + ("NRF52_PLATFORM=1", "OTA_RAK_AUTO_STORE=1", "OTA_RAK15002_SD_STORE=1"), + ("NRF52_PLATFORM=1", "OTA_TOWER_AUTO_STORE=1"), + ("NRF52_PLATFORM=1", "OTA_RAK_AUTO_STORE=1", "OTA_SEEDER_ONLY=1"), + ) + for definitions in variants: + with self.subTest(defines=definitions): + self.compile_and_run(path, source, definitions) + def compile_and_run(self, path, source, defines=(), scenarios=((),)): binary = path / "self-serve-policy.exe" sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined"] diff --git a/tools/lora_ota/README.md b/tools/lora_ota/README.md index 10993fdec..653f3daf3 100644 --- a/tools/lora_ota/README.md +++ b/tools/lora_ota/README.md @@ -7,6 +7,24 @@ command or the Ethernet-only KISS endpoint. See the [mOTA tooling](../mota/READM for container creation and the [OTA protocol](../../docs/ota_protocol.md) for the wire format. +## Companion runner capability checks + +Some deployed adaptive RAK3401, RAK4631 and MeshTower receiver images omit the +bootloader ABI and codec mask from `ota self`. For application updates, +`lora_ota.py` can verify these known diagnostic formats with an additional +read-only `ota bootloader status` query. Hardware, selected storage and the +installed bootloader's board/name/target identity, CRC status and storage +profile must agree. That status contract guarantees codec subset `0x5` (full +image and in-place delta); storage `caps` flags are not interpreted as codecs. +Unsupported, incomplete or conflicting replies still stop preflight. + +This compatibility path applies only to application updates. Bootloader +packages require directly reported ABI and codec-mask fields. New adaptive +firmware reports those actual fields early in `ota self`, before storage +diagnostics, so they survive the remote reply limit and need no fallback. +If required fields cannot be verified, update the application to expose them +before retrying. + ## KISS direct source The default invocation is **read-only probing; no RF transmission**. Serving diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index eac57233c..18d2e70d4 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -289,6 +289,17 @@ class TargetInfo: return self.nrf_sd or self.nrf_qspi +@dataclass(frozen=True) +class BootloaderStatus: + board_id: int + target_id: int + device_name: str + hw_id: str + crc32: int + apply_abi: int + storage_flags: int + + @dataclass(frozen=True) class RadioSettings: frequency: float @@ -2805,14 +2816,13 @@ def bind_contact_selectors(controller: Controller, args: argparse.Namespace) -> args.source_contact_value = source -def query_bootloader_target( - controller: Controller, target: TargetInfo, -) -> TargetInfo: - reply = controller.remote_command(target.name, "ota bootloader status") +def parse_bootloader_status(reply: str) -> BootloaderStatus: + """Validate the installed bootloader's complete status and identity.""" match = re.fullmatch( r"BL board=([0-9A-Fa-f]{8}) target=([0-9A-Fa-f]{8}) name=(\S+) " r"crc=([0-9A-Fa-f]{8}) abi=(\d+) caps=([0-9A-Fa-f]{2}) " - r"\| staged:(?:ready|none) mid=(?:-|[0-9A-Fa-f]{8}) hash=(?:-|[0-9A-Fa-f]{16})", + r"\| staged:(?:none mid=- hash=-|" + r"ready mid=[0-9A-Fa-f]{8} hash=[0-9A-Fa-f]{16})", reply.strip(), ) if match is None: @@ -2820,7 +2830,7 @@ def query_bootloader_target( f"destination cannot stage a bootloader update: {reply}; " "requires exact-board ABI-3 self-update bootloader and capable application" ) - board, target_id, name, _crc, abi, caps = match.groups() + board, target_id, name, crc, abi, caps = match.groups() library = bootloader_library() try: hw_id = ( @@ -2832,14 +2842,100 @@ def query_bootloader_target( raise OtaError(f"invalid destination bootloader identity: {exc}") from exc if derived_target != int(target_id, 16): raise OtaError("destination bootloader board/name does not match its target ID") - if int(abi) < 3 or int(abi) != target.bootloader_abi: + return BootloaderStatus( + int(board, 16), derived_target, name, hw_id, int(crc, 16), + int(abi), int(caps, 16), + ) + + +def query_bootloader_target( + controller: Controller, target: TargetInfo, +) -> TargetInfo: + live = parse_bootloader_status( + controller.remote_command(target.name, "ota bootloader status") + ) + if live.apply_abi < 3 or live.apply_abi != target.bootloader_abi: raise OtaError("destination reports inconsistent or unsupported bootloader ABI") return replace( - target, boot_target_id=derived_target, boot_hw_id=hw_id, - boot_storage=int(caps, 16), + target, boot_target_id=live.target_id, boot_hw_id=live.hw_id, + boot_storage=live.storage_flags, ) +def query_adaptive_bootloader_caps( + controller: Controller, target_name: str, status: str, self_status: str, + hw_id: str | None, +) -> tuple[int, int] | None: + """Recover only capabilities guaranteed by known adaptive firmware. + + Adaptive RAK/Tower `ota self` formats were introduced in 72afac9a4 and + 30fe0d617. Both descend from 3d78d9c84 (required FULL|INPLACE mask 0x5) + and fd98bc90f (status refuses invalid self-update capabilities). Thus a + successful strict `ota bootloader status` attests that subset, independently + of its `caps` field, which describes storage and is NOT a codec mask. + Generic legacy diagnostics predate that contract and must still fail. + """ + prefix = ( + r"self body=[1-9][0-9]* image=[1-9][0-9]* " + r"base_hash=[0-9A-Fa-f]{16} \| maxblk:(?:1024|2048) \| " + ) + storage = None + if hw_id in ("RAK_3401", "RAK4631"): + if re.fullmatch( + prefix + r"internal store:[1-9][0-9]*K \| bootloader: delta apply OK", + self_status.strip(), + ): + storage = "internal" + elif re.fullmatch( + prefix + r"QSPI store:[1-9][0-9]*K \| bootloader: QSPI apply OK", + self_status.strip(), + ): + storage = "QSPI" + elif hw_id == "RAK4631" and re.fullmatch( + prefix + r"SD store \| bootloader: authenticated SD apply OK", + self_status.strip(), + ): + storage = "SD" + name = "3401" if hw_id == "RAK_3401" else "4631" + expected_board = 0x239A0029 + identities = {f"{name}_DFU": 0x0A, f"{name}_AUTO_DFU": 0x1E} + elif hw_id == "Heltec_tower_v2": + shape = re.fullmatch( + prefix + r"store:(SD|internal) \| bootloader:apply OK", + self_status.strip(), + ) + if shape: + storage = shape.group(1) + expected_board = 0x239A0071 + identities = {"TOWER_V2_OTA": 0x09} + else: + return None + if storage is None: + return None + if ( + not status.startswith("OTA | ") + or re.findall(r"\bhw=([^ |]+)", status) != [hw_id] + or re.findall(r"\bbl:([^ |]+)", status) != [storage] + ): + raise OtaError("adaptive destination reports conflicting hardware/storage diagnostics") + live = parse_bootloader_status( + controller.remote_command(target_name, "ota bootloader status", retry=False) + ) + if ( + live.board_id != expected_board + or identities.get(live.device_name) != live.storage_flags + or not 3 <= live.apply_abi < 0xFFFF + or live.crc32 in (0, 0xFFFFFFFF) + ): + raise OtaError("adaptive destination bootloader identity, ABI or storage profile is unverified") + print( + "[bootloader] adaptive diagnostics verified through installed " + f"bootloader status: ABI {live.apply_abi}, guaranteed codec subset 0x5 " + "(storage flags are independent)" + ) + return live.apply_abi, 0x5 + + def query_target( controller: Controller, args: argparse.Namespace, @@ -2929,12 +3025,29 @@ def query_target( bootloader_abi = int(caps_match.group(1)) bootloader_codecs = int(caps_match.group(2), 16) elif platform == "nrf52": - version = bootloader_version or "unknown version" - raise OtaError( - f"destination nRF52 bootloader {version} does not report a " - "compatible mOTA ABI and codec mask in `ota self`; install the " - "exact-board OTAFIX bootloader first" - ) + if getattr(args, "package_kind", "application") != "application": + raise OtaError( + "bootloader packages require a directly reported mOTA ABI and " + "codec mask; upgrade the application to expose complete " + "bootloader diagnostics before retrying" + ) + try: + adaptive_caps = query_adaptive_bootloader_caps( + controller, args.target, status, self_status, hw_id, + ) + except OtaError as exc: + raise OtaError( + "cannot verify the destination's mOTA ABI and codec mask from " + f"adaptive diagnostics: {exc}; upgrade the application to expose " + "complete bootloader diagnostics before retrying" + ) from exc + if adaptive_caps is None: + raise OtaError( + "destination nRF52 application does not report a compatible " + "mOTA ABI and codec mask in `ota self`; upgrade the application " + "to expose complete bootloader diagnostics before retrying" + ) + bootloader_abi, bootloader_codecs = adaptive_caps current_version = None current_version_source = None try: diff --git a/tools/lora_ota/test_lora_ota_adaptive_caps.py b/tools/lora_ota/test_lora_ota_adaptive_caps.py new file mode 100644 index 000000000..16426d32b --- /dev/null +++ b/tools/lora_ota/test_lora_ota_adaptive_caps.py @@ -0,0 +1,193 @@ +"""Read-only capability fallback for deployed adaptive storage diagnostics.""" + +import argparse +import contextlib +import io +import unittest +from unittest import mock + +import lora_ota as ota + + +class AdaptiveBootloaderCapsTests(unittest.TestCase): + prefix = "self body=100 image=156 base_hash=0011223344556677 | maxblk:2048 | " + + def fixture(self, hw="RAK_3401", storage="internal", auto=False, **overrides): + if hw == "Heltec_tower_v2": + board, name, caps = 0x239A0071, "TOWER_V2_OTA", 0x09 + tail = f"store:{storage} | bootloader:apply OK" + else: + family = "3401" if hw == "RAK_3401" else "4631" + board = 0x239A0029 + name = family + ("_AUTO_DFU" if auto else "_DFU") + caps = 0x1E if auto else 0x0A + tail = { + "internal": "internal store:252K | bootloader: delta apply OK", + "QSPI": "QSPI store:2048K | bootloader: QSPI apply OK", + "SD": "SD store | bootloader: authenticated SD apply OK", + }[storage] + replies = { + "ota status": ( + f"OTA | this fw 00112233 (100K) hw={hw} | target:1234ABCD | " + f"maxblk:2048 | bl:{storage} blrc:00 | no download" + ), + "get bootloader.ver": "> OTAFIX2.4.11-preview.126", + "ota self": self.prefix + tail, + "ota bootloader status": self.boot_reply(board, name, caps), + "ota stats": "OTA | fw v1.18.0.1", + } + replies.update(overrides) + controller = mock.Mock() + + def remote(_target, command, **kwargs): + value = replies[command] # Any mutating or unexpected command fails. + if command == "ota bootloader status": + self.assertIs(kwargs.get("retry"), False) + if isinstance(value, Exception): + raise value + return value + + controller.remote_command.side_effect = remote + return controller + + @staticmethod + def boot_reply(board=0x239A0029, name="3401_DFU", caps=0x0A, abi=3): + target = ota.bootloader_library().bootloader_target_id(board, name) + return ( + f"BL board={board:08X} target={target:08X} name={name} " + f"crc=12345678 abi={abi} caps={caps:02X} | staged:none mid=- hash=-" + ) + + def query(self, controller, package_kind="application"): + with contextlib.redirect_stdout(io.StringIO()) as output: + result = ota.query_target( + controller, argparse.Namespace(target="remote", package_kind=package_kind) + ) + return result, output.getvalue() + + def test_adaptive_rak_internal_qspi_and_sd(self): + for hw in ("RAK_3401", "RAK4631"): + for storage in ("internal", "QSPI", "SD"): + if hw == "RAK_3401" and storage == "SD": + continue + for auto in (False, True): + with self.subTest(hw=hw, storage=storage, auto=auto): + controller = self.fixture(hw, storage, auto) + result, log = self.query(controller) + self.assertEqual(result.bootloader_abi, 3) + self.assertEqual(result.bootloader_codecs, 0x5) + self.assertEqual(result.nrf_sd, storage == "SD") + self.assertEqual(result.nrf_qspi, storage == "QSPI") + self.assertIsNone(result.boot_target_id) + self.assertIn("guaranteed codec subset 0x5", log) + self.assertIn("storage flags are independent", log) + self.assertEqual( + sum(c.args[1] == "ota bootloader status" + for c in controller.remote_command.call_args_list), 1, + ) + + def test_adaptive_tower_sd_and_internal(self): + for storage in ("SD", "internal"): + with self.subTest(storage=storage): + result, _ = self.query(self.fixture("Heltec_tower_v2", storage)) + self.assertEqual(result.bootloader_abi, 3) + self.assertEqual(result.bootloader_codecs, 5) + self.assertEqual(result.nrf_sd, storage == "SD") + + def test_direct_fields_remain_authoritative_without_fallback(self): + controller = self.fixture(**{ + "ota self": self.prefix + "bootloader: apply OK (abi=2 codecs=0x4)", + "ota bootloader status": AssertionError("unexpected fallback"), + }) + result, log = self.query(controller) + self.assertEqual((result.bootloader_abi, result.bootloader_codecs), (2, 4)) + self.assertNotIn("guaranteed codec subset", log) + self.assertNotIn("ota bootloader status", [ + c.args[1] for c in controller.remote_command.call_args_list + ]) + + def test_guaranteed_subset_never_substitutes_for_bootloader_exact_mask(self): + controller = self.fixture() + with self.assertRaisesRegex(ota.OtaError, "bootloader packages require a directly reported"): + self.query(controller, package_kind="bootloader") + self.assertNotIn("ota bootloader status", [ + c.args[1] for c in controller.remote_command.call_args_list + ]) + + def test_generic_or_malformed_missing_fields_are_not_inferred(self): + tails = ( + "bootloader: apply OK", "bootloader: SD apply OK", + "internal store:252K | bootloader: delta apply OK (abi=bad codecs=0x5)", + "internal store:252K | bootloader: delta apply OK extra", + "internal store:0K | bootloader: delta apply OK", + "internal store:252K | bootloader: NO delta apply", + "storage unsafe: unavailable", + ) + for tail in tails: + with self.subTest(tail=tail): + controller = self.fixture(**{"ota self": self.prefix + tail}) + with self.assertRaisesRegex(ota.OtaError, "ABI and codec mask.*upgrade the application"): + self.query(controller) + self.assertNotIn("ota bootloader status", [ + c.args[1] for c in controller.remote_command.call_args_list + ]) + + def test_hardware_and_storage_views_must_agree(self): + for status in ( + "OTA | hw=RAK_3401 | target:1234ABCD | maxblk:2048 | bl:QSPI blrc:00", + "OTA | hw=RAK_3401 hw=RAK4631 | target:1234ABCD | maxblk:2048 | bl:internal", + "OTA | hw=RAK_3401 | target:1234ABCD | maxblk:2048 | bl:internal bl:QSPI", + ): + with self.subTest(status=status): + with self.assertRaisesRegex(ota.OtaError, "conflicting hardware/storage"): + self.query(self.fixture(**{"ota status": status})) + with self.assertRaisesRegex(ota.OtaError, "upgrade the application"): + self.query(self.fixture(**{ + "ota status": "OTA | hw=OtherBoard | target:1234ABCD | maxblk:2048 | bl:internal", + })) + + def test_unavailable_or_malformed_status_remains_failure(self): + valid = self.boot_reply() + for reply in ( + "Unknown command", + "Bootloader update unavailable: installed embedded manifest/CRC is invalid", + ota.TransmissionError("missing reply"), + valid[:-1], valid + " unexpected", valid + "\n" + valid, + valid.replace("staged:none mid=- hash=-", "staged:none mid=12345678 hash=-"), + valid.replace("staged:none mid=- hash=-", "staged:ready mid=- hash=-"), + valid.replace("crc=12345678", "crc=00000000"), + valid.replace("crc=12345678", "crc=FFFFFFFF"), + ): + with self.subTest(reply=reply): + with self.assertRaisesRegex(ota.OtaError, "upgrade the application"): + self.query(self.fixture(**{"ota bootloader status": reply})) + + def test_bootloader_identity_abi_and_primary_profile_are_checked(self): + valid = self.boot_reply() + for reply in ( + valid.replace("target=23818A80", "target=00000000"), + self.boot_reply(board=0x239A0071, name="TOWER_V2_OTA", caps=9), + self.boot_reply(name="4631_DFU"), + self.boot_reply(name="OTHER_DFU"), + self.boot_reply(abi=2), self.boot_reply(abi=65535), + self.boot_reply(caps=4), self.boot_reply(caps=9), + self.boot_reply(caps=0x1E), self.boot_reply(caps=0xFF), + ): + with self.subTest(reply=reply): + with self.assertRaisesRegex(ota.OtaError, "upgrade the application"): + self.query(self.fixture(**{"ota bootloader status": reply})) + + def test_missing_version_does_not_replace_live_status_validation(self): + for version in ("> unknown", "Unknown command", ota.OtaError("optional unavailable")): + with self.subTest(version=version): + result, _ = self.query(self.fixture(**{"get bootloader.ver": version})) + self.assertEqual(result.bootloader_codecs, 5) + with self.assertRaisesRegex(ota.OtaError, "upgrade the application"): + self.query(self.fixture(**{ + "get bootloader.ver": version, + "ota bootloader status": "Unknown command", + })) + + +if __name__ == "__main__": + unittest.main()