Skip to content

Fix Z-Wave JS sync loop from stale AVAILABLE events #863

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
raman325 merged 1 commit into from
Feb 5, 2026
Merged

Fix Z-Wave JS sync loop from stale AVAILABLE events #863

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions custom_components/lock_code_manager/coordinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,9 +68,13 @@
if not updates:
return

# Merge updates into a new dict to ensure proper change detection
# (avoids passing the same object reference to listeners)
self.async_set_updated_data({**self.data, **updates})
new_data = {**self.data, **updates}
# Skip update if data hasn't actually changed to avoid redundant logging
# and unnecessary listener notifications
if new_data == self.data:
return

Check warning on line 75 in custom_components/lock_code_manager/coordinator.py

View check run for this annotation

Codecov / codecov/patch

custom_components/lock_code_manager/coordinator.py#L75 

Added line #L75 was not covered by tests

self.async_set_updated_data(new_data)

async def async_get_usercodes(self) -> dict[int, int | str]:
"""Update usercodes."""
Expand Down
88 changes: 65 additions & 23 deletions custom_components/lock_code_manager/providers/zwave_js.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,26 +145,13 @@ def code_slot_in_use(self, code_slot: int) -> bool | None:
except (KeyError, ValueError):
return None

def _resolve_pin_if_masked(self, value: str, code_slot: int) -> str | None:
"""Resolve a PIN value, looking up expected PIN if masked.

Some locks return masked values (all asterisks) instead of the actual PIN.
This method returns the value as-is if not masked, or looks up the expected
PIN from LCM entities if masked.

def _get_slot_entity_states(self, code_slot: int) -> tuple[str, str, str] | None:
"""
slot_in_use = self.code_slot_in_use(code_slot)
Get active state, PIN value, and PIN entity ID for a managed slot.

if value == "0" * len(value) and slot_in_use is False:
# Some locks return all zeros instead of a blank value when cleared - treat
# as unmasked cleared value
return ""

# If not masked, return as-is
if not value or not (value == "*" * len(value) and slot_in_use):
return value

# Masked - look up expected PIN from LCM entities
Returns tuple of (active_state, pin_value, pin_entity_id) if the slot is
managed by LCM and entities exist, None otherwise.
"""
try:
config_entry = next(
entry
Expand All @@ -175,9 +162,6 @@ def _resolve_pin_if_masked(self, value: str, code_slot: int) -> str | None:
except StopIteration:
return None

# Use ATTR_ACTIVE binary sensor (not CONF_ENABLED switch) to match what
# sync logic expects. ATTR_ACTIVE considers conditions, so we only resolve
# masked codes when sync expects a PIN on the lock.
base_unique_id = f"{config_entry.entry_id}|{code_slot}"
active_entity_id = self.ent_reg.async_get_entity_id(
BINARY_SENSOR_DOMAIN, DOMAIN, f"{base_unique_id}|{ATTR_ACTIVE}"
Expand All @@ -195,14 +179,60 @@ def _resolve_pin_if_masked(self, value: str, code_slot: int) -> str | None:
if not active_state or not pin_state:
return None

if active_state.state == STATE_ON:
return (active_state.state, pin_state.state, pin_entity_id)

def _slot_expects_pin(self, code_slot: int) -> bool:
"""
Check if LCM expects a PIN on this slot (active=ON with PIN set).

Used to ignore stale userIdStatus=AVAILABLE events from locks that report
old status after a code was successfully set.
"""
states = self._get_slot_entity_states(code_slot)
if states is None:
return False

active_state, pin_value, _ = states
return active_state == STATE_ON and bool(pin_value)

def _resolve_pin_if_masked(self, value: str, code_slot: int) -> str | None:
"""
Resolve a PIN value, looking up expected PIN if masked.

Some locks return masked values (all asterisks) instead of the actual PIN.
This method returns the value as-is if not masked, or looks up the expected
PIN from LCM entities if masked.

