Add new integration Loqed (#70080)

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

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

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."""

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"

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)

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)},
+        )

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()

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*"
+    }
+  ]
+}

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%]"
+    }
+  }
+}

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."
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -254,6 +254,7 @@ FLOWS = {
         "locative",
         "logi_circle",
         "lookin",
+        "loqed",
         "luftdaten",
         "lutron_caseta",
         "lyric",

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",

homeassistant/generated/zeroconf.py

@@ -433,6 +433,10 @@ ZEROCONF = {
             "domain": "bosch_shc",
             "name": "bosch shc*",
         },
+        {
+            "domain": "loqed",
+            "name": "loqed*",
+        },
         {
             "domain": "nam",
             "name": "nam-*",

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
 

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
 

tests/components/loqed/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Loqed integration."""

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

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**"
+}

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=="
+    }
+  ]
+}

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
+  }
+]

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"
+}

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
+}

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
+}

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
+}

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
+}

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"}

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)

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()