Add Wolflink integration (#34104)

* WOLF Smart-set integration

* Removed translations. Changed device class of timestamp. Added new test for unknown exception

* Remove unit_of_measurement from hours sensor

* Code cleanup. Pull Request comments fixes

* ConnectError import change. Removed DEVICE_CLASS_TIMESTAMP

* Add unique id guard with tests. Use common translations. Move device_id resolution to config_flow.

* Remove debug print

.coveragerc

@@ -934,6 +934,9 @@ omit =
     homeassistant/components/wiffi/*
     homeassistant/components/wink/*
     homeassistant/components/wirelesstag/*
+    homeassistant/components/wolflink/__init__.py
+    homeassistant/components/wolflink/sensor.py
+    homeassistant/components/wolflink/const.py
     homeassistant/components/worldtidesinfo/sensor.py
     homeassistant/components/worxlandroid/sensor.py
     homeassistant/components/x10/light.py

CODEOWNERS

@@ -459,6 +459,7 @@ homeassistant/components/websocket_api/* @home-assistant/core
 homeassistant/components/wiffi/* @mampfes
 homeassistant/components/withings/* @vangorra
 homeassistant/components/wled/* @frenck
+homeassistant/components/wolflink/* @adamkrol93
 homeassistant/components/workday/* @fabaff
 homeassistant/components/worldclock/* @fabaff
 homeassistant/components/xbox_live/* @MartinHjelmare

homeassistant/components/wolflink/__init__.py

@@ -0,0 +1,103 @@
+"""The Wolf SmartSet Service integration."""
+from _datetime import timedelta
+import logging
+
+from httpcore import ConnectError
+from wolf_smartset.token_auth import InvalidAuth
+from wolf_smartset.wolf_client import WolfClient
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    COORDINATOR,
+    DEVICE_GATEWAY,
+    DEVICE_ID,
+    DEVICE_NAME,
+    DOMAIN,
+    PARAMETERS,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup(hass: HomeAssistant, config: dict):
+    """Set up the Wolf SmartSet Service component."""
+    hass.data[DOMAIN] = {}
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Set up Wolf SmartSet Service from a config entry."""
+    username = entry.data[CONF_USERNAME]
+    password = entry.data[CONF_PASSWORD]
+    device_name = entry.data[DEVICE_NAME]
+    device_id = entry.data[DEVICE_ID]
+    gateway_id = entry.data[DEVICE_GATEWAY]
+    _LOGGER.debug(
+        "Setting up wolflink integration for device: %s (id: %s, gateway: %s)",
+        device_name,
+        device_id,
+        gateway_id,
+    )
+
+    wolf_client = WolfClient(username, password)
+
+    parameters = await fetch_parameters(wolf_client, gateway_id, device_id)
+
+    async def async_update_data():
+        """Update all stored entities for Wolf SmartSet."""
+        try:
+            values = await wolf_client.fetch_value(gateway_id, device_id, parameters)
+            return {v.value_id: v.value for v in values}
+        except ConnectError as exception:
+            raise UpdateFailed(f"Error communicating with API: {exception}")
+        except InvalidAuth:
+            raise UpdateFailed("Invalid authentication during update.")
+
+    coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        name="wolflink",
+        update_method=async_update_data,
+        update_interval=timedelta(minutes=1),
+    )
+
+    await coordinator.async_refresh()
+
+    hass.data[DOMAIN][entry.entry_id] = {}
+    hass.data[DOMAIN][entry.entry_id][PARAMETERS] = parameters
+    hass.data[DOMAIN][entry.entry_id][COORDINATOR] = coordinator
+    hass.data[DOMAIN][entry.entry_id][DEVICE_ID] = device_id
+
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, "sensor")
+    )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_forward_entry_unload(entry, "sensor")
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
+
+
+async def fetch_parameters(client: WolfClient, gateway_id: int, device_id: int):
+    """
+    Fetch all available parameters with usage of WolfClient.
+
+    By default Reglertyp entity is removed because API will not provide value for this parameter.
+    """
+    try:
+        fetched_parameters = await client.fetch_parameters(gateway_id, device_id)
+        return [param for param in fetched_parameters if param.name != "Reglertyp"]
+    except ConnectError as exception:
+        raise UpdateFailed(f"Error communicating with API: {exception}")
+    except InvalidAuth:
+        raise UpdateFailed("Invalid authentication during update.")

homeassistant/components/wolflink/config_flow.py

@@ -0,0 +1,93 @@
+"""Config flow for Wolf SmartSet Service integration."""
+import logging
+
+from httpcore import ConnectError
+import voluptuous as vol
+from wolf_smartset.token_auth import InvalidAuth
+from wolf_smartset.wolf_client import WolfClient
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+
+from .const import (  # pylint:disable=unused-import
+    DEVICE_GATEWAY,
+    DEVICE_ID,
+    DEVICE_NAME,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+USER_SCHEMA = vol.Schema(
+    {vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
+)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Wolf SmartSet Service."""
+
+    VERSION = 1
+    CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
+
+    def __init__(self):
+        """Initialize with empty username and password."""
+        self.username = None
+        self.password = None
+        self.fetched_systems = None
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step to get connection parameters."""
+        errors = {}
+        if user_input is not None:
+            wolf_client = WolfClient(
+                user_input[CONF_USERNAME], user_input[CONF_PASSWORD]
+            )
+            try:
+                self.fetched_systems = await wolf_client.fetch_system_list()
+            except ConnectError:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                self.username = user_input[CONF_USERNAME]
+                self.password = user_input[CONF_PASSWORD]
+                return await self.async_step_device()
+        return self.async_show_form(
+            step_id="user", data_schema=USER_SCHEMA, errors=errors
+        )
+
+    async def async_step_device(self, user_input=None):
+        """Allow user to select device from devices connected to specified account."""
+        errors = {}
+        if user_input is not None:
+            device_name = user_input[DEVICE_NAME]
+            system = [
+                device for device in self.fetched_systems if device.name == device_name
+            ]
+            device_id = system[0].id
+            await self.async_set_unique_id(device_id)
+            self._abort_if_unique_id_configured()
+            return self.async_create_entry(
+                title=user_input[DEVICE_NAME],
+                data={
+                    CONF_USERNAME: self.username,
+                    CONF_PASSWORD: self.password,
+                    DEVICE_NAME: device_name,
+                    DEVICE_GATEWAY: system[0].gateway,
+                    DEVICE_ID: device_id,
+                },
+            )
+
+        data_schema = vol.Schema(
+            {
+                vol.Required(DEVICE_NAME): vol.In(
+                    [info.name for info in self.fetched_systems]
+                )
+            }
+        )
+        return self.async_show_form(
+            step_id="device", data_schema=data_schema, errors=errors
+        )

homeassistant/components/wolflink/const.py

@@ -0,0 +1,93 @@
+"""Constants for the Wolf SmartSet Service integration."""
+
+DOMAIN = "wolflink"
+
+COORDINATOR = "coordinator"
+PARAMETERS = "parameters"
+DEVICE_ID = "device_id"
+DEVICE_GATEWAY = "device_gateway"
+DEVICE_NAME = "device_name"
+
+STATES = {
+    "Ein": "ein",
+    "Deaktiviert": "deaktiviert",
+    "Aus": "aus",
+    "Standby": "standby",
+    "Auto": "auto",
+    "Permanent": "permanent",
+    "Initialisierung": "initialisierung",
+    "Antilegionellenfunktion": "antilegionellenfunktion",
+    "Fernschalter ein": "fernschalter_ein",
+    "1x Warmwasser": "1_x_warmwasser",
+    "Bereit, keine Ladung": "bereit_keine_ladung",
+    "Solarbetrieb": "solarbetrieb",
+    "Reduzierter Betrieb": "reduzierter_betrieb",
+    "SmartHome": "smart_home",
+    "SmartGrid": "smart_grid",
+    "Ruhekontakt": "ruhekontakt",
+    "Vorspülen": "vorspulen",
+    "Zünden": "zunden",
+    "Stabilisierung": "stabilisierung",
+    "Ventilprüfung": "ventilprufung",
+    "Nachspülen": "nachspulen",
+    "Softstart": "softstart",
+    "Taktsperre": "taktsperre",
+    "Betrieb ohne Brenner": "betrieb_ohne_brenner",
+    "Abgasklappe": "abgasklappe",
+    "Störung": "storung",
+    "Gradienten Überwachung": "gradienten_uberwachung",
+    "Gasdruck": "gasdruck",
+    "Spreizung hoch": "spreizung_hoch",
+    "Spreizung KF": "spreizung_kf",
+    "Test": "test",
+    "Start": "start",
+    "Frost Heizkreis": "frost_heizkreis",
+    "Frost Warmwasser": "frost_warmwasser",
+    "Schornsteinfeger": "schornsteinfeger",
+    "Kombibetrieb": "kombibetrieb",
+    "Parallelbetrieb": "parallelbetrieb",
+    "Warmwasserbetrieb": "warmwasserbetrieb",
+    "Warmwassernachlauf": "warmwassernachlauf",
+    "Mindest-Kombizeit": "mindest_kombizeit",
+    "Heizbetrieb": "heizbetrieb",
+    "Nachlauf Heizkreispumpe": "nachlauf_heizkreispumpe",
+    "Frostschutz": "frostschutz",
+    "Kaskadenbetrieb": "kaskadenbetrieb",
+    "GLT-Betrieb": "glt_betrieb",
+    "Kalibration": "kalibration",
+    "Kalibration Heizbetrieb": "kalibration_heizbetrieb",
+    "Kalibration Warmwasserbetrieb": "kalibration_warmwasserbetrieb",
+    "Kalibration Kombibetrieb": "kalibration_kombibetrieb",
+    "Warmwasser Schnellstart": "warmwasser_schnellstart",
+    "Externe Deaktivierung": "externe_deaktivierung",
+    "Heizung": "heizung",
+    "Warmwasser": "warmwasser",
+    "Kombigerät": "kombigerat",
+    "Kombigerät mit Solareinbindung": "kombigerat_mit_solareinbindung",
+    "Heizgerät mit Speicher": "heizgerat_mit_speicher",
+    "Nur Heizgerät": "nur_heizgerat",
+    "Aktiviert": "ktiviert",
+    "Sparen": "sparen",
+    "Estrichtrocknung": "estrichtrocknung",
+    "Telefonfernschalter": "telefonfernschalter",
+    "Partymodus": "partymodus",
+    "Urlaubsmodus": "urlaubsmodus",
+    "Automatik ein": "automatik_ein",
+    "Automatik aus": "automatik_aus",
+    "Permanentbetrieb": "permanentbetrieb",
+    "Sparbetrieb": "sparbetrieb",
+    "AutoOnCool": "auto_on_cool",
+    "AutoOffCool": "auto_off_cool",
+    "PermCooling": "perm_cooling",
+    "Absenkbetrieb": "absenkbetrieb",
+    "Eco": "eco",
+    "Absenkstop": "absenkstop",
+    "AT Abschaltung": "at_abschaltung",
+    "RT Abschaltung": "rt_abschaltung",
+    "AT Frostschutz": "at_frostschutz",
+    "RT Frostschutz": "rt_frostschutz",
+    "DHWPrior": "dhw_prior",
+    "Cooling": "cooling",
+    "TPW": "tpw",
+    "Warmwasservorrang": "warmwasservorrang",
+}

homeassistant/components/wolflink/manifest.json

@@ -0,0 +1,8 @@
+{
+  "domain": "wolflink",
+  "name": "Wolf SmartSet Service",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/wolflink",
+  "requirements": ["wolf_smartset==0.1.4"],
+  "codeowners": ["@adamkrol93"]
+}

homeassistant/components/wolflink/sensor.py

@@ -0,0 +1,182 @@
+"""The Wolf SmartSet sensors."""
+import logging
+
+from wolf_smartset.models import (
+    HoursParameter,
+    ListItemParameter,
+    Parameter,
+    PercentageParameter,
+    Pressure,
+    SimpleParameter,
+    Temperature,
+)
+
+from homeassistant.components.wolflink.const import (
+    COORDINATOR,
+    DEVICE_ID,
+    DOMAIN,
+    PARAMETERS,
+    STATES,
+)
+from homeassistant.const import (
+    DEVICE_CLASS_PRESSURE,
+    DEVICE_CLASS_TEMPERATURE,
+    PRESSURE_BAR,
+    TEMP_CELSIUS,
+    TIME_HOURS,
+)
+from homeassistant.helpers.entity import Entity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up all entries for Wolf Platform."""
+
+    coordinator = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR]
+    parameters = hass.data[DOMAIN][config_entry.entry_id][PARAMETERS]
+    device_id = hass.data[DOMAIN][config_entry.entry_id][DEVICE_ID]
+
+    entities = []
+    for parameter in parameters:
+        if isinstance(parameter, Temperature):
+            entities.append(WolfLinkTemperature(coordinator, parameter, device_id))
+        if isinstance(parameter, Pressure):
+            entities.append(WolfLinkPressure(coordinator, parameter, device_id))
+        if isinstance(parameter, PercentageParameter):
+            entities.append(WolfLinkPercentage(coordinator, parameter, device_id))
+        if isinstance(parameter, ListItemParameter):
+            entities.append(WolfLinkState(coordinator, parameter, device_id))
+        if isinstance(parameter, HoursParameter):
+            entities.append(WolfLinkHours(coordinator, parameter, device_id))
+        if isinstance(parameter, SimpleParameter):
+            entities.append(WolfLinkSensor(coordinator, parameter, device_id))
+
+    async_add_entities(entities, True)
+
+
+class WolfLinkSensor(Entity):
+    """Base class for all Wolf entities."""
+
+    def __init__(self, coordinator, wolf_object: Parameter, device_id):
+        """Initialize."""
+        self.coordinator = coordinator
+        self.wolf_object = wolf_object
+        self.device_id = device_id
+
+    @property
+    def name(self):
+        """Return the name."""
+        return f"{self.wolf_object.name}"
+
+    @property
+    def state(self):
+        """Return the state."""
+        return self.coordinator.data[self.wolf_object.value_id]
+
+    @property
+    def device_state_attributes(self):
+        """Return the state attributes."""
+        return {
+            "parameter_id": self.wolf_object.parameter_id,
+            "value_id": self.wolf_object.value_id,
+            "parent": self.wolf_object.parent,
+        }
+
+    @property
+    def unique_id(self):
+        """Return a unique_id for this entity."""
+        return f"{self.device_id}:{self.wolf_object.parameter_id}"
+
+    @property
+    def available(self):
+        """Return True if entity is available."""
+        return self.coordinator.last_update_success
+
+    @property
+    def should_poll(self):
+        """No need to poll. Coordinator notifies entity of updates."""
+        return False
+
+    async def async_added_to_hass(self):
+        """When entity is added to hass."""
+        self.async_on_remove(
+            self.coordinator.async_add_listener(self.async_write_ha_state)
+        )
+
+    async def async_update(self):
+        """Update the sensor."""
+        await self.coordinator.async_request_refresh()
+        _LOGGER.debug("Updating %s", self.coordinator.data[self.wolf_object.value_id])
+
+
+class WolfLinkHours(WolfLinkSensor):
+    """Class for hour based entities."""
+
+    @property
+    def icon(self):
+        """Icon to display in the front Aend."""
+        return "mdi:clock"
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return TIME_HOURS
+
+
+class WolfLinkTemperature(WolfLinkSensor):
+    """Class for temperature based entities."""
+
+    @property
+    def device_class(self):
+        """Return the device_class."""
+        return DEVICE_CLASS_TEMPERATURE
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return TEMP_CELSIUS
+
+
+class WolfLinkPressure(WolfLinkSensor):
+    """Class for pressure based entities."""
+
+    @property
+    def device_class(self):
+        """Return the device_class."""
+        return DEVICE_CLASS_PRESSURE
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return PRESSURE_BAR
+
+
+class WolfLinkPercentage(WolfLinkSensor):
+    """Class for percentage based entities."""
+
+    @property
+    def unit_of_measurement(self):
+        """Return the unit the value is expressed in."""
+        return self.wolf_object.unit
+
+
+class WolfLinkState(WolfLinkSensor):
+    """Class for entities which has defined list of state."""
+
+    @property
+    def device_class(self):
+        """Return the device class."""
+        return "wolflink__state"
+
+    @property
+    def state(self):
+        """Return the state converting with supported values."""
+        state = self.coordinator.data[self.wolf_object.value_id]
+        resolved_state = [
+            item for item in self.wolf_object.items if item.value == int(state)
+        ]
+        if resolved_state:
+            resolved_name = resolved_state[0].name
+            return STATES.get(resolved_name, resolved_name)
+        return state

homeassistant/components/wolflink/strings.json

@@ -0,0 +1,27 @@
+{
+  "config": {
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    },
+    "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%]"
+    },
+    "step": {
+      "user": {
+        "data": {
+          "username": "[%key:common::config_flow::data::username%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "title": "WOLF SmartSet connection"
+      },
+      "device": {
+        "data": {
+          "device_name": "Device"
+        },
+        "title": "Select WOLF device"
+      }
+    }
+  }
+}

homeassistant/components/wolflink/strings.sensor.json

@@ -0,0 +1,87 @@
+{
+  "state": {
+    "wolflink__state": {
+      "ein": "Enabled",
+      "deaktiviert": "Inactive",
+      "aus": "Disabled",
+      "standby": "Standby",
+      "auto": "Auto",
+      "permanent": "Permament",
+      "initialisierung": "Initialization",
+      "antilegionellenfunktion": "Anti-legionella Function",
+      "fernschalter_ein": "Remote control enabled",
+      "1_x_warmwasser": "1 x DHW",
+      "bereit_keine_ladung": "Ready, not loading",
+      "solarbetrieb": "Solar mode",
+      "reduzierter_betrieb": "Limited mode",
+      "smart_home": "SmartHome",
+      "smart_grid": "SmartGrid",
+      "ruhekontakt": "Rest contact",
+      "vorspulen": "Entry rinsing",
+      "zunden": "Ignition",
+      "stabilisierung": "Stablization",
+      "ventilprufung": "Valve test",
+      "nachspulen": "Post-flush",
+      "softstart": "Soft start",
+      "taktsperre": "Anti-cycle",
+      "betrieb_ohne_brenner": "Working without burner",
+      "abgasklappe": "Flue gas damper",
+      "storung": "Fault",
+      "gradienten_uberwachung": "Gradient monitoring",
+      "gasdruck": "Gas pressure",
+      "spreizung_hoch": "dT too wide",
+      "spreizung_kf": "Spread KF",
+      "test": "Test",
+      "start": "Start",
+      "frost_heizkreis": "Heating circuit frost",
+      "frost_warmwasser": "DHW frost",
+      "schornsteinfeger": "Emissions test",
+      "kombibetrieb": "Combi mode",
+      "parallelbetrieb": "Parallel mode",
+      "warmwasserbetrieb": "DHW mode",
+      "warmwassernachlauf": "DHW run-on",
+      "heizbetrieb": "Heating mode",
+      "nachlauf_heizkreispumpe": "Heating circuit pump run-on",
+      "frostschutz": "Frost protection",
+      "kaskadenbetrieb": "Cascade operation",
+      "glt_betrieb": "BMS mode",
+      "kalibration": "Calibration",
+      "kalibration_heizbetrieb": "Heating mode calibration",
+      "kalibration_warmwasserbetrieb": "DHW calibration",
+      "kalibration_kombibetrieb": "Combi mode calibration",
+      "warmwasser_schnellstart": "DHW quick start",
+      "externe_deaktivierung": "External deactivation",
+      "heizung": "Heating",
+      "warmwasser": "DHW",
+      "kombigerat": "Combi boiler",
+      "kombigerat_mit_solareinbindung": "Combi boiler with solar integration",
+      "heizgerat_mit_speicher": "Boiler with cylinder",
+      "nur_heizgerat": "Boiler only",
+      "aktiviert": "Activated",
+      "sparen": "Economy",
+      "estrichtrocknung": "Screed drying",
+      "telefonfernschalter": "Telephone remote switch",
+      "partymodus": "Party mode",
+      "urlaubsmodus": "Holiday mode",
+      "automatik_ein": "Automatic ON",
+      "automatik_aus": "Automatic OFF",
+      "permanentbetrieb": "Permanent mode",
+      "sparbetrieb": "Economy mode",
+      "auto_on_cool": "AutoOnCool",
+      "auto_off_cool": "AutoOffCool",
+      "perm_cooling": "PermCooling",
+      "absenkbetrieb": "Setback mode",
+      "eco": "Eco",
+      "absenkstop": "Setback stop",
+      "at_abschaltung": "OT shutdown",
+      "rt_abschaltung": "RT shutdown",
+      "at_frostschutz": "OT frost protection",
+      "rt_frostschutz": "RT frost protection",
+      "dhw_prior": "DHWPrior",
+      "cooling": "Cooling",
+      "tpw": "TPW",
+      "warmwasservorrang": "DHW priority",
+      "mindest_kombizeit": "Minimum combi time"
+    }
+  }
+}

homeassistant/components/wolflink/translations/en.json

@@ -0,0 +1,28 @@
+{
+  "config": {
+    "abort": {
+      "already_configured": "Device is already configured"
+    },
+    "error": {
+      "cannot_connect": "Failed to connect, please try again",
+      "invalid_auth": "Invalid authentication",
+      "unknown": "Unexpected error"
+    },
+    "step": {
+      "user": {
+        "data": {
+          "username": "Username",
+          "password": "Password"
+        },
+        "title": "WOLF SmartSet connection"
+      },
+      "device": {
+        "data": {
+          "device_name": "Device"
+        },
+        "title": "Select WOLF device"
+      }
+    },
+    "title": "Wolf SmartSet Service"
+  }
+}

homeassistant/generated/config_flows.py

@@ -186,6 +186,7 @@ FLOWS = [
     "wiffi",
     "withings",
     "wled",
+    "wolflink",
     "xiaomi_aqara",
     "xiaomi_miio",
     "zerproc",

requirements_all.txt

@@ -2212,6 +2212,9 @@ withings-api==2.1.6
 # homeassistant.components.wled
 wled==0.4.3
 
+# homeassistant.components.wolflink
+wolf_smartset==0.1.4
+
 # homeassistant.components.xbee
 xbee-helper==0.0.7
 

requirements_test_all.txt

@@ -978,6 +978,9 @@ withings-api==2.1.6
 # homeassistant.components.wled
 wled==0.4.3
 
+# homeassistant.components.wolflink
+wolf_smartset==0.1.4
+
 # homeassistant.components.bluesound
 # homeassistant.components.rest
 # homeassistant.components.startca

tests/components/wolflink/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Wolf SmartSet Service integration."""

tests/components/wolflink/test_config_flow.py

@@ -0,0 +1,139 @@
+"""Test the Wolf SmartSet Service config flow."""
+from httpcore import ConnectError
+from wolf_smartset.models import Device
+from wolf_smartset.token_auth import InvalidAuth
+
+from homeassistant import config_entries, data_entry_flow, setup
+from homeassistant.components.wolflink.const import (
+    DEVICE_GATEWAY,
+    DEVICE_ID,
+    DEVICE_NAME,
+    DOMAIN,
+)
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+
+from tests.async_mock import patch
+from tests.common import MockConfigEntry
+
+CONFIG = {
+    DEVICE_NAME: "test-device",
+    DEVICE_ID: 1234,
+    DEVICE_GATEWAY: 5678,
+    CONF_USERNAME: "test-username",
+    CONF_PASSWORD: "test-password",
+}
+
+INPUT_CONFIG = {
+    CONF_USERNAME: CONFIG[CONF_USERNAME],
+    CONF_PASSWORD: CONFIG[CONF_PASSWORD],
+}
+
+DEVICE = Device(CONFIG[DEVICE_ID], CONFIG[DEVICE_GATEWAY], CONFIG[DEVICE_NAME])
+
+
+async def test_show_form(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"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+
+
+async def test_device_step_form(hass):
+    """Test we get the second step of config."""
+    with patch(
+        "homeassistant.components.wolflink.config_flow.WolfClient.fetch_system_list",
+        return_value=[DEVICE],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}, data=INPUT_CONFIG
+        )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "device"
+
+
+async def test_create_entry(hass):
+    """Test entity creation from device step."""
+    with patch(
+        "homeassistant.components.wolflink.config_flow.WolfClient.fetch_system_list",
+        return_value=[DEVICE],
+    ), patch("homeassistant.components.wolflink.async_setup_entry", return_value=True):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}, data=INPUT_CONFIG
+        )
+
+        result_create_entry = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"device_name": CONFIG[DEVICE_NAME]},
+        )
+
+    assert result_create_entry["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result_create_entry["title"] == CONFIG[DEVICE_NAME]
+    assert result_create_entry["data"] == CONFIG
+
+
+async def test_form_invalid_auth(hass):
+    """Test we handle invalid auth."""
+    with patch(
+        "homeassistant.components.wolflink.config_flow.WolfClient.fetch_system_list",
+        side_effect=InvalidAuth,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}, data=INPUT_CONFIG
+        )
+
+    assert result["type"] == "form"
+    assert result["errors"] == {"base": "invalid_auth"}
+
+
+async def test_form_cannot_connect(hass):
+    """Test we handle cannot connect error."""
+    with patch(
+        "homeassistant.components.wolflink.config_flow.WolfClient.fetch_system_list",
+        side_effect=ConnectError,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}, data=INPUT_CONFIG
+        )
+
+    assert result["type"] == "form"
+    assert result["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_unknown_exception(hass):
+    """Test we handle cannot connect error."""
+    with patch(
+        "homeassistant.components.wolflink.config_flow.WolfClient.fetch_system_list",
+        side_effect=Exception,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}, data=INPUT_CONFIG
+        )
+
+    assert result["type"] == "form"
+    assert result["errors"] == {"base": "unknown"}
+
+
+async def test_already_configured_error(hass):
+    """Test already configured while creating entry."""
+    with patch(
+        "homeassistant.components.wolflink.config_flow.WolfClient.fetch_system_list",
+        return_value=[DEVICE],
+    ), patch("homeassistant.components.wolflink.async_setup_entry", return_value=True):
+
+        MockConfigEntry(
+            domain=DOMAIN, unique_id=CONFIG[DEVICE_ID], data=CONFIG
+        ).add_to_hass(hass)
+
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}, data=INPUT_CONFIG
+        )
+
+        result_create_entry = await hass.config_entries.flow.async_configure(
+            result["flow_id"], {"device_name": CONFIG[DEVICE_NAME]},
+        )
+
+    assert result_create_entry["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result_create_entry["reason"] == "already_configured"