Add copy_slots service to mirror a lock's user-slot profile onto another

Migration helper for swapping a lock's hardware without hand-copying
codes/names in zwave_js: mirrors codes, enabled status, and slot names
from one lock onto another, deleting target-only slots. Not exposed in
the card — service-call only.

README.md

@@ -57,6 +57,16 @@ sync guarantee. When auto-assigning a slot for a group, the lowest slot
 unused by *any* member is chosen, so the same slot number lands on all of
 them.
 
+### `zlock_slots.copy_slots`
+
+A hardware-migration helper for replacing one lock with another, not a
+routine card operation. Takes `source_lock_entity_id` and
+`target_lock_entity_id` (both required, must differ, both must already be
+managed by this integration) and fully mirrors the source's Z-Wave user-slot
+profile — codes, enabled/disabled status, and slot names — onto the target.
+This is a full mirror: any slot occupied on the target with no counterpart
+on the source is deleted.
+
 Slot 0 is reserved by Z-Wave and never shown or written.
 
 ## Lovelace card

custom_components/zlock_slots/services.py

@@ -17,6 +17,7 @@ SERVICE_SET_CODE = "set_code"
 SERVICE_CLEAR_CODE = "clear_code"
 SERVICE_SET_NAME = "set_name"
 SERVICE_SET_SLOT_ENABLED = "set_slot_enabled"
+SERVICE_COPY_SLOTS = "copy_slots"
 
 _TARGET_FIELDS = {
   vol.Exclusive("lock_entity_id", "target"): cv.entity_id,
@@ -59,6 +60,11 @@ SET_SLOT_ENABLED_SCHEMA = vol.All(
   }),
 )
 
+COPY_SLOTS_SCHEMA = vol.Schema({
+  vol.Required("source_lock_entity_id"): cv.entity_id,
+  vol.Required("target_lock_entity_id"): cv.entity_id,
+})
+
 
 def _get_coordinator(hass: HomeAssistant, lock_entity_id: str) -> ZLockCoordinator:
   for entry_data in hass.data.get(DOMAIN, {}).values():
@@ -161,13 +167,67 @@ def register_services(hass: HomeAssistant) -> None:
 
     await _fanout(lock_ids, _set_enabled)
 
+  async def handle_copy_slots(call: ServiceCall) -> None:
+    """Mirror one lock's entire Z-Wave user-slot profile onto another.
+
+    A hardware-migration helper (e.g. replacing lock A with lock B), not a
+    routine card operation: it talks to node.access_control directly rather
+    than the zwave_js.set_lock_usercode/clear_lock_usercode HA services used
+    by set_code/clear_code, and does a full mirror — slots on the target with
+    no counterpart on the source are deleted, not just left alone.
+    """
+    from zwave_js_server.model.access_control import SetUserOptions, UserCredentialType
+    from homeassistant.components.zwave_js.helpers import async_get_node_from_entity_id
+
+    source_lock_entity_id = call.data["source_lock_entity_id"]
+    target_lock_entity_id = call.data["target_lock_entity_id"]
+    if source_lock_entity_id == target_lock_entity_id:
+      raise ServiceValidationError("source_lock_entity_id and target_lock_entity_id must differ")
+
+    coordinator = _get_coordinator(hass, source_lock_entity_id)
+    _get_coordinator(hass, target_lock_entity_id)
+
+    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]
+    target_users = [u for u in await target_node.access_control.get_users() if u.user_id != 0]
+
+    source_slots = {u.user_id for u in source_users}
+    for user in target_users:
+      if user.user_id not in source_slots:
+        await target_node.access_control.delete_user(user.user_id)
+        await coordinator.async_set_name(target_lock_entity_id, user.user_id, "")
+
+    for user in source_users:
+      slot = user.user_id
+      await target_node.access_control.set_user(slot, SetUserOptions(
+        active=user.active,
+        user_type=user.user_type,
+        credential_rule=user.credential_rule,
+        expiring_timeout_minutes=user.expiring_timeout_minutes,
+      ))
+
+      credentials = await source_node.access_control.get_credentials(slot)
+      pin = next((c.data for c in credentials if c.type == UserCredentialType.PIN_CODE), None)
+      if pin is not None:
+        await target_node.access_control.set_credential(slot, UserCredentialType.PIN_CODE, slot, pin)
+
+      name = coordinator.get_name(source_lock_entity_id, slot)
+      await coordinator.async_set_name(target_lock_entity_id, slot, name or "")
+
+    await coordinator.async_refresh_lock(target_lock_entity_id)
+
   hass.services.async_register(DOMAIN, SERVICE_SET_CODE, handle_set_code, SET_CODE_SCHEMA)
   hass.services.async_register(DOMAIN, SERVICE_CLEAR_CODE, handle_clear_code, CLEAR_CODE_SCHEMA)
   hass.services.async_register(DOMAIN, SERVICE_SET_NAME, handle_set_name, SET_NAME_SCHEMA)
   hass.services.async_register(DOMAIN, SERVICE_SET_SLOT_ENABLED, handle_set_slot_enabled, SET_SLOT_ENABLED_SCHEMA)
