"""Tests for zlock_slots service handlers.""" import pytest import voluptuous as vol from unittest.mock import patch, AsyncMock from pytest_homeassistant_custom_component.common import MockConfigEntry from homeassistant.exceptions import ServiceValidationError from custom_components.zlock_slots.coordinator import ZLockCoordinator from custom_components.zlock_slots.services import register_services, unregister_services from custom_components.zlock_slots.const import ( DOMAIN, CONF_LOCK_ENTITY_IDS, CONF_GROUPS, CONF_GROUP_NAME, ) from tests.conftest import LOCK_ENTITY_ID, FakeUserData, FakeNode _ZWAVE_HELPER = "homeassistant.components.zwave_js.helpers.async_get_node_from_entity_id" LOCK_2 = "lock.back_door" @pytest.fixture async def service_env(hass, config_entry, mock_node): """Coordinator wired into hass.data with services registered and mock zwave_js services ready.""" config_entry.add_to_hass(hass) with patch(_ZWAVE_HELPER, return_value=mock_node): coord = ZLockCoordinator(hass, config_entry) await coord.async_setup() hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coord set_lock_usercode = AsyncMock() clear_lock_usercode = AsyncMock() hass.services.async_register("zwave_js", "set_lock_usercode", set_lock_usercode) hass.services.async_register("zwave_js", "clear_lock_usercode", clear_lock_usercode) register_services(hass) yield coord, set_lock_usercode, clear_lock_usercode unregister_services(hass) coord.async_teardown() @pytest.fixture async def group_service_env(hass): """Two-lock group, each backed by its own fake node, services registered.""" entry = MockConfigEntry( domain=DOMAIN, data={CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2]}, options={ CONF_GROUPS: [{ CONF_GROUP_NAME: "All Locks", CONF_LOCK_ENTITY_IDS: [LOCK_ENTITY_ID, LOCK_2], }] }, entry_id="test_group_service_entry", ) entry.add_to_hass(hass) nodes = { LOCK_ENTITY_ID: FakeNode(users=[FakeUserData(1, active=True)]), LOCK_2: FakeNode(users=[FakeUserData(2, active=True)]), } 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 set_lock_usercode = AsyncMock() clear_lock_usercode = AsyncMock() hass.services.async_register("zwave_js", "set_lock_usercode", set_lock_usercode) hass.services.async_register("zwave_js", "clear_lock_usercode", clear_lock_usercode) register_services(hass) with patch(_ZWAVE_HELPER, side_effect=_node_for): yield coord, nodes, set_lock_usercode, clear_lock_usercode unregister_services(hass) coord.async_teardown() # --------------------------------------------------------------------------- # set_code # --------------------------------------------------------------------------- async def test_set_code_calls_zwave_service(hass, service_env, mock_node): coord, set_lock_usercode, _ = service_env with patch(_ZWAVE_HELPER, return_value=mock_node): await hass.services.async_call( DOMAIN, "set_code", {"lock_entity_id": LOCK_ENTITY_ID, "slot": 5, "code": "1234"}, blocking=True, ) set_lock_usercode.assert_awaited_once() service_data = set_lock_usercode.call_args[0][0].data assert service_data["code_slot"] == 5 assert service_data["usercode"] == "1234" assert service_data["entity_id"] == LOCK_ENTITY_ID async def test_set_code_auto_assigns_lowest_free_slot(hass, service_env, mock_node): """Slots 1 and 2 occupied → auto-assign picks 3.""" coord, set_lock_usercode, _ = service_env with patch(_ZWAVE_HELPER, return_value=mock_node): await hass.services.async_call( DOMAIN, "set_code", {"lock_entity_id": LOCK_ENTITY_ID, "code": "5678"}, blocking=True, ) service_data = set_lock_usercode.call_args[0][0].data assert service_data["code_slot"] == 3 async def test_set_code_auto_assign_skips_occupied(hass, service_env, mock_node): """If slot 3 is also occupied, auto-assign picks 4.""" node_with_3 = FakeNode(users=[ FakeUserData(1), FakeUserData(2), FakeUserData(3), ]) coord, set_lock_usercode, _ = service_env # Refresh to get node_with_3 slots into coordinator state. with patch(_ZWAVE_HELPER, return_value=node_with_3): await coord.async_refresh_lock(LOCK_ENTITY_ID) await hass.services.async_call( DOMAIN, "set_code", {"lock_entity_id": LOCK_ENTITY_ID, "code": "9999"}, blocking=True, ) service_data = set_lock_usercode.call_args[0][0].data assert service_data["code_slot"] == 4 async def test_set_code_with_name_saves_name(hass, service_env, mock_node): coord, _, _ = service_env with patch(_ZWAVE_HELPER, return_value=mock_node): await hass.services.async_call( DOMAIN, "set_code", {"lock_entity_id": LOCK_ENTITY_ID, "slot": 5, "code": "4321", "name": "Charlie"}, blocking=True, ) assert coord.get_name(LOCK_ENTITY_ID, 5) == "Charlie" async def test_set_code_unknown_lock_raises(hass, service_env): with pytest.raises(ServiceValidationError): await hass.services.async_call( DOMAIN, "set_code", {"lock_entity_id": "lock.unknown", "slot": 1, "code": "1234"}, blocking=True, ) # --------------------------------------------------------------------------- # clear_code # --------------------------------------------------------------------------- async def test_clear_code_calls_zwave_service(hass, service_env, mock_node): _, _, clear_lock_usercode = service_env with patch(_ZWAVE_HELPER, return_value=mock_node): await hass.services.async_call( DOMAIN, "clear_code", {"lock_entity_id": LOCK_ENTITY_ID, "slot": 2}, blocking=True, ) clear_lock_usercode.assert_awaited_once() service_data = clear_lock_usercode.call_args[0][0].data assert service_data["code_slot"] == 2 assert service_data["entity_id"] == LOCK_ENTITY_ID # --------------------------------------------------------------------------- # set_name # --------------------------------------------------------------------------- async def test_set_name_updates_coordinator(hass, service_env): coord, _, _ = service_env await hass.services.async_call( DOMAIN, "set_name", {"lock_entity_id": LOCK_ENTITY_ID, "slot": 1, "name": "Diana"}, blocking=True, ) assert coord.get_name(LOCK_ENTITY_ID, 1) == "Diana" # --------------------------------------------------------------------------- # set_slot_enabled # --------------------------------------------------------------------------- async def test_set_slot_enabled_calls_node(hass, service_env, mock_node): with patch(_ZWAVE_HELPER, return_value=mock_node): await hass.services.async_call( DOMAIN, "set_slot_enabled", {"lock_entity_id": LOCK_ENTITY_ID, "slot": 1, "enabled": False}, blocking=True, ) mock_node.access_control.set_user.assert_awaited_once() slot_arg, opts_arg = mock_node.access_control.set_user.call_args[0] assert slot_arg == 1 assert opts_arg.active is False async def test_set_slot_enabled_true(hass, service_env, mock_node): with patch(_ZWAVE_HELPER, return_value=mock_node): await hass.services.async_call( DOMAIN, "set_slot_enabled", {"lock_entity_id": LOCK_ENTITY_ID, "slot": 2, "enabled": True}, blocking=True, ) _, opts_arg = mock_node.access_control.set_user.call_args[0] assert opts_arg.active is True # --------------------------------------------------------------------------- # Group targeting: fan-out and target validation # --------------------------------------------------------------------------- async def test_set_code_group_fanout(hass, group_service_env): _, _, set_lock_usercode, _ = group_service_env await hass.services.async_call( DOMAIN, "set_code", {"group_name": "All Locks", "slot": 5, "code": "1234"}, blocking=True, ) assert set_lock_usercode.await_count == 2 entity_ids = {c[0][0].data["entity_id"] for c in set_lock_usercode.call_args_list} assert entity_ids == {LOCK_ENTITY_ID, LOCK_2} async def test_set_code_group_auto_assign_uses_union_of_members(hass, group_service_env): """LOCK_ENTITY_ID has slot 1 occupied, LOCK_2 has slot 2 — union picks 3 for both.""" _, _, set_lock_usercode, _ = group_service_env await hass.services.async_call( DOMAIN, "set_code", {"group_name": "All Locks", "code": "5678"}, blocking=True, ) slots_used = {c[0][0].data["code_slot"] for c in set_lock_usercode.call_args_list} assert slots_used == {3} async def test_clear_code_group_fanout(hass, group_service_env): _, _, _, clear_lock_usercode = group_service_env await hass.services.async_call( DOMAIN, "clear_code", {"group_name": "All Locks", "slot": 1}, blocking=True, ) assert clear_lock_usercode.await_count == 2 entity_ids = {c[0][0].data["entity_id"] for c in clear_lock_usercode.call_args_list} assert entity_ids == {LOCK_ENTITY_ID, LOCK_2} async def test_set_name_group_fanout(hass, group_service_env): coord, _, _, _ = group_service_env await hass.services.async_call( DOMAIN, "set_name", {"group_name": "All Locks", "slot": 1, "name": "Everyone"}, blocking=True, ) assert coord.get_name(LOCK_ENTITY_ID, 1) == "Everyone" assert coord.get_name(LOCK_2, 1) == "Everyone" async def test_set_slot_enabled_group_fanout(hass, group_service_env): coord, nodes, _, _ = group_service_env await hass.services.async_call( DOMAIN, "set_slot_enabled", {"group_name": "All Locks", "slot": 1, "enabled": False}, blocking=True, ) nodes[LOCK_ENTITY_ID].access_control.set_user.assert_awaited_once() nodes[LOCK_2].access_control.set_user.assert_awaited_once() async def test_group_fanout_continues_after_one_member_fails(hass, group_service_env): """One lock erroring must not block the operation from reaching the others.""" coord, nodes, _, _ = group_service_env nodes[LOCK_ENTITY_ID].access_control.set_user.side_effect = RuntimeError("boom") await hass.services.async_call( DOMAIN, "set_slot_enabled", {"group_name": "All Locks", "slot": 1, "enabled": False}, blocking=True, ) nodes[LOCK_ENTITY_ID].access_control.set_user.assert_awaited_once() nodes[LOCK_2].access_control.set_user.assert_awaited_once() async def test_unknown_group_raises(hass, group_service_env): with pytest.raises(ServiceValidationError): await hass.services.async_call( DOMAIN, "set_code", {"group_name": "Nonexistent", "slot": 1, "code": "1234"}, blocking=True, ) async def test_both_lock_and_group_target_rejected(hass, group_service_env): with pytest.raises(vol.Invalid): await hass.services.async_call( DOMAIN, "set_code", {"lock_entity_id": LOCK_ENTITY_ID, "group_name": "All Locks", "slot": 1, "code": "1234"}, blocking=True, ) async def test_neither_lock_nor_group_target_rejected(hass, group_service_env): with pytest.raises(vol.Invalid): await hass.services.async_call( DOMAIN, "set_code", {"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, )