dilim/homeassistant-rental-control
1
0 Fork 0
mirror of https://github.com/tykeal/homeassistant-rental-control.git synced 2026-11-06 05:17:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert: Restore 3.4.6 slot management behavior

Browse Source 
Revert the 3.5.0 slot-reconciliation redesign (PRs #598,
#599, #600). At real-world scale the first-upgrade adoption
failed to rematch existing slots, mass-cleared working lock
codes, and then permanently wedged (all slots stuck
pending_clear, "selected but no free managed slot
available", persisted across restarts). Releases are
immutable, so restore the proven 3.4.6 behavior for 3.5.1.
The #571 explicit-entry-guards and #572 module
decomposition refactors are intentionally preserved. The
reconciliation redesign will be re-engineered separately
with real-scale upgrade/clear-confirmation tests.

Co-authored-by: Claude <claude@anthropic.com>
Signed-off-by: Andrew Grimberg <tykeal@bardicgrove.org>
This commit is contained in:
Andrew GrimbergandClaude committed 2026-06-20 16:21:49 -07:00
1 parent ecda8cdc91
commit 06220576d0
17 files changed
+505 -17642
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

custom_components/rental_control/__init__.py
-17
View File
@@ -72,23 +72,6 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
config_entry=config_entry,
)
# Load Store before Keymaster bootstrap (ordering fix for #597)
await coordinator.async_load_slot_store()
# Inject persisted mappings or adopt slots on first upgrade
persisted = coordinator.get_persisted_slot_mappings()
if persisted:
if coordinator.event_overrides is not None:
try:
coordinator.event_overrides.load_persisted_mappings(persisted)
except ValueError:
_LOGGER.exception(
"Failed to load persisted slot mappings for %s",
config_entry.entry_id,
)
elif coordinator.lockname:
await coordinator.async_adopt_keymaster_slots()
# Bootstrap Keymaster slot overrides from current HA state before
# first refresh so overrides are checked against the initial data
await coordinator.async_setup_keymaster_overrides()
custom_components/rental_control/const.py
-14
View File
@@ -134,17 +134,3 @@ CHECKIN_STATE_CHECKED_OUT = "checked_out"
# Early checkout grace period in minutes
EARLY_CHECKOUT_GRACE_MINUTES = 15
# Store constants
STORE_SLOT_MAPPINGS_KEY = "rental_control.slot_mappings"
STORE_SCHEMA_VERSION = 1
# Slot status string constants (mirror SlotStatus enum values)
SLOT_STATUS_OCCUPIED = "occupied"
SLOT_STATUS_PENDING_SET = "pending_set"
SLOT_STATUS_PENDING_CLEAR = "pending_clear"
SLOT_STATUS_BLOCKED = "blocked"
# Operation kind constants
OPERATION_KIND_SET = "set"
OPERATION_KIND_CLEAR = "clear"
custom_components/rental_control/coordinator.py
+3 -1152
View File
File diff suppressed because it is too large. Load diff
custom_components/rental_control/event_overrides.py
+9 -775
View File
@@ -20,27 +20,15 @@ from datetime import date
from datetime import datetime
from datetime import time
import logging
from time import monotonic
from typing import TYPE_CHECKING
from typing import Any
from typing import NamedTuple
from typing import TypedDict
import uuid
from homeassistant.util import dt
from .const import DEFAULT_MAX_RETRY_CYCLES
from .const import SLOT_STATUS_OCCUPIED
from .const import SLOT_STATUS_PENDING_CLEAR
from .reconciliation import ActionKind
from .reconciliation import DesiredPlan
from .reconciliation import Reservation
from .reconciliation import SlotAction
from .util import EventIdentity
from .util import OperationResult
from .util import async_fire_clear_code
from .util import async_fire_set_code
from .util import async_fire_update_times
from .util import get_event_identities
from .util import normalize_uid
from .util import trim_name
@@ -51,7 +39,6 @@ if TYPE_CHECKING:
_LOGGER = logging.getLogger(__name__)
SLOT_MISS_THRESHOLD = 2
SUPPRESSED_STATE_CHANGE_TTL = 5.0
def _to_utc(value: datetime) -> datetime:
@@ -75,18 +62,6 @@ def _to_utc(value: datetime) -> datetime:
return utc
def _states_match(expected: str, actual: str) -> bool:
"""Compare raw HA state strings, normalizing datetimes when possible."""
if expected == actual:
return True
expected_dt = dt.parse_datetime(expected)
actual_dt = dt.parse_datetime(actual)
if expected_dt is not None and actual_dt is not None:
return _to_utc(expected_dt) == _to_utc(actual_dt)
return False
def _strip_prefix(slot_name: str, prefix: str) -> str:
"""Remove a leading prefix and space from slot_name.
@@ -129,25 +104,6 @@ class EventOverride(TypedDict):
end_time: datetime
class _SlotEvent:
"""Minimal event adapter for util slot-operation helpers."""
def __init__(
self,
slot_name: str,
slot_code: str,
start: datetime,
end: datetime,
) -> None:
"""Populate the attributes expected by util service helpers."""
self.extra_state_attributes: dict[str, Any] = {
"slot_name": slot_name,
"slot_code": slot_code,
"start": start,
"end": end,
}
class EventOverrides:
"""Event Overrides object and methods."""
@@ -168,20 +124,6 @@ class EventOverrides:
self._max_name_length: int = 0
self._prefix_length: int = 0
# Store-backed fields (T018)
self._persisted_mappings: dict[str, dict[str, Any]] = {}
self._pending_clear_slots: dict[int, str] = {}
self._pending_fences: dict[int, str] = {}
self._actual_state_cache: dict[int, dict[str, Any]] = {}
self._reconciliation_active: bool = False
# Per-slot error tracking for diagnostics (T096)
self._last_slot_errors: dict[int, str] = {}
# Coordinator-originated state feedback awaiting callback suppression
self._suppressed_state_changes: dict[int, dict[str, tuple[str, float]]] = {}
# Latest diagnostics snapshot for HA diagnostics collection (T096)
self._diagnostics_snapshot: dict[str, Any] = {}
@property
def max_slots(self) -> int:
"""Return the max_slots known."""
@@ -189,16 +131,7 @@ class EventOverrides:
@property
def next_slot(self) -> int | None:
"""Return the next available slot index for the greedy path.
.. deprecated::
This property is part of the retired greedy slot-assignment path.
The reconciliation coordinator (``compute_desired_plan`` /
``async_apply_plan``) owns slot selection and does not consult
``_next_slot``. The property is retained as a backward-compatible
shim for tests that exercise :meth:`async_reserve_or_get_slot`
directly; production code must not read it.
"""
"""Return the next_slot available."""
return self._next_slot
@property
@@ -246,238 +179,8 @@ class EventOverrides:
"""Set the event prefix length including separator."""
self._prefix_length = value
@property
def persisted_mappings(self) -> dict[str, dict[str, Any]]:
"""Return a read-only copy of the persisted slot mappings."""
return dict(self._persisted_mappings)
@property
def pending_clear_slots(self) -> dict[int, str]:
"""Return a read-only copy of the pending-clear slot map."""
return dict(self._pending_clear_slots)
@property
def pending_fences(self) -> dict[int, str]:
"""Return a read-only copy of pending operation fences."""
return dict(self._pending_fences)
@property
def reconciliation_active(self) -> bool:
"""True while async_apply_plan is executing."""
return self._reconciliation_active
def suppress_state_changes(self, slot: int, changes: dict[str, Any]) -> None:
"""Mark coordinator-originated state changes to ignore in callbacks."""
now = monotonic()
pending = self._suppressed_state_changes.setdefault(slot, {})
for entity_id, value in changes.items():
pending[entity_id] = (str(value), now)
def should_suppress_state_change(
self, slot: int, entity_id: str, state: str
) -> bool:
"""Return True when a callback is feedback from our own service call."""
pending = self._suppressed_state_changes.get(slot)
if not pending:
return False
now = monotonic()
for pending_entity_id, (_, created) in list(pending.items()):
if now - created > SUPPRESSED_STATE_CHANGE_TTL:
pending.pop(pending_entity_id, None)
expected = pending.get(entity_id)
if expected is None:
if not pending:
self._suppressed_state_changes.pop(slot, None)
return False
expected_state, _ = expected
if _states_match(expected_state, state):
pending.pop(entity_id, None)
if not pending:
self._suppressed_state_changes.pop(slot, None)
return True
return False
@property
def diagnostics_snapshot(self) -> dict[str, Any]:
"""Return the latest diagnostics snapshot for HA diagnostics collection.
The snapshot is built after each :meth:`async_apply_plan` call and
captures matched slots, pending corrections, blocked clear reasons,
retry counts, and last errors. Raw slot codes are never included.
Returns:
A shallow copy of the current diagnostics snapshot dict, or an
empty dict if no plan has been applied yet.
"""
return dict(self._diagnostics_snapshot)
def get_last_slot_error(self, slot: int) -> str | None:
"""Return the last error string for *slot*, or ``None`` if no error.
Args:
slot: Keymaster slot number.
Returns:
The last recorded error string for the slot, or ``None``.
"""
return self._last_slot_errors.get(slot)
def _record_slot_error(self, slot: int, error: str) -> None:
"""Record a failed operation error for *slot*.
Args:
slot: Keymaster slot number.
error: Human-readable error description.
"""
self._last_slot_errors[slot] = error
def _clear_slot_error(self, slot: int) -> None:
"""Clear the recorded error for *slot* on a successful operation.
Args:
slot: Keymaster slot number.
"""
self._last_slot_errors.pop(slot, None)
def update_diagnostics_snapshot(self, plan: "DesiredPlan") -> None:
"""Build and store a diagnostics snapshot from the completed plan.
Called after :meth:`async_apply_plan` completes. Captures matched
slots (those with desired assignments), pending corrections (slots
with ``retry_clear`` or ``blocked`` actions), manual drift slots
(those with ``overwrite_manual_change`` actions), blocked clear
reasons, per-slot retry counts, and last errors. Raw slot codes
are never included.
Args:
plan: The :class:`~.reconciliation.DesiredPlan` that was just applied.
"""
matched: dict[int, dict[str, Any]] = {}
pending_corrections: dict[int, dict[str, Any]] = {}
manual_drift_slots: dict[int, dict[str, Any]] = {}
for slot_num, ps in plan.slots.items():
if ps.desired_identity_key is not None:
matched[slot_num] = {
"identity_key": ps.desired_identity_key,
"action": ps.action.value,
}
if ps.action.value in (
ActionKind.RETRY_CLEAR.value,
ActionKind.BLOCKED.value,
):
pending_corrections[slot_num] = {
"action": ps.action.value,
"blocked_reason": ps.pending_reason,
"retry_count": ps.retry_count,
}
if ps.action is ActionKind.OVERWRITE_MANUAL_CHANGE:
drift_fields: list[str] = []
if ps.pending_reason and ps.pending_reason.startswith(
"drifted fields: "
):
drift_fields = [
f.strip()
for f in ps.pending_reason[len("drifted fields: ") :].split(",")
if f.strip()
]
manual_drift_slots[slot_num] = {
"action": ps.action.value,
"identity_key": ps.desired_identity_key,
"drift_fields": drift_fields,
}
self._diagnostics_snapshot = {
"plan_id": plan.plan_id,
"generated_at": plan.generated_at.isoformat(),
"matched_slots": matched,
"pending_corrections": pending_corrections,
"manual_drift_slots": manual_drift_slots,
"pending_clear_slots": sorted(self._pending_clear_slots.keys()),
"slot_retry_counts": {
slot: self._retry_counts.get(slot, 0)
for slot in range(self._start_slot, self._start_slot + self._max_slots)
},
"last_slot_errors": dict(self._last_slot_errors),
}
def load_persisted_mappings(self, mappings: dict[str, dict[str, Any]]) -> None:
"""Load persisted slot mappings from the HA Store.
Validates that no two mappings both claim the same slot with
status ``occupied``; raises ``ValueError`` on conflict.
Args:
mappings: Identity-key → mapping dict from the HA Store.
Raises:
ValueError: If two mappings claim the same slot and both
have ``occupied`` status.
"""
slot_owners: dict[int, str] = {}
for identity_key, mapping in mappings.items():
slot = mapping.get("slot")
status = mapping.get("status")
if slot is not None and status == SLOT_STATUS_OCCUPIED:
if slot in slot_owners:
raise ValueError(
f"Duplicate occupied slot {slot}: claimed by both "
f"{slot_owners[slot]!r} and {identity_key!r}"
)
slot_owners[slot] = identity_key
self._persisted_mappings = {k: dict(v) for k, v in mappings.items()}
self._pending_clear_slots = {}
for identity_key, mapping in mappings.items():
if mapping.get("status") == SLOT_STATUS_PENDING_CLEAR:
slot = mapping.get("slot")
if slot is not None:
operation_id = mapping.get("operation_id")
self._pending_clear_slots[slot] = (
operation_id if operation_id is not None else identity_key
)
def update_actual_state(self, slot: int, state: dict[str, Any]) -> None:
"""Store the observed Keymaster state snapshot for a slot.
Args:
slot: Keymaster slot number.
state: Dict capturing the observed entity states for the slot.
"""
self._actual_state_cache[slot] = state
def get_actual_state(self, slot: int) -> dict[str, Any] | None:
"""Return the cached actual state for a slot, or None.
Args:
slot: Keymaster slot number.
Returns:
The cached state dict, or ``None`` if not yet observed.
"""
return self._actual_state_cache.get(slot)
def __assign_next_slot(self) -> None:
"""Recompute ``_next_slot`` for the deprecated greedy path.
Called only from the greedy-path methods (:meth:`update`,
:meth:`async_update`, :meth:`async_reserve_or_get_slot`,
:meth:`async_check_overrides`) so that :attr:`next_slot` stays
accurate for callers that still use the legacy API.
Reconciliation methods (:meth:`_apply_clear`, :meth:`_apply_set`,
:meth:`_apply_overwrite_manual_change`) do *not* call this helper;
slot selection in the reconciliation path is determined by
``compute_desired_plan``, not by ``_next_slot``.
.. deprecated::
Do not call from new code. Will be removed when the greedy
path shims are retired.
"""
"""Assign the next slot."""
_LOGGER.debug("In EventOverrides.assign_next_slot")
@@ -707,19 +410,10 @@ class EventOverrides:
uid: str | None = None,
prefix: str | None = None,
) -> ReserveResult:
"""Atomically find existing slot or reserve next available (greedy path).
"""Atomically find existing slot or reserve next available.
All work is performed under ``_lock`` so concurrent callers
are serialised.
.. deprecated::
This method is the retired greedy slot-assignment path. The
reconciliation coordinator (``compute_desired_plan`` /
``async_apply_plan``) is now the sole authority for slot
assignments and does not call this method. The method is
retained as a backward-compatible shim for tests that exercise
the greedy path directly; it must not be called from production
coordinator or sensor code.
"""
async with self._lock:
if prefix is None:
@@ -846,430 +540,6 @@ class EventOverrides:
self._retry_counts[slot] = 0
self._escalated[slot] = False
async def async_apply_plan(
self,
coordinator: Any,
plan: DesiredPlan,
res_by_key: dict[str, Reservation],
) -> list[OperationResult]:
"""Apply a desired plan by executing slot actions."""
async with self._lock:
self._reconciliation_active = True
results: list[OperationResult] = []
try:
for action in plan.actions:
if action.kind in {ActionKind.NOOP, ActionKind.BLOCKED}:
continue
slot = action.slot
identity_key = action.identity_key
if action.kind in {ActionKind.CLEAR, ActionKind.RETRY_CLEAR}:
_clear_reason = action.reason
if _clear_reason == "duplicate_non_canonical":
_LOGGER.warning(
"Duplicate collapse: clearing non-canonical slot %d "
"(duplicate actual assignment)",
slot,
)
elif _clear_reason == "phantom":
_LOGGER.warning(
"Phantom recovery: clearing slot %d "
"(name-only state, no usable PIN)",
slot,
)
elif _clear_reason == "stale":
_LOGGER.warning(
"Stale correction: clearing slot %d "
"(expired or absent reservation)",
slot,
)
elif _clear_reason == "mis_assigned":
_LOGGER.warning(
"Mis-assignment correction: clearing slot %d "
"(wrong reservation in slot)",
slot,
)
result = await self._apply_clear(coordinator, slot)
elif action.kind is ActionKind.SET:
res = res_by_key.get(identity_key) if identity_key else None
if res is None:
_LOGGER.warning(
"SET action for slot %d has no reservation; skipping", slot
)
continue
result = await self._apply_set(coordinator, slot, res, plan.plan_id)
elif action.kind is ActionKind.UPDATE_TIMES:
res = res_by_key.get(identity_key) if identity_key else None
if res is None:
_LOGGER.warning(
"UPDATE_TIMES action for slot %d has no reservation; "
"skipping",
slot,
)
continue
result = await self._apply_update_times(coordinator, slot, res)
elif action.kind is ActionKind.OVERWRITE_MANUAL_CHANGE:
res = res_by_key.get(identity_key) if identity_key else None
if res is None:
_LOGGER.warning(
"OVERWRITE_MANUAL_CHANGE action for slot %d has no "
"reservation; skipping",
slot,
)
continue
result = await self._apply_overwrite_manual_change(
coordinator, slot, res, action
)
else:
continue
results.append(result)
finally:
self.update_diagnostics_snapshot(plan)
async with self._lock:
self._reconciliation_active = False
return results
async def _apply_clear(
self,
coordinator: Any,
slot: int,
) -> OperationResult:
"""Apply a CLEAR or RETRY_CLEAR action for one slot."""
operation_id = str(uuid.uuid4())
expected_name: str | None = None
async with self._lock:
self._pending_fences[slot] = operation_id
self._pending_clear_slots[slot] = operation_id
override = self._overrides.get(slot)
if override is not None:
expected_name = override.get("slot_name")
result = await async_fire_clear_code(
coordinator, slot, expected_name=expected_name
)
async with self._lock:
current_token = self._pending_fences.get(slot)
if current_token != operation_id:
_LOGGER.warning(
"Stale clear token for slot %d "
"(expected %s, got %s); discarding result",
slot,
operation_id,
current_token,
)
return OperationResult(kind="clear", slot=slot, unconfirmed=True)
if result.confirmed:
_LOGGER.debug("Clear confirmed for slot %d; marking free", slot)
self._pending_fences.pop(slot, None)
self._pending_clear_slots.pop(slot, None)
self._overrides[slot] = None
self._slot_uids.pop(slot, None)
self._slot_miss_counts.pop(slot, None)
self._clear_slot_error(slot)
# NOTE: __assign_next_slot() is intentionally NOT called here.
# Slot selection in the reconciliation path is determined by
# compute_desired_plan(), not by the deprecated _next_slot field.
elif result.failed:
_LOGGER.warning(
"Clear failed for slot %d (error: %s); slot remains pending-clear",
slot,
result.error,
)
self._record_slot_error(slot, result.error or "clear failed")
elif result.lingering_name or result.lingering_pin:
_LOGGER.warning(
"Clear not fully confirmed for slot %d "
"(lingering_name=%s, lingering_pin=%s); "
"slot remains pending-clear",
slot,
result.lingering_name,
result.lingering_pin,
)
self._record_slot_error(
slot,
f"lingering state after clear: "
f"name={result.lingering_name} pin={result.lingering_pin}",
)
else:
_LOGGER.debug(
"Clear unconfirmed for slot %d; slot remains pending-clear", slot
)
return result
async def _apply_set(
self,
coordinator: Any,
slot: int,
res: Reservation,
plan_id: str,
) -> OperationResult:
"""Apply a SET action for one slot."""
import hashlib as _hashlib
operation_id = (
f"{plan_id}-set-{slot}-"
f"{_hashlib.sha256(res.identity_key.encode()).hexdigest()[:8]}"
)
async with self._lock:
self._pending_fences[slot] = operation_id
self._overrides[slot] = {
"slot_name": res.slot_name,
"slot_code": res.slot_code,
"start_time": res.buffered_start,
"end_time": res.buffered_end,
}
self._slot_miss_counts.pop(slot, None)
self.suppress_state_changes(
slot,
{
f"switch.{coordinator.lockname}_code_slot_"
f"{slot}_use_date_range_limits": "on",
f"text.{coordinator.lockname}_code_slot_{slot}_name": (
res.display_slot_name
),
f"text.{coordinator.lockname}_code_slot_{slot}_pin": res.slot_code,
f"datetime.{coordinator.lockname}_code_slot_"
f"{slot}_date_range_start": res.buffered_start.isoformat(),
f"datetime.{coordinator.lockname}_code_slot_"
f"{slot}_date_range_end": res.buffered_end.isoformat(),
},
)
event = _SlotEvent(
slot_name=res.slot_name,
slot_code=res.slot_code,
start=res.start,
end=res.end,
)
result = await async_fire_set_code(coordinator, event, slot)
async with self._lock:
current_token = self._pending_fences.get(slot)
if current_token != operation_id:
_LOGGER.warning("Stale set token for slot %d; discarding result", slot)
return OperationResult(kind="set", slot=slot, unconfirmed=True)
if result.confirmed:
_LOGGER.debug(
"Set confirmed for slot %d for reservation %s",
slot,
res.identity_key,
)
self._pending_fences.pop(slot, None)
self._clear_slot_error(slot)
# NOTE: __assign_next_slot() is intentionally NOT called here.
# Slot selection in the reconciliation path is determined by
# compute_desired_plan(), not by the deprecated _next_slot field.
elif result.failed:
_LOGGER.warning(
"Set failed for slot %d (error: %s); reverting pre-assignment",
slot,
result.error,
)
self._pending_fences.pop(slot, None)
self._overrides[slot] = None
self._slot_uids.pop(slot, None)
self._record_slot_error(slot, result.error or "set failed")
# NOTE: __assign_next_slot() is intentionally NOT called here.
# Slot selection in the reconciliation path is determined by
# compute_desired_plan(), not by the deprecated _next_slot field.
else:
_LOGGER.debug(
"Set unconfirmed for slot %d; keeping tentative assignment", slot
)
self._pending_fences.pop(slot, None)
return result
async def _apply_update_times(
self,
coordinator: Any,
slot: int,
res: Reservation,
) -> OperationResult:
"""Apply an UPDATE_TIMES action for one slot."""
async with self._lock:
self.suppress_state_changes(
slot,
{
f"datetime.{coordinator.lockname}_code_slot_"
f"{slot}_date_range_start": res.buffered_start.isoformat(),
f"datetime.{coordinator.lockname}_code_slot_"
f"{slot}_date_range_end": res.buffered_end.isoformat(),
},
)
event = _SlotEvent(
slot_name=res.slot_name,
slot_code=res.slot_code,
start=res.start,
end=res.end,
)
result = await async_fire_update_times(coordinator, event, slot)
if result.confirmed:
async with self._lock:
override = self._overrides.get(slot)
if override is not None:
override["start_time"] = res.buffered_start
override["end_time"] = res.buffered_end
_LOGGER.debug(
"update_times confirmed for slot %d; in-memory dates updated",
slot,
)
return result
async def _apply_overwrite_manual_change(
self,
coordinator: Any,
slot: int,
res: Reservation,
action: "SlotAction",
) -> OperationResult:
"""Apply an OVERWRITE_MANUAL_CHANGE action for one slot.
Logs a warning with all drift details — slot number, changed field
names, desired reservation identity, observed name and
classification from the actual-state cache — and then restores the
desired state via :func:`~.util.async_fire_set_code`. Raw PIN
values are never written to logs; only code *presence* (boolean) is
reported.
Unmanaged slots never trigger this path because
:func:`~.reconciliation.compute_desired_plan` iterates only over
managed slots.
Args:
coordinator: The active coordinator instance.
slot: Keymaster slot number to overwrite.
res: Desired :class:`~.reconciliation.Reservation` for this slot.
action: The :class:`~.reconciliation.SlotAction` carrying the
``OVERWRITE_MANUAL_CHANGE`` kind and drift reason string.
Returns:
An :class:`~.util.OperationResult` from the underlying
:func:`~.util.async_fire_set_code` call.
"""
import hashlib as _hashlib
drift_fields: list[str] = []
if action.reason and action.reason.startswith("drifted fields: "):
drift_fields = [
f.strip()
for f in action.reason[len("drifted fields: ") :].split(",")
if f.strip()
]
# Gather observed state for the log message. The actual-state cache
# is populated by the coordinator before async_apply_plan is called.
# Raw PIN values are never stored in the cache; only has_code (bool).
actual = self._actual_state_cache.get(slot) or {}
observed_name: str = actual.get("name_state") or "(unknown)"
observed_classification: str = actual.get("classification") or "(unknown)"
observed_has_code: bool | None = actual.get("has_code")
_LOGGER.warning(
"Manual/external drift detected on managed slot %d "
"(reservation %s, desired name %r): "
"changed fields=%s, "
"observed name=%r, observed classification=%s, "
"observed has_code=%s; "
"restoring desired state.",
slot,
res.identity_key,
res.display_slot_name,
drift_fields,
observed_name,
observed_classification,
observed_has_code,
)
operation_id = (
f"overwrite-{slot}-"
f"{_hashlib.sha256(res.identity_key.encode()).hexdigest()[:8]}"
)
async with self._lock:
self._pending_fences[slot] = operation_id
self._overrides[slot] = {
"slot_name": res.slot_name,
"slot_code": res.slot_code,
"start_time": res.buffered_start,
"end_time": res.buffered_end,
}
self._slot_miss_counts.pop(slot, None)
self.suppress_state_changes(
slot,
{
f"switch.{coordinator.lockname}_code_slot_"
f"{slot}_use_date_range_limits": "on",
f"text.{coordinator.lockname}_code_slot_{slot}_name": (
res.display_slot_name
),
f"text.{coordinator.lockname}_code_slot_{slot}_pin": res.slot_code,
f"datetime.{coordinator.lockname}_code_slot_"
f"{slot}_date_range_start": res.buffered_start.isoformat(),
f"datetime.{coordinator.lockname}_code_slot_"
f"{slot}_date_range_end": res.buffered_end.isoformat(),
},
)
event = _SlotEvent(
slot_name=res.slot_name,
slot_code=res.slot_code,
start=res.start,
end=res.end,
)
result = await async_fire_set_code(coordinator, event, slot)
async with self._lock:
current_token = self._pending_fences.get(slot)
if current_token != operation_id:
_LOGGER.warning(
"Stale overwrite token for slot %d; discarding result", slot
)
return OperationResult(kind="set", slot=slot, unconfirmed=True)
if result.confirmed:
_LOGGER.debug(
"Overwrite confirmed for slot %d; desired state restored "
"for reservation %s.",
slot,
res.identity_key,
)
self._pending_fences.pop(slot, None)
self._clear_slot_error(slot)
# NOTE: __assign_next_slot() is intentionally NOT called here.
# Slot selection in the reconciliation path is determined by
# compute_desired_plan(), not by the deprecated _next_slot field.
elif result.failed:
_LOGGER.warning(
"Overwrite failed for slot %d (error: %s); "
"slot may remain drifted.",
slot,
result.error,
)
self._pending_fences.pop(slot, None)
self._record_slot_error(slot, result.error or "overwrite failed")
else:
_LOGGER.debug(
"Overwrite unconfirmed for slot %d; keeping tentative assignment.",
slot,
)
self._pending_fences.pop(slot, None)
return result
def _slot_has_matching_event(
self,
slot: int,
@@ -1493,20 +763,11 @@ class EventOverrides:
coordinator,
calendar: list[CalendarEvent] | None = None,
) -> None:
"""Check overrides and fire clear-code events for stale slots (greedy path).
"""Check if overrides need to have a clear_code event fired.
When called from within _async_update_data, pass the fresh
calendar list directly because coordinator.data has not been
updated yet by the DUC framework.
.. deprecated::
This method implements the retired greedy cleanup policy. Stale
slot detection and clearing is now handled entirely by the
reconciliation coordinator via ``compute_desired_plan`` /
``async_apply_plan`` (``ActionKind.CLEAR`` / ``RETRY_CLEAR``).
The coordinator no longer calls this method. The method is
retained as a backward-compatible shim; production code must
not invoke it.
"""
_LOGGER.debug("In EventOverrides.async_check_overrides")
@@ -1596,44 +857,17 @@ class EventOverrides:
if clear_code:
_LOGGER.debug("Firing clear code for slot %s", slot)
try:
result = await async_fire_clear_code(
await async_fire_clear_code(
coordinator, slot, expected_name=self.get_slot_name(slot)
)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception(
"Unexpected error firing clear code for slot %d; "
"slot remains occupied to prevent "
"double-assignment.",
"Failed to fire clear code for slot %d; "
"freeing slot so new events can be assigned. "
"The stale code may be overwritten when a "
"future event is programmed into this slot.",
slot,
)
continue
if not isinstance(result, OperationResult):
result = OperationResult(
kind="clear",
slot=slot,
unconfirmed=True,
)
if (
not result.confirmed
or result.failed
or result.lingering_name
or result.lingering_pin
):
_LOGGER.warning(
"Clear not confirmed for slot %d "
"(failed=%s, unconfirmed=%s, "
"lingering_name=%s, lingering_pin=%s); "
"slot remains occupied.",
slot,
result.failed,
result.unconfirmed,
result.lingering_name,
result.lingering_pin,
)
continue
self._overrides[slot] = None
self._slot_uids.pop(slot, None)
custom_components/rental_control/reconciliation.py
-1579
View File
File diff suppressed because it is too large. Load diff
custom_components/rental_control/sensors/calsensor.py
+109 -39
View File
@@ -22,9 +22,12 @@ from ..const import ICON
from ..const import NAME
from ..const import SECONDS_PER_HOUR
from ..const import SECONDS_PER_MINUTE
from ..reconciliation import make_reservation_fingerprint
from ..util import async_fire_clear_code
from ..util import async_fire_set_code
from ..util import async_fire_update_times
from ..util import gen_uuid
from ..util import get_slot_name
from ..util import normalize_uid
if TYPE_CHECKING:
from ..coordinator import RentalControlCoordinator
@@ -413,27 +416,19 @@ class RentalControlCalSensor(CoordinatorEntity["RentalControlCoordinator"]):
)
self._event_attributes["slot_name"] = slot_name
# Read slot assignment and code from coordinator reconciliation
# state. The coordinator owns all slot mutations; the sensor is
# purely read-only. Fall back to a locally generated code when
# reconciliation has not yet produced a result (e.g. no lock
# configured, or first refresh before reconciliation ran).
slot_number: int | None = None
slot_code: str | None = None
if self.coordinator.event_overrides is not None and slot_name is not None:
identity_key = make_reservation_fingerprint(
self.coordinator.entry_id,
slot_name,
event.start,
event.end,
)
slot_number = self.coordinator.get_slot_assignment(identity_key)
slot_code = self.coordinator.get_slot_code(identity_key)
slot_code = self._generate_door_code()
self._event_attributes["slot_number"] = None
if slot_code is None:
slot_code = self._generate_door_code()
self._event_attributes["slot_number"] = slot_number
# Read-only lookup for display: show existing override code
# immediately rather than the generated fallback. This is
# safe because it is not used for slot-assignment decisions.
overrides = self.coordinator.event_overrides
if overrides and slot_name is not None:
existing = overrides.get_slot_with_name(slot_name)
if existing and existing["slot_code"]:
slot_code = str(existing["slot_code"])
slot_key = overrides.get_slot_key_by_name(slot_name)
self._event_attributes["slot_number"] = slot_key if slot_key else None
self._event_attributes["slot_code"] = slot_code
# attributes parsed from description
@@ -472,6 +467,27 @@ class RentalControlCalSensor(CoordinatorEntity["RentalControlCoordinator"]):
self._parsed_attributes = parsed_attributes
# Schedule atomic slot assignment with a snapshot of
# event data so the async task is immune to subsequent
# coordinator updates mutating self._event_attributes.
# Gate on overrides.ready to avoid spurious overflow
# warnings during bootstrap when _next_slot is still None.
if overrides and overrides.ready and slot_name is not None:
event_uid: str | None = normalize_uid(
event.uid if hasattr(event, "uid") else None
)
self.hass.async_create_task(
self._async_handle_slot_assignment(
slot_name=slot_name,
slot_code=slot_code,
start_time=event.start,
end_time=event.end,
uid=event_uid,
prefix=self.coordinator.event_prefix or "",
eta_days=eta_days,
)
)
else:
# No reservations
_LOGGER.debug(
@@ -513,24 +529,78 @@ class RentalControlCalSensor(CoordinatorEntity["RentalControlCoordinator"]):
prefix: str,
eta_days: int | None,
) -> None:
"""No-op backward-compatible shim; slot assignment is owned by the coordinator.
"""Atomically reserve or locate an existing slot for the current event.
The per-event slot-assignment scheduling path — where
``_handle_coordinator_update`` would call this method to invoke
:meth:`~..event_overrides.EventOverrides.async_reserve_or_get_slot`
or fire set-code / clear-code / update-times events — has been
removed. All slot mutation is now performed exclusively by
:class:`~..coordinator.RentalControlCoordinator` through the
reconciliation cycle (``compute_desired_plan`` / ``async_apply_plan``).
All event data is captured at call-time so the coroutine is
immune to subsequent coordinator updates that could mutate
``self._event_attributes`` before execution.
This shim is retained so that any external call sites produce a
harmless no-op rather than an ``AttributeError``. It is verified
by ``test_async_handle_slot_assignment_is_noop`` as a regression
guard. The method must never be called or scheduled from
:meth:`_handle_coordinator_update`.
.. deprecated::
The per-event scheduling and mutation fallback path has been
retired. This shim will be removed in a future release once
all external references are eliminated.
Uses ``async_reserve_or_get_slot`` to eliminate the
check-then-act race that previously existed between
``get_slot_with_name`` and ``next_slot`` reads.
"""
overrides = self.coordinator.event_overrides
if overrides is None:
return
result = await overrides.async_reserve_or_get_slot(
slot_name=slot_name,
slot_code=slot_code,
start_time=start_time,
end_time=end_time,
uid=uid,
prefix=prefix,
)
if result.slot is None:
return
# Staleness guard: a subsequent coordinator update may have
# changed this sensor to a different event. The reservation
# itself is still valid, but Keymaster side effects must not
# fire for a stale event.
current_attrs = self._event_attributes
current_uid = normalize_uid(
str(current_attrs["uid"]) if current_attrs.get("uid") is not None else None
)
if (
current_attrs.get("slot_name") != slot_name
or current_attrs.get("start") != start_time
or current_attrs.get("end") != end_time
or current_uid != uid
):
_LOGGER.debug(
"Slot assignment for '%s' is stale, skipping "
"Keymaster side effects for sensor %s",
slot_name,
self.name,
)
return
self._event_attributes["slot_number"] = result.slot
if result.is_new:
await async_fire_set_code(self.coordinator, self, result.slot)
return
# Existing slot — update displayed code from the override
override = overrides.overrides.get(result.slot)
if override and override["slot_code"]:
self._event_attributes["slot_code"] = str(override["slot_code"])
self.async_write_ha_state()
if result.times_updated:
if (
self.coordinator.code_generator == "date_based"
and self.coordinator.should_update_code
and eta_days
and eta_days > 0
):
_LOGGER.debug(
"Clearing slot %s for sensor %s due to date shift",
result.slot,
self.name,
)
await async_fire_clear_code(self.coordinator, result.slot)
else:
await async_fire_update_times(self.coordinator, self, result.slot)
custom_components/rental_control/util.py
+49 -201
View File
@@ -18,14 +18,12 @@ from __future__ import annotations
import asyncio
from collections.abc import Coroutine
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import date
from datetime import datetime
from datetime import time
from datetime import timedelta
from datetime import tzinfo
import hashlib
import inspect
import logging
from pathlib import Path
import re
@@ -61,27 +59,6 @@ from .const import NAME
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, slots=True)
class OperationResult:
"""Result of a physical Keymaster slot service operation.
Returned by ``async_fire_clear_code()``, ``async_fire_set_code()``,
and ``async_fire_update_times()``. Successful helper paths set one
of ``confirmed``, ``unconfirmed``, or ``failed`` to ``True``.
``lingering_name`` and ``lingering_pin`` are supplementary flags for
clear operations. Raw PIN values are never stored.
"""
kind: str
slot: int
confirmed: bool = False
unconfirmed: bool = False
failed: bool = False
lingering_name: bool = False
lingering_pin: bool = False
error: str | None = None
def get_entry_data(hass: HomeAssistant, entry_id: str) -> dict[str, Any] | None:
"""Return Rental Control entry data when domain and entry data exist."""
domain_data = cast("dict[str, dict[str, Any]] | None", hass.data.get(DOMAIN))
@@ -195,7 +172,7 @@ def delete_folder(absolute_path: str | Path, *relative_paths: str) -> None:
async def async_fire_clear_code(
coordinator, slot: int, expected_name: str | None = None
) -> OperationResult:
) -> None:
"""Fire a clear_code signal."""
_LOGGER.debug(
"In async_fire_clear_code - slot: %s, name: %s", slot, coordinator.name
@@ -204,7 +181,7 @@ async def async_fire_clear_code(
reset_entity = f"{BUTTON}.{coordinator.lockname}_code_slot_{slot}_reset"
if not coordinator.lockname:
return OperationResult(kind="clear", slot=slot, unconfirmed=True)
return
if (
expected_name is not None
@@ -215,7 +192,7 @@ async def async_fire_clear_code(
slot,
expected_name,
)
return OperationResult(kind="clear", slot=slot, unconfirmed=True)
return
try:
# Reset the slot
@@ -225,9 +202,7 @@ async def async_fire_clear_code(
target={"entity_id": reset_entity},
blocking=True,
)
except asyncio.CancelledError:
raise
except Exception as exc:
except Exception:
escalated = coordinator.event_overrides.record_retry_failure(slot)
if escalated:
pn_create(
@@ -237,26 +212,19 @@ async def async_fire_clear_code(
title="Rental Control: Lock Command Failure",
notification_id=f"rental_control_slot_{slot}_clear_failure",
)
return OperationResult(
kind="clear",
slot=slot,
failed=True,
error=str(exc),
)
raise
# Give Keymaster time to propagate the state change
await asyncio.sleep(0.5)
unconfirmed = False
lingering_name = False
lingering_pin = False
# Verify the slot name was actually cleared
name_entity = f"{TEXT}.{coordinator.lockname}_code_slot_{slot}_name"
pin_entity = f"{TEXT}.{coordinator.lockname}_code_slot_{slot}_pin"
name_state = hass.states.get(name_entity)
if name_state is None:
unconfirmed = True
elif name_state.state not in ("", "unknown", "unavailable"):
if name_state is not None and name_state.state not in (
"",
"unknown",
"unavailable",
):
_LOGGER.warning(
"Slot %d name '%s' persisted after reset; "
"forcing name clear via text.set_value",
@@ -271,35 +239,11 @@ async def async_fire_clear_code(
service_data={"value": ""},
blocking=True,
)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception(
"Failed to force-clear name for slot %d",
slot,
)
name_state = hass.states.get(name_entity)
if name_state is None:
unconfirmed = True
elif name_state.state not in ("", "unknown", "unavailable"):
lingering_name = True
pin_state = hass.states.get(pin_entity)
if pin_state is None:
unconfirmed = True
elif pin_state.state not in ("", "unknown", "unavailable"):
lingering_pin = True
if lingering_name or lingering_pin:
return OperationResult(
kind="clear",
slot=slot,
unconfirmed=True,
lingering_name=lingering_name,
lingering_pin=lingering_pin,
)
if unconfirmed:
return OperationResult(kind="clear", slot=slot, unconfirmed=True)
was_escalated = coordinator.event_overrides._escalated.get(slot, False)
coordinator.event_overrides.record_retry_success(slot)
@@ -308,7 +252,6 @@ async def async_fire_clear_code(
hass,
notification_id=f"rental_control_slot_{slot}_clear_failure",
)
return OperationResult(kind="clear", slot=slot, confirmed=True)
def trim_name(name: str, max_length: int) -> str:
@@ -378,7 +321,7 @@ def apply_buffer(
return dt_start, dt_end
async def async_fire_set_code(coordinator, event, slot: int) -> OperationResult:
async def async_fire_set_code(coordinator, event, slot: int) -> None:
"""Set codes into a slot."""
_LOGGER.debug("In async_fire_set_code - slot: %s", slot)
_LOGGER.debug("Event: %s", event)
@@ -388,7 +331,7 @@ async def async_fire_set_code(coordinator, event, slot: int) -> OperationResult:
coro: list[Coroutine] = []
if not lockname:
return OperationResult(kind="set", slot=slot, unconfirmed=True)
return
if coordinator.event_prefix:
prefix = f"{coordinator.event_prefix} "
@@ -409,7 +352,7 @@ async def async_fire_set_code(coordinator, event, slot: int) -> OperationResult:
slot,
expected_name,
)
return OperationResult(kind="set", slot=slot, unconfirmed=True)
return
try:
# Disable the slot, this should help avoid notices from Keymaster
@@ -504,9 +447,7 @@ async def async_fire_set_code(coordinator, event, slot: int) -> OperationResult:
results = await asyncio.gather(*coro, return_exceptions=True)
check_gather_results(results, "Lock slot operation")
_raise_first_gather_exception(results)
except asyncio.CancelledError:
raise
except Exception as exc:
except Exception:
escalated = coordinator.event_overrides.record_retry_failure(slot)
if escalated:
pn_create(
@@ -516,35 +457,18 @@ async def async_fire_set_code(coordinator, event, slot: int) -> OperationResult:
title="Rental Control: Lock Command Failure",
notification_id=f"rental_control_slot_{slot}_failure",
)
return OperationResult(
kind="set",
slot=slot,
failed=True,
error=str(exc),
raise
was_escalated = coordinator.event_overrides._escalated.get(slot, False)
coordinator.event_overrides.record_retry_success(slot)
if was_escalated:
pn_dismiss(
coordinator.hass,
notification_id=f"rental_control_slot_{slot}_failure",
)
name_entity = f"{TEXT}.{lockname}_code_slot_{slot}_name"
block_till_done = getattr(coordinator.hass, "async_block_till_done", None)
if block_till_done is not None:
done_result = block_till_done()
if inspect.isawaitable(done_result):
await done_result
name_state = coordinator.hass.states.get(name_entity)
if name_state is None:
return OperationResult(kind="set", slot=slot, unconfirmed=True)
if name_state.state == slot_name:
was_escalated = coordinator.event_overrides._escalated.get(slot, False)
coordinator.event_overrides.record_retry_success(slot)
if was_escalated:
pn_dismiss(
coordinator.hass,
notification_id=f"rental_control_slot_{slot}_failure",
)
return OperationResult(kind="set", slot=slot, confirmed=True)
return OperationResult(kind="set", slot=slot, unconfirmed=True)
async def async_fire_update_times(coordinator, event, slot: int) -> OperationResult:
async def async_fire_update_times(coordinator, event, slot: int) -> None:
"""Update times on slot."""
lockname: str = coordinator.lockname
@@ -552,7 +476,7 @@ async def async_fire_update_times(coordinator, event, slot: int) -> OperationRes
slot_name: str = event.extra_state_attributes["slot_name"]
if not slot or not lockname:
return OperationResult(kind="update_times", slot=slot, unconfirmed=True)
return
if not coordinator.event_overrides.verify_slot_ownership(slot, slot_name):
_LOGGER.warning(
@@ -560,7 +484,7 @@ async def async_fire_update_times(coordinator, event, slot: int) -> OperationRes
slot,
slot_name,
)
return OperationResult(kind="update_times", slot=slot, unconfirmed=True)
return
# Compute buffered validity window for Keymaster
buffered_start, buffered_end = apply_buffer(
@@ -590,18 +514,6 @@ async def async_fire_update_times(coordinator, event, slot: int) -> OperationRes
)
results = await asyncio.gather(*coro, return_exceptions=True)
check_gather_results(results, "Lock slot operation")
try:
_raise_first_gather_exception(results)
except asyncio.CancelledError:
raise
except Exception as exc:
return OperationResult(
kind="update_times",
slot=slot,
failed=True,
error=str(exc),
)
return OperationResult(kind="update_times", slot=slot, confirmed=True)
def _ensure_datetime(value: date | datetime, rc) -> datetime:
@@ -781,11 +693,6 @@ async def handle_state_change(
await asyncio.sleep(0.1)
entity_id = event.data["entity_id"]
event_new_state = event.data.get("new_state")
event_new_value = (
getattr(event_new_state, "state", None) if event_new_state is not None else None
)
event_has_new_value = event_new_value is not None
_LOGGER.debug(
"Handling state change for %s in %s with event: %s",
@@ -799,9 +706,6 @@ async def handle_state_change(
_LOGGER.warning("Could not extract slot number from entity_id: %s", entity_id)
return
slot_num = int(slot_match.group(1))
existing_override: dict[str, Any] | None = None
if event_has_new_value and coordinator.event_overrides:
existing_override = coordinator.event_overrides.overrides.get(slot_num)
if "_reset" in entity_id:
_LOGGER.debug("Resetting overrides %s for %s.", slot_num, lockname)
@@ -810,29 +714,12 @@ async def handle_state_change(
)
return
if event_has_new_value and coordinator.event_overrides.should_suppress_state_change(
slot_num, entity_id, event_new_value
):
_LOGGER.debug(
"Ignoring coordinator feedback for %s slot %s.",
lockname,
slot_num,
)
return
slot_state = hass.states.get(f"switch.{lockname}_code_slot_{slot_num}_enabled")
_LOGGER.debug("Slot %s state: %s", slot_num, slot_state)
if slot_state is None:
return
slot_enabled_entity_id = f"switch.{lockname}_code_slot_{slot_num}_enabled"
slot_enabled_state = (
event_new_value
if event_has_new_value and entity_id == slot_enabled_entity_id
else slot_state.state
)
if slot_enabled_state != "on":
if slot_state.state != "on":
_LOGGER.debug(
"Slot %s is not enabled, skipping update for %s.",
slot_num,
@@ -840,86 +727,47 @@ async def handle_state_change(
)
return
slot_code_entity_id = f"text.{lockname}_code_slot_{slot_num}_pin"
slot_name_entity_id = f"text.{lockname}_code_slot_{slot_num}_name"
slot_code = hass.states.get(slot_code_entity_id)
slot_name = hass.states.get(slot_name_entity_id)
slot_code = hass.states.get(f"text.{lockname}_code_slot_{slot_num}_pin")
slot_name = hass.states.get(f"text.{lockname}_code_slot_{slot_num}_name")
use_date_range_entity_id = (
use_date_range = hass.states.get(
f"switch.{lockname}_code_slot_{slot_num}_use_date_range_limits"
)
use_date_range = hass.states.get(use_date_range_entity_id)
_LOGGER.debug("Use Date Range: %s", use_date_range)
use_date_range_state = (
event_new_value
if event_has_new_value and entity_id == use_date_range_entity_id
else use_date_range.state
if use_date_range
else None
)
if use_date_range_state == "on":
start_time_entity_id = (
if use_date_range and use_date_range.state == "on":
g_start_time = hass.states.get(
f"datetime.{lockname}_code_slot_{slot_num}_date_range_start"
)
end_time_entity_id = f"datetime.{lockname}_code_slot_{slot_num}_date_range_end"
g_start_time = hass.states.get(start_time_entity_id)
g_end_time = hass.states.get(end_time_entity_id)
g_end_time = hass.states.get(
f"datetime.{lockname}_code_slot_{slot_num}_date_range_end"
)
else:
start_time_entity_id = ""
end_time_entity_id = ""
g_start_time = None
g_end_time = None
if slot_code is None:
return
if event_has_new_value and entity_id == slot_code_entity_id:
slot_code_value = (
"" if event_new_value in ("unknown", "unavailable") else event_new_value
)
elif event_has_new_value and existing_override is not None:
slot_code_value = str(existing_override["slot_code"])
else:
slot_code_value = (
"" if slot_code.state in ("unknown", "unavailable") else slot_code.state
)
slot_code_value = (
"" if slot_code.state in ("unknown", "unavailable") else slot_code.state
)
if slot_name is None:
return
if event_has_new_value and entity_id == slot_name_entity_id:
slot_name_value = (
"" if event_new_value in ("unknown", "unavailable") else event_new_value
)
elif event_has_new_value and existing_override is not None:
slot_name_value = str(existing_override["slot_name"])
else:
slot_name_value = (
"" if slot_name.state in ("unknown", "unavailable") else slot_name.state
)
slot_name_value = (
"" if slot_name.state in ("unknown", "unavailable") else slot_name.state
)
start_time = dt.start_of_local_day()
end_time = dt.start_of_local_day()
if event_has_new_value and existing_override is not None:
start_time = existing_override["start_time"]
end_time = existing_override["end_time"]
if g_start_time is not None:
if event_has_new_value and entity_id == start_time_entity_id:
p_start_time = dt.parse_datetime(event_new_value)
if p_start_time:
start_time = p_start_time
elif not (event_has_new_value and existing_override is not None):
p_start_time = dt.parse_datetime(g_start_time.state)
if p_start_time:
start_time = p_start_time
p_start_time = dt.parse_datetime(g_start_time.state)
if p_start_time:
start_time = p_start_time
if g_end_time is not None:
if event_has_new_value and entity_id == end_time_entity_id:
p_end_time = dt.parse_datetime(event_new_value)
if p_end_time:
end_time = p_end_time
elif not (event_has_new_value and existing_override is not None):
p_end_time = dt.parse_datetime(g_end_time.state)
if p_end_time:
end_time = p_end_time
p_end_time = dt.parse_datetime(g_end_time.state)
if p_end_time:
end_time = p_end_time
_LOGGER.debug(
"updating overrides for %s slot %s. "
@@ -965,8 +813,8 @@ async def handle_state_change(
start_time,
end_time,
)
# Reconciliation runs exclusively from coordinator apply_plan (R-007);
# callbacks must not launch reconciliation.
await coordinator.event_overrides.async_check_overrides(coordinator)
async def async_reload_package_platforms(hass: HomeAssistant) -> bool:
tests/conftest.py
-91
View File
@@ -6,21 +6,12 @@
from __future__ import annotations
from datetime import time
import inspect
import json
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Any
from unittest.mock import MagicMock
from aiohttp import hdrs
from aiohttp.client_reqrep import ClientResponse
from aiohttp.client_reqrep import RequestInfo
from aiohttp.helpers import TimerNoop
from aioresponses import aioresponses
import aioresponses.core as aioresponses_core
from multidict import CIMultiDict
from multidict import CIMultiDictProxy
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
@@ -38,88 +29,6 @@ if TYPE_CHECKING:
# Enable custom components for testing
pytest_plugins = ("pytest_homeassistant_custom_component",)
def _aioresponses_build_response_aiohttp313( # noqa: PLR0913
self,
url,
method=hdrs.METH_GET,
request_headers=None,
status=200,
body="",
content_type="application/json",
payload=None,
headers=None,
response_class=None,
reason=None,
):
"""Build a mock ClientResponse compatible with aiohttp >= 3.13.
aiohttp 3.13 added a required ``stream_writer`` keyword-only
argument to ``ClientResponse.__init__``. This replacement for
``aioresponses.core.RequestMatch._build_response`` injects the
extra argument so that ``aioresponses``-based coordinator tests
pass without modification.
"""
if response_class is None:
response_class = ClientResponse
if payload is not None:
body = json.dumps(payload)
if not isinstance(body, bytes):
body = str.encode(body)
if request_headers is None:
request_headers = {}
loop = MagicMock()
loop.get_debug.return_value = True
kwargs: dict[str, Any] = {}
kwargs["request_info"] = RequestInfo(
url=url,
method=method,
headers=CIMultiDictProxy(CIMultiDict(**request_headers)),
real_url=url,
)
kwargs["writer"] = None
kwargs["continue100"] = None
kwargs["timer"] = TimerNoop()
kwargs["traces"] = []
kwargs["loop"] = loop
kwargs["session"] = None
kwargs["stream_writer"] = MagicMock(output_size=0)
_headers = CIMultiDict({hdrs.CONTENT_TYPE: content_type})
if headers:
_headers.update(headers)
raw_headers = self._build_raw_headers(_headers)
resp = response_class(method, url, **kwargs)
for header in _headers.getall(hdrs.SET_COOKIE, ()):
resp.cookies.load(header)
resp._headers = _headers
resp._raw_headers = raw_headers
resp.status = status
resp.reason = reason
resp.content = aioresponses_core.stream_reader_factory(loop)
resp.content.feed_data(body)
resp.content.feed_eof()
return resp
def _patch_aioresponses_for_aiohttp_313() -> None:
"""Apply aioresponses patch when running with aiohttp >= 3.13."""
if "stream_writer" not in inspect.signature(ClientResponse).parameters:
return # pragma: no cover
if (
"stream_writer"
in inspect.signature(aioresponses_core.RequestMatch._build_response).parameters
):
return # pragma: no cover
aioresponses_core.RequestMatch._build_response = ( # type: ignore[method-assign]
_aioresponses_build_response_aiohttp313
)
_patch_aioresponses_for_aiohttp_313()
_TESTS_ROOT = Path(__file__).parent
tests/integration/test_refresh_cycle.py
-2857
View File
File diff suppressed because it is too large. Load diff
tests/integration/test_slot_concurrency.py
-141
View File
@@ -582,144 +582,3 @@ class TestSingleReservationRegression:
)
assert r3.slot == 1
assert r3.is_new is True
class TestCallbackReentrancyFencing:
"""Callbacks cannot launch reconciliation during a pending operation."""
@pytest.mark.asyncio
async def test_reconciliation_active_during_apply_plan(self) -> None:
"""reconciliation_active is True during apply_plan execution."""
from unittest.mock import MagicMock
from unittest.mock import patch
from custom_components.rental_control.reconciliation import ActionKind
from custom_components.rental_control.reconciliation import DesiredPlan
from custom_components.rental_control.reconciliation import SlotAction
from custom_components.rental_control.util import OperationResult
eo = _populated_eo(start_slot=1, max_slots=2)
eo.update(1, "c1", "Guest", _make_dt(2026, 1, 1), _make_dt(2026, 1, 2))
observed_active_states: list[bool] = []
async def mock_clear_code(*args, **kwargs) -> OperationResult:
"""Record reconciliation_active during the clear call."""
observed_active_states.append(eo.reconciliation_active)
return OperationResult(kind="clear", slot=1, confirmed=True)
now = _make_dt(2026, 8, 1)
plan = DesiredPlan(plan_id="t064-active", generated_at=now)
plan.actions = [SlotAction(kind=ActionKind.CLEAR, slot=1, identity_key=None)]
coordinator = MagicMock()
coordinator.lockname = "test_lock"
with patch(
"custom_components.rental_control.event_overrides.async_fire_clear_code",
side_effect=mock_clear_code,
):
await eo.async_apply_plan(coordinator, plan, {})
assert len(observed_active_states) == 1
assert observed_active_states[0] is True
assert eo.reconciliation_active is False
@pytest.mark.asyncio
async def test_handle_state_change_does_not_start_reconciliation(self) -> None:
"""handle_state_change no longer calls async_check_overrides."""
from unittest.mock import AsyncMock
from unittest.mock import MagicMock
from unittest.mock import patch
from custom_components.rental_control.const import COORDINATOR
from custom_components.rental_control.const import DOMAIN
from custom_components.rental_control.util import handle_state_change
lockname = "test_lock"
slot_num = 1
mock_coordinator = MagicMock()
mock_coordinator.lockname = lockname
mock_coordinator.trim_names = False
mock_coordinator.event_overrides = MagicMock()
mock_coordinator.event_overrides.async_check_overrides = AsyncMock()
mock_coordinator.update_event_overrides = AsyncMock()
name_state = MagicMock()
name_state.state = "Guest"
pin_state = MagicMock()
pin_state.state = "1234"
enabled_state = MagicMock()
enabled_state.state = "on"
def states_get(entity_id: str):
"""Return the mocked Keymaster state for the requested entity."""
if "enabled" in entity_id:
return enabled_state
if "pin" in entity_id:
return pin_state
if "name" in entity_id:
return name_state
return None
hass = MagicMock()
hass.data = {DOMAIN: {"entry_id": {COORDINATOR: mock_coordinator}}}
hass.states.get = states_get
config_entry = MagicMock()
config_entry.entry_id = "entry_id"
event = MagicMock()
event.data = {"entity_id": f"switch.{lockname}_code_slot_{slot_num}_enabled"}
with patch(
"custom_components.rental_control.util.asyncio.sleep",
new_callable=AsyncMock,
):
await handle_state_change(hass, config_entry, event)
mock_coordinator.event_overrides.async_check_overrides.assert_not_called()
mock_coordinator.update_event_overrides.assert_awaited_once()
@pytest.mark.asyncio
async def test_pending_token_survives_state_change_callback(self) -> None:
"""A concurrent callback does not clear the active fence token."""
from unittest.mock import MagicMock
from unittest.mock import patch
from custom_components.rental_control.reconciliation import ActionKind
from custom_components.rental_control.reconciliation import DesiredPlan
from custom_components.rental_control.reconciliation import SlotAction
from custom_components.rental_control.util import OperationResult
eo = _populated_eo(start_slot=1, max_slots=2)
eo.update(1, "c1", "Guest", _make_dt(2026, 1, 1), _make_dt(2026, 1, 2))
token_during_service_call: list[str | None] = []
async def mock_clear_with_callback(*args, **kwargs) -> OperationResult:
"""Record the token and simulate a concurrent callback update."""
token_during_service_call.append(eo.pending_fences.get(1))
await eo.async_update(
slot=2,
slot_code="",
slot_name="",
start_time=_make_dt(2026, 1, 1),
end_time=_make_dt(2026, 1, 1),
)
return OperationResult(kind="clear", slot=1, confirmed=True)
now = _make_dt(2026, 8, 1)
plan = DesiredPlan(plan_id="t064-token", generated_at=now)
plan.actions = [SlotAction(kind=ActionKind.CLEAR, slot=1, identity_key=None)]
coordinator = MagicMock()
coordinator.lockname = "test_lock"
with patch(
"custom_components.rental_control.event_overrides.async_fire_clear_code",
side_effect=mock_clear_with_callback,
):
results = await eo.async_apply_plan(coordinator, plan, {})
assert token_during_service_call[0] is not None
assert eo.overrides[1] is None
assert results[0].confirmed is True
tests/unit/test_checkin_sensor.py
-130
View File
@@ -4111,7 +4111,6 @@ class TestSupportingComponentAbsence:
assert sensor._is_keymaster_monitoring_enabled() is expected
@freeze_time("2025-07-15T10:00:00+00:00")
async def test_checkout_missing_early_expiry_switch_continues(
self,
hass: HomeAssistant,
@@ -4192,7 +4191,6 @@ class TestMissingEntryDataFallbacks:
assert hass.data[DOMAIN] is domain_data
@pytest.mark.parametrize("remove_domain", [True, False])
@freeze_time("2025-07-15T10:00:00+00:00")
async def test_checkout_missing_entry_data_continues_without_early_expiry(
self,
hass: HomeAssistant,
@@ -4664,12 +4662,6 @@ class TestEventTrackingStability:
state oscillation when coordinator data reorders.
"""
@pytest.fixture(autouse=True)
def _freeze_midday(self) -> Generator[None, None, None]:
"""Pin now away from day boundaries for stable follow-on timing tests."""
with freeze_time("2025-06-19T12:00:00+00:00"):
yield
async def test_find_tracked_event_at_position_zero(
self,
hass: HomeAssistant,
@@ -5889,125 +5881,3 @@ class TestCheckedInSelfHealing:
# Should remain checked_in
assert sensor._state == CHECKIN_STATE_CHECKED_IN
class TestCheckinTrackingReconciliationRegression:
"""T105 regression: Check-in state machine preserved under reconciliation.
Pins two behaviours: the sensor stays checked_in when the reconciler
reassigns the tracked reservation to a different slot (the slot_name
match is what matters, not the slot number), and the checked_out
state is preserved through coordinator updates.
"""
@freeze_time("2025-07-01T12:00:00+00:00")
async def test_checked_in_preserved_when_slot_reassigned(
self,
hass: HomeAssistant,
mock_checkin_coordinator: MagicMock,
mock_checkin_config_entry: MockConfigEntry,
) -> None:
"""Checked-in state survives reconciliation slot reassignment."""
sensor = _create_sensor(
hass, mock_checkin_coordinator, mock_checkin_config_entry
)
now = dt_util.now()
tracked_event = _make_event(
summary="Reserved - John Smith",
start=now - timedelta(hours=2),
end=now + timedelta(days=2),
)
sensor._state = CHECKIN_STATE_CHECKED_IN
sensor._tracked_event_summary = tracked_event.summary
sensor._tracked_event_start = tracked_event.start
sensor._tracked_event_end = tracked_event.end
sensor._tracked_event_slot_name = "John Smith"
mock_checkin_coordinator.get_slot_assignment.return_value = 12
mock_checkin_coordinator.data = [tracked_event]
sensor._handle_coordinator_update()
assert sensor._state == CHECKIN_STATE_CHECKED_IN
assert sensor._tracked_event_slot_name == "John Smith"
@freeze_time("2025-07-01T12:00:00+00:00")
async def test_checked_out_state_preserved_through_coordinator_update(
self,
hass: HomeAssistant,
mock_checkin_coordinator: MagicMock,
mock_checkin_config_entry: MockConfigEntry,
) -> None:
"""Checked-out state is preserved during later coordinator refreshes."""
sensor = _create_sensor(
hass, mock_checkin_coordinator, mock_checkin_config_entry
)
sensor._state = CHECKIN_STATE_CHECKED_OUT
sensor._tracked_event_slot_name = "John Smith"
sensor._checkout_time = dt_util.now() - timedelta(hours=1)
sensor._unsub_timer = MagicMock()
mock_checkin_coordinator.data = []
sensor._handle_coordinator_update()
assert sensor._state == CHECKIN_STATE_CHECKED_OUT
async def test_apply_checkin_protection_marks_checked_in_reservation(
self,
hass: HomeAssistant,
mock_checkin_coordinator: MagicMock,
mock_checkin_config_entry: MockConfigEntry,
) -> None:
"""Active checked-in guests are marked protected for reconciliation."""
from custom_components.rental_control.coordinator import (
RentalControlCoordinator,
)
from custom_components.rental_control.reconciliation import Reservation
sensor = _create_sensor(
hass, mock_checkin_coordinator, mock_checkin_config_entry
)
sensor._state = CHECKIN_STATE_CHECKED_IN
sensor._tracked_event_slot_name = "John Smith"
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][mock_checkin_config_entry.entry_id] = {
CHECKIN_SENSOR: sensor,
COORDINATOR: mock_checkin_coordinator,
}
mock_checkin_coordinator.hass = hass
mock_checkin_coordinator._entry_id = mock_checkin_config_entry.entry_id
now = dt_util.now()
reservations = [
Reservation(
identity_key="res-john",
start=now,
end=now + timedelta(days=2),
buffered_start=now,
buffered_end=now + timedelta(days=2),
summary="Reserved - John Smith",
slot_name="John Smith",
display_slot_name="RC John Smith",
slot_code="1234",
),
Reservation(
identity_key="res-other",
start=now,
end=now + timedelta(days=2),
buffered_start=now,
buffered_end=now + timedelta(days=2),
summary="Reserved - Other Guest",
slot_name="Other Guest",
display_slot_name="RC Other Guest",
slot_code="5678",
),
]
RentalControlCoordinator._apply_checkin_protection(
mock_checkin_coordinator,
reservations,
)
assert reservations[0].protected_active is True
assert reservations[1].protected_active is False
tests/unit/test_coordinator.py
+1 -2438
View File
File diff suppressed because it is too large. Load diff
tests/unit/test_event_overrides.py
+31 -2823
View File
File diff suppressed because it is too large. Load diff
tests/unit/test_keymaster_event_diagnostics.py
-131
View File
@@ -524,134 +524,3 @@ class TestDiagnosticsAttribute:
assert "keymaster_event_diagnostics" in attrs
assert isinstance(attrs["keymaster_event_diagnostics"], list)
assert attrs["keymaster_event_diagnostics"][0]["disposition"] == "accepted"
# ---------------------------------------------------------------------------
# T094: Diagnostics redaction compatibility tests
# ---------------------------------------------------------------------------
class TestRedactionCompatibility:
"""T094: Verify that neither the keymaster event diagnostics ring buffer
nor the new slot-plan diagnostics snapshot expose raw slot codes or
sensitive reservation metadata."""
def test_keymaster_event_buffer_has_no_slot_code_field(self) -> None:
"""Entries in keymaster_event_diagnostics never contain slot_code."""
from collections import deque
buf: deque = deque(maxlen=10)
entry = {
"timestamp": "2026-05-15T13:08:11+00:00",
"lockname": "Front Door",
"lockname_slug": "front_door",
"state": "unlocked",
"code_slot_num": 11,
"disposition": "accepted",
}
buf.append(entry)
for item in list(buf):
assert "slot_code" not in item
assert "pin" not in item
assert "code_value" not in item
def test_keymaster_event_buffer_fields_are_non_sensitive(self) -> None:
"""Accepted entries contain only lockname, state, slot num, disposition, timestamp."""
from collections import deque
buf: deque = deque(maxlen=10)
entry = {
"timestamp": "2026-05-15T13:08:11+00:00",
"lockname": "Front Door",
"lockname_slug": "front_door",
"state": "unlocked",
"code_slot_num": 11,
"disposition": "accepted",
}
buf.append(entry)
allowed_keys = {
"timestamp",
"lockname",
"lockname_slug",
"state",
"code_slot_num",
"disposition",
}
for item in list(buf):
assert set(item.keys()).issubset(allowed_keys)
def test_event_overrides_diagnostics_snapshot_has_no_raw_codes(self) -> None:
"""EventOverrides.diagnostics_snapshot contains no slot_code or PIN data."""
from datetime import datetime
from datetime import timedelta
from datetime import timezone
from custom_components.rental_control.event_overrides import EventOverrides
from custom_components.rental_control.reconciliation import ActionKind
from custom_components.rental_control.reconciliation import DesiredPlan
from custom_components.rental_control.reconciliation import PlannedSlot
_TZ = timezone.utc
now = datetime(2026, 8, 1, tzinfo=_TZ)
eo = EventOverrides(start_slot=10, max_slots=2)
eo.update(10, "SECRETPIN123", "Guest A", now, now + timedelta(days=7))
plan = DesiredPlan(plan_id="redact-test", generated_at=now)
plan.slots[10] = PlannedSlot(
slot=10,
desired_identity_key="res-a",
actual_classification="occupied",
action=ActionKind.NOOP,
)
plan.slots[11] = PlannedSlot(
slot=11,
desired_identity_key=None,
actual_classification="free",
action=ActionKind.NOOP,
)
eo.update_diagnostics_snapshot(plan)
snap = eo.diagnostics_snapshot
snap_str = str(snap)
assert "SECRETPIN123" not in snap_str
assert "slot_code" not in snap_str
def test_compute_desired_plan_diagnostics_has_no_slot_code(self) -> None:
"""compute_desired_plan diagnostics exclude slot_code from all entries."""
from datetime import datetime
from datetime import timedelta
from datetime import timezone
from custom_components.rental_control.reconciliation import ManagedSlot
from custom_components.rental_control.reconciliation import Reservation
from custom_components.rental_control.reconciliation import SlotStatus
from custom_components.rental_control.reconciliation import compute_desired_plan
_TZ = timezone.utc
start = datetime(2026, 8, 1, 14, tzinfo=_TZ)
end = start + timedelta(days=7)
r = Reservation(
identity_key="r-redact",
start=start,
end=end,
buffered_start=start,
buffered_end=end,
summary="Guest Redact",
slot_name="Guest Redact",
display_slot_name="RC Guest Redact",
slot_code="RAWPIN4567",
)
ms = ManagedSlot(slot=10, managed=True, status=SlotStatus.FREE)
plan = compute_desired_plan(
[r],
[ms],
max_events=3,
plan_id="p-redact",
generated_at=start,
)
diag_str = str(plan.diagnostics)
assert "RAWPIN4567" not in diag_str
assert "slot_code" not in diag_str
tests/unit/test_sensors.py
+270 -547
View File
@@ -8,6 +8,7 @@ from __future__ import annotations
from datetime import datetime
from datetime import timezone
import random
from unittest.mock import AsyncMock
from unittest.mock import MagicMock
from unittest.mock import call
from unittest.mock import patch
@@ -19,6 +20,7 @@ from custom_components.rental_control.const import COORDINATOR
from custom_components.rental_control.const import DOMAIN
from custom_components.rental_control.const import ICON
from custom_components.rental_control.const import NAME
from custom_components.rental_control.event_overrides import ReserveResult
from custom_components.rental_control.sensor import async_setup_entry
from custom_components.rental_control.sensor import async_setup_platform
from custom_components.rental_control.sensors.calsensor import RentalControlCalSensor
@@ -55,7 +57,6 @@ def _make_coordinator(
*,
name: str = "Test Rental",
unique_id: str = "test_unique_id",
entry_id: str = "test_entry_id",
last_update_success: bool = True,
data: list | None = None,
event_prefix: str = "",
@@ -63,14 +64,11 @@ def _make_coordinator(
code_length: int = 4,
event_overrides: MagicMock | None = None,
should_update_code: bool = True,
slot_assignment: int | None = None,
slot_code: str | None = None,
) -> MagicMock:
"""Create a mock coordinator for sensor testing."""
coordinator = MagicMock()
coordinator.name = name
coordinator.unique_id = unique_id
coordinator.entry_id = entry_id
coordinator.last_update_success = last_update_success
coordinator.data = data
coordinator.event_prefix = event_prefix
@@ -78,8 +76,6 @@ def _make_coordinator(
coordinator.code_length = code_length
coordinator.event_overrides = event_overrides
coordinator.should_update_code = should_update_code
coordinator.get_slot_assignment.return_value = slot_assignment
coordinator.get_slot_code.return_value = slot_code
coordinator.device_info = {
"identifiers": {(DOMAIN, unique_id)},
"name": f"{NAME} {name}",
@@ -1354,33 +1350,70 @@ class TestHandleCoordinatorUpdateNoEvents:
# ---------------------------------------------------------------------------
# Override interaction tests — reconciliation read-only path
# Override interaction tests
# ---------------------------------------------------------------------------
class TestHandleCoordinatorUpdateOverrides:
"""Tests for _handle_coordinator_update reading from reconciliation state."""
"""Tests for _handle_coordinator_update interactions with event_overrides."""
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_code_from_reconciliation(self, hass) -> None:
"""Verify slot_code is read from coordinator reconciliation state."""
async def test_fires_set_code_when_new_reservation(self, hass) -> None:
"""Verify async_fire_set_code is called when slot is newly reserved."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_code="5555",
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = None
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, True, False)
)
coordinator = _make_coordinator(data=[event], event_overrides=overrides)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_set_code",
new_callable=AsyncMock,
) as mock_set_code:
sensor._handle_coordinator_update()
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
mock_set_code.assert_called_once_with(coordinator, sensor, 10)
@freeze_time("2025-03-10T12:00:00+00:00")
async def test_uses_override_slot_code(self, hass) -> None:
"""Verify slot_code from override is displayed immediately in sync path."""
event = _make_event()
override = {
"slot_code": "5555",
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, False, False)
)
overrides.overrides = {10: override}
coordinator = _make_coordinator(data=[event], event_overrides=overrides)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
# Code is available immediately from the sync read-only lookup
assert sensor.extra_state_attributes["slot_code"] == "5555"
# Await the scheduled task to avoid unawaited coroutine warning
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
@freeze_time("2025-03-10T12:00:00+00:00")
def test_generates_code_when_no_override(self, hass) -> None:
"""Verify code is generated when event_overrides is None."""
"""Verify code is always generated synchronously in coordinator update."""
event = _make_event()
coordinator = _make_coordinator(
data=[event], event_overrides=None, code_generator="date_based"
@@ -1396,14 +1429,23 @@ class TestHandleCoordinatorUpdateOverrides:
assert attrs["slot_code"].isdigit()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_falls_back_to_generated_when_no_reconciliation_code(self, hass) -> None:
"""Verify code is generated when reconciliation has no code for the event."""
async def test_generates_code_when_override_has_no_code(self, hass) -> None:
"""Verify code is generated when override exists but has no slot_code."""
event = _make_event()
override = {
"slot_code": None,
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, False, False)
)
overrides.overrides = {10: override}
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
code_generator="date_based",
slot_code=None,
data=[event], event_overrides=overrides, code_generator="date_based"
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
@@ -1411,13 +1453,93 @@ class TestHandleCoordinatorUpdateOverrides:
sensor._handle_coordinator_update()
# Code is generated since override has no slot_code
attrs = sensor.extra_state_attributes
assert attrs["slot_code"] is not None
assert attrs["slot_code"].isdigit()
# Await the scheduled task to avoid unawaited coroutine warning
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
@freeze_time("2025-03-10T12:00:00+00:00")
def test_no_reconciliation_lookup_when_overrides_none(self, hass) -> None:
"""Verify fingerprint/reconciliation lookup is skipped when no lock configured."""
async def test_fires_update_times_when_dates_change(self, hass) -> None:
"""Verify async_fire_update_times is called when times_updated is True."""
event = _make_event(
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
)
override = {
"slot_code": "1234",
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, False, True)
)
overrides.overrides = {10: override}
coordinator = _make_coordinator(
data=[event],
event_overrides=overrides,
code_generator="static_random",
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_update_times",
new_callable=AsyncMock,
) as mock_update_times:
sensor._handle_coordinator_update()
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
mock_update_times.assert_called_once_with(coordinator, sensor, 10)
@freeze_time("2025-03-10T12:00:00+00:00")
async def test_fires_clear_code_on_date_shift_date_based(self, hass) -> None:
"""Verify async_fire_clear_code is called on date shift with date_based generator."""
event = _make_event(
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
)
override = {
"slot_code": "1234",
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, False, True)
)
overrides.overrides = {10: override}
coordinator = _make_coordinator(
data=[event],
event_overrides=overrides,
code_generator="date_based",
should_update_code=True,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_clear_code",
new_callable=AsyncMock,
) as mock_clear_code:
sensor._handle_coordinator_update()
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
mock_clear_code.assert_called_once_with(coordinator, 10)
@freeze_time("2025-03-10T12:00:00+00:00")
def test_no_override_interactions_when_overrides_none(self, hass) -> None:
"""Verify no set_code/update_times calls when event_overrides is None."""
event = _make_event()
coordinator = _make_coordinator(data=[event], event_overrides=None)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
@@ -1425,44 +1547,154 @@ class TestHandleCoordinatorUpdateOverrides:
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.make_reservation_fingerprint",
) as mock_fingerprint:
"custom_components.rental_control.sensors.calsensor.async_fire_set_code",
new_callable=AsyncMock,
) as mock_set_code:
sensor._handle_coordinator_update()
mock_fingerprint.assert_not_called()
mock_set_code.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_number_from_reconciliation(self, hass) -> None:
"""Verify slot_number is read from coordinator reconciliation state."""
async def test_no_set_code_when_existing_reservation(self, hass) -> None:
"""Verify set_code is not called when reservation already has a slot."""
event = _make_event()
override = {
"slot_code": "1234",
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, False, False)
)
overrides.overrides = {10: override}
coordinator = _make_coordinator(data=[event], event_overrides=overrides)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_set_code",
new_callable=AsyncMock,
) as mock_set_code:
sensor._handle_coordinator_update()
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
mock_set_code.assert_not_called()
@freeze_time("2025-03-15T12:00:00+00:00")
async def test_updates_times_not_clear_when_eta_days_zero(self, hass) -> None:
"""Verify update_times (not clear_code) is called when eta_days is 0.
When event starts today (eta_days=0), the clear_code branch requires
eta_days > 0. With eta_days=0 the condition is false, so update_times
is called instead.
"""
event = _make_event(
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
)
override = {
"slot_code": "1234",
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, False, True)
)
overrides.overrides = {10: override}
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_assignment=12,
event_overrides=overrides,
code_generator="date_based",
should_update_code=True,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with (
patch(
"custom_components.rental_control.sensors.calsensor.async_fire_clear_code",
new_callable=AsyncMock,
) as mock_clear_code,
patch(
"custom_components.rental_control.sensors.calsensor.async_fire_update_times",
new_callable=AsyncMock,
) as mock_update_times,
):
sensor._handle_coordinator_update()
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
mock_clear_code.assert_not_called()
mock_update_times.assert_called_once_with(coordinator, sensor, 10)
@freeze_time("2025-03-10T12:00:00+00:00")
async def test_slot_number_set_on_assignment(self, hass) -> None:
"""Verify slot_number is exposed after slot assignment."""
event = _make_event()
override = {
"slot_code": "4321",
"start_time": event.start,
"end_time": event.end,
}
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = override
overrides.get_slot_key_by_name.return_value = 12
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(12, False, False)
)
overrides.overrides = {12: override}
coordinator = _make_coordinator(data=[event], event_overrides=overrides)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
# Sync path sets slot_number from read-only lookup
assert sensor.extra_state_attributes["slot_number"] == 12
# Async path also sets slot_number from result.slot
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
assert sensor.extra_state_attributes["slot_number"] == 12
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_number_none_when_not_assigned(self, hass) -> None:
"""Verify slot_number is None when reconciliation has no assignment."""
async def test_slot_number_set_after_new_assignment(self, hass) -> None:
"""Verify slot_number is populated after a new slot assignment."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_assignment=None,
overrides = MagicMock()
overrides.ready = True
overrides.get_slot_with_name.return_value = None
overrides.get_slot_key_by_name.return_value = 0
overrides.async_reserve_or_get_slot = AsyncMock(
return_value=ReserveResult(10, True, False)
)
coordinator = _make_coordinator(data=[event], event_overrides=overrides)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_set_code",
new_callable=AsyncMock,
):
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["slot_number"] is None
# Sync path: get_slot_key_by_name returned 0, so None
assert sensor.extra_state_attributes["slot_number"] is None
coro = sensor.hass.async_create_task.call_args[0][0]
await coro
# Async path sets slot_number from result.slot
assert sensor.extra_state_attributes["slot_number"] == 10
def test_slot_number_cleared_when_no_events(self, hass) -> None:
"""Verify slot_number is reset to None when no events."""
@@ -1475,512 +1707,3 @@ class TestHandleCoordinatorUpdateOverrides:
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["slot_number"] is None
# ---------------------------------------------------------------------------
# T099: Sensor read-only — prove slot-mutation functions are never called
# ---------------------------------------------------------------------------
class TestSensorReadOnly:
"""T099: Prove _handle_coordinator_update never calls slot-mutation helpers.
After the reconciliation transition the sensor is purely read-only.
These tests act as regression guards: if any side-effect helper is
accidentally re-introduced in _handle_coordinator_update they will fail.
"""
@freeze_time("2025-03-10T12:00:00+00:00")
def test_does_not_call_async_reserve_or_get_slot(self, hass) -> None:
"""Verify async_reserve_or_get_slot is never called during update."""
event = _make_event()
overrides = MagicMock()
overrides.ready = True
coordinator = _make_coordinator(data=[event], event_overrides=overrides)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
overrides.async_reserve_or_get_slot.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_does_not_call_async_fire_set_code(self, hass) -> None:
"""Verify async_fire_set_code is never called during coordinator update."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_assignment=None,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_set_code",
create=True,
) as mock_set_code:
sensor._handle_coordinator_update()
mock_set_code.assert_not_called()
# No async task should be scheduled either
sensor.hass.async_create_task.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_does_not_call_async_fire_clear_code(self, hass) -> None:
"""Verify async_fire_clear_code is never called during coordinator update."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_assignment=5,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_clear_code",
create=True,
) as mock_clear_code:
sensor._handle_coordinator_update()
mock_clear_code.assert_not_called()
sensor.hass.async_create_task.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_does_not_call_async_fire_update_times(self, hass) -> None:
"""Verify async_fire_update_times is never called during coordinator update."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_assignment=5,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.async_fire_update_times",
create=True,
) as mock_update_times:
sensor._handle_coordinator_update()
mock_update_times.assert_not_called()
sensor.hass.async_create_task.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_no_async_task_scheduled(self, hass) -> None:
"""Verify _handle_coordinator_update never schedules an async task."""
event = _make_event()
overrides = MagicMock()
overrides.ready = True
coordinator = _make_coordinator(
data=[event],
event_overrides=overrides,
slot_assignment=7,
slot_code="1234",
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
sensor.hass.async_create_task.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_no_async_task_when_overrides_none(self, hass) -> None:
"""Verify no async task is scheduled even when event_overrides is None."""
event = _make_event()
coordinator = _make_coordinator(data=[event], event_overrides=None)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
sensor.hass.async_create_task.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
async def test_async_handle_slot_assignment_is_noop(self, hass) -> None:
"""Verify _async_handle_slot_assignment is a harmless no-op."""
coordinator = _make_coordinator(data=[], event_overrides=MagicMock())
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
await sensor._async_handle_slot_assignment(
slot_name="test",
slot_code="1234",
start_time=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end_time=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
uid=None,
prefix="",
eta_days=5,
)
coordinator.event_overrides.async_reserve_or_get_slot.assert_not_called()
# ---------------------------------------------------------------------------
# T100: event_N attribute regression tests
# ---------------------------------------------------------------------------
class TestEventNAttributeRegression:
"""T100: Regression tests for all public event_N sensor attributes.
Ensures slot_number, slot_code, summary, start, end, ETA, UID,
slot_name, and no-reservation state are correctly populated after the
read-only reconciliation transition.
"""
@freeze_time("2025-03-10T12:00:00+00:00")
def test_summary_attribute(self, hass) -> None:
"""Verify summary attribute is populated from the event."""
event = _make_event(summary="Reserved - Jane Smith")
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["summary"] == "Reserved - Jane Smith"
@freeze_time("2025-03-10T12:00:00+00:00")
def test_start_and_end_attributes(self, hass) -> None:
"""Verify start and end attributes are populated from the event."""
start = datetime(2025, 4, 1, 15, 0, tzinfo=timezone.utc)
end = datetime(2025, 4, 7, 11, 0, tzinfo=timezone.utc)
event = _make_event(start=start, end=end)
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
attrs = sensor.extra_state_attributes
assert attrs["start"] == start
assert attrs["end"] == end
@freeze_time("2025-03-10T12:00:00+00:00")
def test_eta_attributes_populated(self, hass) -> None:
"""Verify eta_days, eta_hours, and eta_minutes are populated for future events."""
start = datetime(2025, 3, 15, 12, 0, tzinfo=timezone.utc) # 5 days ahead
end = datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc)
event = _make_event(start=start, end=end)
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
attrs = sensor.extra_state_attributes
assert attrs["eta_days"] == 5
assert attrs["eta_hours"] is not None
assert attrs["eta_minutes"] is not None
@freeze_time("2025-03-20T12:00:00+00:00")
def test_eta_attributes_none_for_past_events(self, hass) -> None:
"""Verify ETA attributes are None for past events."""
start = datetime(2025, 3, 15, 12, 0, tzinfo=timezone.utc)
end = datetime(2025, 3, 18, 11, 0, tzinfo=timezone.utc)
event = _make_event(start=start, end=end)
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
attrs = sensor.extra_state_attributes
assert attrs["eta_days"] is None
assert attrs["eta_hours"] is None
assert attrs["eta_minutes"] is None
@freeze_time("2025-03-10T12:00:00+00:00")
def test_uid_attribute(self, hass) -> None:
"""Verify uid attribute is populated from the event."""
event = _make_event(uid="test-reservation-uid-42")
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["uid"] == "test-reservation-uid-42"
@freeze_time("2025-03-10T12:00:00+00:00")
def test_uid_none_when_event_has_no_uid(self, hass) -> None:
"""Verify uid attribute is None when the event has no UID."""
event = _make_event(uid=None)
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["uid"] is None
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_name_attribute(self, hass) -> None:
"""Verify slot_name attribute is populated via get_slot_name."""
event = _make_event(summary="Reserved - Alice Wonder")
coordinator = _make_coordinator(data=[event])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.get_slot_name",
return_value="Alice Wonder",
):
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["slot_name"] == "Alice Wonder"
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_number_from_reconciliation_state(self, hass) -> None:
"""Verify slot_number is read synchronously from reconciliation state."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_assignment=9,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
# slot_number is set synchronously — no async task needed
assert sensor.extra_state_attributes["slot_number"] == 9
sensor.hass.async_create_task.assert_not_called()
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_code_from_reconciliation_state(self, hass) -> None:
"""Verify slot_code is read synchronously from reconciliation state."""
event = _make_event()
coordinator = _make_coordinator(
data=[event],
event_overrides=MagicMock(),
slot_code="9876",
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["slot_code"] == "9876"
sensor.hass.async_create_task.assert_not_called()
def test_no_reservation_state(self, hass) -> None:
"""Verify no-reservation state resets all event_N attributes correctly."""
coordinator = _make_coordinator(data=[], event_prefix="")
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
attrs = sensor.extra_state_attributes
assert attrs["summary"] == "No reservation"
assert attrs["start"] is None
assert attrs["end"] is None
assert attrs["uid"] is None
assert attrs["eta_days"] is None
assert attrs["eta_hours"] is None
assert attrs["eta_minutes"] is None
assert attrs["slot_name"] is None
assert attrs["slot_code"] is None
assert attrs["slot_number"] is None
assert sensor.state == "No reservation"
def test_no_reservation_state_with_prefix(self, hass) -> None:
"""Verify no-reservation state includes event prefix when configured."""
coordinator = _make_coordinator(data=[], event_prefix="MyPlace")
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
attrs = sensor.extra_state_attributes
assert attrs["summary"] == "MyPlace No reservation"
assert sensor.state == "MyPlace No reservation"
@freeze_time("2025-03-10T12:00:00+00:00")
def test_slot_number_cleared_when_no_events(self, hass) -> None:
"""Verify slot_number is reset to None when event list is empty."""
coordinator = _make_coordinator(data=[])
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor._event_attributes["slot_number"] = 7
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
sensor._handle_coordinator_update()
assert sensor.extra_state_attributes["slot_number"] is None
@freeze_time("2025-03-10T12:00:00+00:00")
def test_identity_key_uses_coordinator_entry_id(self, hass) -> None:
"""Verify reconciliation lookup uses coordinator.entry_id for fingerprinting."""
event = _make_event()
overrides = MagicMock()
coordinator = _make_coordinator(
data=[event],
event_overrides=overrides,
entry_id="specific-entry-id",
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", 0)
sensor.hass = MagicMock()
sensor.async_write_ha_state = MagicMock()
with patch(
"custom_components.rental_control.sensors.calsensor.make_reservation_fingerprint",
return_value="fake-key",
) as mock_fp:
sensor._handle_coordinator_update()
# entry_id is always the first arg to make_reservation_fingerprint
args = mock_fp.call_args[0]
assert args[0] == "specific-entry-id"
class TestCodeRegenerationRegression:
"""T104 regression: Code generation semantics preserved.
Pins date-based code shift, static-random stability, and
date-based fallback when no description is available.
"""
def setup_method(self) -> None:
"""Save RNG state before each test."""
self._rng_state = random.getstate()
def teardown_method(self) -> None:
"""Restore RNG state after each test."""
random.setstate(self._rng_state)
@staticmethod
def _make_sensor(
hass,
*,
code_generator: str,
start: datetime,
end: datetime,
uid: str | None,
description: str | None,
event_number: int = 0,
) -> RentalControlCalSensor:
"""Build a sensor with event attributes primed for code generation."""
coordinator = _make_coordinator(
code_generator=code_generator,
code_length=4,
)
sensor = RentalControlCalSensor(hass, coordinator, f"{NAME} Test", event_number)
sensor._event_attributes["start"] = start
sensor._event_attributes["end"] = end
sensor._event_attributes["uid"] = uid
sensor._event_attributes["description"] = description
return sensor
def test_date_based_code_shifts_when_dates_change(self, hass) -> None:
"""Date-based codes change when reservation dates shift."""
sensor_a = self._make_sensor(
hass,
code_generator="date_based",
start=datetime(2025, 1, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 1, 17, 11, 0, tzinfo=timezone.utc),
uid="date-a",
description="Reservation A",
)
sensor_b = self._make_sensor(
hass,
code_generator="date_based",
start=datetime(2025, 1, 16, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 1, 18, 11, 0, tzinfo=timezone.utc),
uid="date-b",
description="Reservation B",
event_number=1,
)
assert sensor_a._generate_door_code() != sensor_b._generate_door_code()
def test_date_based_code_stable_when_dates_unchanged(self, hass) -> None:
"""Date-based codes stay stable when dates do not change."""
sensor_a = self._make_sensor(
hass,
code_generator="date_based",
start=datetime(2025, 1, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 1, 17, 11, 0, tzinfo=timezone.utc),
uid="same-a",
description="Reservation A",
)
sensor_b = self._make_sensor(
hass,
code_generator="date_based",
start=datetime(2025, 1, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 1, 17, 11, 0, tzinfo=timezone.utc),
uid="same-b",
description="Reservation B",
event_number=1,
)
assert sensor_a._generate_door_code() == sensor_b._generate_door_code()
def test_static_random_code_stable_across_refreshes(self, hass) -> None:
"""Static-random codes are deterministic for the same UID."""
sensor = self._make_sensor(
hass,
code_generator="static_random",
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
uid="stable-uid",
description="Original description",
)
first = sensor._generate_door_code()
second = sensor._generate_door_code()
assert first == second
def test_static_random_uid_beats_description_for_stability(self, hass) -> None:
"""UID stability wins even when descriptions change across refreshes."""
sensor_a = self._make_sensor(
hass,
code_generator="static_random",
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
uid="shared-uid",
description="Guest: Alice",
)
sensor_b = self._make_sensor(
hass,
code_generator="static_random",
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
uid="shared-uid",
description="Guest: Alice - updated note",
event_number=1,
)
assert sensor_a._generate_door_code() == sensor_b._generate_door_code()
def test_date_based_fallback_when_description_none(self, hass) -> None:
"""Missing description and UID force date-based generation."""
sensor = self._make_sensor(
hass,
code_generator="static_random",
start=datetime(2025, 3, 15, 16, 0, tzinfo=timezone.utc),
end=datetime(2025, 3, 20, 11, 0, tzinfo=timezone.utc),
uid=None,
description=None,
)
assert sensor._generate_door_code() == "1520"
tests/unit/test_slot_reconciliation.py
-4198
View File
File diff suppressed because it is too large. Load diff
tests/unit/test_util.py
+33 -509
View File
@@ -5,7 +5,6 @@
from __future__ import annotations
import asyncio
from datetime import date
from datetime import datetime
from datetime import timedelta
@@ -29,9 +28,7 @@ from custom_components.rental_control.const import DEFAULT_PATH
from custom_components.rental_control.const import DOMAIN
from custom_components.rental_control.const import NAME
from custom_components.rental_control.util import EventIdentity
from custom_components.rental_control.util import OperationResult
from custom_components.rental_control.util import add_call
from custom_components.rental_control.util import apply_buffer
from custom_components.rental_control.util import async_fire_clear_code
from custom_components.rental_control.util import async_fire_set_code
from custom_components.rental_control.util import async_fire_update_times
@@ -1565,16 +1562,14 @@ class TestAsyncFireSetCode:
]
assert len(name_calls) == 1
async def test_gather_exception_returns_failed_result_for_retry(
async def test_gather_exception_propagates_for_retry(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Verify a failing gather call is logged and returned as failed."""
"""Verify a failing gather call is logged and re-raised for retry."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.event_prefix = ""
coordinator.trim_names = False
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides.record_retry_failure.return_value = False
@@ -1582,50 +1577,17 @@ class TestAsyncFireSetCode:
coordinator.hass.services.async_call = AsyncMock(side_effect=error)
event = self._make_event()
with caplog.at_level(
logging.ERROR, logger="custom_components.rental_control.util"
with (
caplog.at_level(
logging.ERROR, logger="custom_components.rental_control.util"
),
pytest.raises(ServiceNotFound),
):
result = await async_fire_set_code(coordinator, event, 10)
await async_fire_set_code(coordinator, event, 10)
assert "Lock slot operation" in caplog.text
assert result.failed is True
assert result.error is not None
coordinator.event_overrides.record_retry_failure.assert_called_once_with(10)
async def test_set_code_flushes_before_verification(self) -> None:
"""Verify set_code flushes HA jobs before reading state."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.event_prefix = ""
coordinator.trim_names = False
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {}
coordinator.event_overrides.record_retry_success.return_value = None
coordinator.hass.services.async_call = AsyncMock()
coordinator.hass.async_block_till_done = AsyncMock()
coordinator.hass.states.get.return_value = MagicMock(state="Guest")
result = await async_fire_set_code(coordinator, self._make_event(), 10)
coordinator.hass.async_block_till_done.assert_awaited_once()
assert result.confirmed is True
async def test_set_code_cancelled_error_propagates(self) -> None:
"""Verify set_code does not swallow task cancellation."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.event_prefix = ""
coordinator.trim_names = False
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.hass.services.async_call = AsyncMock(
side_effect=asyncio.CancelledError()
)
with pytest.raises(asyncio.CancelledError):
await async_fire_set_code(coordinator, self._make_event(), 10)
# ---------------------------------------------------------------------------
# async_fire_update_times tests (T021)
@@ -1707,19 +1669,6 @@ class TestAsyncFireUpdateTimes:
assert "Lock slot operation" in caplog.text
async def test_cancelled_error_propagates(self) -> None:
"""Verify update_times does not swallow task cancellation."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.hass.services.async_call = AsyncMock(
side_effect=asyncio.CancelledError()
)
with pytest.raises(asyncio.CancelledError):
await async_fire_update_times(coordinator, self._make_event(), 10)
# ---------------------------------------------------------------------------
# Pre-execution verification abort tests (T020)
@@ -1841,7 +1790,6 @@ class TestRetryEscalation:
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.hass.services.async_call = AsyncMock()
coordinator.hass.states.get.return_value = MagicMock(state="Guest")
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {10: True}
@@ -1873,7 +1821,6 @@ class TestRetryEscalation:
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.hass.services.async_call = AsyncMock()
coordinator.hass.states.get.return_value = MagicMock(state="Guest")
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {}
@@ -1914,14 +1861,15 @@ class TestRetryEscalation:
"end": "2025-01-17T11:00:00",
}
with patch(
"custom_components.rental_control.util.pn_create",
) as mock_create:
result = await async_fire_set_code(coordinator, event, 10)
with (
patch(
"custom_components.rental_control.util.pn_create",
) as mock_create,
pytest.raises(Exception, match="service unavailable"),
):
await async_fire_set_code(coordinator, event, 10)
coordinator.event_overrides.record_retry_failure.assert_called_once_with(10)
assert result.failed is True
assert result.error is not None
mock_create.assert_called_once()
call_kwargs = mock_create.call_args
assert call_kwargs[1]["notification_id"] == "rental_control_slot_10_failure"
@@ -1946,14 +1894,15 @@ class TestRetryEscalation:
"end": "2025-01-17T11:00:00",
}
with patch(
"custom_components.rental_control.util.pn_create",
) as mock_create:
result = await async_fire_set_code(coordinator, event, 10)
with (
patch(
"custom_components.rental_control.util.pn_create",
) as mock_create,
pytest.raises(Exception, match="service unavailable"),
):
await async_fire_set_code(coordinator, event, 10)
coordinator.event_overrides.record_retry_failure.assert_called_once_with(10)
assert result.failed is True
assert result.error is not None
mock_create.assert_not_called()
async def test_clear_code_records_success_and_dismisses(self) -> None:
@@ -1962,7 +1911,7 @@ class TestRetryEscalation:
coordinator.name = "Test Rental"
coordinator.lockname = "front_door"
coordinator.hass.services.async_call = AsyncMock()
coordinator.hass.states.get.return_value = MagicMock(state="")
coordinator.hass.states.get.return_value = None
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {10: True}
@@ -1977,55 +1926,6 @@ class TestRetryEscalation:
notification_id="rental_control_slot_10_clear_failure",
)
async def test_set_code_unconfirmed_does_not_reset_retry(self) -> None:
"""Verify unconfirmed set_code leaves retry tracking intact."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.event_prefix = ""
coordinator.trim_names = False
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.hass.services.async_call = AsyncMock()
coordinator.hass.states.get.return_value = MagicMock(state="Other")
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {10: True}
event = MagicMock()
event.extra_state_attributes = {
"slot_name": "Guest",
"slot_code": "1234",
"start": "2025-01-15T16:00:00",
"end": "2025-01-17T11:00:00",
}
with patch(
"custom_components.rental_control.util.pn_dismiss",
) as mock_dismiss:
result = await async_fire_set_code(coordinator, event, 10)
assert result.unconfirmed is True
coordinator.event_overrides.record_retry_success.assert_not_called()
mock_dismiss.assert_not_called()
async def test_clear_code_unconfirmed_does_not_reset_retry(self) -> None:
"""Verify unconfirmed clear_code leaves retry tracking intact."""
coordinator = MagicMock()
coordinator.name = "Test Rental"
coordinator.lockname = "front_door"
coordinator.hass.services.async_call = AsyncMock()
coordinator.hass.states.get.return_value = None
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {10: True}
with patch(
"custom_components.rental_control.util.pn_dismiss",
) as mock_dismiss:
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.unconfirmed is True
coordinator.event_overrides.record_retry_success.assert_not_called()
mock_dismiss.assert_not_called()
async def test_clear_code_failure_escalates(self) -> None:
"""Verify clear_code failure creates notification on escalation."""
coordinator = MagicMock()
@@ -2038,22 +1938,23 @@ class TestRetryEscalation:
error = Exception("lock offline")
coordinator.hass.services.async_call = AsyncMock(side_effect=error)
with patch(
"custom_components.rental_control.util.pn_create",
) as mock_create:
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
with (
patch(
"custom_components.rental_control.util.pn_create",
) as mock_create,
pytest.raises(Exception, match="lock offline"),
):
await async_fire_clear_code(coordinator, 10, expected_name="Guest")
coordinator.event_overrides.record_retry_failure.assert_called_once_with(10)
assert result.failed is True
assert result.error is not None
mock_create.assert_called_once()
call_kwargs = mock_create.call_args
assert (
call_kwargs[1]["notification_id"] == "rental_control_slot_10_clear_failure"
)
async def test_clear_code_failure_returns_failed_result(self) -> None:
"""Verify failure is returned after recording retry state."""
async def test_clear_code_failure_reraises(self) -> None:
"""Verify exception is re-raised after recording failure."""
coordinator = MagicMock()
coordinator.name = "Test Rental"
coordinator.lockname = "front_door"
@@ -2063,21 +1964,7 @@ class TestRetryEscalation:
error = RuntimeError("hardware fault")
coordinator.hass.services.async_call = AsyncMock(side_effect=error)
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.failed is True
assert result.error == "hardware fault"
async def test_clear_code_cancelled_error_propagates(self) -> None:
"""Verify clear_code does not swallow task cancellation."""
coordinator = MagicMock()
coordinator.name = "Test Rental"
coordinator.lockname = "front_door"
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.hass.services.async_call = AsyncMock(
side_effect=asyncio.CancelledError()
)
with pytest.raises(asyncio.CancelledError):
with pytest.raises(RuntimeError, match="hardware fault"):
await async_fire_clear_code(coordinator, 10, expected_name="Guest")
@@ -2648,366 +2535,3 @@ class TestBufferInUpdateTimes:
assert event.extra_state_attributes["start"] == original_start
assert event.extra_state_attributes["end"] == original_end
class TestAsyncFireClearCodeOperationResult:
"""Tests for async_fire_clear_code OperationResult outcomes."""
def _make_coordinator(self) -> MagicMock:
"""Return a coordinator mock for clear-code tests."""
coordinator = MagicMock()
coordinator.name = "Test Rental"
coordinator.lockname = "front_door"
coordinator.hass.services.async_call = AsyncMock()
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {}
return coordinator
async def test_confirmed_when_name_and_pin_cleared(self) -> None:
"""Clear is confirmed when both name and PIN entities are cleared."""
coordinator = self._make_coordinator()
def states_get(entity_id: str) -> MagicMock:
"""Return a cleared mock state for any requested entity."""
state = MagicMock()
state.state = ""
return state
coordinator.hass.states.get.side_effect = states_get
with patch(
"custom_components.rental_control.util.asyncio.sleep",
new_callable=AsyncMock,
):
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result == OperationResult(kind="clear", slot=10, confirmed=True)
async def test_unconfirmed_when_name_state_none(self) -> None:
"""Clear is unconfirmed when the name entity cannot be read."""
coordinator = self._make_coordinator()
coordinator.hass.states.get.return_value = None
with patch(
"custom_components.rental_control.util.asyncio.sleep",
new_callable=AsyncMock,
):
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.unconfirmed is True
assert result.confirmed is False
async def test_lingering_name_when_name_persists(self) -> None:
"""Persistent name after reset yields lingering_name."""
coordinator = self._make_coordinator()
def _state(value: str) -> MagicMock:
"""Return a mock state with the provided value."""
state = MagicMock()
state.state = value
return state
name_reads = [_state("Ghost"), _state("Ghost")]
pin_state = _state("")
def states_get(entity_id: str) -> MagicMock:
"""Return persistent name reads and a cleared PIN state."""
if entity_id.endswith("_name"):
return name_reads.pop(0)
return pin_state
coordinator.hass.states.get.side_effect = states_get
with patch(
"custom_components.rental_control.util.asyncio.sleep",
new_callable=AsyncMock,
):
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.unconfirmed is True
assert result.lingering_name is True
assert result.confirmed is False
async def test_lingering_pin_when_pin_persists(self) -> None:
"""Persistent PIN after reset yields lingering_pin."""
coordinator = self._make_coordinator()
def states_get(entity_id: str) -> MagicMock:
"""Return a cleared name state and lingering PIN state."""
state = MagicMock()
if entity_id.endswith("_name"):
state.state = ""
else:
state.state = "5678"
return state
coordinator.hass.states.get.side_effect = states_get
with patch(
"custom_components.rental_control.util.asyncio.sleep",
new_callable=AsyncMock,
):
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.unconfirmed is True
assert result.lingering_pin is True
assert result.confirmed is False
async def test_service_failure_returns_failed(self) -> None:
"""Button press failure is returned as failed."""
coordinator = self._make_coordinator()
coordinator.hass.services.async_call = AsyncMock(
side_effect=RuntimeError("lock offline")
)
coordinator.event_overrides.record_retry_failure.return_value = False
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.failed is True
assert result.error == "lock offline"
async def test_no_lockname_returns_unconfirmed(self) -> None:
"""Missing lockname returns an unconfirmed result."""
coordinator = self._make_coordinator()
coordinator.lockname = ""
result = await async_fire_clear_code(coordinator, 10)
assert result.unconfirmed is True
async def test_ownership_failure_returns_unconfirmed(self) -> None:
"""Ownership mismatch returns an unconfirmed result."""
coordinator = self._make_coordinator()
coordinator.event_overrides.verify_slot_ownership.return_value = False
result = await async_fire_clear_code(coordinator, 10, expected_name="Guest")
assert result.unconfirmed is True
class TestAsyncFireSetCodeOperationResult:
"""Tests for async_fire_set_code OperationResult outcomes."""
@staticmethod
def _make_event() -> MagicMock:
"""Return a set-code event payload."""
event = MagicMock()
event.extra_state_attributes = {
"slot_name": "Guest",
"slot_code": "1234",
"start": "2025-01-15T16:00:00",
"end": "2025-01-17T11:00:00",
}
return event
def _make_coordinator(self) -> MagicMock:
"""Return a coordinator mock for set-code tests."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.event_prefix = ""
coordinator.trim_names = False
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.hass.services.async_call = AsyncMock()
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {}
return coordinator
async def test_confirmed_when_name_matches(self) -> None:
"""Set is confirmed when the written name can be read back."""
coordinator = self._make_coordinator()
state = MagicMock()
state.state = "Guest"
coordinator.hass.states.get.return_value = state
result = await async_fire_set_code(coordinator, self._make_event(), 10)
assert result == OperationResult(kind="set", slot=10, confirmed=True)
async def test_unconfirmed_when_name_state_none(self) -> None:
"""Set is unconfirmed when the name entity is unreadable."""
coordinator = self._make_coordinator()
coordinator.hass.states.get.return_value = None
result = await async_fire_set_code(coordinator, self._make_event(), 10)
assert result.unconfirmed is True
async def test_service_failure_returns_failed(self) -> None:
"""Service failure is returned as failed."""
coordinator = self._make_coordinator()
coordinator.hass.services.async_call = AsyncMock(
side_effect=RuntimeError("service unavailable")
)
coordinator.event_overrides.record_retry_failure.return_value = False
result = await async_fire_set_code(coordinator, self._make_event(), 10)
assert result.failed is True
assert result.error == "service unavailable"
async def test_no_lockname_returns_unconfirmed(self) -> None:
"""Missing lockname returns an unconfirmed result."""
coordinator = self._make_coordinator()
coordinator.lockname = ""
result = await async_fire_set_code(coordinator, self._make_event(), 10)
assert result.unconfirmed is True
class TestAsyncFireUpdateTimesOperationResult:
"""Tests for async_fire_update_times OperationResult outcomes."""
@staticmethod
def _make_event() -> MagicMock:
"""Return an update-times event payload."""
event = MagicMock()
event.extra_state_attributes = {
"slot_name": "Guest",
"start": "2025-01-15T16:00:00",
"end": "2025-01-17T11:00:00",
}
return event
def _make_coordinator(self) -> MagicMock:
"""Return a coordinator mock for update-times tests."""
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.code_buffer_before = 0
coordinator.code_buffer_after = 0
coordinator.hass.services.async_call = AsyncMock()
coordinator.event_overrides.verify_slot_ownership.return_value = True
return coordinator
async def test_confirmed_on_success(self) -> None:
"""Successful service calls return confirmed."""
coordinator = self._make_coordinator()
result = await async_fire_update_times(coordinator, self._make_event(), 10)
assert result == OperationResult(
kind="update_times",
slot=10,
confirmed=True,
)
async def test_failed_on_service_exception(self) -> None:
"""Gather failures are returned as failed."""
coordinator = self._make_coordinator()
coordinator.hass.services.async_call = AsyncMock(
side_effect=ServiceNotFound("datetime", "set_value")
)
result = await async_fire_update_times(coordinator, self._make_event(), 10)
assert result.failed is True
assert result.error is not None
async def test_no_lockname_returns_unconfirmed(self) -> None:
"""Missing lockname returns an unconfirmed result."""
coordinator = self._make_coordinator()
coordinator.lockname = ""
result = await async_fire_update_times(coordinator, self._make_event(), 10)
assert result.unconfirmed is True
class TestBufferRegressionSemantics:
"""T102 regression: Lock-code buffer semantics preserved after reconciliation.
These tests pin the apply_buffer behaviour: the buffered window
sent to Keymaster differs from the original event window, and the
original event attributes are never mutated.
"""
@staticmethod
def _make_coordinator() -> MagicMock:
"""Return a coordinator mock for buffered set-code tests."""
from zoneinfo import ZoneInfo
coordinator = MagicMock()
coordinator.lockname = "front_door"
coordinator.timezone = ZoneInfo("America/New_York")
coordinator.trim_names = False
coordinator.max_name_length = 0
coordinator.event_prefix = ""
coordinator.code_buffer_before = 30
coordinator.code_buffer_after = 15
coordinator.hass.services.async_call = AsyncMock()
state = MagicMock()
state.state = "Guest"
coordinator.hass.states.get.return_value = state
coordinator.event_overrides.verify_slot_ownership.return_value = True
coordinator.event_overrides._escalated = {}
return coordinator
def test_apply_buffer_zero_returns_inputs_unchanged(self) -> None:
"""Zero-minute buffers return the original objects unchanged."""
rc = MagicMock()
start = datetime(2025, 1, 15, 16, 0, tzinfo=dt_util.UTC)
end = datetime(2025, 1, 17, 11, 0, tzinfo=dt_util.UTC)
buffered_start, buffered_end = apply_buffer(start, end, 0, 0, rc)
assert buffered_start is start
assert buffered_end is end
def test_apply_buffer_before_shifts_start_only(self) -> None:
"""Before-buffer moves only the start backward."""
rc = MagicMock()
start = datetime(2025, 1, 15, 16, 0, tzinfo=dt_util.UTC)
end = datetime(2025, 1, 17, 11, 0, tzinfo=dt_util.UTC)
buffered_start, buffered_end = apply_buffer(start, end, 60, 0, rc)
assert buffered_start == start - timedelta(minutes=60)
assert buffered_end == end
def test_apply_buffer_after_extends_end_only(self) -> None:
"""After-buffer moves only the end forward."""
rc = MagicMock()
start = datetime(2025, 1, 15, 16, 0, tzinfo=dt_util.UTC)
end = datetime(2025, 1, 17, 11, 0, tzinfo=dt_util.UTC)
buffered_start, buffered_end = apply_buffer(start, end, 0, 30, rc)
assert buffered_start == start
assert buffered_end == end + timedelta(minutes=30)
def test_apply_buffer_converts_date_to_datetime(self) -> None:
"""Date-only values are normalised before buffer arithmetic."""
from zoneinfo import ZoneInfo
rc = MagicMock()
rc.timezone = ZoneInfo("America/New_York")
buffered_start, buffered_end = apply_buffer(
date(2025, 1, 15),
date(2025, 1, 17),
30,
15,
rc,
)
assert buffered_start == datetime(2025, 1, 14, 23, 30, tzinfo=rc.timezone)
assert buffered_end == datetime(2025, 1, 17, 0, 15, tzinfo=rc.timezone)
async def test_set_code_event_attributes_unchanged_after_buffer(self) -> None:
"""Buffered writes do not mutate the original event attributes."""
coordinator = self._make_coordinator()
original_start = datetime(2025, 1, 15, 16, 0, tzinfo=dt_util.UTC)
original_end = datetime(2025, 1, 17, 11, 0, tzinfo=dt_util.UTC)
event = MagicMock()
event.extra_state_attributes = {
"slot_name": "Guest",
"slot_code": "1234",
"start": original_start,
"end": original_end,
}
await async_fire_set_code(coordinator, event, 10)
assert event.extra_state_attributes["start"] == original_start
assert event.extra_state_attributes["end"] == original_end