"""
slot_in_use = self.code_slot_in_use(code_slot)

if value == "0" * len(value) and slot_in_use is False:
# Some locks return all zeros instead of a blank value when cleared - treat
# as unmasked cleared value
return ""

# If not masked, return as-is
if not value or not (value == "*" * len(value) and slot_in_use):
return value

# Masked - look up expected PIN from LCM entities
# Use ATTR_ACTIVE binary sensor (not CONF_ENABLED switch) to match what
# sync logic expects. ATTR_ACTIVE considers conditions, so we only resolve
# masked codes when sync expects a PIN on the lock.
states = self._get_slot_entity_states(code_slot)
if states is None:
return None

active_state, pin_value, pin_entity_id = states

if active_state == STATE_ON:
_LOGGER.debug(
"PIN is masked for lock %s code slot %s, assuming value from PIN entity %s",
self.lock.entity_id,
code_slot,
pin_entity_id,
)
return pin_state.state
return pin_value

# Fall back to returning masked value if active state is not ON (e.g. slot not
# enabled) - we don't care what the value is, just that there is one so that
Expand Down Expand Up @@ -251,6 +281,18 @@ def on_value_updated(event: dict[str, Any]) -> None:
if property_name == LOCK_USERCODE_STATUS_PROPERTY:
status = args.get("newValue")
if status == CodeSlotStatus.AVAILABLE:
# Ignore AVAILABLE status if LCM expects a PIN on this slot.
# Some locks send stale AVAILABLE events after a code was set,
# which would cause infinite sync loops.
if self._slot_expects_pin(code_slot):
_LOGGER.debug(
"Lock %s: ignoring userIdStatus=AVAILABLE for slot %s "
"(LCM expects PIN on this slot)",
self.lock.entity_id,
code_slot,
)
return

# Slot was cleared - update coordinator if needed
if self.coordinator and self.coordinator.data.get(code_slot) != "":
_LOGGER.debug(
Expand Down
217 changes: 217 additions & 0 deletions tests/providers/test_zwave_js.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1529,3 +1529,220 @@ async def test_resolve_pin_if_masked_all_zeros_slot_unknown(
# All zeros with unknown slot status → returned as-is
result = zwave_js_lock._resolve_pin_if_masked("0000", 1)
assert result == "0000"


# _slot_expects_pin tests


async def test_slot_expects_pin_returns_true_when_active_with_pin(
hass: HomeAssistant,
zwave_js_lock: ZWaveJSLock,
zwave_integration: MockConfigEntry,
) -> None:
"""Test _slot_expects_pin returns True when active=ON and PIN is set."""
lcm_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCKS: [zwave_js_lock.lock.entity_id],
CONF_SLOTS: {"2": {}},
},
)
lcm_entry.add_to_hass(hass)
await zwave_js_lock.async_setup(lcm_entry)

# Create the active and PIN entities
ent_reg = er.async_get(hass)
base_unique_id = f"{lcm_entry.entry_id}|2"

active_entry = ent_reg.async_get_or_create(
"binary_sensor",
DOMAIN,
f"{base_unique_id}|active",
config_entry=lcm_entry,
)
pin_entry = ent_reg.async_get_or_create(
"text",
DOMAIN,
f"{base_unique_id}|pin",
config_entry=lcm_entry,
)

# Set states: active=ON, pin="1234"
hass.states.async_set(active_entry.entity_id, "on")
hass.states.async_set(pin_entry.entity_id, "1234")
await hass.async_block_till_done()

assert zwave_js_lock._slot_expects_pin(2) is True

await zwave_js_lock.async_unload(False)


async def test_slot_expects_pin_returns_false_when_inactive(
hass: HomeAssistant,
zwave_js_lock: ZWaveJSLock,
zwave_integration: MockConfigEntry,
) -> None:
"""Test _slot_expects_pin returns False when active=OFF."""
lcm_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCKS: [zwave_js_lock.lock.entity_id],
CONF_SLOTS: {"2": {}},
},
)
lcm_entry.add_to_hass(hass)
await zwave_js_lock.async_setup(lcm_entry)

# Create the active and PIN entities
ent_reg = er.async_get(hass)
base_unique_id = f"{lcm_entry.entry_id}|2"

active_entry = ent_reg.async_get_or_create(
"binary_sensor",
DOMAIN,
f"{base_unique_id}|active",
config_entry=lcm_entry,
)
pin_entry = ent_reg.async_get_or_create(
"text",
DOMAIN,
f"{base_unique_id}|pin",
config_entry=lcm_entry,
)

# Set states: active=OFF, pin="1234"
hass.states.async_set(active_entry.entity_id, "off")
hass.states.async_set(pin_entry.entity_id, "1234")
await hass.async_block_till_done()

assert zwave_js_lock._slot_expects_pin(2) is False

await zwave_js_lock.async_unload(False)


async def test_slot_expects_pin_returns_false_for_unmanaged_slot(
hass: HomeAssistant,
zwave_js_lock: ZWaveJSLock,
zwave_integration: MockConfigEntry,
) -> None:
"""Test _slot_expects_pin returns False for unmanaged slot."""
lcm_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCKS: [zwave_js_lock.lock.entity_id],
CONF_SLOTS: {"1": {}}, # Only slot 1 managed
},
)
lcm_entry.add_to_hass(hass)
await zwave_js_lock.async_setup(lcm_entry)

# Slot 99 is not managed
assert zwave_js_lock._slot_expects_pin(99) is False

await zwave_js_lock.async_unload(False)


async def test_push_update_user_id_status_available_ignored_when_slot_expects_pin(
hass: HomeAssistant,
zwave_js_lock: ZWaveJSLock,
zwave_integration: MockConfigEntry,
lock_schlage_be469: Node,
) -> None:
"""Test that userIdStatus=AVAILABLE is ignored when slot expects a PIN.

