fix: add timeout and fallback for lock_info probe

src/yalexs_ble/lock.py

@@ -48,6 +48,8 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+LOCK_INFO_TIMEOUT = 5
+
 AA_BATTERY_VOLTAGE_TO_PERCENTAGE = (
     (1.55, 100),
     (1.549, 97),
@@ -305,20 +307,43 @@ async def lock_info(self) -> LockInfo:
         """Probe the lock for information."""
         _LOGGER.debug("%s: Probing the lock", self.name)
         assert self.client is not None  # nosec
-        lock_info = []
-        for char_uuid in (
-            MANUFACTURER_NAME_CHARACTERISTIC,
+        # Read model first since it drives battery and door sense behavior.
+        # Order: model, manufacturer, serial, firmware.
+        char_uuids = (
             MODEL_NUMBER_CHARACTERISTIC,
+            MANUFACTURER_NAME_CHARACTERISTIC,
             SERIAL_NUMBER_CHARACTERISTIC,
             FIRMWARE_REVISION_CHARACTERISTIC,
-        ):
-            char = self.client.services.get_characteristic(char_uuid)
-            if not char:
-                await self._handle_missing_characteristic(char_uuid)
-            lock_info.append(
-                (await self.client.read_gatt_char(char)).decode().split("\0")[0]
-            )
-        self._lock_info = LockInfo(*lock_info)
+        )
+        results: dict[str, str] = {}
+        async with util.asyncio_timeout(LOCK_INFO_TIMEOUT):
+            for char_uuid in char_uuids:
+                char = self.client.services.get_characteristic(char_uuid)
+                if not char:
+                    _LOGGER.warning(
+                        "%s: Characteristic %s not found", self.name, char_uuid
+                    )
+                    continue
+                try:
+                    results[char_uuid] = (
+                        (await self.client.read_gatt_char(char)).decode().split("\0")[0]
+                    )
+                except BleakError as err:
+                    _LOGGER.warning(
+                        "%s: Failed to read characteristic %s: %s",
+                        self.name,
+                        char_uuid,
+                        err,
+                    )
+        # Use the BLE address as fallback serial to keep devices unique
+        # in Home Assistant when the characteristic read fails.
+        serial_fallback = self.ble_device_callback().address
+        self._lock_info = LockInfo(
+            manufacturer=results.get(MANUFACTURER_NAME_CHARACTERISTIC, "Unknown"),
+            model=results.get(MODEL_NUMBER_CHARACTERISTIC, ""),
+            serial=results.get(SERIAL_NUMBER_CHARACTERISTIC, serial_fallback),
+            firmware=results.get(FIRMWARE_REVISION_CHARACTERISTIC, "Unknown"),
+        )
         return self._lock_info
 
     @raise_if_not_connected

src/yalexs_ble/push.py

@@ -863,6 +863,25 @@ async def _poll_battery(
 
         return state, True
 
+    async def _probe_lock_info(self, lock: Lock) -> LockInfo:
+        """Probe the lock for info, falling back to defaults on failure."""
+        try:
+            lock_info = await lock.lock_info()
+        except (TimeoutError, BleakError) as err:
+            _LOGGER.warning(
+                "%s: Failed to probe lock info (%s), continuing with defaults",
+                self.name,
+                err,
+            )
+            lock_info = LockInfo(
+                manufacturer="Unknown",
+                model="",
+                serial=self.address,
+                firmware="Unknown",
+            )
+        _LOGGER.debug("Obtained lock info: %s", lock_info)
+        return lock_info
+
     @operation_lock
     @retry_bluetooth_connection_error
     async def _update(self) -> LockState:
@@ -874,8 +893,7 @@ async def _update(self) -> LockState:
         )
         lock = await self._ensure_connected()
         if not self._lock_info:
-            self._lock_info = await lock.lock_info()
-            _LOGGER.debug("Obtained lock info: %s", self._lock_info)
+            self._lock_info = await self._probe_lock_info(lock)
         # Asking for battery first seems to be reduce the chance of the lock
         # getting into a bad state.
         state = self._get_current_state()

tests/test_lock.py

@@ -3,13 +3,23 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from bleak.exc import BleakError
 from bleak_retry_connector import BLEDevice
 
-from yalexs_ble.const import LockOperationRemoteType, LockOperationSource, LockStatus
+from yalexs_ble.const import (
+    FIRMWARE_REVISION_CHARACTERISTIC,
+    MANUFACTURER_NAME_CHARACTERISTIC,
+    MODEL_NUMBER_CHARACTERISTIC,
+    SERIAL_NUMBER_CHARACTERISTIC,
+    LockInfo,
+    LockOperationRemoteType,
+    LockOperationSource,
+    LockStatus,
+)
 from yalexs_ble.lock import Lock
 
 
-def test_create_lock():
+def test_create_lock() -> None:
     Lock(
         lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock"),
         "0800200c9a66",
@@ -20,7 +30,7 @@ def test_create_lock():
 
 
 @pytest.mark.asyncio
-async def test_connection_canceled_on_disconnect():
+async def test_connection_canceled_on_disconnect() -> None:
     disconnect_mock = AsyncMock()
     mock_client = MagicMock(connected=True, disconnect=disconnect_mock)
     lock = Lock(
@@ -32,7 +42,7 @@ async def test_connection_canceled_on_disconnect():
     )
     lock.client = mock_client
 
-    async def connect_and_wait():
+    async def connect_and_wait() -> None:
         await lock.connect()
         await asyncio.sleep(2)
 
@@ -47,7 +57,7 @@ async def connect_and_wait():
     assert task.cancelled() is True
 
 
-def test_parse_operation_source():
+def test_parse_operation_source() -> None:
     """Test parsing operation source and remote type."""
     lock = Lock(
         lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock"),
@@ -93,7 +103,7 @@ def test_parse_operation_source():
     assert remote_type is LockOperationRemoteType.UNKNOWN
 
 
-def test_parse_lock_command_response_jammed():
+def test_parse_lock_command_response_jammed() -> None:
     """Test parsing LOCK command response with JAMMED status."""
     lock = Lock(
         lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock"),
@@ -114,7 +124,7 @@ def test_parse_lock_command_response_jammed():
     assert result_list[0] is LockStatus.JAMMED
 
 
-def test_parse_lock_command_response_unlocked():
+def test_parse_lock_command_response_unlocked() -> None:
     """Test parsing LOCK command response with UNLOCKED (jam as unlocked)."""
     lock = Lock(
         lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock"),
@@ -135,7 +145,7 @@ def test_parse_lock_command_response_unlocked():
     assert result_list[0] is LockStatus.UNLOCKED
 
 
-def test_parse_unlock_command_response():
+def test_parse_unlock_command_response() -> None:
     """Test parsing UNLOCK command response."""
     lock = Lock(
         lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock"),
@@ -156,7 +166,7 @@ def test_parse_unlock_command_response():
     assert result_list[0] is LockStatus.UNLOCKED
 
 
-def test_parse_lock_command_response_locked_success():
+def test_parse_lock_command_response_locked_success() -> None:
     """Test parsing LOCK command response with successful LOCKED status."""
     lock = Lock(
         lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock"),
@@ -175,3 +185,163 @@ def test_parse_lock_command_response_locked_success():
     result_list = list(result)
     assert len(result_list) == 1
     assert result_list[0] is LockStatus.LOCKED
+
+
+_CHAR_DATA: dict[str, bytes] = {
+    MODEL_NUMBER_CHARACTERISTIC: b"ASL-03",
+    MANUFACTURER_NAME_CHARACTERISTIC: b"August",
+    SERIAL_NUMBER_CHARACTERISTIC: b"12345",
+    FIRMWARE_REVISION_CHARACTERISTIC: b"2.0.0",
+}
+
+# Model is read first, then manufacturer, serial, firmware.
+_CHAR_ORDER: tuple[str, ...] = (
+    MODEL_NUMBER_CHARACTERISTIC,
+    MANUFACTURER_NAME_CHARACTERISTIC,
+    SERIAL_NUMBER_CHARACTERISTIC,
+    FIRMWARE_REVISION_CHARACTERISTIC,
+)
+
+
+def _make_lock_with_mock_client(
+    side_effects: dict[str, Exception] | None = None,
+) -> tuple[Lock, MagicMock]:
+    """Create a Lock with a mock BLE client for lock_info tests."""
+    lock = Lock(
+        lambda: BLEDevice("aa:bb:cc:dd:ee:ff", "lock", details=None),
+        "0800200c9a66",
+        1,
+        "mylock",
+        lambda _: None,
+    )
+    mock_client = MagicMock()
+    mock_client.is_connected = True
+    lock.client = mock_client
+    lock.session = MagicMock()
+    lock.secure_session = MagicMock()
+
+    effects = side_effects or {}
+
+    # Map each characteristic UUID to a unique mock object so
+    # read_gatt_char can identify which UUID is being read.
+    char_mocks: dict[str, MagicMock] = {}
+    mock_to_uuid: dict[int, str] = {}
+    for uuid in _CHAR_ORDER:
+        m = MagicMock()
+        char_mocks[uuid] = m
+        mock_to_uuid[id(m)] = uuid
+
+    mock_client.services.get_characteristic = char_mocks.get
+
+    async def read_gatt_char(char: MagicMock) -> bytes:
+        uuid = mock_to_uuid[id(char)]
+        if uuid in effects:
+            raise effects[uuid]
+        return _CHAR_DATA[uuid]
+
+    mock_client.read_gatt_char = read_gatt_char
+    mock_client._mock_to_uuid = mock_to_uuid
+    return lock, mock_client
+
+
+@pytest.mark.asyncio
+async def test_lock_info_success() -> None:
+    """Test lock_info reads all characteristics successfully."""
+    lock, _ = _make_lock_with_mock_client()
+
+    info = await lock.lock_info()
+
+    assert info == LockInfo(
+        manufacturer="August",
+        model="ASL-03",
+        serial="12345",
+        firmware="2.0.0",
+    )
+
+
+@pytest.mark.asyncio
+async def test_lock_info_partial_failure() -> None:
+    """Test lock_info continues when individual reads fail."""
+    lock, _ = _make_lock_with_mock_client(
+        side_effects={SERIAL_NUMBER_CHARACTERISTIC: BleakError("Connection dropped")}
+    )
+
+    info = await lock.lock_info()
+
+    assert info.manufacturer == "August"
+    assert info.model == "ASL-03"
+    assert info.serial == "aa:bb:cc:dd:ee:ff"
+    assert info.firmware == "2.0.0"
+
+
+@pytest.mark.asyncio
+async def test_lock_info_all_reads_fail() -> None:
+    """Test lock_info returns all Unknown when every read fails."""
+    lock, _ = _make_lock_with_mock_client(
+        side_effects={uuid: BleakError("Failed") for uuid in _CHAR_ORDER}
+    )
+
+    info = await lock.lock_info()
+
+    assert info == LockInfo(
+        manufacturer="Unknown",
+        model="",
+        serial="aa:bb:cc:dd:ee:ff",
+        firmware="Unknown",
+    )
+
+
+@pytest.mark.asyncio
+async def test_lock_info_timeout() -> None:
+    """Test lock_info raises TimeoutError when reads hang."""
+    lock, mock_client = _make_lock_with_mock_client()
+
+    async def hang_forever(char: MagicMock) -> bytes:
+        await asyncio.sleep(999)
+        return b""  # unreachable
+
+    mock_client.read_gatt_char = hang_forever
+
+    with patch("yalexs_ble.lock.LOCK_INFO_TIMEOUT", 0), pytest.raises(TimeoutError):
+        await lock.lock_info()
+
+
+@pytest.mark.asyncio
+async def test_lock_info_missing_characteristic() -> None:
+    """Test lock_info skips missing characteristics instead of aborting."""
+    lock, mock_client = _make_lock_with_mock_client()
+
+    original_get = mock_client.services.get_characteristic
+
+    def get_char_skip_serial(uuid: str) -> MagicMock | None:
+        if uuid == SERIAL_NUMBER_CHARACTERISTIC:
+            return None
+        return original_get(uuid)
+
+    mock_client.services.get_characteristic = get_char_skip_serial
+
+    info = await lock.lock_info()
+
+    assert info.manufacturer == "August"
+    assert info.model == "ASL-03"
+    assert info.serial == "aa:bb:cc:dd:ee:ff"
+    assert info.firmware == "2.0.0"
+
+
+@pytest.mark.asyncio
+async def test_lock_info_reads_model_first() -> None:
+    """Test that model is read first so it's available as early as possible."""
+    lock, mock_client = _make_lock_with_mock_client()
+    call_order: list[str] = []
+    original_read = mock_client.read_gatt_char
+    mock_to_uuid = mock_client._mock_to_uuid
+
+    async def tracking_read(char: MagicMock) -> bytes:
+        call_order.append(mock_to_uuid[id(char)])
+        return await original_read(char)
+
+    mock_client.read_gatt_char = tracking_read
+
+    await lock.lock_info()
+
+    assert call_order[0] == MODEL_NUMBER_CHARACTERISTIC