Migrate loqed to use runtime_data (#147478)

* Migrate loqed to use runtime_data

* Fix tests
This commit is contained in:
epenet 2025-06-25 08:07:17 +02:00 committed by GitHub
parent f22b623968
commit 51da1bc25a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 22 additions and 37 deletions

View File

@ -2,28 +2,22 @@
from __future__ import annotations
import logging
import re
import aiohttp
from loqedAPI import loqed
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .coordinator import LoqedDataCoordinator
from .coordinator import LoqedConfigEntry, LoqedDataCoordinator
PLATFORMS: list[str] = [Platform.LOCK, Platform.SENSOR]
PLATFORMS: list[Platform] = [Platform.LOCK, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: LoqedConfigEntry) -> bool:
"""Set up loqed from a config entry."""
websession = async_get_clientsession(hass)
host = entry.data["bridge_ip"]
@ -49,19 +43,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: LoqedConfigEntry) -> bool:
"""Unload a config entry."""
coordinator: LoqedDataCoordinator = hass.data[DOMAIN][entry.entry_id]
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
await coordinator.remove_webhooks()
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
await entry.runtime_data.remove_webhooks()
return unload_ok

View File

@ -17,6 +17,8 @@ from .const import CONF_CLOUDHOOK_URL, DOMAIN
_LOGGER = logging.getLogger(__name__)
type LoqedConfigEntry = ConfigEntry[LoqedDataCoordinator]
class BatteryMessage(TypedDict):
"""Properties in a battery update message."""
@ -71,12 +73,12 @@ class StatusMessage(TypedDict):
class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
"""Data update coordinator for the loqed platform."""
config_entry: ConfigEntry
config_entry: LoqedConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: LoqedConfigEntry,
api: loqed.LoqedAPI,
lock: loqed.Lock,
) -> None:
@ -166,7 +168,9 @@ class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
await self.lock.deleteWebhook(webhook_index)
async def async_cloudhook_generate_url(hass: HomeAssistant, entry: ConfigEntry) -> str:
async def async_cloudhook_generate_url(
hass: HomeAssistant, entry: LoqedConfigEntry
) -> str:
"""Generate the full URL for a webhook_id."""
if CONF_CLOUDHOOK_URL not in entry.data:
webhook_url = await cloud.async_create_cloudhook(

View File

@ -6,12 +6,10 @@ import logging
from typing import Any
from homeassistant.components.lock import LockEntity, LockEntityFeature
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import LoqedDataCoordinator
from .const import DOMAIN
from .coordinator import LoqedConfigEntry, LoqedDataCoordinator
from .entity import LoqedEntity
WEBHOOK_API_ENDPOINT = "/api/loqed/webhook"
@ -21,13 +19,11 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: LoqedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Loqed lock platform."""
coordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([LoqedLock(coordinator)])
async_add_entities([LoqedLock(entry.runtime_data)])
class LoqedLock(LoqedEntity, LockEntity):

View File

@ -8,7 +8,6 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
@ -17,8 +16,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import LoqedDataCoordinator, StatusMessage
from .coordinator import LoqedConfigEntry, LoqedDataCoordinator, StatusMessage
from .entity import LoqedEntity
SENSORS: Final[tuple[SensorEntityDescription, ...]] = (
@ -43,11 +41,11 @@ SENSORS: Final[tuple[SensorEntityDescription, ...]] = (
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: LoqedConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Loqed lock platform."""
coordinator = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
async_add_entities(LoqedSensor(coordinator, sensor) for sensor in SENSORS)

View File

@ -8,8 +8,7 @@ from unittest.mock import AsyncMock, Mock, patch
from loqedAPI import loqed
import pytest
from homeassistant.components.loqed import DOMAIN
from homeassistant.components.loqed.const import CONF_CLOUDHOOK_URL
from homeassistant.components.loqed.const import CONF_CLOUDHOOK_URL, DOMAIN
from homeassistant.const import CONF_API_TOKEN, CONF_NAME, CONF_WEBHOOK_ID
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component

View File

@ -3,8 +3,6 @@
from loqedAPI import loqed
from homeassistant.components.lock import LockState
from homeassistant.components.loqed import LoqedDataCoordinator
from homeassistant.components.loqed.const import DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_LOCK,
@ -33,7 +31,7 @@ async def test_lock_responds_to_bolt_state_updates(
hass: HomeAssistant, integration: MockConfigEntry, lock: loqed.Lock
) -> None:
"""Tests the lock responding to updates."""
coordinator: LoqedDataCoordinator = hass.data[DOMAIN][integration.entry_id]
coordinator = integration.runtime_data
lock.bolt_state = "night_lock"
coordinator.async_update_listeners()