diff --git a/src/opendisplay/ota.py b/src/opendisplay/ota.py index 080b304..54f0dfd 100644 --- a/src/opendisplay/ota.py +++ b/src/opendisplay/ota.py @@ -3,23 +3,43 @@ from __future__ import annotations import asyncio -from collections.abc import Callable +from collections.abc import Callable, Sequence from typing import TYPE_CHECKING from .exceptions import OTAError +# Nordic Legacy DFU GATT service, advertised while in bootloader mode. Kept in +# sync with ``nrf_ota._const.LEGACY_DFU_SERVICE_UUID`` and duplicated here so +# ``find_nrf_dfu_device`` doesn't need the optional ``nrf-ota`` dependency. +LEGACY_DFU_SERVICE_UUID = "00001530-1212-efde-1523-785feabcd123" -def _increment_mac(address: str) -> str: - """Return the BLE MAC address incremented by 1, carrying across octets. - The nRF DFU bootloader advertises at ``original + 1``. Incrementing only the - last octet (``+1 & 0xFF``) fails to carry when it is 0xFF (e.g. ...:FF -> ...:00 - instead of rolling into the previous octet), causing a silent 30 s scan miss. +def _increment_mac(address: str) -> str: + """Return the BLE MAC with its last octet incremented by 1 (no carry). + + Nordic DFU bootloaders advertise at ``original + 1``, incrementing only the + last octet as a ``uint8``: it wraps 0xFF -> 0x00 without carrying into the + previous octet (e.g. ``AA:BB:CC:DD:EE:FF`` -> ``AA:BB:CC:DD:EE:00``). This + matches the standalone ``nrf-ota`` scanner's convention and the documented + Nordic bootloader behaviour. The exact wrap behaviour should be validated on + real hardware; the name/service-UUID fallback in ``find_nrf_dfu_device`` + keeps discovery robust even if this guess is off. """ parts = address.upper().split(":") - mac_int = int("".join(parts), 16) - mac_int = (mac_int + 1) & 0xFFFFFFFFFFFF - return ":".join(f"{(mac_int >> (8 * (5 - i))) & 0xFF:02X}" for i in range(6)) + parts[-1] = f"{(int(parts[-1], 16) + 1) & 0xFF:02X}" + return ":".join(parts) + + +def _advertises_dfu_service(service_uuids: Sequence[str]) -> bool: + """Return True if *service_uuids* contain the Legacy DFU service UUID. + + Mirrors the service-UUID half of ``nrf_ota.scan._is_dfu_advertisement``. + Selection deliberately does NOT match on device name: a name that merely + contains "DFU"/"DfuTarg" would pick the wrong tag when two devices are in + bootloader mode at once (finding MJ-23), so only the unique per-device MAC+1 + or the DFU service UUID are used to select the target. + """ + return any(LEGACY_DFU_SERVICE_UUID.lower() in s.lower() for s in service_uuids) if TYPE_CHECKING: @@ -57,7 +77,7 @@ async def perform_nrf_dfu( """ try: from bleak_retry_connector import BleakClientWithServiceCache, establish_connection - from nrf_ota._const import DEFAULT_PRN, LEGACY_DFU_SERVICE_UUID, TYPE_APPLICATION + from nrf_ota._const import DEFAULT_PRN, TYPE_APPLICATION from nrf_ota._zip import _parse_zip_bytes from nrf_ota.dfu import LegacyDFU except ImportError as exc: @@ -178,9 +198,13 @@ async def find_nrf_dfu_device(original_address: str) -> BLEDevice | None: """Poll the BLE scanner for an nRF DFU-mode device. Call this after ``OpenDisplayDevice.trigger_dfu_bootloader()`` disconnects. - Checks MAC+1 first (Nordic DFU bootloaders commonly increment the last - byte of the address). Falls back to the original address after 10 s in - case this particular bootloader keeps the same address. + Checks MAC+1 first (Nordic DFU bootloaders increment the last byte of the + address, wrapping without carry). As a fallback it also matches any device + advertising the Legacy DFU service UUID, so discovery still succeeds if the + MAC+1 guess is off. Selection never matches on device name — a DFU-ish name + would pick the wrong tag when two are in bootloader mode (finding MJ-23). + Falls back to the original address after 10 s in case this particular + bootloader keeps the same address. Works in both plain bleak environments and HA's cached scanner — in HA, BleakScanner.discover() returns the passive-scan cache, so repeated calls @@ -206,10 +230,16 @@ async def find_nrf_dfu_device(original_address: str) -> BLEDevice | None: if attempt >= 5: candidates.append(original_address.upper()) - devices = await BleakScanner.discover(timeout=0.1) - addr_map = {dev.address.upper(): dev for dev in devices} - for addr in candidates: - if addr in addr_map: - return addr_map[addr] + # return_adv=True gives the advertisement data (service UUIDs) needed for + # the DFU service-UUID fallback below. + discovered = await BleakScanner.discover(timeout=0.1, return_adv=True) + for device, adv in discovered.values(): + if device.address.upper() in candidates: + return device + # Fallback: a device advertising the Legacy DFU service is safe to + # return even before the 10 s mark — the stale app-mode entry doesn't + # expose that service, so it can't be matched here by mistake. + if _advertises_dfu_service(adv.service_uuids or []): + return device return None diff --git a/tests/unit/test_ota.py b/tests/unit/test_ota.py index 407cb8c..0918170 100644 --- a/tests/unit/test_ota.py +++ b/tests/unit/test_ota.py @@ -27,12 +27,25 @@ def _make_ble_device(address: str = "AA:BB:CC:DD:EE:FF") -> MagicMock: # --------------------------------------------------------------------------- -def _make_scanner_device(address: str) -> MagicMock: +def _make_scanner_device(address: str, name: str | None = None) -> MagicMock: dev = MagicMock() dev.address = address + dev.name = name return dev +def _make_adv(local_name: str | None = None, service_uuids: list[str] | None = None) -> MagicMock: + adv = MagicMock() + adv.local_name = local_name + adv.service_uuids = service_uuids or [] + return adv + + +def _discovered(*pairs: tuple[MagicMock, MagicMock]) -> dict[str, tuple[MagicMock, MagicMock]]: + """Build the ``address -> (device, adv)`` dict that discover(return_adv=True) returns.""" + return {dev.address: (dev, adv) for dev, adv in pairs} + + @pytest.mark.asyncio async def test_find_nrf_dfu_device_mac_plus1() -> None: """Device found at MAC+1 on the first attempt.""" @@ -42,7 +55,7 @@ async def test_find_nrf_dfu_device_mac_plus1() -> None: patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=[dfu_dev]) + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, _make_adv()))) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") assert result is dfu_dev @@ -54,9 +67,9 @@ async def test_find_nrf_dfu_device_original_address_after_5_attempts() -> None: original_dev = _make_scanner_device("AA:BB:CC:DD:EE:01") attempt = 0 - async def _discover(timeout: float) -> list[MagicMock]: + async def _discover(timeout: float, return_adv: bool = False) -> dict[str, tuple[MagicMock, MagicMock]]: nonlocal attempt - result = [original_dev] if attempt >= 5 else [] + result = _discovered((original_dev, _make_adv())) if attempt >= 5 else {} attempt += 1 return result @@ -78,7 +91,7 @@ async def test_find_nrf_dfu_device_not_found_returns_none() -> None: patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=[]) + scanner_cls.discover = AsyncMock(return_value={}) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") assert result is None @@ -86,19 +99,56 @@ async def test_find_nrf_dfu_device_not_found_returns_none() -> None: @pytest.mark.asyncio async def test_find_nrf_dfu_device_mac_plus1_wraps_ff() -> None: - """MAC+1 carries across octets: EE:FF → EF:00 (not EE:00).""" - dfu_dev = _make_scanner_device("AA:BB:CC:DD:EF:00") + """MAC+1 wraps the last octet without carry: EE:FF → EE:00 (not EF:00).""" + dfu_dev = _make_scanner_device("AA:BB:CC:DD:EE:00") with ( patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=[dfu_dev]) + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, _make_adv()))) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:FF") assert result is dfu_dev +@pytest.mark.asyncio +async def test_find_nrf_dfu_device_service_uuid_fallback() -> None: + """A device advertising the Legacy DFU service UUID is found even if its MAC doesn't match.""" + dfu_dev = _make_scanner_device("11:22:33:44:55:66") # not MAC+1 of the original + adv = _make_adv(service_uuids=["00001530-1212-efde-1523-785feabcd123"]) + + with ( + patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), + patch("bleak.BleakScanner") as scanner_cls, + ): + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, adv))) + result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") + + assert result is dfu_dev + + +@pytest.mark.asyncio +async def test_find_nrf_dfu_device_dfu_name_alone_not_selected() -> None: + """A DFU-ish name with no matching UUID/MAC must NOT be selected (finding MJ-23). + + Selecting on a name that merely contains "DFU"/"DfuTarg" picks the wrong tag + when two devices are in bootloader mode at once, so name is never a + selection discriminator — only MAC+1 or the DFU service UUID are. + """ + dfu_named = _make_scanner_device("11:22:33:44:55:66", name="DfuTarg") + adv = _make_adv(local_name="DfuTarg") # no DFU service UUID advertised + + with ( + patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), + patch("bleak.BleakScanner") as scanner_cls, + ): + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_named, adv))) + result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") + + assert result is None + + # --------------------------------------------------------------------------- # perform_* wrappers — optional-dependency import guards # --------------------------------------------------------------------------- diff --git a/tests/unit/test_ota_mac_increment.py b/tests/unit/test_ota_mac_increment.py index fe7b98b..0f0be65 100644 --- a/tests/unit/test_ota_mac_increment.py +++ b/tests/unit/test_ota_mac_increment.py @@ -1,4 +1,9 @@ -"""Test nRF DFU MAC increment carries across octets (§4).""" +"""Test nRF DFU MAC increment wraps the last octet without carry (§4). + +Nordic DFU bootloaders advertise at ``original + 1``, incrementing only the last +octet as a ``uint8`` (wrap 0xFF -> 0x00, no carry into the previous octet). This +matches the standalone ``nrf-ota`` scanner's convention. +""" from __future__ import annotations @@ -11,10 +16,10 @@ ("addr", "expected"), [ ("AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"), - ("AA:BB:CC:DD:EE:FF", "AA:BB:CC:DD:EF:00"), # carry into previous octet - ("AA:BB:CC:DD:FF:FF", "AA:BB:CC:DE:00:00"), # double carry - ("FF:FF:FF:FF:FF:FF", "00:00:00:00:00:00"), # wraps around + ("AA:BB:CC:DD:EE:FF", "AA:BB:CC:DD:EE:00"), # wraps, no carry + ("AA:BB:CC:DD:FF:FF", "AA:BB:CC:DD:FF:00"), # only the last octet wraps + ("FF:FF:FF:FF:FF:FF", "FF:FF:FF:FF:FF:00"), # previous octets untouched ], ) -def test_increment_mac_carries(addr: str, expected: str) -> None: +def test_increment_mac_no_carry(addr: str, expected: str) -> None: assert _increment_mac(addr) == expected