diff --git a/CODEOWNERS b/CODEOWNERS index 8b575d11e6c..3f8f27187f3 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -703,6 +703,8 @@ build.json @home-assistant/supervisor /tests/components/logi_circle/ @evanjd /homeassistant/components/lookin/ @ANMalko @bdraco /tests/components/lookin/ @ANMalko @bdraco +/homeassistant/components/loqed/ @mikewoudenberg +/tests/components/loqed/ @mikewoudenberg /homeassistant/components/lovelace/ @home-assistant/frontend /tests/components/lovelace/ @home-assistant/frontend /homeassistant/components/luci/ @mzdrale diff --git a/homeassistant/components/loqed/__init__.py b/homeassistant/components/loqed/__init__.py new file mode 100644 index 00000000000..1248c75612f --- /dev/null +++ b/homeassistant/components/loqed/__init__.py @@ -0,0 +1,55 @@ +"""The loqed integration.""" +from __future__ import annotations + +import logging +import re + +from loqedAPI import loqed + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import DOMAIN +from .coordinator import LoqedDataCoordinator + +PLATFORMS: list[str] = [Platform.LOCK] + + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up loqed from a config entry.""" + websession = async_get_clientsession(hass) + host = entry.data["bridge_ip"] + apiclient = loqed.APIClient(websession, f"http://{host}") + api = loqed.LoqedAPI(apiclient) + + lock = await api.async_get_lock( + entry.data["lock_key_key"], + entry.data["bridge_key"], + int(entry.data["lock_key_local_id"]), + re.sub(r"LOQED-([a-f0-9]+)\.local", r"\1", entry.data["bridge_mdns_hostname"]), + ) + coordinator = LoqedDataCoordinator(hass, api, lock, entry) + await coordinator.ensure_webhooks() + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator + + await coordinator.async_config_entry_first_refresh() + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> 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() + + return unload_ok diff --git a/homeassistant/components/loqed/config_flow.py b/homeassistant/components/loqed/config_flow.py new file mode 100644 index 00000000000..c757d2f0080 --- /dev/null +++ b/homeassistant/components/loqed/config_flow.py @@ -0,0 +1,166 @@ +"""Config flow for loqed integration.""" +from __future__ import annotations + +import logging +import re +from typing import Any + +import aiohttp +from loqedAPI import cloud_loqed, loqed +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.components import webhook +from homeassistant.components.zeroconf import ZeroconfServiceInfo +from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_NAME, CONF_WEBHOOK_ID +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResult +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Loqed.""" + + VERSION = 1 + DOMAIN = DOMAIN + _host: str | None = None + + async def validate_input( + self, hass: HomeAssistant, data: dict[str, Any] + ) -> dict[str, Any]: + """Validate the user input allows us to connect.""" + + # 1. Checking loqed-connection + try: + session = async_get_clientsession(hass) + cloud_api_client = cloud_loqed.CloudAPIClient( + session, + data[CONF_API_TOKEN], + ) + cloud_client = cloud_loqed.LoqedCloudAPI(cloud_api_client) + lock_data = await cloud_client.async_get_locks() + except aiohttp.ClientError: + _LOGGER.error("HTTP Connection error to loqed API") + raise CannotConnect from aiohttp.ClientError + + try: + selected_lock = next( + lock + for lock in lock_data["data"] + if lock["bridge_ip"] == self._host or lock["name"] == data.get("name") + ) + + apiclient = loqed.APIClient(session, f"http://{selected_lock['bridge_ip']}") + api = loqed.LoqedAPI(apiclient) + lock = await api.async_get_lock( + selected_lock["backend_key"], + selected_lock["bridge_key"], + selected_lock["local_id"], + selected_lock["bridge_ip"], + ) + + # checking getWebooks to check the bridgeKey + await lock.getWebhooks() + return { + "lock_key_key": selected_lock["key_secret"], + "bridge_key": selected_lock["bridge_key"], + "lock_key_local_id": selected_lock["local_id"], + "bridge_mdns_hostname": selected_lock["bridge_hostname"], + "bridge_ip": selected_lock["bridge_ip"], + "name": selected_lock["name"], + "id": selected_lock["id"], + } + except StopIteration: + raise InvalidAuth from StopIteration + except aiohttp.ClientError: + _LOGGER.error("HTTP Connection error to loqed lock") + raise CannotConnect from aiohttp.ClientError + + async def async_step_zeroconf( + self, discovery_info: ZeroconfServiceInfo + ) -> FlowResult: + """Handle zeroconf discovery.""" + host = discovery_info.host + self._host = host + + session = async_get_clientsession(self.hass) + apiclient = loqed.APIClient(session, f"http://{host}") + api = loqed.LoqedAPI(apiclient) + lock_data = await api.async_get_lock_details() + + # Check if already exists + await self.async_set_unique_id(lock_data["bridge_mac_wifi"]) + self._abort_if_unique_id_configured({CONF_HOST: host}) + + return await self.async_step_user() + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Show userform to user.""" + user_data_schema = ( + vol.Schema( + { + vol.Required(CONF_API_TOKEN): str, + } + ) + if self._host + else vol.Schema( + { + vol.Required(CONF_NAME): str, + vol.Required(CONF_API_TOKEN): str, + } + ) + ) + + if user_input is None: + return self.async_show_form( + step_id="user", + data_schema=user_data_schema, + description_placeholders={ + "config_url": "https://integrations.production.loqed.com/personal-access-tokens", + }, + ) + + errors = {} + + try: + info = await self.validate_input(self.hass, user_input) + except CannotConnect: + errors["base"] = "cannot_connect" + except InvalidAuth: + errors["base"] = "invalid_auth" + else: + await self.async_set_unique_id( + re.sub(r"LOQED-([a-f0-9]+)\.local", r"\1", info["bridge_mdns_hostname"]) + ) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title="LOQED Touch Smart Lock", + data=( + user_input | {CONF_WEBHOOK_ID: webhook.async_generate_id()} | info + ), + ) + + return self.async_show_form( + step_id="user", + data_schema=user_data_schema, + errors=errors, + description_placeholders={ + "config_url": "https://integrations.production.loqed.com/personal-access-tokens", + }, + ) + + +class CannotConnect(HomeAssistantError): + """Error to indicate we cannot connect.""" + + +class InvalidAuth(HomeAssistantError): + """Error to indicate there is invalid auth.""" diff --git a/homeassistant/components/loqed/const.py b/homeassistant/components/loqed/const.py new file mode 100644 index 00000000000..0374e72d5f0 --- /dev/null +++ b/homeassistant/components/loqed/const.py @@ -0,0 +1,6 @@ +"""Constants for the loqed integration.""" + + +DOMAIN = "loqed" +OAUTH2_AUTHORIZE = "https://app.loqed.com/API/integration_oauth3/login.php" +OAUTH2_TOKEN = "https://app.loqed.com/API/integration_oauth3/token.php" diff --git a/homeassistant/components/loqed/coordinator.py b/homeassistant/components/loqed/coordinator.py new file mode 100644 index 00000000000..ec7d5467b49 --- /dev/null +++ b/homeassistant/components/loqed/coordinator.py @@ -0,0 +1,152 @@ +"""Provides the coordinator for a LOQED lock.""" +import logging +from typing import TypedDict + +from aiohttp.web import Request +import async_timeout +from loqedAPI import loqed + +from homeassistant.components import webhook +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_WEBHOOK_ID +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +class BatteryMessage(TypedDict): + """Properties in a battery update message.""" + + mac_wifi: str + mac_ble: str + battery_type: str + battery_percentage: int + + +class StateReachedMessage(TypedDict): + """Properties in a battery update message.""" + + requested_state: str + requested_state_numeric: int + event_type: str + key_local_id: int + mac_wifi: str + mac_ble: str + + +class TransitionMessage(TypedDict): + """Properties in a battery update message.""" + + go_to_state: str + go_to_state_numeric: int + event_type: str + key_local_id: int + mac_wifi: str + mac_ble: str + + +class StatusMessage(TypedDict): + """Properties returned by the status endpoint of the bridhge.""" + + battery_percentage: int + battery_type: str + battery_type_numeric: int + battery_voltage: float + bolt_state: str + bolt_state_numeric: int + bridge_mac_wifi: str + bridge_mac_ble: str + lock_online: int + webhooks_number: int + ip_address: str + up_timestamp: int + wifi_strength: int + ble_strength: int + + +class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]): + """Data update coordinator for the loqed platform.""" + + def __init__( + self, + hass: HomeAssistant, + api: loqed.LoqedAPI, + lock: loqed.Lock, + entry: ConfigEntry, + ) -> None: + """Initialize the Loqed Data Update coordinator.""" + super().__init__(hass, _LOGGER, name="Loqed sensors") + self._hass = hass + self._api = api + self._entry = entry + self.lock = lock + + async def _async_update_data(self) -> StatusMessage: + """Fetch data from API endpoint.""" + async with async_timeout.timeout(10): + return await self._api.async_get_lock_details() + + @callback + async def _handle_webhook( + self, hass: HomeAssistant, webhook_id: str, request: Request + ) -> None: + """Handle incoming Loqed messages.""" + _LOGGER.debug("Callback received: %s", request.headers) + received_ts = request.headers["TIMESTAMP"] + received_hash = request.headers["HASH"] + body = await request.text() + + _LOGGER.debug("Callback body: %s", body) + + event_data = await self.lock.receiveWebhook(body, received_hash, received_ts) + if "error" in event_data: + _LOGGER.warning("Incorrect callback received:: %s", event_data) + return + + self.async_update_listeners() + + async def ensure_webhooks(self) -> None: + """Register webhook on LOQED bridge.""" + webhook_id = self._entry.data[CONF_WEBHOOK_ID] + + webhook.async_register( + self.hass, DOMAIN, "Loqed", webhook_id, self._handle_webhook + ) + webhook_url = webhook.async_generate_url(self.hass, webhook_id) + _LOGGER.info("Webhook URL: %s", webhook_url) + + webhooks = await self.lock.getWebhooks() + + webhook_index = next( + (x["id"] for x in webhooks if x["url"] == webhook_url), None + ) + + if not webhook_index: + await self.lock.registerWebhook(webhook_url) + webhooks = await self.lock.getWebhooks() + webhook_index = next(x["id"] for x in webhooks if x["url"] == webhook_url) + + _LOGGER.info("Webhook got index %s", webhook_index) + + async def remove_webhooks(self) -> None: + """Remove webhook from LOQED bridge.""" + webhook_id = self._entry.data[CONF_WEBHOOK_ID] + webhook_url = webhook.async_generate_url(self.hass, webhook_id) + + webhook.async_unregister( + self.hass, + webhook_id, + ) + _LOGGER.info("Webhook URL: %s", webhook_url) + + webhooks = await self.lock.getWebhooks() + + webhook_index = next( + (x["id"] for x in webhooks if x["url"] == webhook_url), None + ) + + if webhook_index: + await self.lock.deleteWebhook(webhook_index) diff --git a/homeassistant/components/loqed/entity.py b/homeassistant/components/loqed/entity.py new file mode 100644 index 00000000000..1b1731815b4 --- /dev/null +++ b/homeassistant/components/loqed/entity.py @@ -0,0 +1,29 @@ +"""Base entity for the LOQED integration.""" +from __future__ import annotations + +from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import DOMAIN +from .coordinator import LoqedDataCoordinator + + +class LoqedEntity(CoordinatorEntity[LoqedDataCoordinator]): + """Defines a LOQED entity.""" + + _attr_has_entity_name = True + + def __init__(self, coordinator: LoqedDataCoordinator) -> None: + """Initialize the LOQED entity.""" + super().__init__(coordinator=coordinator) + + lock_id = coordinator.lock.id + + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, lock_id)}, + manufacturer="LOQED", + name="LOQED Lock", + model="Touch Smart Lock", + connections={(CONNECTION_NETWORK_MAC, lock_id)}, + ) diff --git a/homeassistant/components/loqed/lock.py b/homeassistant/components/loqed/lock.py new file mode 100644 index 00000000000..5a7540ba89e --- /dev/null +++ b/homeassistant/components/loqed/lock.py @@ -0,0 +1,85 @@ +"""LOQED lock integration for Home Assistant.""" +from __future__ import annotations + +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 AddEntitiesCallback + +from . import LoqedDataCoordinator +from .const import DOMAIN +from .entity import LoqedEntity + +WEBHOOK_API_ENDPOINT = "/api/loqed/webhook" + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: + """Set up the Loqed lock platform.""" + coordinator = hass.data[DOMAIN][entry.entry_id] + + async_add_entities([LoqedLock(coordinator, entry.data["name"])]) + + +class LoqedLock(LoqedEntity, LockEntity): + """Representation of a loqed lock.""" + + _attr_supported_features = LockEntityFeature.OPEN + + def __init__(self, coordinator: LoqedDataCoordinator, name: str) -> None: + """Initialize the lock.""" + super().__init__(coordinator) + self._lock = coordinator.lock + self._attr_unique_id = self._lock.id + self._attr_name = name + + @property + def changed_by(self) -> str: + """Return internal ID of last used key.""" + return "KeyID " + str(self._lock.last_key_id) + + @property + def is_locking(self) -> bool | None: + """Return true if lock is locking.""" + return self._lock.bolt_state == "locking" + + @property + def is_unlocking(self) -> bool | None: + """Return true if lock is unlocking.""" + return self._lock.bolt_state == "unlocking" + + @property + def is_jammed(self) -> bool | None: + """Return true if lock is jammed.""" + return self._lock.bolt_state == "motor_stall" + + @property + def is_locked(self) -> bool | None: + """Return true if lock is locked.""" + return self._lock.bolt_state in ["night_lock_remote", "night_lock"] + + async def async_lock(self, **kwargs: Any) -> None: + """Lock the lock.""" + await self._lock.lock() + + async def async_unlock(self, **kwargs: Any) -> None: + """Unlock the lock.""" + await self._lock.unlock() + + async def async_open(self, **kwargs: Any) -> None: + """Open the door latch.""" + await self._lock.open() + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + _LOGGER.debug(self.coordinator.data) + if "bolt_state" in self.coordinator.data: + self._lock.updateState(self.coordinator.data["bolt_state"]).close() + self.async_write_ha_state() diff --git a/homeassistant/components/loqed/manifest.json b/homeassistant/components/loqed/manifest.json new file mode 100644 index 00000000000..1000d8f804d --- /dev/null +++ b/homeassistant/components/loqed/manifest.json @@ -0,0 +1,16 @@ +{ + "domain": "loqed", + "name": "LOQED Touch Smart Lock", + "codeowners": ["@mikewoudenberg"], + "config_flow": true, + "dependencies": ["webhook"], + "documentation": "https://www.home-assistant.io/integrations/loqed", + "iot_class": "local_push", + "requirements": ["loqedAPI==2.1.7"], + "zeroconf": [ + { + "type": "_http._tcp.local.", + "name": "loqed*" + } + ] +} diff --git a/homeassistant/components/loqed/strings.json b/homeassistant/components/loqed/strings.json new file mode 100644 index 00000000000..5448f01b7f9 --- /dev/null +++ b/homeassistant/components/loqed/strings.json @@ -0,0 +1,22 @@ +{ + "config": { + "flow_title": "LOQED Touch Smartlock setup", + "step": { + "user": { + "description": "Login at {config_url} and: \n* Create an API-key by clicking 'Create' \n* Copy the created access token.", + "data": { + "name": "Name of your lock in the LOQED app.", + "api_key": "[%key:common::config_flow::data::api_key%]" + } + } + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + } + } +} diff --git a/homeassistant/components/loqed/translations/en.json b/homeassistant/components/loqed/translations/en.json new file mode 100644 index 00000000000..a961f10cb1b --- /dev/null +++ b/homeassistant/components/loqed/translations/en.json @@ -0,0 +1,22 @@ +{ + "config": { + "abort": { + "already_configured": "Device is already configured" + }, + "error": { + "cannot_connect": "Failed to connect", + "invalid_auth": "Invalid authentication", + "unknown": "Unexpected error" + }, + "flow_title": "LOQED Touch Smartlock setup", + "step": { + "user": { + "data": { + "api_key": "API Key", + "name": "Name of your lock in the LOQED app." + }, + "description": "Login at {config_url} and: \n* Create an API-key by clicking 'Create' \n* Copy the created access token." + } + } + } +} \ No newline at end of file diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index fe7586e5890..3145c5cdc49 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -254,6 +254,7 @@ FLOWS = { "locative", "logi_circle", "lookin", + "loqed", "luftdaten", "lutron_caseta", "lyric", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 5e10fdf86ff..26833d62368 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -3085,6 +3085,12 @@ "config_flow": true, "iot_class": "local_push" }, + "loqed": { + "name": "LOQED Touch Smart Lock", + "integration_type": "hub", + "config_flow": true, + "iot_class": "local_push" + }, "luftdaten": { "name": "Sensor.Community", "integration_type": "device", diff --git a/homeassistant/generated/zeroconf.py b/homeassistant/generated/zeroconf.py index ea4c1c92816..ccf35e384bb 100644 --- a/homeassistant/generated/zeroconf.py +++ b/homeassistant/generated/zeroconf.py @@ -433,6 +433,10 @@ ZEROCONF = { "domain": "bosch_shc", "name": "bosch shc*", }, + { + "domain": "loqed", + "name": "loqed*", + }, { "domain": "nam", "name": "nam-*", diff --git a/requirements_all.txt b/requirements_all.txt index a5a681624e0..85793819eca 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1145,6 +1145,9 @@ logi-circle==0.2.3 # homeassistant.components.london_underground london-tube-status==0.5 +# homeassistant.components.loqed +loqedAPI==2.1.7 + # homeassistant.components.luftdaten luftdaten==0.7.4 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 864a425f0be..b528e286af2 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -874,6 +874,9 @@ life360==5.5.0 # homeassistant.components.logi_circle logi-circle==0.2.3 +# homeassistant.components.loqed +loqedAPI==2.1.7 + # homeassistant.components.luftdaten luftdaten==0.7.4 diff --git a/tests/components/loqed/__init__.py b/tests/components/loqed/__init__.py new file mode 100644 index 00000000000..c40c0f7c8f5 --- /dev/null +++ b/tests/components/loqed/__init__.py @@ -0,0 +1 @@ +"""Tests for the Loqed integration.""" diff --git a/tests/components/loqed/conftest.py b/tests/components/loqed/conftest.py new file mode 100644 index 00000000000..da7009a5744 --- /dev/null +++ b/tests/components/loqed/conftest.py @@ -0,0 +1,72 @@ +"""Contains fixtures for Loqed tests.""" + +from collections.abc import AsyncGenerator +import json +from typing import Any +from unittest.mock import AsyncMock, Mock, patch + +from loqedAPI import loqed +import pytest + +from homeassistant.components.loqed import 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 + +from tests.common import MockConfigEntry, load_fixture + + +@pytest.fixture(name="config_entry") +def config_entry_fixture() -> MockConfigEntry: + """Mock config entry.""" + + config = load_fixture("loqed/integration_config.json") + json_config = json.loads(config) + return MockConfigEntry( + version=1, + domain=DOMAIN, + data={ + "id": "Foo", + "bridge_ip": json_config["bridge_ip"], + "bridge_mdns_hostname": json_config["bridge_mdns_hostname"], + "bridge_key": json_config["bridge_key"], + "lock_key_local_id": int(json_config["lock_key_local_id"]), + "lock_key_key": json_config["lock_key_key"], + CONF_WEBHOOK_ID: "Webhook_id", + CONF_API_TOKEN: "Token", + CONF_NAME: "Home", + }, + ) + + +@pytest.fixture(name="lock") +def lock_fixture() -> loqed.Lock: + """Set up a mock implementation of a Lock.""" + webhooks_fixture = json.loads(load_fixture("loqed/get_all_webhooks.json")) + + mock_lock = Mock(spec=loqed.Lock, id="Foo", last_key_id=2) + mock_lock.name = "LOQED smart lock" + mock_lock.getWebhooks = AsyncMock(return_value=webhooks_fixture) + mock_lock.bolt_state = "locked" + return mock_lock + + +@pytest.fixture(name="integration") +async def integration_fixture( + hass: HomeAssistant, config_entry: MockConfigEntry, lock: loqed.Lock +) -> AsyncGenerator[MockConfigEntry, None]: + """Set up the loqed integration with a config entry.""" + config: dict[str, Any] = {DOMAIN: {CONF_API_TOKEN: ""}} + config_entry.add_to_hass(hass) + + lock_status = json.loads(load_fixture("loqed/status_ok.json")) + + with patch("loqedAPI.loqed.LoqedAPI.async_get_lock", return_value=lock), patch( + "loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_status + ), patch( + "homeassistant.components.webhook.async_generate_url", + return_value="http://hook_id", + ): + await async_setup_component(hass, DOMAIN, config) + await hass.async_block_till_done() + yield config_entry diff --git a/tests/components/loqed/fixtures/battery_update.json b/tests/components/loqed/fixtures/battery_update.json new file mode 100644 index 00000000000..be62d7f14f8 --- /dev/null +++ b/tests/components/loqed/fixtures/battery_update.json @@ -0,0 +1,6 @@ +{ + "battery_type": "NICKEL_METAL_HYDRIDE", + "battery_percentage": 88, + "mac_wifi": "**REDACTED**", + "mac_ble": "**REDACTED**" +} diff --git a/tests/components/loqed/fixtures/get_all_locks.json b/tests/components/loqed/fixtures/get_all_locks.json new file mode 100644 index 00000000000..bd8489c5c87 --- /dev/null +++ b/tests/components/loqed/fixtures/get_all_locks.json @@ -0,0 +1,26 @@ +{ + "data": [ + { + "id": "Foo", + "name": "MyLock", + "battery_percentage": 64, + "battery_type": "nickel_metal_hydride", + "bolt_state": "day_lock", + "party_mode": false, + "guest_access_mode": false, + "twist_assist": false, + "touch_to_connect": true, + "lock_direction": "clockwise", + "mortise_lock_type": "cylinder_operated_no_handle_on_the_outside", + "supported_lock_states": ["open", "day_lock", "night_lock"], + "online": true, + "bridge_ip": "192.168.12.34", + "bridge_hostname": "LOQED-aabbccddeeff.local", + "local_id": 1, + "key_secret": "SGFsbG8gd2VyZWxk", + "backend_key": "aGVsbG8gd29ybGQ=", + "bridge_webhook_count": 1, + "bridge_key": "Ym9uam91ciBtb25kZQ==" + } + ] +} diff --git a/tests/components/loqed/fixtures/get_all_webhooks.json b/tests/components/loqed/fixtures/get_all_webhooks.json new file mode 100644 index 00000000000..cf53fcf56a9 --- /dev/null +++ b/tests/components/loqed/fixtures/get_all_webhooks.json @@ -0,0 +1,15 @@ +[ + { + "id": 1, + "url": "http://hook_id", + "trigger_state_changed_open": 1, + "trigger_state_changed_latch": 1, + "trigger_state_changed_night_lock": 1, + "trigger_state_changed_unknown": 1, + "trigger_state_goto_open": 1, + "trigger_state_goto_latch": 1, + "trigger_state_goto_night_lock": 1, + "trigger_battery": 1, + "trigger_online_status": 1 + } +] diff --git a/tests/components/loqed/fixtures/integration_config.json b/tests/components/loqed/fixtures/integration_config.json new file mode 100644 index 00000000000..c4d9f87162c --- /dev/null +++ b/tests/components/loqed/fixtures/integration_config.json @@ -0,0 +1,9 @@ +{ + "lock_id": "**REDACTED**", + "lock_key_local_id": 1, + "lock_key_key": "SGFsbG8gd2VyZWxk", + "backend_key": "aGVsbG8gd29ybGQ=", + "bridge_key": "Ym9uam91ciBtb25kZQ==", + "bridge_ip": "192.168.12.34", + "bridge_mdns_hostname": "LOQED-aabbccddeeff.local" +} diff --git a/tests/components/loqed/fixtures/lock_going_to_daylock.json b/tests/components/loqed/fixtures/lock_going_to_daylock.json new file mode 100644 index 00000000000..a07ba43d25c --- /dev/null +++ b/tests/components/loqed/fixtures/lock_going_to_daylock.json @@ -0,0 +1,8 @@ +{ + "go_to_state": "DAY_LOCK", + "go_to_state_numeric": 2, + "mac_wifi": "**REDACTED**", + "mac_ble": "**REDACTED**", + "event_type": "GO_TO_STATE_TWIST_ASSIST_LATCH", + "key_local_id": 255 +} diff --git a/tests/components/loqed/fixtures/lock_going_to_nightlock.json b/tests/components/loqed/fixtures/lock_going_to_nightlock.json new file mode 100644 index 00000000000..8f7ce2ec025 --- /dev/null +++ b/tests/components/loqed/fixtures/lock_going_to_nightlock.json @@ -0,0 +1,8 @@ +{ + "go_to_state": "NIGHT_LOCK", + "go_to_state_numeric": 3, + "mac_wifi": "**REDACTED**", + "mac_ble": "**REDACTED**", + "event_type": "GO_TO_STATE_TWIST_ASSIST_LATCH", + "key_local_id": 255 +} diff --git a/tests/components/loqed/fixtures/nightlock_reached.json b/tests/components/loqed/fixtures/nightlock_reached.json new file mode 100644 index 00000000000..f8ce0d45cae --- /dev/null +++ b/tests/components/loqed/fixtures/nightlock_reached.json @@ -0,0 +1,8 @@ +{ + "requested_state": "NIGHT_LOCK", + "requested_state_numeric": 3, + "mac_wifi": "**REDACTED**", + "mac_ble": "**REDACTED**", + "event_type": "STATE_CHANGED_LATCH", + "key_local_id": 255 +} diff --git a/tests/components/loqed/fixtures/status_ok.json b/tests/components/loqed/fixtures/status_ok.json new file mode 100644 index 00000000000..4e5362507e6 --- /dev/null +++ b/tests/components/loqed/fixtures/status_ok.json @@ -0,0 +1,16 @@ +{ + "battery_percentage": 78, + "battery_type": "NICKEL_METAL_HYDRIDE", + "battery_type_numeric": 1, + "battery_voltage": 10.37, + "bolt_state": "day_lock", + "bolt_state_numeric": 2, + "bridge_mac_wifi": "***REDACTED***", + "bridge_mac_ble": "***REDACTED***", + "lock_online": 1, + "webhooks_number": 1, + "ip_address": "192.168.42.12", + "up_timestamp": 1653041994, + "wifi_strength": 73, + "ble_strength": 20 +} diff --git a/tests/components/loqed/test_config_flow.py b/tests/components/loqed/test_config_flow.py new file mode 100644 index 00000000000..c9c577e7199 --- /dev/null +++ b/tests/components/loqed/test_config_flow.py @@ -0,0 +1,220 @@ +"""Test the Loqed config flow.""" +import json +from unittest.mock import Mock, patch + +import aiohttp +from loqedAPI import loqed + +from homeassistant import config_entries +from homeassistant.components import zeroconf +from homeassistant.components.loqed.const import DOMAIN +from homeassistant.const import CONF_API_TOKEN, CONF_NAME, CONF_WEBHOOK_ID +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from tests.common import load_fixture +from tests.test_util.aiohttp import AiohttpClientMocker + +zeroconf_data = zeroconf.ZeroconfServiceInfo( + host="192.168.12.34", + addresses=["127.0.0.1"], + hostname="LOQED-ffeeddccbbaa.local", + name="mock_name", + port=9123, + properties={}, + type="mock_type", +) + + +async def test_create_entry_zeroconf(hass: HomeAssistant) -> None: + """Test we get can create a lock via zeroconf.""" + lock_result = json.loads(load_fixture("loqed/status_ok.json")) + + with patch( + "loqedAPI.loqed.LoqedAPI.async_get_lock_details", + return_value=lock_result, + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=zeroconf_data, + ) + + assert result["type"] == FlowResultType.FORM + assert result["errors"] is None + + mock_lock = Mock(spec=loqed.Lock, id="Foo") + webhook_id = "Webhook_ID" + all_locks_response = json.loads(load_fixture("loqed/get_all_locks.json")) + + with patch( + "loqedAPI.cloud_loqed.LoqedCloudAPI.async_get_locks", + return_value=all_locks_response, + ), patch( + "loqedAPI.loqed.LoqedAPI.async_get_lock", + return_value=mock_lock, + ), patch( + "homeassistant.components.loqed.async_setup_entry", + return_value=True, + ), patch( + "homeassistant.components.webhook.async_generate_id", return_value=webhook_id + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_API_TOKEN: "eyadiuyfasiuasf", + }, + ) + await hass.async_block_till_done() + found_lock = all_locks_response["data"][0] + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "LOQED Touch Smart Lock" + assert result2["data"] == { + "id": "Foo", + "lock_key_key": found_lock["key_secret"], + "bridge_key": found_lock["bridge_key"], + "lock_key_local_id": found_lock["local_id"], + "bridge_mdns_hostname": found_lock["bridge_hostname"], + "bridge_ip": found_lock["bridge_ip"], + "name": found_lock["name"], + CONF_WEBHOOK_ID: webhook_id, + CONF_API_TOKEN: "eyadiuyfasiuasf", + } + mock_lock.getWebhooks.assert_awaited() + + +async def test_create_entry_user( + hass: HomeAssistant, aioclient_mock: AiohttpClientMocker +) -> None: + """Test we can create a lock via manual entry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert result["type"] == FlowResultType.FORM + assert result["errors"] is None + + lock_result = json.loads(load_fixture("loqed/status_ok.json")) + mock_lock = Mock(spec=loqed.Lock, id="Foo") + webhook_id = "Webhook_ID" + all_locks_response = json.loads(load_fixture("loqed/get_all_locks.json")) + found_lock = all_locks_response["data"][0] + + with patch( + "loqedAPI.cloud_loqed.LoqedCloudAPI.async_get_locks", + return_value=all_locks_response, + ), patch( + "loqedAPI.loqed.LoqedAPI.async_get_lock", + return_value=mock_lock, + ), patch( + "homeassistant.components.loqed.async_setup_entry", + return_value=True, + ), patch( + "homeassistant.components.webhook.async_generate_id", return_value=webhook_id + ), patch( + "loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_result + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_API_TOKEN: "eyadiuyfasiuasf", CONF_NAME: "MyLock"}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "LOQED Touch Smart Lock" + assert result2["data"] == { + "id": "Foo", + "lock_key_key": found_lock["key_secret"], + "bridge_key": found_lock["bridge_key"], + "lock_key_local_id": found_lock["local_id"], + "bridge_mdns_hostname": found_lock["bridge_hostname"], + "bridge_ip": found_lock["bridge_ip"], + "name": found_lock["name"], + CONF_WEBHOOK_ID: webhook_id, + CONF_API_TOKEN: "eyadiuyfasiuasf", + } + mock_lock.getWebhooks.assert_awaited() + + +async def test_cannot_connect( + hass: HomeAssistant, aioclient_mock: AiohttpClientMocker +) -> None: + """Test we handle cannot connect error.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert result["type"] == FlowResultType.FORM + assert result["errors"] is None + + with patch( + "loqedAPI.cloud_loqed.LoqedCloudAPI.async_get_locks", + side_effect=aiohttp.ClientError, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_API_TOKEN: "eyadiuyfasiuasf", CONF_NAME: "MyLock"}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["errors"] == {"base": "cannot_connect"} + + +async def test_invalid_auth_when_lock_not_found( + hass: HomeAssistant, aioclient_mock: AiohttpClientMocker +) -> None: + """Test we handle a situation where the user enters an invalid lock name.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert result["type"] == FlowResultType.FORM + assert result["errors"] is None + + all_locks_response = json.loads(load_fixture("loqed/get_all_locks.json")) + + with patch( + "loqedAPI.cloud_loqed.LoqedCloudAPI.async_get_locks", + return_value=all_locks_response, + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_API_TOKEN: "eyadiuyfasiuasf", CONF_NAME: "MyLock2"}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["errors"] == {"base": "invalid_auth"} + + +async def test_cannot_connect_when_lock_not_reachable( + hass: HomeAssistant, aioclient_mock: AiohttpClientMocker +) -> None: + """Test we handle a situation where the user enters an invalid lock name.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert result["type"] == FlowResultType.FORM + assert result["errors"] is None + + all_locks_response = json.loads(load_fixture("loqed/get_all_locks.json")) + + with patch( + "loqedAPI.cloud_loqed.LoqedCloudAPI.async_get_locks", + return_value=all_locks_response, + ), patch("loqedAPI.loqed.LoqedAPI.async_get_lock", side_effect=aiohttp.ClientError): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {CONF_API_TOKEN: "eyadiuyfasiuasf", CONF_NAME: "MyLock"}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["errors"] == {"base": "cannot_connect"} diff --git a/tests/components/loqed/test_init.py b/tests/components/loqed/test_init.py new file mode 100644 index 00000000000..89b67ee3258 --- /dev/null +++ b/tests/components/loqed/test_init.py @@ -0,0 +1,106 @@ +"""Tests the init part of the Loqed integration.""" + +import json +from typing import Any +from unittest.mock import AsyncMock, patch + +from loqedAPI import loqed + +from homeassistant.components.loqed.const import DOMAIN +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import CONF_WEBHOOK_ID +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry, load_fixture + + +async def test_webhook_rejects_invalid_message( + hass: HomeAssistant, + hass_client_no_auth, + integration: MockConfigEntry, + lock: loqed.Lock, +): + """Test webhook called with invalid message.""" + await async_setup_component(hass, "http", {"http": {}}) + client = await hass_client_no_auth() + + coordinator = hass.data[DOMAIN][integration.entry_id] + lock.receiveWebhook = AsyncMock(return_value={"error": "invalid hash"}) + + with patch.object(coordinator, "async_set_updated_data") as mock: + message = load_fixture("loqed/battery_update.json") + timestamp = 1653304609 + await client.post( + f"/api/webhook/{integration.data[CONF_WEBHOOK_ID]}", + data=message, + headers={"timestamp": str(timestamp), "hash": "incorrect hash"}, + ) + + mock.assert_not_called() + + +async def test_webhook_accepts_valid_message( + hass: HomeAssistant, + hass_client_no_auth, + integration: MockConfigEntry, + lock: loqed.Lock, +): + """Test webhook called with valid message.""" + await async_setup_component(hass, "http", {"http": {}}) + client = await hass_client_no_auth() + processed_message = json.loads(load_fixture("loqed/battery_update.json")) + coordinator = hass.data[DOMAIN][integration.entry_id] + lock.receiveWebhook = AsyncMock(return_value=processed_message) + + with patch.object(coordinator, "async_update_listeners") as mock: + message = load_fixture("loqed/battery_update.json") + timestamp = 1653304609 + await client.post( + f"/api/webhook/{integration.data[CONF_WEBHOOK_ID]}", + data=message, + headers={"timestamp": str(timestamp), "hash": "incorrect hash"}, + ) + + mock.assert_called() + + +async def test_setup_webhook_in_bridge( + hass: HomeAssistant, config_entry: MockConfigEntry, lock: loqed.Lock +): + """Test webhook setup in loqed bridge.""" + config: dict[str, Any] = {DOMAIN: {}} + config_entry.add_to_hass(hass) + + lock_status = json.loads(load_fixture("loqed/status_ok.json")) + webhooks_fixture = json.loads(load_fixture("loqed/get_all_webhooks.json")) + lock.getWebhooks = AsyncMock(side_effect=[[], webhooks_fixture]) + + with patch("loqedAPI.loqed.LoqedAPI.async_get_lock", return_value=lock), patch( + "loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_status + ), patch( + "homeassistant.components.webhook.async_generate_url", + return_value="http://hook_id", + ): + await async_setup_component(hass, DOMAIN, config) + await hass.async_block_till_done() + + lock.registerWebhook.assert_called_with("http://hook_id") + + +async def test_unload_entry(hass, integration: MockConfigEntry, lock: loqed.Lock): + """Test successful unload of entry.""" + + assert await hass.config_entries.async_unload(integration.entry_id) + await hass.async_block_till_done() + + lock.deleteWebhook.assert_called_with(1) + assert integration.state is ConfigEntryState.NOT_LOADED + assert not hass.data.get(DOMAIN) + + +async def test_unload_entry_fails(hass, integration: MockConfigEntry, lock: loqed.Lock): + """Test unsuccessful unload of entry.""" + lock.deleteWebhook = AsyncMock(side_effect=Exception) + + assert not await hass.config_entries.async_unload(integration.entry_id) diff --git a/tests/components/loqed/test_lock.py b/tests/components/loqed/test_lock.py new file mode 100644 index 00000000000..422b7ab6830 --- /dev/null +++ b/tests/components/loqed/test_lock.py @@ -0,0 +1,87 @@ +"""Tests the lock platform of the Loqed integration.""" +from loqedAPI import loqed + +from homeassistant.components.loqed import LoqedDataCoordinator +from homeassistant.components.loqed.const import DOMAIN +from homeassistant.const import ( + ATTR_ENTITY_ID, + SERVICE_LOCK, + SERVICE_OPEN, + SERVICE_UNLOCK, + STATE_LOCKED, + STATE_UNLOCKED, +) +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + + +async def test_lock_entity( + hass: HomeAssistant, + integration: MockConfigEntry, +) -> None: + """Test the lock entity.""" + entity_id = "lock.loqed_lock_home" + + state = hass.states.get(entity_id) + + assert state + assert state.state == STATE_UNLOCKED + + +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] + lock.bolt_state = "night_lock" + coordinator.async_update_listeners() + + entity_id = "lock.loqed_lock_home" + + state = hass.states.get(entity_id) + + assert state + assert state.state == STATE_LOCKED + + +async def test_lock_transition_to_unlocked( + hass: HomeAssistant, integration: MockConfigEntry, lock: loqed.Lock +) -> None: + """Tests the lock transitions to unlocked state.""" + + entity_id = "lock.loqed_lock_home" + + await hass.services.async_call( + "lock", SERVICE_UNLOCK, {ATTR_ENTITY_ID: entity_id}, blocking=True + ) + await hass.async_block_till_done() + lock.unlock.assert_called() + + +async def test_lock_transition_to_locked( + hass: HomeAssistant, integration: MockConfigEntry, lock: loqed.Lock +) -> None: + """Tests the lock transitions to locked state.""" + + entity_id = "lock.loqed_lock_home" + + await hass.services.async_call( + "lock", SERVICE_LOCK, {ATTR_ENTITY_ID: entity_id}, blocking=True + ) + await hass.async_block_till_done() + lock.lock.assert_called() + + +async def test_lock_transition_to_open( + hass: HomeAssistant, integration: MockConfigEntry, lock: loqed.Lock +) -> None: + """Tests the lock transitions to open state.""" + + entity_id = "lock.loqed_lock_home" + + await hass.services.async_call( + "lock", SERVICE_OPEN, {ATTR_ENTITY_ID: entity_id}, blocking=True + ) + await hass.async_block_till_done() + lock.open.assert_called()