Przeglądaj źródła

Read copy_slots source data from cache, not a live query

get_users()/get_credentials() (fresh) walk the device's linked list of
users via nextUserId, so a slot the firmware didn't chain in correctly
(e.g. slot 250 on a real lock, confirmed via zwave-js's AccessControl.ts)
gets silently missed. get_users_cached()/get_credentials_cached() scan
every ID up to the reported max instead, matching what the slots sensor
already relies on and catching such orphaned entries.
ZiB 3 miesięcy temu
rodzic
commit
8bc4326dd7
2 zmienionych plików z 10 dodań i 3 usunięć
  1. 7 3
      custom_components/zlock_slots/services.py
  2. 3 0
      tests/conftest.py

+ 7 - 3
custom_components/zlock_slots/services.py

@@ -198,9 +198,13 @@ def register_services(hass: HomeAssistant) -> None:
     source_node = async_get_node_from_entity_id(hass, source_lock_entity_id)
     target_node = async_get_node_from_entity_id(hass, target_lock_entity_id)
 
-    source_users = [u for u in await source_node.access_control.get_users() if u.user_id != 0]
+    # Cached, not fresh: on some devices a live get_users() only enumerates a
+    # "normal" slot range and drops out-of-range entries (e.g. a slot number
+    # reused as a backup/admin code), while get_users_cached() — the same
+    # call the slots sensor uses — reports them correctly.
+    source_users = [u for u in await source_node.access_control.get_users_cached() if u.user_id != 0]
     target_slot_ids = {
-      u.user_id for u in await target_node.access_control.get_users() if u.user_id != 0
+      u.user_id for u in await target_node.access_control.get_users_cached() if u.user_id != 0
     }
 
     _LOGGER.info(
@@ -229,7 +233,7 @@ def register_services(hass: HomeAssistant) -> None:
         credential_rule=user.credential_rule,
         expiring_timeout_minutes=user.expiring_timeout_minutes,
       )
-      credentials = await source_node.access_control.get_credentials(slot)
+      credentials = await source_node.access_control.get_credentials_cached(slot)
       pin = next((c.data for c in credentials if c.type == UserCredentialType.PIN_CODE), None)
       credential = (
         AddUserCredential(credential_type=UserCredentialType.PIN_CODE, credential_slot=slot, data=pin)

+ 3 - 0
tests/conftest.py

@@ -51,6 +51,9 @@ class FakeAccessControl:
       return [FakeCredentialData(user_id, user_id, data=self._credentials[user_id])]
     return []
 
+  async def get_credentials_cached(self, user_id):
+    return await self.get_credentials(user_id)
+
   async def _set_user(self, user_id, options):
     self._users = [u for u in self._users if u.user_id != user_id]
     self._users.append(FakeUserData(