Add bosch_shc supporting Bosch Smart Home Controller (#34063)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

.coveragerc

@@ -112,6 +112,10 @@ omit =
     homeassistant/components/bmw_connected_drive/lock.py
     homeassistant/components/bmw_connected_drive/notify.py
     homeassistant/components/bmw_connected_drive/sensor.py
+    homeassistant/components/bosch_shc/__init__.py
+    homeassistant/components/bosch_shc/const.py
+    homeassistant/components/bosch_shc/binary_sensor.py
+    homeassistant/components/bosch_shc/entity.py
     homeassistant/components/braviatv/__init__.py
     homeassistant/components/braviatv/const.py
     homeassistant/components/braviatv/media_player.py

CODEOWNERS

@@ -70,6 +70,7 @@ homeassistant/components/blueprint/* @home-assistant/core
 homeassistant/components/bmp280/* @belidzs
 homeassistant/components/bmw_connected_drive/* @gerard33 @rikroe
 homeassistant/components/bond/* @prystupa
+homeassistant/components/bosch_shc/* @tschamm
 homeassistant/components/braviatv/* @bieniu
 homeassistant/components/broadlink/* @danielhiversen @felipediel
 homeassistant/components/brother/* @bieniu

homeassistant/components/bosch_shc/__init__.py

@@ -0,0 +1,94 @@
+"""The Bosch Smart Home Controller integration."""
+import logging
+
+from boschshcpy import SHCSession
+from boschshcpy.exceptions import SHCAuthenticationError, SHCConnectionError
+
+from homeassistant.components.zeroconf import async_get_instance
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers import device_registry as dr
+
+from .const import (
+    CONF_SSL_CERTIFICATE,
+    CONF_SSL_KEY,
+    DATA_POLLING_HANDLER,
+    DATA_SESSION,
+    DOMAIN,
+)
+
+PLATFORMS = [
+    "binary_sensor",
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Bosch SHC from a config entry."""
+    data = entry.data
+
+    zeroconf = await async_get_instance(hass)
+    try:
+        session = await hass.async_add_executor_job(
+            SHCSession,
+            data[CONF_HOST],
+            data[CONF_SSL_CERTIFICATE],
+            data[CONF_SSL_KEY],
+            False,
+            zeroconf,
+        )
+    except SHCAuthenticationError as err:
+        raise ConfigEntryAuthFailed from err
+    except SHCConnectionError as err:
+        raise ConfigEntryNotReady from err
+
+    shc_info = session.information
+    if shc_info.updateState.name == "UPDATE_AVAILABLE":
+        _LOGGER.warning("Please check for software updates in the Bosch Smart Home App")
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = {
+        DATA_SESSION: session,
+    }
+
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        connections={(dr.CONNECTION_NETWORK_MAC, dr.format_mac(shc_info.unique_id))},
+        identifiers={(DOMAIN, shc_info.unique_id)},
+        manufacturer="Bosch",
+        name=entry.title,
+        model="SmartHomeController",
+        sw_version=shc_info.version,
+    )
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    async def stop_polling(event):
+        """Stop polling service."""
+        await hass.async_add_executor_job(session.stop_polling)
+
+    await hass.async_add_executor_job(session.start_polling)
+    hass.data[DOMAIN][entry.entry_id][
+        DATA_POLLING_HANDLER
+    ] = hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_polling)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    session: SHCSession = hass.data[DOMAIN][entry.entry_id][DATA_SESSION]
+
+    hass.data[DOMAIN][entry.entry_id][DATA_POLLING_HANDLER]()
+    hass.data[DOMAIN][entry.entry_id].pop(DATA_POLLING_HANDLER)
+    await hass.async_add_executor_job(session.stop_polling)
+
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/bosch_shc/binary_sensor.py

@@ -0,0 +1,49 @@
+"""Platform for binarysensor integration."""
+from boschshcpy import SHCSession, SHCShutterContact
+
+from homeassistant.components.binary_sensor import (
+    DEVICE_CLASS_DOOR,
+    DEVICE_CLASS_WINDOW,
+    BinarySensorEntity,
+)
+
+from .const import DATA_SESSION, DOMAIN
+from .entity import SHCEntity
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up the SHC binary sensor platform."""
+    entities = []
+    session: SHCSession = hass.data[DOMAIN][config_entry.entry_id][DATA_SESSION]
+
+    for binary_sensor in session.device_helper.shutter_contacts:
+        entities.append(
+            ShutterContactSensor(
+                device=binary_sensor,
+                parent_id=session.information.unique_id,
+                entry_id=config_entry.entry_id,
+            )
+        )
+
+    if entities:
+        async_add_entities(entities)
+
+
+class ShutterContactSensor(SHCEntity, BinarySensorEntity):
+    """Representation of a SHC shutter contact sensor."""
+
+    @property
+    def is_on(self):
+        """Return the state of the sensor."""
+        return self._device.state == SHCShutterContact.ShutterContactService.State.OPEN
+
+    @property
+    def device_class(self):
+        """Return the class of this device, from component DEVICE_CLASSES."""
+        switcher = {
+            "ENTRANCE_DOOR": DEVICE_CLASS_DOOR,
+            "REGULAR_WINDOW": DEVICE_CLASS_WINDOW,
+            "FRENCH_WINDOW": DEVICE_CLASS_DOOR,
+            "GENERIC": DEVICE_CLASS_WINDOW,
+        }
+        return switcher.get(self._device.device_class, DEVICE_CLASS_WINDOW)

homeassistant/components/bosch_shc/config_flow.py

@@ -0,0 +1,227 @@
+"""Config flow for Bosch Smart Home Controller integration."""
+import logging
+from os import makedirs
+
+from boschshcpy import SHCRegisterClient, SHCSession
+from boschshcpy.exceptions import (
+    SHCAuthenticationError,
+    SHCConnectionError,
+    SHCRegistrationError,
+    SHCSessionError,
+)
+import voluptuous as vol
+
+from homeassistant import config_entries, core
+from homeassistant.components.zeroconf import async_get_instance
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_TOKEN
+
+from .const import (
+    CONF_HOSTNAME,
+    CONF_SHC_CERT,
+    CONF_SHC_KEY,
+    CONF_SSL_CERTIFICATE,
+    CONF_SSL_KEY,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+HOST_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+    }
+)
+
+
+def write_tls_asset(hass: core.HomeAssistant, filename: str, asset: bytes) -> None:
+    """Write the tls assets to disk."""
+    makedirs(hass.config.path(DOMAIN), exist_ok=True)
+    with open(hass.config.path(DOMAIN, filename), "w") as file_handle:
+        file_handle.write(asset.decode("utf-8"))
+
+
+def create_credentials_and_validate(hass, host, user_input, zeroconf):
+    """Create and store credentials and validate session."""
+    helper = SHCRegisterClient(host, user_input[CONF_PASSWORD])
+    result = helper.register(host, "HomeAssistant")
+
+    if result is not None:
+        write_tls_asset(hass, CONF_SHC_CERT, result["cert"])
+        write_tls_asset(hass, CONF_SHC_KEY, result["key"])
+
+        session = SHCSession(
+            host,
+            hass.config.path(DOMAIN, CONF_SHC_CERT),
+            hass.config.path(DOMAIN, CONF_SHC_KEY),
+            True,
+            zeroconf,
+        )
+        session.authenticate()
+
+    return result
+
+
+def get_info_from_host(hass, host, zeroconf):
+    """Get information from host."""
+    session = SHCSession(
+        host,
+        "",
+        "",
+        True,
+        zeroconf,
+    )
+    information = session.mdns_info()
+    return {"title": information.name, "unique_id": information.unique_id}
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Bosch SHC."""
+
+    VERSION = 1
+    info = None
+    host = None
+    hostname = None
+
+    async def async_step_reauth(self, user_input=None):
+        """Perform reauth upon an API authentication error."""
+        return await self.async_step_reauth_confirm()
+
+    async def async_step_reauth_confirm(self, user_input=None):
+        """Dialog that informs the user that reauth is required."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="reauth_confirm",
+                data_schema=HOST_SCHEMA,
+            )
+        self.host = host = user_input[CONF_HOST]
+        self.info = await self._get_info(host)
+        return await self.async_step_credentials()
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
+        errors = {}
+        if user_input is not None:
+            host = user_input[CONF_HOST]
+            try:
+                self.info = info = await self._get_info(host)
+            except SHCConnectionError:
+                errors["base"] = "cannot_connect"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(info["unique_id"])
+                self._abort_if_unique_id_configured({CONF_HOST: host})
+                self.host = host
+                return await self.async_step_credentials()
+
+        return self.async_show_form(
+            step_id="user", data_schema=HOST_SCHEMA, errors=errors
+        )
+
+    async def async_step_credentials(self, user_input=None):
+        """Handle the credentials step."""
+        errors = {}
+        if user_input is not None:
+            zeroconf = await async_get_instance(self.hass)
+            try:
+                result = await self.hass.async_add_executor_job(
+                    create_credentials_and_validate,
+                    self.hass,
+                    self.host,
+                    user_input,
+                    zeroconf,
+                )
+            except SHCAuthenticationError:
+                errors["base"] = "invalid_auth"
+            except SHCConnectionError:
+                errors["base"] = "cannot_connect"
+            except SHCSessionError as err:
+                _LOGGER.warning("Session error: %s", err.message)
+                errors["base"] = "session_error"
+            except SHCRegistrationError as err:
+                _LOGGER.warning("Registration error: %s", err.message)
+                errors["base"] = "pairing_failed"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                entry_data = {
+                    CONF_SSL_CERTIFICATE: self.hass.config.path(DOMAIN, CONF_SHC_CERT),
+                    CONF_SSL_KEY: self.hass.config.path(DOMAIN, CONF_SHC_KEY),
+                    CONF_HOST: self.host,
+                    CONF_TOKEN: result["token"],
+                    CONF_HOSTNAME: result["token"].split(":", 1)[1],
+                }
+                existing_entry = await self.async_set_unique_id(self.info["unique_id"])
+                if existing_entry:
+                    self.hass.config_entries.async_update_entry(
+                        existing_entry,
+                        data=entry_data,
+                    )
+                    await self.hass.config_entries.async_reload(existing_entry.entry_id)
+                    return self.async_abort(reason="reauth_successful")
+
+                return self.async_create_entry(
+                    title=self.info["title"],
+                    data=entry_data,
+                )
+        else:
+            user_input = {}
+
+        schema = vol.Schema(
+            {
+                vol.Required(
+                    CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
+                ): str,
+            }
+        )
+
+        return self.async_show_form(
+            step_id="credentials", data_schema=schema, errors=errors
+        )
+
+    async def async_step_zeroconf(self, discovery_info):
+        """Handle zeroconf discovery."""
+        if not discovery_info.get("name", "").startswith("Bosch SHC"):
+            return self.async_abort(reason="not_bosch_shc")
+
+        try:
+            self.info = info = await self._get_info(discovery_info["host"])
+        except SHCConnectionError:
+            return self.async_abort(reason="cannot_connect")
+
+        local_name = discovery_info["hostname"][:-1]
+        node_name = local_name[: -len(".local")]
+
+        await self.async_set_unique_id(info["unique_id"])
+        self._abort_if_unique_id_configured({CONF_HOST: discovery_info["host"]})
+        self.host = discovery_info["host"]
+        self.context["title_placeholders"] = {"name": node_name}
+        return await self.async_step_confirm_discovery()
+
+    async def async_step_confirm_discovery(self, user_input=None):
+        """Handle discovery confirm."""
+        errors = {}
+        if user_input is not None:
+            return await self.async_step_credentials()
+
+        return self.async_show_form(
+            step_id="confirm_discovery",
+            description_placeholders={
+                "model": "Bosch SHC",
+                "host": self.host,
+            },
+            errors=errors,
+        )
+
+    async def _get_info(self, host):
+        """Get additional information."""
+        zeroconf = await async_get_instance(self.hass)
+
+        return await self.hass.async_add_executor_job(
+            get_info_from_host,
+            self.hass,
+            host,
+            zeroconf,
+        )

homeassistant/components/bosch_shc/const.py

@@ -0,0 +1,12 @@
+"""Constants for the Bosch SHC integration."""
+
+CONF_HOSTNAME = "hostname"
+CONF_SHC_CERT = "bosch_shc-cert.pem"
+CONF_SHC_KEY = "bosch_shc-key.pem"
+CONF_SSL_CERTIFICATE = "ssl_certificate"
+CONF_SSL_KEY = "ssl_key"
+
+DATA_SESSION = "session"
+DATA_POLLING_HANDLER = "polling_handler"
+
+DOMAIN = "bosch_shc"

homeassistant/components/bosch_shc/entity.py

@@ -0,0 +1,92 @@
+"""Bosch Smart Home Controller base entity."""
+from boschshcpy.device import SHCDevice
+
+from homeassistant.helpers.device_registry import async_get as get_dev_reg
+from homeassistant.helpers.entity import Entity
+
+from .const import DOMAIN
+
+
+async def async_remove_devices(hass, entity, entry_id):
+    """Get item that is removed from session."""
+    dev_registry = get_dev_reg(hass)
+    device = dev_registry.async_get_device(
+        identifiers={(DOMAIN, entity.device_id)}, connections=set()
+    )
+    if device is not None:
+        dev_registry.async_update_device(device.id, remove_config_entry_id=entry_id)
+
+
+class SHCEntity(Entity):
+    """Representation of a SHC base entity."""
+
+    def __init__(self, device: SHCDevice, parent_id: str, entry_id: str) -> None:
+        """Initialize the generic SHC device."""
+        self._device = device
+        self._parent_id = parent_id
+        self._entry_id = entry_id
+
+    async def async_added_to_hass(self):
+        """Subscribe to SHC events."""
+        await super().async_added_to_hass()
+
+        def on_state_changed():
+            self.schedule_update_ha_state()
+
+        def update_entity_information():
+            if self._device.deleted:
+                self.hass.add_job(async_remove_devices(self.hass, self, self._entry_id))
+            else:
+                self.schedule_update_ha_state()
+
+        for service in self._device.device_services:
+            service.subscribe_callback(self.entity_id, on_state_changed)
+        self._device.subscribe_callback(self.entity_id, update_entity_information)
+
+    async def async_will_remove_from_hass(self):
+        """Unsubscribe from SHC events."""
+        await super().async_will_remove_from_hass()
+        for service in self._device.device_services:
+            service.unsubscribe_callback(self.entity_id)
+        self._device.unsubscribe_callback(self.entity_id)
+
+    @property
+    def unique_id(self):
+        """Return the unique ID of the device."""
+        return self._device.serial
+
+    @property
+    def name(self):
+        """Name of the entity."""
+        return self._device.name
+
+    @property
+    def device_id(self):
+        """Device id of the entity."""
+        return self._device.id
+
+    @property
+    def device_info(self):
+        """Return the device info."""
+        return {
+            "identifiers": {(DOMAIN, self.device_id)},
+            "name": self._device.name,
+            "manufacturer": self._device.manufacturer,
+            "model": self._device.device_model,
+            "via_device": (
+                DOMAIN,
+                self._device.parent_device_id
+                if self._device.parent_device_id is not None
+                else self._parent_id,
+            ),
+        }
+
+    @property
+    def available(self):
+        """Return false if status is unavailable."""
+        return self._device.status == "AVAILABLE"
+
+    @property
+    def should_poll(self):
+        """Report polling mode. SHC Entity is communicating via long polling."""
+        return False

homeassistant/components/bosch_shc/manifest.json

@@ -0,0 +1,13 @@
+{
+  "domain": "bosch_shc",
+  "name": "Bosch SHC",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/bosch_shc",
+  "requirements": ["boschshcpy==0.2.17"],
+  "zeroconf": [
+    {"type": "_http._tcp.local.", "name": "bosch shc*"}
+  ],
+  "iot_class": "local_push",
+  "codeowners": ["@tschamm"],
+  "after_dependencies": ["zeroconf"]
+}

homeassistant/components/bosch_shc/strings.json

@@ -0,0 +1,38 @@
+{
+  "title": "Bosch SHC",
+  "config": {
+    "step": {
+      "user": {
+        "description": "Set up your Bosch Smart Home Controller to allow monitoring and control with Home Assistant.",
+        "title": "SHC authentication parameters",
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]"
+        }
+      },
+      "credentials": {
+        "data": {
+          "password": "Password of the Smart Home Controller"
+        }
+      },
+      "confirm_discovery": {
+        "description": "Please press the Bosch Smart Home Controller's front-side button until LED starts flashing.\nReady to continue to set up {model} @ {host} with Home Assistant?"
+      },
+      "reauth_confirm": {
+        "title": "[%key:common::config_flow::title::reauth%]",
+        "description": "The bosch_shc integration needs to re-authenticate your account"
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "pairing_failed": "Pairing failed; please check the Bosch Smart Home Controller is in pairing mode (LED flashing) as well as your password is correct.",
+      "session_error": "Session error: API return Non-OK result.",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
+    },
+    "flow_title": "Bosch SHC: {name}"
+  }
+}

homeassistant/components/bosch_shc/translations/en.json

@@ -0,0 +1,41 @@
+{
+    "title": "Bosch SHC",
+    "config": {
+        "flow_title": "Bosch SHC: {name}",
+        "step": {
+            "confirm_discovery": {
+                "description": "Please press the Bosch Smart Home Controller's front-side button until LED starts flashing.\nReady to continue to set up {model} @ {host} with Home Assistant?"
+            },
+            "credentials": {
+                "data": {
+                    "password": "Password of the Smart Home Controller"
+                }
+            },
+            "user": {
+                "description": "Set up your Bosch Smart Home Controller to allow monitoring and control with Home Assistant.",
+                "title": "SHC authentication parameters",
+                "data": {
+                    "host": "Host"
+                }
+            },
+            "reauth_confirm": {
+                "title": "SHC authentication parameters",
+                "description": "The bosch_shc integration needs to re-authenticate your account",
+                "data": {
+                    "host": "Host"
+                }
+            }
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "invalid_auth": "Invalid authentication",
+            "pairing_failed": "Pairing failed; please check the Bosch Smart Home Controller is in pairing mode (LED flashing) as well as your password is correct.",
+            "session_error": "Session error: API return Non-OK result.",
+            "unknown": "Unexpected error"
+        },
+        "abort": {
+            "already_configured": "Device is already configured",
+            "reauth_successful": "Re-authentication was successful"
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -33,6 +33,7 @@ FLOWS = [
     "blink",
     "bmw_connected_drive",
     "bond",
+    "bosch_shc",
     "braviatv",
     "broadlink",
     "brother",

homeassistant/generated/zeroconf.py

@@ -97,6 +97,10 @@ ZEROCONF = {
         }
     ],
     "_http._tcp.local.": [
+        {
+            "domain": "bosch_shc",
+            "name": "bosch shc*"
+        },
         {
             "domain": "nam",
             "name": "nam-*"

requirements_all.txt

@@ -382,6 +382,9 @@ blockchain==1.4.4
 # homeassistant.components.bond
 bond-api==0.1.12
 
+# homeassistant.components.bosch_shc
+boschshcpy==0.2.17
+
 # homeassistant.components.amazon_polly
 # homeassistant.components.route53
 boto3==1.16.52

requirements_test_all.txt

@@ -219,6 +219,9 @@ blinkpy==0.17.0
 # homeassistant.components.bond
 bond-api==0.1.12
 
+# homeassistant.components.bosch_shc
+boschshcpy==0.2.17
+
 # homeassistant.components.braviatv
 bravia-tv==1.0.8
 

tests/components/bosch_shc/__init__.py

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

tests/components/bosch_shc/test_config_flow.py

@@ -0,0 +1,625 @@
+"""Test the Bosch SHC config flow."""
+from unittest.mock import PropertyMock, mock_open, patch
+
+from boschshcpy.exceptions import (
+    SHCAuthenticationError,
+    SHCConnectionError,
+    SHCRegistrationError,
+    SHCSessionError,
+)
+from boschshcpy.information import SHCInformation
+
+from homeassistant import config_entries, setup
+from homeassistant.components.bosch_shc.config_flow import write_tls_asset
+from homeassistant.components.bosch_shc.const import CONF_SHC_CERT, CONF_SHC_KEY, DOMAIN
+
+from tests.common import MockConfigEntry
+
+MOCK_SETTINGS = {
+    "name": "Test name",
+    "device": {"mac": "test-mac", "hostname": "test-host"},
+}
+DISCOVERY_INFO = {
+    "host": "1.1.1.1",
+    "port": 0,
+    "hostname": "shc012345.local.",
+    "type": "_http._tcp.local.",
+    "name": "Bosch SHC [test-mac]._http._tcp.local.",
+}
+
+
+async def test_form_user(hass):
+    """Test we get the form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+    assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate"
+    ) as mock_authenticate, patch(
+        "homeassistant.components.bosch_shc.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "create_entry"
+    assert result3["title"] == "shc012345"
+    assert result3["data"] == {
+        "host": "1.1.1.1",
+        "ssl_certificate": hass.config.path(DOMAIN, CONF_SHC_CERT),
+        "ssl_key": hass.config.path(DOMAIN, CONF_SHC_KEY),
+        "token": "abc:123",
+        "hostname": "123",
+    }
+
+    assert len(mock_authenticate.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_form_get_info_connection_error(hass):
+    """Test we handle connection error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        side_effect=SHCConnectionError,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "host": "1.1.1.1",
+            },
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "user"
+    assert result2["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_get_info_exception(hass):
+    """Test we handle exceptions."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        side_effect=Exception,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                "host": "1.1.1.1",
+            },
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "user"
+    assert result2["errors"] == {"base": "unknown"}
+
+
+async def test_form_pairing_error(hass):
+    """Test we handle pairing error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+    assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        side_effect=SHCRegistrationError(""),
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "form"
+    assert result3["step_id"] == "credentials"
+    assert result3["errors"] == {"base": "pairing_failed"}
+
+
+async def test_form_user_invalid_auth(hass):
+    """Test we handle invalid auth."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+    assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate",
+        side_effect=SHCAuthenticationError,
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "form"
+    assert result3["step_id"] == "credentials"
+    assert result3["errors"] == {"base": "invalid_auth"}
+
+
+async def test_form_validate_connection_error(hass):
+    """Test we handle connection error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+    assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate",
+        side_effect=SHCConnectionError,
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "form"
+    assert result3["step_id"] == "credentials"
+    assert result3["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_validate_session_error(hass):
+    """Test we handle session error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+    assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate",
+        side_effect=SHCSessionError(""),
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "form"
+    assert result3["step_id"] == "credentials"
+    assert result3["errors"] == {"base": "session_error"}
+
+
+async def test_form_validate_exception(hass):
+    """Test we handle exception."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+    assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate",
+        side_effect=Exception,
+    ):
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "form"
+    assert result3["step_id"] == "credentials"
+    assert result3["errors"] == {"base": "unknown"}
+
+
+async def test_form_already_configured(hass):
+    """Test we get the form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    entry = MockConfigEntry(
+        domain="bosch_shc", unique_id="test-mac", data={"host": "0.0.0.0"}
+    )
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "1.1.1.1"},
+        )
+
+        assert result2["type"] == "abort"
+        assert result2["reason"] == "already_configured"
+
+    # Test config entry got updated with latest IP
+    assert entry.data["host"] == "1.1.1.1"
+
+
+async def test_zeroconf(hass):
+    """Test we get the form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_ZEROCONF},
+        )
+        assert result["type"] == "form"
+        assert result["step_id"] == "confirm_discovery"
+        assert result["errors"] == {}
+        context = next(
+            flow["context"]
+            for flow in hass.config_entries.flow.async_progress()
+            if flow["flow_id"] == result["flow_id"]
+        )
+        assert context["title_placeholders"]["name"] == "shc012345"
+
+    result2 = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {},
+    )
+    assert result2["type"] == "form"
+    assert result2["step_id"] == "credentials"
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate",
+    ), patch(
+        "homeassistant.components.bosch_shc.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "create_entry"
+    assert result3["title"] == "shc012345"
+    assert result3["data"] == {
+        "host": "1.1.1.1",
+        "ssl_certificate": hass.config.path(DOMAIN, CONF_SHC_CERT),
+        "ssl_key": hass.config.path(DOMAIN, CONF_SHC_KEY),
+        "token": "abc:123",
+        "hostname": "123",
+    }
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_zeroconf_already_configured(hass):
+    """Test we get the form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    entry = MockConfigEntry(
+        domain="bosch_shc", unique_id="test-mac", data={"host": "0.0.0.0"}
+    )
+    entry.add_to_hass(hass)
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_ZEROCONF},
+        )
+
+        assert result["type"] == "abort"
+        assert result["reason"] == "already_configured"
+
+    # Test config entry got updated with latest IP
+    assert entry.data["host"] == "1.1.1.1"
+
+
+async def test_zeroconf_cannot_connect(hass):
+    """Test we get the form."""
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info", side_effect=SHCConnectionError
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            data=DISCOVERY_INFO,
+            context={"source": config_entries.SOURCE_ZEROCONF},
+        )
+        assert result["type"] == "abort"
+        assert result["reason"] == "cannot_connect"
+
+
+async def test_zeroconf_not_bosch_shc(hass):
+    """Test we filter out non-bosch_shc devices."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        data={"host": "1.1.1.1", "name": "notboschshc"},
+        context={"source": config_entries.SOURCE_ZEROCONF},
+    )
+    assert result["type"] == "abort"
+    assert result["reason"] == "not_bosch_shc"
+
+
+async def test_reauth(hass):
+    """Test we get the form."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    mock_config = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id="test-mac",
+        data={
+            "host": "1.1.1.1",
+            "hostname": "test-mac",
+            "ssl_certificate": "test-cert.pem",
+            "ssl_key": "test-key.pem",
+        },
+        title="shc012345",
+    )
+    mock_config.add_to_hass(hass)
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_REAUTH},
+        data=mock_config.data,
+    )
+    assert result["type"] == "form"
+    assert result["step_id"] == "reauth_confirm"
+
+    with patch(
+        "boschshcpy.session.SHCSession.mdns_info",
+        return_value=SHCInformation,
+    ), patch(
+        "boschshcpy.information.SHCInformation.name",
+        new_callable=PropertyMock,
+        return_value="shc012345",
+    ), patch(
+        "boschshcpy.information.SHCInformation.unique_id",
+        new_callable=PropertyMock,
+        return_value="test-mac",
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {"host": "2.2.2.2"},
+        )
+
+        assert result2["type"] == "form"
+        assert result2["step_id"] == "credentials"
+        assert result2["errors"] == {}
+
+    with patch(
+        "boschshcpy.register_client.SHCRegisterClient.register",
+        return_value={
+            "token": "abc:123",
+            "cert": b"content_cert",
+            "key": b"content_key",
+        },
+    ), patch("os.mkdir"), patch("builtins.open"), patch(
+        "boschshcpy.session.SHCSession.authenticate"
+    ), patch(
+        "homeassistant.components.bosch_shc.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {"password": "test"},
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == "abort"
+    assert result3["reason"] == "reauth_successful"
+
+    assert mock_config.data["host"] == "2.2.2.2"
+
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_tls_assets_writer(hass):
+    """Test we write tls assets to correct location."""
+    assets = {
+        "token": "abc:123",
+        "cert": b"content_cert",
+        "key": b"content_key",
+    }
+    with patch("os.mkdir"), patch("builtins.open", mock_open()) as mocked_file:
+        write_tls_asset(hass, CONF_SHC_CERT, assets["cert"])
+        mocked_file.assert_called_with(hass.config.path(DOMAIN, CONF_SHC_CERT), "w")
+        mocked_file().write.assert_called_with("content_cert")
+
+        write_tls_asset(hass, CONF_SHC_KEY, assets["key"])
+        mocked_file.assert_called_with(hass.config.path(DOMAIN, CONF_SHC_KEY), "w")
+        mocked_file().write.assert_called_with("content_key")