test: add tests for accesslimit_count behavior; refactor entity helpers

custom_components/keymaster/entity.py

@@ -4,7 +4,7 @@
 
 from dataclasses import dataclass
 import logging
-from typing import Any
+from typing import Any, Literal
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -35,13 +35,14 @@ def __init__(self, entity_description: KeymasterEntityDescription) -> None:
         self.coordinator: KeymasterCoordinator = entity_description.coordinator
         self._config_entry: ConfigEntry = entity_description.config_entry
         self.entity_description: KeymasterEntityDescription = entity_description
-        self._attr_available = False
         self._property: str = entity_description.key  # <Platform>.<Property>.<SubProperty>:<Slot Number*>.<SubProperty>:<Slot Number*>  *Only if needed
         self._kmlock: KeymasterLock | None = self.coordinator.sync_get_lock_by_config_entry_id(
             self._config_entry.entry_id
         )
-        if self._kmlock:
-            self._attr_name: str | None = f"{self._kmlock.lock_name} {self.entity_description.name}"
+        self._code_slot = self._get_x_num("code_slots")
+        self._day_of_week_num = self._get_x_num("accesslimit_day_of_week")
+
+        self._attr_available = False
         # _LOGGER.debug(
         #     "[Entity init] entity_description.name: %s, name: %s",
         #     self.entity_description.name,
@@ -54,18 +55,14 @@ def __init__(self, entity_description: KeymasterEntityDescription) -> None:
         #     self._property,
         #     self.unique_id,
         # )
-        self._code_slot: None | int = None
-        if ".code_slots" in self._property:
-            self._code_slot = self._get_code_slots_num()
-        self._day_of_week_num: None | int = None
-        if "accesslimit_day_of_week" in self._property:
-            self._day_of_week_num = self._get_day_of_week_num()
         self._attr_extra_state_attributes: dict[str, Any] = {}
         self._attr_device_info: DeviceInfo = {
             "identifiers": {(DOMAIN, self._config_entry.entry_id)},
         }
-        # _LOGGER.debug(f"[Entity init] Entity created: {self.name}, device_info: {self.device_info}")
+        if self._kmlock:
+            self._attr_name: str | None = f"{self._kmlock.lock_name} {self.entity_description.name}"
         super().__init__(self.coordinator, self._attr_unique_id)
+        # _LOGGER.debug(f"[Entity init] Entity created: {self.name}, device_info: {self.device_info}")
 
     @property
     def available(self) -> bool:
@@ -121,27 +118,19 @@ def _set_property_value(self, value: Any) -> bool:
         )
         return True
 
-    def _get_code_slots_num(self) -> None | int:
-        if ".code_slots" not in self._property:
-            return None
-        slots: list[str] = self._property.split(".")
-        for slot in slots:
-            if slot.startswith("code_slots"):
-                if ":" not in slot:
-                    return None
-                return int(slot.split(":")[1])
-        return None
-
-    def _get_day_of_week_num(self) -> None | int:
-        if "accesslimit_day_of_week" not in self._property:
+    def _get_x_num(
+        self, property_name: Literal["code_slots", "accesslimit_day_of_week"]
+    ) -> None | int:
+        """Return the code slot or day of week number from the property string."""
+        try:
+            slot_str = next(
+                part
+                for part in self._property.split(".")
+                if part.startswith(property_name) and ":" in part
+            )
+        except StopIteration:
             return None
-        slots: list[str] = self._property.split(".")
-        for slot in slots:
-            if slot.startswith("accesslimit_day_of_week"):
-                if ":" not in slot:
-                    return None
-                return int(slot.split(":")[1])
-        return None
+        return int(slot_str.split(":")[1])
 
 
 @dataclass(frozen=True, kw_only=True)

custom_components/keymaster/number.py

@@ -111,6 +111,11 @@ def __init__(
         )
         self._attr_native_value: float | None = None
 
+    @property
+    def _is_accesslimit_count(self) -> bool:
+        """Return True if this Number is for accesslimit_count."""
+        return self._property.split(".")[-1] == "accesslimit_count"
+
     @callback
     def _handle_coordinator_update(self) -> None:
         # _LOGGER.debug(f"[Number handle_coordinator_update] self.coordinator.data: {self.coordinator.data}")
