fix(ota): align nRF DFU MAC increment to nrf-ota no-carry convention + add discovery fallback

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

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,27 +91,64 @@ 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
 
 
 @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
 # ---------------------------------------------------------------------------

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