This prevents sync loops where the lock sends stale AVAILABLE status
after a code was successfully set.
"""
lcm_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCKS: [zwave_js_lock.lock.entity_id],
CONF_SLOTS: {"2": {}},
},
)
lcm_entry.add_to_hass(hass)
await zwave_js_lock.async_setup(lcm_entry)

# Set up a mock coordinator with existing PIN
mock_coordinator = MagicMock()
mock_coordinator.data = {2: "1234"} # Slot has a PIN
zwave_js_lock.coordinator = mock_coordinator

# Subscribe to push updates
zwave_js_lock.subscribe_push_updates()

# Mock _slot_expects_pin to return True (LCM expects a PIN on this slot)
with patch.object(zwave_js_lock, "_slot_expects_pin", return_value=True):
# Simulate userIdStatus=AVAILABLE event (stale status from lock)
event = ZwaveEvent(
type="value updated",
data={
"args": {
"commandClass": CommandClass.USER_CODE,
"property": LOCK_USERCODE_STATUS_PROPERTY,
"propertyKey": 2,
"newValue": CodeSlotStatus.AVAILABLE,
},
},
)
lock_schlage_be469.emit("value updated", event.data)
await hass.async_block_till_done()

# Coordinator should NOT be updated (AVAILABLE ignored)
mock_coordinator.push_update.assert_not_called()

zwave_js_lock.unsubscribe_push_updates()
await zwave_js_lock.async_unload(False)


async def test_push_update_user_id_status_available_clears_when_slot_inactive(
hass: HomeAssistant,
zwave_js_lock: ZWaveJSLock,
zwave_integration: MockConfigEntry,
lock_schlage_be469: Node,
) -> None:
"""Test that userIdStatus=AVAILABLE clears slot when LCM doesn't expect a PIN.

When the slot is inactive (active=OFF), AVAILABLE status should clear
the coordinator as expected.
"""
lcm_entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOCKS: [zwave_js_lock.lock.entity_id],
CONF_SLOTS: {"2": {}},
},
)
lcm_entry.add_to_hass(hass)
await zwave_js_lock.async_setup(lcm_entry)

# Set up a mock coordinator with existing PIN
mock_coordinator = MagicMock()
mock_coordinator.data = {2: "1234"} # Slot has a PIN
zwave_js_lock.coordinator = mock_coordinator

# Subscribe to push updates
zwave_js_lock.subscribe_push_updates()

# Mock _slot_expects_pin to return False (slot is inactive)
with patch.object(zwave_js_lock, "_slot_expects_pin", return_value=False):
# Simulate userIdStatus=AVAILABLE event
event = ZwaveEvent(
type="value updated",
data={
"args": {
"commandClass": CommandClass.USER_CODE,
"property": LOCK_USERCODE_STATUS_PROPERTY,
"propertyKey": 2,
"newValue": CodeSlotStatus.AVAILABLE,
},
},
)
lock_schlage_be469.emit("value updated", event.data)
await hass.async_block_till_done()

# Coordinator SHOULD be updated (slot cleared)
mock_coordinator.push_update.assert_called_once_with({2: ""})

zwave_js_lock.unsubscribe_push_updates()
await zwave_js_lock.async_unload(False)
Loading