Add Ekey Bionyx integration (#139132)

Co-authored-by: Erik Montnemery <erik@montnemery.com>

CODEOWNERS

@@ -410,6 +410,8 @@ build.json @home-assistant/supervisor
 /homeassistant/components/egardia/ @jeroenterheerdt
 /homeassistant/components/eheimdigital/ @autinerd
 /tests/components/eheimdigital/ @autinerd
+/homeassistant/components/ekeybionyx/ @richardpolzer
+/tests/components/ekeybionyx/ @richardpolzer
 /homeassistant/components/electrasmart/ @jafar-atili
 /tests/components/electrasmart/ @jafar-atili
 /homeassistant/components/electric_kiwi/ @mikey0000

homeassistant/components/ekeybionyx/__init__.py

@@ -0,0 +1,24 @@
+"""The Ekey Bionyx integration."""
+
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+PLATFORMS: list[Platform] = [Platform.EVENT]
+
+
+type EkeyBionyxConfigEntry = ConfigEntry
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: EkeyBionyxConfigEntry) -> bool:
+    """Set up the Ekey Bionyx config entry."""
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: EkeyBionyxConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/ekeybionyx/application_credentials.py

@@ -0,0 +1,14 @@
+"""application_credentials platform the Ekey Bionyx integration."""
+
+from homeassistant.components.application_credentials import AuthorizationServer
+from homeassistant.core import HomeAssistant
+
+from .const import OAUTH2_AUTHORIZE, OAUTH2_TOKEN
+
+
+async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
+    """Return authorization server."""
+    return AuthorizationServer(
+        authorize_url=OAUTH2_AUTHORIZE,
+        token_url=OAUTH2_TOKEN,
+    )

homeassistant/components/ekeybionyx/config_flow.py

