zlock_slots/tests/test_services.py

"""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,
    )