@@ -140,7 +145,7 @@ def _handle_coordinator_update(self) -> None:
             self.async_write_ha_state()
             return
 
-        if self._property.endswith(".accesslimit_count") and (
+        if self._is_accesslimit_count and (
             not self._kmlock.code_slots
             or not self._code_slot
             or not self._kmlock.code_slots[self._code_slot].accesslimit_count_enabled
@@ -150,8 +155,7 @@ def _handle_coordinator_update(self) -> None:
             return
 
         if (
-            self._property.endswith(".autolock_min_day")
-            or self._property.endswith(".autolock_min_night")
+            self._property.split(".")[-1].startswith("autolock")
         ) and not self._kmlock.autolock_enabled:
             self._attr_available = False
             self.async_write_ha_state()
@@ -169,7 +173,7 @@ async def async_set_native_value(self, value: float) -> None:
             value,
         )
         if (
-            self._property.endswith(".accesslimit_count")
+            self._is_accesslimit_count
             and self._kmlock
             and self._kmlock.parent_name
             and (
@@ -185,7 +189,7 @@ async def async_set_native_value(self, value: float) -> None:
             )
             return
         # Convert to int for accesslimit_count (NumberEntity returns float)
-        if self._property.split(".")[-1] == "accesslimit_count":
+        if self._is_accesslimit_count:
             value = int(value)
         if self._set_property_value(value):
             self._attr_native_value = value

tests/test_coordinator_events.py

@@ -6,7 +6,7 @@
 import pytest
 
 from custom_components.keymaster.coordinator import KeymasterCoordinator
-from custom_components.keymaster.lock import KeymasterLock
+from custom_components.keymaster.lock import KeymasterCodeSlot, KeymasterLock
 from homeassistant.components.lock.const import LockState
 from homeassistant.core import Event
 
@@ -148,3 +148,167 @@ async def test_handle_lock_state_change_entity(hass, mock_coordinator, mock_lock
     assert kwargs["source"] == "entity_state"
     # Should use label from map based on alarm_type 18
     assert kwargs["event_label"] == "Keypad Lock"
+
+
+@pytest.fixture
+def coordinator_for_unlock_test(hass):
+    """Create a coordinator for testing _lock_unlocked method directly."""
+    with (
+        patch("custom_components.keymaster.coordinator.dr.async_get"),
+        patch("custom_components.keymaster.coordinator.er.async_get"),
+        patch("custom_components.keymaster.coordinator.Path"),
+    ):
+        coord = KeymasterCoordinator(hass)
+
+    # Mock throttle to always allow
+    coord._throttle = MagicMock()
+    coord._throttle.is_allowed.return_value = True
+
+    # Set initial setup done event so get_lock_by_config_entry_id doesn't block
+    coord._initial_setup_done_event.set()
+
+    return coord
+
+
+async def test_lock_unlocked_decrements_accesslimit_count(hass, coordinator_for_unlock_test):
+    """Test that accesslimit_count is decremented when lock is unlocked with a code slot.
+
+    When a lock is unlocked using a code slot that has accesslimit_count_enabled=True
+    and accesslimit_count > 0, the count should be decremented by 1.
+    """
+    coordinator = coordinator_for_unlock_test
+
+    # Create a lock with code slot having accesslimit enabled and count > 0
+    kmlock = KeymasterLock(
+        lock_name="test_lock",
+        lock_entity_id="lock.test_lock",
+        keymaster_config_entry_id="test_entry",
+    )
+    kmlock.lock_state = LockState.LOCKED  # Must be locked initially
+    kmlock.code_slots = {
+        1: KeymasterCodeSlot(
+            number=1,
+            enabled=True,
+            accesslimit_count_enabled=True,
+            accesslimit_count=5,  # Start with 5 uses remaining
+        )
+    }
+    coordinator.kmlocks["test_entry"] = kmlock
+
+    # Mock methods that would normally interact with HA
+    with (
+        patch.object(coordinator, "async_refresh", new=AsyncMock()),
+        patch.object(coordinator, "update_slot_active_state", new=AsyncMock()),
+        patch.object(coordinator, "clear_pin_from_lock", new=AsyncMock()),
+    ):
+        # Unlock with code slot 1
+        await coordinator._lock_unlocked(
+            kmlock=kmlock,
+            code_slot_num=1,
+            source="event",
+            event_label="Keypad Unlock",
+        )
+
+        # Verify the count was decremented from 5 to 4
+        assert kmlock.code_slots[1].accesslimit_count == 4
+        assert isinstance(kmlock.code_slots[1].accesslimit_count, int)
+
+
+async def test_lock_unlocked_decrements_parent_lock_accesslimit_count(
+    hass, coordinator_for_unlock_test
+):
+    """Test that parent lock's accesslimit_count is decremented for child locks.
+
+    When a child lock (with parent_name set) is unlocked using a code slot that
+    doesn't override the parent, the parent lock's accesslimit_count should be
+    decremented instead.
+    """
+    coordinator = coordinator_for_unlock_test
+
+    # Create parent lock
+    parent_kmlock = KeymasterLock(
+        lock_name="parent_lock",
+        lock_entity_id="lock.parent_lock",
+        keymaster_config_entry_id="parent_entry",
+    )
+    parent_kmlock.code_slots = {
+        1: KeymasterCodeSlot(
+            number=1,
+            enabled=True,
+            accesslimit_count_enabled=True,
+            accesslimit_count=10,  # Parent starts with 10
+        )
+    }
+    coordinator.kmlocks["parent_entry"] = parent_kmlock
+
+    # Create child lock that references parent
+    child_kmlock = KeymasterLock(
+        lock_name="child_lock",
+        lock_entity_id="lock.child_lock",
+        keymaster_config_entry_id="child_entry",
+    )
+    child_kmlock.lock_state = LockState.LOCKED
+    child_kmlock.parent_name = "parent_lock"
+    child_kmlock.parent_config_entry_id = "parent_entry"
+    child_kmlock.code_slots = {
+        1: KeymasterCodeSlot(
+            number=1,
+            enabled=True,
+            override_parent=False,  # NOT overriding parent
+            accesslimit_count_enabled=False,  # Child's own limit not enabled
+        )
+    }
+    coordinator.kmlocks["child_entry"] = child_kmlock
+
+    # Mock methods
+    with (
+        patch.object(coordinator, "async_refresh", new=AsyncMock()),
+        patch.object(coordinator, "update_slot_active_state", new=AsyncMock()),
+        patch.object(coordinator, "clear_pin_from_lock", new=AsyncMock()),
+    ):
+        # Unlock child lock with code slot 1
+        await coordinator._lock_unlocked(
+            kmlock=child_kmlock,
+            code_slot_num=1,
+            source="event",
+            event_label="Keypad Unlock",
+        )
+
+        # Verify parent's count was decremented from 10 to 9
+        assert parent_kmlock.code_slots[1].accesslimit_count == 9
+
+
+async def test_lock_unlocked_does_not_decrement_when_count_zero(hass, coordinator_for_unlock_test):
+    """Test that accesslimit_count is not decremented when already at 0."""
+    coordinator = coordinator_for_unlock_test
+
+    kmlock = KeymasterLock(
+        lock_name="test_lock",
+        lock_entity_id="lock.test_lock",
+        keymaster_config_entry_id="test_entry",
+    )
+    kmlock.lock_state = LockState.LOCKED
+    kmlock.code_slots = {
+        1: KeymasterCodeSlot(
+            number=1,
+            enabled=True,
+            accesslimit_count_enabled=True,
+            accesslimit_count=0,  # Already at 0
+        )
+    }
+    coordinator.kmlocks["test_entry"] = kmlock
+
+    with (
+        patch.object(coordinator, "async_refresh", new=AsyncMock()),
+        patch.object(coordinator, "update_slot_active_state", new=AsyncMock()),
+        patch.object(coordinator, "clear_pin_from_lock", new=AsyncMock()),
+    ):
+        await coordinator._lock_unlocked(
+            kmlock=kmlock,
+            code_slot_num=1,
+            source="event",
+            event_label="Keypad Unlock",
+        )
+
+        # Count should remain at 0 (not go negative)
+        assert kmlock.code_slots[1].accesslimit_count == 0