@@ -0,0 +1,271 @@
+"""Config flow for ekey bionyx."""
+
+import asyncio
+import json
+import logging
+import re
+import secrets
+from typing import Any, NotRequired, TypedDict
+
+import aiohttp
+import ekey_bionyxpy
+import voluptuous as vol
+
+from homeassistant.components.webhook import (
+    async_generate_id as webhook_generate_id,
+    async_generate_path as webhook_generate_path,
+)
+from homeassistant.config_entries import ConfigFlowResult
+from homeassistant.const import CONF_TOKEN, CONF_URL
+from homeassistant.helpers import config_entry_oauth2_flow, config_validation as cv
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.network import get_url
+from homeassistant.helpers.selector import SelectOptionDict, SelectSelector
+
+from .const import API_URL, DOMAIN, INTEGRATION_NAME, SCOPE
+
+# Valid webhook name: starts with letter or underscore, contains letters, digits, spaces, dots, and underscores, does not end with space or dot
+VALID_NAME_PATTERN = re.compile(r"^(?![\d\s])[\w\d \.]*[\w\d]$")
+
+
+class ConfigFlowEkeyApi(ekey_bionyxpy.AbstractAuth):
+    """ekey bionyx authentication before a ConfigEntry exists.
+
+    This implementation directly provides the token without supporting refresh.
+    """
+
+    def __init__(
+        self,
+        websession: aiohttp.ClientSession,
+        token: dict[str, Any],
+    ) -> None:
+        """Initialize ConfigFlowEkeyApi."""
+        super().__init__(websession, API_URL)
+        self._token = token
+
+    async def async_get_access_token(self) -> str:
+        """Return the token for the Ekey API."""
+        return self._token["access_token"]
+
+
+class EkeyFlowData(TypedDict):
+    """Type for Flow Data."""
+
+    api: NotRequired[ekey_bionyxpy.BionyxAPI]
+    system: NotRequired[ekey_bionyxpy.System]
+    systems: NotRequired[list[ekey_bionyxpy.System]]
+
+
+class OAuth2FlowHandler(
+    config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
+):
+    """Config flow to handle ekey bionyx OAuth2 authentication."""
+
+    DOMAIN = DOMAIN
+
+    check_deletion_task: asyncio.Task[None] | None = None
+
+    def __init__(self) -> None:
+        """Initialize OAuth2FlowHandler."""
+        super().__init__()
+        self._data: EkeyFlowData = {}
+
+    @property
+    def logger(self) -> logging.Logger:
+        """Return logger."""
+        return logging.getLogger(__name__)
+
+    @property
+    def extra_authorize_data(self) -> dict[str, Any]:
+        """Extra data that needs to be appended to the authorize url."""
+        return {"scope": SCOPE}
+
+    async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
+        """Start the user facing flow by initializing the API and getting the systems."""
+        client = ConfigFlowEkeyApi(async_get_clientsession(self.hass), data[CONF_TOKEN])
+        ap = ekey_bionyxpy.BionyxAPI(client)
+        self._data["api"] = ap
+        try:
+            system_res = await ap.get_systems()
+        except aiohttp.ClientResponseError:
+            return self.async_abort(
+                reason="cannot_connect",
+                description_placeholders={"ekeybionyx": INTEGRATION_NAME},
+            )
+        system = [s for s in system_res if s.own_system]
+        if len(system) == 0:
+            return self.async_abort(reason="no_own_systems")
+        self._data["systems"] = system
+        if len(system) == 1:
+            # skipping choose_system since there is only one
+            self._data["system"] = system[0]
+            return await self.async_step_check_system(user_input=None)
+        return await self.async_step_choose_system(user_input=None)
+
+    async def async_step_choose_system(
+        self, user_input: dict[str, Any] | None
+    ) -> ConfigFlowResult:
+        """Dialog to choose System if multiple systems are present."""
+        if user_input is None:
+            options: list[SelectOptionDict] = [
+                {"value": s.system_id, "label": s.system_name}
+                for s in self._data["systems"]
+            ]
+            data_schema = {vol.Required("system"): SelectSelector({"options": options})}
+            return self.async_show_form(
+                step_id="choose_system",
+                data_schema=vol.Schema(data_schema),
+                description_placeholders={"ekeybionyx": INTEGRATION_NAME},
+            )
+        self._data["system"] = [
+            s for s in self._data["systems"] if s.system_id == user_input["system"]
+        ][0]
+        return await self.async_step_check_system(user_input=None)
+
+    async def async_step_check_system(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Check if system has open webhooks."""
+        system = self._data["system"]
+        await self.async_set_unique_id(system.system_id)
+        self._abort_if_unique_id_configured()
+
+        if (
+            system.function_webhook_quotas["free"] == 0
+            and system.function_webhook_quotas["used"] == 0
+        ):
+            return self.async_abort(
+                reason="no_available_webhooks",
+                description_placeholders={"ekeybionyx": INTEGRATION_NAME},
+            )
+
+        if system.function_webhook_quotas["used"] > 0:
+            return await self.async_step_delete_webhooks()
+        return await self.async_step_webhooks(user_input=None)
+
+    async def async_step_webhooks(
+        self, user_input: dict[str, Any] | None
+    ) -> ConfigFlowResult:
+        """Dialog to setup webhooks."""
+        system = self._data["system"]
+
+        errors: dict[str, str] | None = None
+        if user_input is not None:
+            errors = {}
+            for key, webhook_name in user_input.items():
+                if key == CONF_URL:
+                    continue
+                if not re.match(VALID_NAME_PATTERN, webhook_name):
+                    errors.update({key: "invalid_name"})
+            try:
+                cv.url(user_input[CONF_URL])
+            except vol.Invalid:
+                errors[CONF_URL] = "invalid_url"
+            if set(user_input) == {CONF_URL}:
+                errors["base"] = "no_webhooks_provided"
+
+            if not errors:
+                webhook_data = [
+                    {
+                        "auth": secrets.token_hex(32),
+                        "name": webhook_name,
+                        "webhook_id": webhook_generate_id(),
+                    }
+                    for key, webhook_name in user_input.items()
+                    if key != CONF_URL
+                ]
+                for webhook in webhook_data:
+                    wh_def: ekey_bionyxpy.WebhookData = {
+                        "integrationName": "Home Assistant",
+                        "functionName": webhook["name"],
+                        "locationName": "Home Assistant",
+                        "definition": {
+                            "url": user_input[CONF_URL]
+                            + webhook_generate_path(webhook["webhook_id"]),
+                            "authentication": {"apiAuthenticationType": "None"},
+                            "securityLevel": "AllowHttp",
+                            "method": "Post",
+                            "body": {
+                                "contentType": "application/json",
+                                "content": json.dumps({"auth": webhook["auth"]}),
+                            },
+                        },
+                    }
+                    webhook["ekey_id"] = (await system.add_webhook(wh_def)).webhook_id
+                return self.async_create_entry(
+                    title=self._data["system"].system_name,
+                    data={"webhooks": webhook_data},
+                )
+
+        data_schema: dict[Any, Any] = {
+            vol.Optional(f"webhook{i + 1}"): vol.All(str, vol.Length(max=50))
+            for i in range(self._data["system"].function_webhook_quotas["free"])
+        }
+        data_schema[vol.Required(CONF_URL)] = str
+        return self.async_show_form(
+            step_id="webhooks",
+            data_schema=self.add_suggested_values_to_schema(
+                vol.Schema(data_schema),
+                {
+                    CONF_URL: get_url(
+                        self.hass,
+                        allow_ip=True,
+                        prefer_external=False,
+                    )
+                }
+                | (user_input or {}),
+            ),
+            errors=errors,
+            description_placeholders={
+                "webhooks_available": str(
+                    self._data["system"].function_webhook_quotas["free"]
+                ),
+                "ekeybionyx": INTEGRATION_NAME,
+            },
+        )
+
+    async def async_step_delete_webhooks(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Form to delete Webhooks."""
+        if user_input is None:
+            return self.async_show_form(step_id="delete_webhooks")
+        for webhook in await self._data["system"].get_webhooks():
+            await webhook.delete()
+        return await self.async_step_wait_for_deletion(user_input=None)
+
+    async def async_step_wait_for_deletion(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Wait for webhooks to be deleted in another flow."""
+        uncompleted_task: asyncio.Task[None] | None = None
+
+        if not self.check_deletion_task:
+            self.check_deletion_task = self.hass.async_create_task(
+                self.async_check_deletion_status()
+            )
+        if not self.check_deletion_task.done():
+            progress_action = "check_deletion_status"
+            uncompleted_task = self.check_deletion_task
+        if uncompleted_task:
+            return self.async_show_progress(
+                step_id="wait_for_deletion",
+                description_placeholders={"ekeybionyx": INTEGRATION_NAME},
+                progress_action=progress_action,
+                progress_task=uncompleted_task,
+            )
+        self.check_deletion_task = None
+        return self.async_show_progress_done(next_step_id="webhooks")
+
+    async def async_check_deletion_status(self) -> None:
+        """Check if webhooks have been deleted."""
+        while True:
+            self._data["systems"] = await self._data["api"].get_systems()
+            self._data["system"] = [
+                s
+                for s in self._data["systems"]
+                if s.system_id == self._data["system"].system_id
+            ][0]
+            if self._data["system"].function_webhook_quotas["used"] == 0:
+                break
+            await asyncio.sleep(5)

homeassistant/components/ekeybionyx/const.py

@@ -0,0 +1,13 @@
+"""Constants for the Ekey Bionyx integration."""
+
+import logging
+
+DOMAIN = "ekeybionyx"
+INTEGRATION_NAME = "ekey bionyx"
+
+LOGGER = logging.getLogger(__package__)
+
+OAUTH2_AUTHORIZE = "https://ekeybionyxprod.b2clogin.com/ekeybionyxprod.onmicrosoft.com/B2C_1_sign_in_v2/oauth2/v2.0/authorize"
+OAUTH2_TOKEN = "https://ekeybionyxprod.b2clogin.com/ekeybionyxprod.onmicrosoft.com/B2C_1_sign_in_v2/oauth2/v2.0/token"
+API_URL = "https://api.bionyx.io/3rd-party/api"
+SCOPE = "https://ekeybionyxprod.onmicrosoft.com/3rd-party-api/api-access"

homeassistant/components/ekeybionyx/event.py

@@ -0,0 +1,70 @@
+"""Event platform for ekey bionyx integration."""
+
+from aiohttp.hdrs import METH_POST
+from aiohttp.web import Request, Response
+
+from homeassistant.components.event import EventDeviceClass, EventEntity
+from homeassistant.components.webhook import (
+    async_register as webhook_register,
+    async_unregister as webhook_unregister,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import EkeyBionyxConfigEntry
+from .const import DOMAIN
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: EkeyBionyxConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Ekey event."""
+    async_add_entities(EkeyEvent(data) for data in entry.data["webhooks"])
+
+
+class EkeyEvent(EventEntity):
+    """Ekey Event."""
+
+    _attr_device_class = EventDeviceClass.BUTTON
+    _attr_event_types = ["event happened"]
+
+    def __init__(
+        self,
+        data: dict[str, str],
+    ) -> None:
+        """Initialise a Ekey event entity."""
+        self._attr_name = data["name"]
+        self._attr_unique_id = data["ekey_id"]
+        self._webhook_id = data["webhook_id"]
+        self._auth = data["auth"]
+
+    @callback
+    def _async_handle_event(self) -> None:
+        """Handle the webhook event."""
+        self._trigger_event("event happened")
+        self.async_write_ha_state()
+
+    async def async_added_to_hass(self) -> None:
+        """Register callbacks with your device API/library."""
+
+        async def async_webhook_handler(
+            hass: HomeAssistant, webhook_id: str, request: Request
+        ) -> Response | None:
+            if (await request.json())["auth"] == self._auth:
+                self._async_handle_event()
+            return None
+
+        webhook_register(
+            self.hass,
+            DOMAIN,
+            f"Ekey {self._attr_name}",
+            self._webhook_id,
+            async_webhook_handler,
+            allowed_methods=[METH_POST],
+        )
+
+    async def async_will_remove_from_hass(self) -> None:
+        """Unregister Webhook."""
+        webhook_unregister(self.hass, self._webhook_id)

homeassistant/components/ekeybionyx/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "ekeybionyx",
+  "name": "ekey bionyx",
+  "codeowners": ["@richardpolzer"],
+  "config_flow": true,
+  "dependencies": ["application_credentials", "http"],
+  "documentation": "https://www.home-assistant.io/integrations/ekeybionyx",
+  "iot_class": "local_push",
+  "quality_scale": "bronze",
+  "requirements": ["ekey-bionyxpy==1.0.0"]
+}

homeassistant/components/ekeybionyx/quality_scale.yaml

@@ -0,0 +1,92 @@
+rules:
+  # Bronze
+  action-setup:
+    status: exempt
+    comment: This integration does not provide actions.
+  appropriate-polling:
+    status: exempt
+    comment: This integration does not poll.
+  brands: done
+  common-modules: done
+  config-flow: done
+  config-flow-test-coverage: done
+  dependency-transparency: done
+  docs-actions:
+    status: exempt
+    comment: This integration does not provide actions.
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup: done
+  entity-unique-id: done
+  has-entity-name: done
+  runtime-data:
+    status: exempt
+    comment: This integration does not connect to any device or service.
+  test-before-configure: done
+  test-before-setup:
+    status: exempt
+    comment: This integration does not connect to any device or service.
+  unique-config-entry: done
+
+  # Silver
+  action-exceptions:
+    status: exempt
+    comment: This integration does not provide actions.
+  config-entry-unloading: done
+  docs-configuration-parameters: todo
+  docs-installation-parameters: todo
+  entity-unavailable:
+    status: exempt
+    comment: This integration has no way of knowing if the fingerprint reader is offline.
+  integration-owner: done
+  log-when-unavailable:
+    status: exempt
+    comment: This integration has no way of knowing if the fingerprint reader is offline.
+  parallel-updates:
+    status: exempt
+    comment: This integration does not poll.
+  reauthentication-flow:
+    status: exempt
+    comment: This integration does not store the tokens.
+  test-coverage: todo
+
+  # Gold
+  devices:
+    status: exempt
+    comment: This integration does not connect to any device or service.
+  diagnostics: todo
+  discovery-update-info:
+    status: exempt
+    comment: This integration does not support discovery.
+  discovery:
+    status: exempt
+    comment: This integration does not support discovery.
+  docs-data-update: todo
+  docs-examples: todo
+  docs-known-limitations: done
+  docs-supported-devices: todo
+  docs-supported-functions: todo
+  docs-troubleshooting: todo
+  docs-use-cases: todo
+  dynamic-devices:
+    status: exempt
+    comment: This integration does not connect to any device or service.
+  entity-category: todo
+  entity-device-class: done
+  entity-disabled-by-default:
+    status: exempt
+    comment: This integration has no entities that should be disabled by default.
+  entity-translations: todo
+  exception-translations: todo
+  icon-translations: todo
+  reconfiguration-flow: todo
+  repair-issues: todo
+  stale-devices:
+    status: exempt
+    comment: This integration does not connect to any device or service.
+
+  # Platinum
+  async-dependency: done
+  inject-websession: done
+  strict-typing: todo

homeassistant/components/ekeybionyx/strings.json

@@ -0,0 +1,66 @@
+{
+  "config": {
+    "step": {
+      "pick_implementation": {
+        "title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
+      },
+      "choose_system": {
+        "data": {
+          "system": "System"
+        },
+        "data_description": {
+          "system": "System the event entities should be set up for."
+        },
+        "description": "Please select the {ekeybionyx} system which you want to connect to Home Assistant."
+      },
+      "webhooks": {
+        "description": "Please name your event entities. These event entities will be mapped as functions in the {ekeybionyx} app. You can configure up to {webhooks_available} event entities. Leaving a name empty will skip the setup of that event entity.",
+        "data": {
+          "webhook1": "Event entity 1",
+          "webhook2": "Event entity 2",
+          "webhook3": "Event entity 3",
+          "webhook4": "Event entity 4",
+          "webhook5": "Event entity 5",
+          "url": "Home Assistant URL"
+        },
+        "data_description": {
+          "webhook1": "Name of event entity 1 that will be mapped into a function",
+          "webhook2": "Name of event entity 2 that will be mapped into a function",
+          "webhook3": "Name of event entity 3 that will be mapped into a function",
+          "webhook4": "Name of event entity 4 that will be mapped into a function",
+          "webhook5": "Name of event entity 5 that will be mapped into a function",
+          "url": "Home Assistant instance URL which can be reached from the fingerprint controller"
+        }
+      },
+      "delete_webhooks": {
+        "description": "This system has already been connected to Home Assistant. If you continue, the previously configured functions will be deleted."
+      }
+    },
+    "progress": {
+      "check_deletion_status": "Please go to the {ekeybionyx} app and confirm the deletion of the functions."
+    },
+    "error": {
+      "invalid_name": "Name is invalid",
+      "invalid_url": "URL is invalid",
+      "no_webhooks_provided": "No event names provided"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
+      "oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
+      "oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
+      "oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
+      "missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
+      "authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
+      "no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
+      "user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]",
+      "no_available_webhooks": "There are no available webhooks in the {ekeybionyx} plattform. Please delete some and try again.",
+      "no_own_systems": "Your account does not have admin access to any systems.",
+      "cannot_connect": "Connection to {ekeybionyx} failed. Please check your Internet connection and try again."
+    },
+    "create_entry": {
+      "default": "[%key:common::config_flow::create_entry::authenticated%]"
+    }
+  }
+}

homeassistant/generated/application_credentials.py

@@ -6,6 +6,7 @@ To update, run python3 -m script.hassfest
 APPLICATION_CREDENTIALS = [
     "aladdin_connect",
     "august",
+    "ekeybionyx",
     "electric_kiwi",
     "fitbit",
     "geocaching",

homeassistant/generated/config_flows.py

@@ -168,6 +168,7 @@ FLOWS = {
         "edl21",
         "efergy",
         "eheimdigital",
+        "ekeybionyx",
         "electrasmart",
         "electric_kiwi",
         "elevenlabs",

homeassistant/generated/integrations.json

@@ -1609,6 +1609,12 @@
       "config_flow": true,
       "iot_class": "local_polling"
     },
+    "ekeybionyx": {
+      "name": "ekey bionyx",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_push"
+    },
     "electrasmart": {
       "name": "Electra Smart",
       "integration_type": "hub",

requirements_all.txt

@@ -852,6 +852,9 @@ ecoaliface==0.4.0
 # homeassistant.components.eheimdigital
 eheimdigital==1.3.0
 
+# homeassistant.components.ekeybionyx
+ekey-bionyxpy==1.0.0
+
 # homeassistant.components.electric_kiwi
 electrickiwi-api==0.9.14
 

requirements_test_all.txt

@@ -743,6 +743,9 @@ easyenergy==2.1.2
 # homeassistant.components.eheimdigital
 eheimdigital==1.3.0
 
+# homeassistant.components.ekeybionyx
+ekey-bionyxpy==1.0.0
+
 # homeassistant.components.electric_kiwi
 electrickiwi-api==0.9.14
 

tests/components/ekeybionyx/__init__.py

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

tests/components/ekeybionyx/conftest.py

@@ -0,0 +1,173 @@
+"""Conftest module for ekeybionyx."""
+
+from http import HTTPStatus
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.ekeybionyx.const import DOMAIN
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+def dummy_systems(
+    num_systems: int, free_wh: int, used_wh: int, own_system: bool = True
+) -> list[dict]:
+    """Create dummy systems."""
+    return [
+        {
+            "systemName": f"System {i + 1}",
+            "systemId": f"946DA01F-9ABD-4D9D-80C7-02AF85C822A{i + 8}",
+            "ownSystem": own_system,
+            "functionWebhookQuotas": {"free": free_wh, "used": used_wh},
+        }
+        for i in range(num_systems)
+    ]
+
+
+@pytest.fixture(name="system")
+def mock_systems(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems",
+        json=dummy_systems(2, 5, 0),
+    )
+
+
+@pytest.fixture(name="no_own_system")
+def mock_no_own_systems(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems",
+        json=dummy_systems(1, 1, 0, False),
+    )
+
+
+@pytest.fixture(name="no_response")
+def mock_no_response(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems",
+        status=HTTPStatus.INTERNAL_SERVER_ERROR,
+    )
+
+
+@pytest.fixture(name="no_available_webhooks")
+def mock_no_available_webhooks(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems",
+        json=dummy_systems(1, 0, 0),
+    )
+
+
+@pytest.fixture(name="already_set_up")
+def mock_already_set_up(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems",
+        json=dummy_systems(1, 0, 1),
+    )
+
+
+@pytest.fixture(name="webhooks")
+def mock_webhooks(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems/946DA01F-9ABD-4D9D-80C7-02AF85C822A8/function-webhooks",
+        json=[
+            {
+                "functionWebhookId": "946DA01F-9ABD-4D9D-80C7-02AF85C822B9",
+                "integrationName": "Home Assistant",
+                "locationName": "A simple string containing 0 to 128 word, space and punctuation characters.",
+                "functionName": "A simple string containing 0 to 50 word, space and punctuation characters.",
+                "expiresAt": "2022-05-16T04:11:28.0000000+00:00",
+                "modificationState": None,
+            }
+        ],
+    )
+
+
+@pytest.fixture(name="webhook_deletion")
+def mock_webhook_deletion(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.delete(
+        "https://api.bionyx.io/3rd-party/api/systems/946DA01F-9ABD-4D9D-80C7-02AF85C822A8/function-webhooks/946DA01F-9ABD-4D9D-80C7-02AF85C822B9",
+        status=HTTPStatus.ACCEPTED,
+    )
+
+
+@pytest.fixture(name="add_webhook", autouse=True)
+def mock_add_webhook(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Fixture to setup fake requests made to Ekey Bionyx API during config flow."""
+    aioclient_mock.post(
+        "https://api.bionyx.io/3rd-party/api/systems/946DA01F-9ABD-4D9D-80C7-02AF85C822A8/function-webhooks",
+        status=HTTPStatus.CREATED,
+        json={
+            "functionWebhookId": "946DA01F-9ABD-4D9D-80C7-02AF85C822A8",
+            "integrationName": "Home Assistant",
+            "locationName": "Home Assistant",
+            "functionName": "Test",
+            "expiresAt": "2022-05-16T04:11:28.0000000+00:00",
+            "modificationState": None,
+        },
+    )
+
+
+@pytest.fixture(name="webhook_id")
+def mock_webhook_id():
+    """Mock webhook_id."""
+    with patch(
+        "homeassistant.components.webhook.async_generate_id", return_value="1234567890"
+    ):
+        yield
+
+
+@pytest.fixture(name="token_hex")
+def mock_token_hex():
+    """Mock auth property."""
+    with patch(
+        "secrets.token_hex",
+        return_value="f2156edca7fc6871e13845314a6fc68622e5ad7c58f17663a487ed28cac247f7",
+    ):
+        yield
+
+
+@pytest.fixture(name="config_entry")
+def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
+    """Create mocked config entry."""
+    return MockConfigEntry(
+        title="test@test.com",
+        domain=DOMAIN,
+        data={
+            "webhooks": [
+                {
+                    "webhook_id": "a2156edca7fb6671e13845314f6fc68622e5dd7c58f17663a487bd28cac247e7",
+                    "name": "Test1",
+                    "auth": "f2156edca7fc6871e13845314a6fc68622e5ad7c58f17663a487ed28cac247f7",
+                    "ekey_id": "946DA01F-9ABD-4D9D-80C7-02AF85C822A8",
+                }
+            ]
+        },
+        unique_id="946DA01F-9ABD-4D9D-80C7-02AF85C822A8",
+        version=1,
+        minor_version=1,
+    )

tests/components/ekeybionyx/test_config_flow.py

@@ -0,0 +1,360 @@
+"""Test the ekey bionyx config flow."""
+
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.application_credentials import (
+    ClientCredential,
+    async_import_client_credential,
+)
+from homeassistant.components.ekeybionyx.const import (
+    DOMAIN,
+    OAUTH2_AUTHORIZE,
+    OAUTH2_TOKEN,
+    SCOPE,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+from homeassistant.helpers import config_entry_oauth2_flow
+from homeassistant.setup import async_setup_component
+
+from .conftest import dummy_systems
+
+from tests.test_util.aiohttp import AiohttpClientMocker
+from tests.typing import ClientSessionGenerator
+
+CLIENT_ID = "1234"
+CLIENT_SECRET = "5678"
+
+
+@pytest.fixture
+async def setup_credentials(hass: HomeAssistant) -> None:
+    """Fixture to setup credentials."""
+    assert await async_setup_component(hass, "application_credentials", {})
+    await async_import_client_credential(
+        hass,
+        DOMAIN,
+        ClientCredential(CLIENT_ID, CLIENT_SECRET),
+    )
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+async def test_full_flow(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    setup_credentials: None,
+    webhook_id: None,
+    system: None,
+    token_hex: None,
+) -> None:
+    """Check full flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}"
+        f"&scope={SCOPE}"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+    flow = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert flow.get("step_id") == "choose_system"
+
+    flow2 = await hass.config_entries.flow.async_configure(
+        flow["flow_id"], {"system": "946DA01F-9ABD-4D9D-80C7-02AF85C822A8"}
+    )
+    assert flow2.get("step_id") == "webhooks"
+
+    flow3 = await hass.config_entries.flow.async_configure(
+        flow2["flow_id"],
+        {
+            "url": "localhost:8123",
+        },
+    )
+
+    assert flow3.get("errors") == {"base": "no_webhooks_provided", "url": "invalid_url"}
+
+    flow4 = await hass.config_entries.flow.async_configure(
+        flow3["flow_id"],
+        {
+            "webhook1": "Test ",
+            "webhook2": " Invalid",
+            "webhook3": "1Invalid",
+            "webhook4": "Also@Invalid",
+            "webhook5": "Invalid-Name",
+            "url": "localhost:8123",
+        },
+    )
+
+    assert flow4.get("errors") == {
+        "url": "invalid_url",
+        "webhook1": "invalid_name",
+        "webhook2": "invalid_name",
+        "webhook3": "invalid_name",
+        "webhook4": "invalid_name",
+        "webhook5": "invalid_name",
+    }
+
+    with patch(
+        "homeassistant.components.ekeybionyx.async_setup_entry", return_value=True
+    ) as mock_setup:
+        flow5 = await hass.config_entries.flow.async_configure(
+            flow2["flow_id"],
+            {
+                "webhook1": "Test",
+                "url": "http://localhost:8123",
+            },
+        )
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+    assert hass.config_entries.async_entries(DOMAIN)[0].data == {
+        "webhooks": [
+            {
+                "webhook_id": "1234567890",
+                "name": "Test",
+                "auth": "f2156edca7fc6871e13845314a6fc68622e5ad7c58f17663a487ed28cac247f7",
+                "ekey_id": "946DA01F-9ABD-4D9D-80C7-02AF85C822A8",
+            }
+        ]
+    }
+
+    assert flow5.get("type") is FlowResultType.CREATE_ENTRY
+
+    assert len(mock_setup.mock_calls) == 1
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+async def test_no_own_system(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    setup_credentials: None,
+    no_own_system: None,
+) -> None:
+    """Check no own System flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}"
+        f"&scope={SCOPE}"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+    flow = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 0
+
+    assert flow.get("type") is FlowResultType.ABORT
+    assert flow.get("reason") == "no_own_systems"
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+async def test_no_available_webhooks(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    setup_credentials: None,
+    no_available_webhooks: None,
+) -> None:
+    """Check no own System flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}"
+        f"&scope={SCOPE}"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+    flow = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 0
+
+    assert flow.get("type") is FlowResultType.ABORT
+    assert flow.get("reason") == "no_available_webhooks"
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+async def test_cleanup(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    setup_credentials: None,
+    already_set_up: None,
+    webhooks: None,
+    webhook_deletion: None,
+) -> None:
+    """Check no own System flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}"
+        f"&scope={SCOPE}"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+
+    flow = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert flow.get("step_id") == "delete_webhooks"
+
+    flow2 = await hass.config_entries.flow.async_configure(flow["flow_id"], {})
+    assert flow2.get("type") is FlowResultType.SHOW_PROGRESS
+
+    aioclient_mock.clear_requests()
+
+    aioclient_mock.get(
+        "https://api.bionyx.io/3rd-party/api/systems",
+        json=dummy_systems(1, 1, 0),
+    )
+
+    await hass.async_block_till_done()
+
+    assert (
+        hass.config_entries.flow.async_get(flow2["flow_id"]).get("step_id")
+        == "webhooks"
+    )
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+async def test_error_on_setup(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    aioclient_mock: AiohttpClientMocker,
+    setup_credentials: None,
+    no_response: None,
+) -> None:
+    """Check no own System flow."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}"
+        f"&scope={SCOPE}"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+    flow = await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 0
+
+    assert flow.get("type") is FlowResultType.ABORT
+    assert flow.get("reason") == "cannot_connect"

tests/components/ekeybionyx/test_init.py

@@ -0,0 +1,30 @@
+"""Module contains tests for the ekeybionyx component's initialization.
+
+Functions:
+    test_async_setup_entry(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
+        Test a successful setup entry and unload of entry.
+"""
+
+from homeassistant.components.ekeybionyx.const import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def test_async_setup_entry(
+    hass: HomeAssistant, config_entry: MockConfigEntry
+) -> None:
+    """Test a successful setup entry and unload of entry."""
+    config_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert config_entry.state is ConfigEntryState.NOT_LOADED