Add IoTaWatt integration (#55364)
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
parent
daa9c8d856
commit
3bd9be2f6d
17 changed files with 778 additions and 0 deletions
|
@ -248,6 +248,7 @@ homeassistant/components/integration/* @dgomes
|
|||
homeassistant/components/intent/* @home-assistant/core
|
||||
homeassistant/components/intesishome/* @jnimmo
|
||||
homeassistant/components/ios/* @robbiet480
|
||||
homeassistant/components/iotawatt/* @gtdiehl
|
||||
homeassistant/components/iperf3/* @rohankapoorcom
|
||||
homeassistant/components/ipma/* @dgomes @abmantis
|
||||
homeassistant/components/ipp/* @ctalkington
|
||||
|
|
24
homeassistant/components/iotawatt/__init__.py
Normal file
24
homeassistant/components/iotawatt/__init__.py
Normal file
|
@ -0,0 +1,24 @@
|
|||
"""The iotawatt integration."""
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import IotawattUpdater
|
||||
|
||||
PLATFORMS = ("sensor",)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up iotawatt from a config entry."""
|
||||
coordinator = IotawattUpdater(hass, entry)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
return unload_ok
|
107
homeassistant/components/iotawatt/config_flow.py
Normal file
107
homeassistant/components/iotawatt/config_flow.py
Normal file
|
@ -0,0 +1,107 @@
|
|||
"""Config flow for iotawatt integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from iotawattpy.iotawatt import Iotawatt
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, core, exceptions
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.helpers import httpx_client
|
||||
|
||||
from .const import CONNECTION_ERRORS, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def validate_input(
|
||||
hass: core.HomeAssistant, data: dict[str, str]
|
||||
) -> dict[str, str]:
|
||||
"""Validate the user input allows us to connect."""
|
||||
iotawatt = Iotawatt(
|
||||
"",
|
||||
data[CONF_HOST],
|
||||
httpx_client.get_async_client(hass),
|
||||
data.get(CONF_USERNAME),
|
||||
data.get(CONF_PASSWORD),
|
||||
)
|
||||
try:
|
||||
is_connected = await iotawatt.connect()
|
||||
except CONNECTION_ERRORS:
|
||||
return {"base": "cannot_connect"}
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
return {"base": "unknown"}
|
||||
|
||||
if not is_connected:
|
||||
return {"base": "invalid_auth"}
|
||||
|
||||
return {}
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for iotawatt."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize."""
|
||||
self._data = {}
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle the initial step."""
|
||||
if user_input is None:
|
||||
user_input = {}
|
||||
|
||||
schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST, default=user_input.get(CONF_HOST, "")): str,
|
||||
}
|
||||
)
|
||||
if not user_input:
|
||||
return self.async_show_form(step_id="user", data_schema=schema)
|
||||
|
||||
if not (errors := await validate_input(self.hass, user_input)):
|
||||
return self.async_create_entry(title=user_input[CONF_HOST], data=user_input)
|
||||
|
||||
if errors == {"base": "invalid_auth"}:
|
||||
self._data.update(user_input)
|
||||
return await self.async_step_auth()
|
||||
|
||||
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
|
||||
|
||||
async def async_step_auth(self, user_input=None):
|
||||
"""Authenticate user if authentication is enabled on the IoTaWatt device."""
|
||||
if user_input is None:
|
||||
user_input = {}
|
||||
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")
|
||||
): str,
|
||||
vol.Required(
|
||||
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
|
||||
): str,
|
||||
}
|
||||
)
|
||||
if not user_input:
|
||||
return self.async_show_form(step_id="auth", data_schema=data_schema)
|
||||
|
||||
data = {**self._data, **user_input}
|
||||
|
||||
if errors := await validate_input(self.hass, data):
|
||||
return self.async_show_form(
|
||||
step_id="auth", data_schema=data_schema, errors=errors
|
||||
)
|
||||
|
||||
return self.async_create_entry(title=data[CONF_HOST], data=data)
|
||||
|
||||
|
||||
class CannotConnect(exceptions.HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class InvalidAuth(exceptions.HomeAssistantError):
|
||||
"""Error to indicate there is invalid auth."""
|
12
homeassistant/components/iotawatt/const.py
Normal file
12
homeassistant/components/iotawatt/const.py
Normal file
|
@ -0,0 +1,12 @@
|
|||
"""Constants for the IoTaWatt integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import httpx
|
||||
|
||||
DOMAIN = "iotawatt"
|
||||
VOLT_AMPERE_REACTIVE = "VAR"
|
||||
VOLT_AMPERE_REACTIVE_HOURS = "VARh"
|
||||
|
||||
CONNECTION_ERRORS = (KeyError, json.JSONDecodeError, httpx.HTTPError)
|
56
homeassistant/components/iotawatt/coordinator.py
Normal file
56
homeassistant/components/iotawatt/coordinator.py
Normal file
|
@ -0,0 +1,56 @@
|
|||
"""IoTaWatt DataUpdateCoordinator."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from iotawattpy.iotawatt import Iotawatt
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import httpx_client
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import CONNECTION_ERRORS
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IotawattUpdater(DataUpdateCoordinator):
|
||||
"""Class to manage fetching update data from the IoTaWatt Energy Device."""
|
||||
|
||||
api: Iotawatt | None = None
|
||||
|
||||
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
|
||||
"""Initialize IotaWattUpdater object."""
|
||||
self.entry = entry
|
||||
super().__init__(
|
||||
hass=hass,
|
||||
logger=_LOGGER,
|
||||
name=entry.title,
|
||||
update_interval=timedelta(seconds=30),
|
||||
)
|
||||
|
||||
async def _async_update_data(self):
|
||||
"""Fetch sensors from IoTaWatt device."""
|
||||
if self.api is None:
|
||||
api = Iotawatt(
|
||||
self.entry.title,
|
||||
self.entry.data[CONF_HOST],
|
||||
httpx_client.get_async_client(self.hass),
|
||||
self.entry.data.get(CONF_USERNAME),
|
||||
self.entry.data.get(CONF_PASSWORD),
|
||||
)
|
||||
try:
|
||||
is_authenticated = await api.connect()
|
||||
except CONNECTION_ERRORS as err:
|
||||
raise UpdateFailed("Connection failed") from err
|
||||
|
||||
if not is_authenticated:
|
||||
raise UpdateFailed("Authentication error")
|
||||
|
||||
self.api = api
|
||||
|
||||
await self.api.update()
|
||||
return self.api.getSensors()
|
13
homeassistant/components/iotawatt/manifest.json
Normal file
13
homeassistant/components/iotawatt/manifest.json
Normal file
|
@ -0,0 +1,13 @@
|
|||
{
|
||||
"domain": "iotawatt",
|
||||
"name": "IoTaWatt",
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/iotawatt",
|
||||
"requirements": [
|
||||
"iotawattpy==0.0.8"
|
||||
],
|
||||
"codeowners": [
|
||||
"@gtdiehl"
|
||||
],
|
||||
"iot_class": "local_polling"
|
||||
}
|
213
homeassistant/components/iotawatt/sensor.py
Normal file
213
homeassistant/components/iotawatt/sensor.py
Normal file
|
@ -0,0 +1,213 @@
|
|||
"""Support for IoTaWatt Energy monitor."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
|
||||
from iotawattpy.sensor import Sensor
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
STATE_CLASS_MEASUREMENT,
|
||||
STATE_CLASS_TOTAL_INCREASING,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
DEVICE_CLASS_CURRENT,
|
||||
DEVICE_CLASS_ENERGY,
|
||||
DEVICE_CLASS_POWER,
|
||||
DEVICE_CLASS_POWER_FACTOR,
|
||||
DEVICE_CLASS_VOLTAGE,
|
||||
ELECTRIC_CURRENT_AMPERE,
|
||||
ELECTRIC_POTENTIAL_VOLT,
|
||||
ENERGY_WATT_HOUR,
|
||||
FREQUENCY_HERTZ,
|
||||
PERCENTAGE,
|
||||
POWER_VOLT_AMPERE,
|
||||
POWER_WATT,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers import entity, entity_registry, update_coordinator
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
|
||||
|
||||
from .const import DOMAIN, VOLT_AMPERE_REACTIVE, VOLT_AMPERE_REACTIVE_HOURS
|
||||
from .coordinator import IotawattUpdater
|
||||
|
||||
|
||||
@dataclass
|
||||
class IotaWattSensorEntityDescription(SensorEntityDescription):
|
||||
"""Class describing IotaWatt sensor entities."""
|
||||
|
||||
value: Callable | None = None
|
||||
|
||||
|
||||
ENTITY_DESCRIPTION_KEY_MAP: dict[str, IotaWattSensorEntityDescription] = {
|
||||
"Amps": IotaWattSensorEntityDescription(
|
||||
"Amps",
|
||||
native_unit_of_measurement=ELECTRIC_CURRENT_AMPERE,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
device_class=DEVICE_CLASS_CURRENT,
|
||||
),
|
||||
"Hz": IotaWattSensorEntityDescription(
|
||||
"Hz",
|
||||
native_unit_of_measurement=FREQUENCY_HERTZ,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
icon="mdi:flash",
|
||||
),
|
||||
"PF": IotaWattSensorEntityDescription(
|
||||
"PF",
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
device_class=DEVICE_CLASS_POWER_FACTOR,
|
||||
value=lambda value: value * 100,
|
||||
),
|
||||
"Watts": IotaWattSensorEntityDescription(
|
||||
"Watts",
|
||||
native_unit_of_measurement=POWER_WATT,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
device_class=DEVICE_CLASS_POWER,
|
||||
),
|
||||
"WattHours": IotaWattSensorEntityDescription(
|
||||
"WattHours",
|
||||
native_unit_of_measurement=ENERGY_WATT_HOUR,
|
||||
state_class=STATE_CLASS_TOTAL_INCREASING,
|
||||
device_class=DEVICE_CLASS_ENERGY,
|
||||
),
|
||||
"VA": IotaWattSensorEntityDescription(
|
||||
"VA",
|
||||
native_unit_of_measurement=POWER_VOLT_AMPERE,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
icon="mdi:flash",
|
||||
),
|
||||
"VAR": IotaWattSensorEntityDescription(
|
||||
"VAR",
|
||||
native_unit_of_measurement=VOLT_AMPERE_REACTIVE,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
icon="mdi:flash",
|
||||
),
|
||||
"VARh": IotaWattSensorEntityDescription(
|
||||
"VARh",
|
||||
native_unit_of_measurement=VOLT_AMPERE_REACTIVE_HOURS,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
icon="mdi:flash",
|
||||
),
|
||||
"Volts": IotaWattSensorEntityDescription(
|
||||
"Volts",
|
||||
native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
|
||||
state_class=STATE_CLASS_MEASUREMENT,
|
||||
device_class=DEVICE_CLASS_VOLTAGE,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Add sensors for passed config_entry in HA."""
|
||||
coordinator: IotawattUpdater = hass.data[DOMAIN][config_entry.entry_id]
|
||||
created = set()
|
||||
|
||||
@callback
|
||||
def _create_entity(key: str) -> IotaWattSensor:
|
||||
"""Create a sensor entity."""
|
||||
created.add(key)
|
||||
return IotaWattSensor(
|
||||
coordinator=coordinator,
|
||||
key=key,
|
||||
mac_address=coordinator.data["sensors"][key].hub_mac_address,
|
||||
name=coordinator.data["sensors"][key].getName(),
|
||||
entity_description=ENTITY_DESCRIPTION_KEY_MAP.get(
|
||||
coordinator.data["sensors"][key].getUnit(),
|
||||
IotaWattSensorEntityDescription("base_sensor"),
|
||||
),
|
||||
)
|
||||
|
||||
async_add_entities(_create_entity(key) for key in coordinator.data["sensors"])
|
||||
|
||||
@callback
|
||||
def new_data_received():
|
||||
"""Check for new sensors."""
|
||||
entities = [
|
||||
_create_entity(key)
|
||||
for key in coordinator.data["sensors"]
|
||||
if key not in created
|
||||
]
|
||||
if entities:
|
||||
async_add_entities(entities)
|
||||
|
||||
coordinator.async_add_listener(new_data_received)
|
||||
|
||||
|
||||
class IotaWattSensor(update_coordinator.CoordinatorEntity, SensorEntity):
|
||||
"""Defines a IoTaWatt Energy Sensor."""
|
||||
|
||||
entity_description: IotaWattSensorEntityDescription
|
||||
_attr_force_update = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator,
|
||||
key,
|
||||
mac_address,
|
||||
name,
|
||||
entity_description: IotaWattSensorEntityDescription,
|
||||
):
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator=coordinator)
|
||||
|
||||
self._key = key
|
||||
data = self._sensor_data
|
||||
if data.getType() == "Input":
|
||||
self._attr_unique_id = (
|
||||
f"{data.hub_mac_address}-input-{data.getChannel()}-{data.getUnit()}"
|
||||
)
|
||||
self.entity_description = entity_description
|
||||
|
||||
@property
|
||||
def _sensor_data(self) -> Sensor:
|
||||
"""Return sensor data."""
|
||||
return self.coordinator.data["sensors"][self._key]
|
||||
|
||||
@property
|
||||
def name(self) -> str | None:
|
||||
"""Return name of the entity."""
|
||||
return self._sensor_data.getName()
|
||||
|
||||
@property
|
||||
def device_info(self) -> entity.DeviceInfo | None:
|
||||
"""Return device info."""
|
||||
return {
|
||||
"connections": {
|
||||
(CONNECTION_NETWORK_MAC, self._sensor_data.hub_mac_address)
|
||||
},
|
||||
"manufacturer": "IoTaWatt",
|
||||
"model": "IoTaWatt",
|
||||
}
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle updated data from the coordinator."""
|
||||
if self._key not in self.coordinator.data["sensors"]:
|
||||
if self._attr_unique_id:
|
||||
entity_registry.async_get(self.hass).async_remove(self.entity_id)
|
||||
else:
|
||||
self.hass.async_create_task(self.async_remove())
|
||||
return
|
||||
|
||||
super()._handle_coordinator_update()
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
"""Return the extra state attributes of the entity."""
|
||||
data = self._sensor_data
|
||||
attrs = {"type": data.getType()}
|
||||
if attrs["type"] == "Input":
|
||||
attrs["channel"] = data.getChannel()
|
||||
|
||||
return attrs
|
||||
|
||||
@property
|
||||
def native_value(self) -> entity.StateType:
|
||||
"""Return the state of the sensor."""
|
||||
if func := self.entity_description.value:
|
||||
return func(self._sensor_data.getValue())
|
||||
|
||||
return self._sensor_data.getValue()
|
23
homeassistant/components/iotawatt/strings.json
Normal file
23
homeassistant/components/iotawatt/strings.json
Normal file
|
@ -0,0 +1,23 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]"
|
||||
}
|
||||
},
|
||||
"auth": {
|
||||
"data": {
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
},
|
||||
"description": "The IoTawatt device requires authentication. Please enter the username and password and click the Submit button."
|
||||
}
|
||||
},
|
||||
"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%]"
|
||||
}
|
||||
}
|
||||
}
|
24
homeassistant/components/iotawatt/translations/en.json
Normal file
24
homeassistant/components/iotawatt/translations/en.json
Normal file
|
@ -0,0 +1,24 @@
|
|||
{
|
||||
"config": {
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect",
|
||||
"invalid_auth": "Invalid authentication",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"auth": {
|
||||
"data": {
|
||||
"password": "Password",
|
||||
"username": "Username"
|
||||
},
|
||||
"description": "The IoTawatt device requires authentication. Please enter the username and password and click the Submit button."
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "Host"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"title": "iotawatt"
|
||||
}
|
|
@ -130,6 +130,7 @@ FLOWS = [
|
|||
"ifttt",
|
||||
"insteon",
|
||||
"ios",
|
||||
"iotawatt",
|
||||
"ipma",
|
||||
"ipp",
|
||||
"iqvia",
|
||||
|
|
|
@ -864,6 +864,9 @@ influxdb-client==1.14.0
|
|||
# homeassistant.components.influxdb
|
||||
influxdb==5.2.3
|
||||
|
||||
# homeassistant.components.iotawatt
|
||||
iotawattpy==0.0.8
|
||||
|
||||
# homeassistant.components.iperf3
|
||||
iperf3==0.1.11
|
||||
|
||||
|
|
|
@ -504,6 +504,9 @@ influxdb-client==1.14.0
|
|||
# homeassistant.components.influxdb
|
||||
influxdb==5.2.3
|
||||
|
||||
# homeassistant.components.iotawatt
|
||||
iotawattpy==0.0.8
|
||||
|
||||
# homeassistant.components.gogogate2
|
||||
ismartgate==4.0.0
|
||||
|
||||
|
|
21
tests/components/iotawatt/__init__.py
Normal file
21
tests/components/iotawatt/__init__.py
Normal file
|
@ -0,0 +1,21 @@
|
|||
"""Tests for the IoTaWatt integration."""
|
||||
from iotawattpy.sensor import Sensor
|
||||
|
||||
INPUT_SENSOR = Sensor(
|
||||
channel="1",
|
||||
name="My Sensor",
|
||||
io_type="Input",
|
||||
unit="WattHours",
|
||||
value="23",
|
||||
begin="",
|
||||
mac_addr="mock-mac",
|
||||
)
|
||||
OUTPUT_SENSOR = Sensor(
|
||||
channel="N/A",
|
||||
name="My WattHour Sensor",
|
||||
io_type="Output",
|
||||
unit="WattHours",
|
||||
value="243",
|
||||
begin="",
|
||||
mac_addr="mock-mac",
|
||||
)
|
27
tests/components/iotawatt/conftest.py
Normal file
27
tests/components/iotawatt/conftest.py
Normal file
|
@ -0,0 +1,27 @@
|
|||
"""Test fixtures for IoTaWatt."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.iotawatt import DOMAIN
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def entry(hass):
|
||||
"""Mock config entry added to HA."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={"host": "1.2.3.4"})
|
||||
entry.add_to_hass(hass)
|
||||
return entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_iotawatt(entry):
|
||||
"""Mock iotawatt."""
|
||||
with patch("homeassistant.components.iotawatt.coordinator.Iotawatt") as mock:
|
||||
instance = mock.return_value
|
||||
instance.connect = AsyncMock(return_value=True)
|
||||
instance.update = AsyncMock()
|
||||
instance.getSensors.return_value = {"sensors": {}}
|
||||
yield instance
|
143
tests/components/iotawatt/test_config_flow.py
Normal file
143
tests/components/iotawatt/test_config_flow.py
Normal file
|
@ -0,0 +1,143 @@
|
|||
"""Test the IoTawatt config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import httpx
|
||||
|
||||
from homeassistant import config_entries, setup
|
||||
from homeassistant.components.iotawatt.const import DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
|
||||
|
||||
|
||||
async def test_form(hass: HomeAssistant) -> None:
|
||||
"""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"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == config_entries.SOURCE_USER
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.iotawatt.async_setup_entry",
|
||||
return_value=True,
|
||||
), patch(
|
||||
"homeassistant.components.iotawatt.config_flow.Iotawatt.connect",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"host": "1.1.1.1",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result2["data"] == {
|
||||
"host": "1.1.1.1",
|
||||
}
|
||||
|
||||
|
||||
async def test_form_auth(hass: HomeAssistant) -> None:
|
||||
"""Test we handle auth."""
|
||||
|
||||
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"] == RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.iotawatt.config_flow.Iotawatt.connect",
|
||||
return_value=False,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"host": "1.1.1.1"},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["step_id"] == "auth"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.iotawatt.config_flow.Iotawatt.connect",
|
||||
return_value=False,
|
||||
):
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "mock-user",
|
||||
"password": "mock-pass",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result3["type"] == RESULT_TYPE_FORM
|
||||
assert result3["step_id"] == "auth"
|
||||
assert result3["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.iotawatt.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.iotawatt.config_flow.Iotawatt.connect",
|
||||
return_value=True,
|
||||
):
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "mock-user",
|
||||
"password": "mock-pass",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result4["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
assert result4["data"] == {
|
||||
"host": "1.1.1.1",
|
||||
"username": "mock-user",
|
||||
"password": "mock-pass",
|
||||
}
|
||||
|
||||
|
||||
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
|
||||
"""Test we handle cannot connect error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.iotawatt.config_flow.Iotawatt.connect",
|
||||
side_effect=httpx.HTTPError("any"),
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"host": "1.1.1.1"},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_form_setup_exception(hass: HomeAssistant) -> None:
|
||||
"""Test we handle broad exception."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.iotawatt.config_flow.Iotawatt.connect",
|
||||
side_effect=Exception,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"host": "1.1.1.1"},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
31
tests/components/iotawatt/test_init.py
Normal file
31
tests/components/iotawatt/test_init.py
Normal file
|
@ -0,0 +1,31 @@
|
|||
"""Test init."""
|
||||
import httpx
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import INPUT_SENSOR
|
||||
|
||||
|
||||
async def test_setup_unload(hass, mock_iotawatt, entry):
|
||||
"""Test we can setup and unload an entry."""
|
||||
mock_iotawatt.getSensors.return_value["sensors"]["my_sensor_key"] = INPUT_SENSOR
|
||||
assert await async_setup_component(hass, "iotawatt", {})
|
||||
await hass.async_block_till_done()
|
||||
assert await hass.config_entries.async_unload(entry.entry_id)
|
||||
|
||||
|
||||
async def test_setup_connection_failed(hass, mock_iotawatt, entry):
|
||||
"""Test connection error during startup."""
|
||||
mock_iotawatt.connect.side_effect = httpx.ConnectError("")
|
||||
assert await async_setup_component(hass, "iotawatt", {})
|
||||
await hass.async_block_till_done()
|
||||
assert entry.state == ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_setup_auth_failed(hass, mock_iotawatt, entry):
|
||||
"""Test auth error during startup."""
|
||||
mock_iotawatt.connect.return_value = False
|
||||
assert await async_setup_component(hass, "iotawatt", {})
|
||||
await hass.async_block_till_done()
|
||||
assert entry.state == ConfigEntryState.SETUP_RETRY
|
76
tests/components/iotawatt/test_sensor.py
Normal file
76
tests/components/iotawatt/test_sensor.py
Normal file
|
@ -0,0 +1,76 @@
|
|||
"""Test setting up sensors."""
|
||||
from datetime import timedelta
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
ATTR_STATE_CLASS,
|
||||
DEVICE_CLASS_ENERGY,
|
||||
STATE_CLASS_TOTAL_INCREASING,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_DEVICE_CLASS,
|
||||
ATTR_FRIENDLY_NAME,
|
||||
ATTR_UNIT_OF_MEASUREMENT,
|
||||
ENERGY_WATT_HOUR,
|
||||
)
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from . import INPUT_SENSOR, OUTPUT_SENSOR
|
||||
|
||||
from tests.common import async_fire_time_changed
|
||||
|
||||
|
||||
async def test_sensor_type_input(hass, mock_iotawatt):
|
||||
"""Test input sensors work."""
|
||||
assert await async_setup_component(hass, "iotawatt", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_entity_ids()) == 0
|
||||
|
||||
# Discover this sensor during a regular update.
|
||||
mock_iotawatt.getSensors.return_value["sensors"]["my_sensor_key"] = INPUT_SENSOR
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_entity_ids()) == 1
|
||||
|
||||
state = hass.states.get("sensor.my_sensor")
|
||||
assert state is not None
|
||||
assert state.state == "23"
|
||||
assert state.attributes[ATTR_STATE_CLASS] == STATE_CLASS_TOTAL_INCREASING
|
||||
assert state.attributes[ATTR_FRIENDLY_NAME] == "My Sensor"
|
||||
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == ENERGY_WATT_HOUR
|
||||
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_ENERGY
|
||||
assert state.attributes["channel"] == "1"
|
||||
assert state.attributes["type"] == "Input"
|
||||
|
||||
mock_iotawatt.getSensors.return_value["sensors"].pop("my_sensor_key")
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get("sensor.my_sensor") is None
|
||||
|
||||
|
||||
async def test_sensor_type_output(hass, mock_iotawatt):
|
||||
"""Tests the sensor type of Output."""
|
||||
mock_iotawatt.getSensors.return_value["sensors"][
|
||||
"my_watthour_sensor_key"
|
||||
] = OUTPUT_SENSOR
|
||||
assert await async_setup_component(hass, "iotawatt", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_entity_ids()) == 1
|
||||
|
||||
state = hass.states.get("sensor.my_watthour_sensor")
|
||||
assert state is not None
|
||||
assert state.state == "243"
|
||||
assert state.attributes[ATTR_FRIENDLY_NAME] == "My WattHour Sensor"
|
||||
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == ENERGY_WATT_HOUR
|
||||
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_ENERGY
|
||||
assert state.attributes["type"] == "Output"
|
||||
|
||||
mock_iotawatt.getSensors.return_value["sensors"].pop("my_watthour_sensor_key")
|
||||
async_fire_time_changed(hass, utcnow() + timedelta(seconds=30))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get("sensor.my_watthour_sensor") is None
|
Loading…
Add table
Reference in a new issue