+  hass.services.async_register(DOMAIN, SERVICE_COPY_SLOTS, handle_copy_slots, COPY_SLOTS_SCHEMA)
 
 
 def unregister_services(hass: HomeAssistant) -> None:
-  for service in (SERVICE_SET_CODE, SERVICE_CLEAR_CODE, SERVICE_SET_NAME, SERVICE_SET_SLOT_ENABLED):
+  for service in (
+    SERVICE_SET_CODE, SERVICE_CLEAR_CODE, SERVICE_SET_NAME, SERVICE_SET_SLOT_ENABLED, SERVICE_COPY_SLOTS,
+  ):
     if hass.services.has_service(DOMAIN, service):
       hass.services.async_remove(DOMAIN, service)

custom_components/zlock_slots/services.yaml

@@ -134,3 +134,29 @@ set_slot_enabled:
       required: true
       selector:
         boolean:
+
+copy_slots:
+  name: Copy Slots
+  description: >-
+    Mirror one lock's entire Z-Wave user-slot profile (codes, enabled/disabled
+    status, and slot names) onto another lock. A hardware-migration helper for
+    replacing one lock with another — this is a full mirror: slots occupied on
+    the target with no counterpart on the source are deleted. Both locks must
+    already be managed by this integration.
+  fields:
+    source_lock_entity_id:
+      name: Source lock
+      description: The lock to copy the user-slot profile from.
+      required: true
+      selector:
+        entity:
+          domain: lock
+          integration: zwave_js
+    target_lock_entity_id:
+      name: Target lock
+      description: The lock to overwrite with the source lock's user-slot profile.
+      required: true
+      selector:
+        entity:
+          domain: lock
+          integration: zwave_js

tests/conftest.py

@@ -13,23 +13,64 @@ DOOR_SENSOR_ENTITY = "binary_sensor.front_door_contact"
 
 
 class FakeUserData:
-  def __init__(self, user_id: int, active: bool = True):
+  def __init__(self, user_id: int, active: bool = True, user_type=0,
+               credential_rule=None, expiring_timeout_minutes=None):
     self.user_id = user_id
     self.active = active
+    self.user_type = user_type
+    self.credential_rule = credential_rule
+    self.expiring_timeout_minutes = expiring_timeout_minutes
+
+
+class FakeCredentialData:
+  def __init__(self, user_id: int, slot: int, type=1, data=None):
+    self.user_id = user_id
+    self.slot = slot
+    self.type = type
+    self.data = data
 
 
 class FakeAccessControl:
-  def __init__(self, users=None):
+  def __init__(self, users=None, credentials=None):
     self._users = list(users or [])
-    self.set_user = AsyncMock()
+    # user_id -> PIN code string
+    self._credentials: dict[int, str] = dict(credentials or {})
+    self.set_user = AsyncMock(side_effect=self._set_user)
+    self.set_credential = AsyncMock(side_effect=self._set_credential)
+    self.delete_user = AsyncMock(side_effect=self._delete_user)
 
   async def get_users_cached(self):
     return list(self._users)
 
+  async def get_users(self):
+    return list(self._users)
+
+  async def get_credentials(self, user_id):
+    if user_id in self._credentials:
+      return [FakeCredentialData(user_id, user_id, data=self._credentials[user_id])]
+    return []
+
+  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(
+      user_id,
+      active=options.active,
+      user_type=options.user_type,
+      credential_rule=options.credential_rule,
+      expiring_timeout_minutes=options.expiring_timeout_minutes,
+    ))
+
+  async def _set_credential(self, user_id, credential_type, credential_slot, data):
+    self._credentials[user_id] = data
+
+  async def _delete_user(self, user_id):
+    self._users = [u for u in self._users if u.user_id != user_id]
+    self._credentials.pop(user_id, None)
+
 
 class FakeNode:
-  def __init__(self, users=None):
-    self.access_control = FakeAccessControl(users)
+  def __init__(self, users=None, credentials=None):
+    self.access_control = FakeAccessControl(users, credentials)
 
 
 @pytest.fixture(autouse=True)

tests/test_services.py

@@ -325,3 +325,98 @@ async def test_neither_lock_nor_group_target_rejected(hass, group_service_env):
       {"slot": 1, "code": "1234"},
       blocking=True,
     )
+
+
+# ---------------------------------------------------------------------------
+# copy_slots
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+async def copy_slots_env(hass):
+  """Two locks (source populated, target empty), services registered."""
+  entry = MockConfigEntry(
+    domain=DOMAIN,
+    data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]},
+    options={},
+    entry_id="test_copy_slots_entry",
+  )
+  entry.add_to_hass(hass)
+
+  nodes = {
+    LOCK_ENTITY_ID: FakeNode(
+      users=[FakeUserData(1, active=True), FakeUserData(2, active=False)],
+      credentials={1: "1234", 2: "5678"},
+    ),
+    LOCK_2: FakeNode(users=[FakeUserData(9, active=True)], credentials={9: "0000"}),
+  }
+
+  def _node_for(hass_, entity_id):
+    return nodes[entity_id]
+
+  with patch(_ZWAVE_HELPER, side_effect=_node_for):
+    coord = ZLockCoordinator(hass, entry)
+    await coord.async_setup()
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coord
+    await coord.async_set_name(LOCK_ENTITY_ID, 1, "Alice")
+
+    register_services(hass)
+    with patch(_ZWAVE_HELPER, side_effect=_node_for):
+      yield coord, nodes
+
+  unregister_services(hass)
+  coord.async_teardown()
+
+
+async def test_copy_slots_mirrors_users_and_codes(hass, copy_slots_env):
+  coord, nodes = copy_slots_env
+  await hass.services.async_call(
+    DOMAIN, "copy_slots",
+    {"source_lock_entity_id": LOCK_ENTITY_ID, "target_lock_entity_id": LOCK_2},
+    blocking=True,
+  )
+
+  target_ac = nodes[LOCK_2].access_control
+  target_ids = {u.user_id for u in await target_ac.get_users()}
+  assert target_ids == {1, 2}
+
+  target_slot1 = next(u for u in await target_ac.get_users() if u.user_id == 1)
+  assert target_slot1.active is True
+  target_slot2 = next(u for u in await target_ac.get_users() if u.user_id == 2)
+  assert target_slot2.active is False
+
+  assert target_ac._credentials[1] == "1234"
+  assert target_ac._credentials[2] == "5678"
+  assert coord.get_name(LOCK_2, 1) == "Alice"
+
+
+async def test_copy_slots_deletes_target_only_slots(hass, copy_slots_env):
+  """Target's slot 9 has no counterpart on the source and must be removed."""
+  coord, nodes = copy_slots_env
+  await hass.services.async_call(
+    DOMAIN, "copy_slots",
+    {"source_lock_entity_id": LOCK_ENTITY_ID, "target_lock_entity_id": LOCK_2},
+    blocking=True,
+  )
+
+  target_ac = nodes[LOCK_2].access_control
+  target_ids = {u.user_id for u in await target_ac.get_users()}
+  assert 9 not in target_ids
+  assert 9 not in target_ac._credentials
+
+
+async def test_copy_slots_rejects_same_source_and_target(hass, copy_slots_env):
+  with pytest.raises(ServiceValidationError):
+    await hass.services.async_call(
+      DOMAIN, "copy_slots",
+      {"source_lock_entity_id": LOCK_ENTITY_ID, "target_lock_entity_id": LOCK_ENTITY_ID},
+      blocking=True,
+    )
+
+
+async def test_copy_slots_rejects_unmanaged_lock(hass, copy_slots_env):
+  with pytest.raises(ServiceValidationError):
+    await hass.services.async_call(
+      DOMAIN, "copy_slots",
+      {"source_lock_entity_id": LOCK_ENTITY_ID, "target_lock_entity_id": "lock.unknown"},
+      blocking=True,
+    )