github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add new integration to control Electra Smart HVAC devices (#70361) 

* Added new integration to support Electra Smart (HVAC)

* fixes + option to set scan interval

* renamed the module to electrasmart and added unittests

* added non tested files to .coveragerc

* changed the usage from UpdateCoordinator to each entity updates it self

* small fixes

* increased pypi package version, increased polling timeout to 60 seconds, improved error handling

* PARALLEL_UPDATE=1 to prevent multi access to the API

* code improvements

* aligned with the new HA APIs

* fixes

* fixes

* more

* fixes

* more

* more

* handled re-atuh flow

* fixed test

* removed hvac action

* added shabat mode

* tests: 100% coverage

* ran hassfest

* Update homeassistant/components/electrasmart/manifest.json

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* Update homeassistant/components/electrasmart/manifest.json

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* Update homeassistant/components/electrasmart/manifest.json

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* Update homeassistant/components/electrasmart/climate.py

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* address Shay's comments

* address Shay's comments

* address more comments

---------

Co-authored-by: Shay Levy <levyshay1@gmail.com>
This commit is contained in:
Jafar Atili 2023-05-20 13:13:32 +03:00 committed by GitHub
parent 09a8479cf0
commit 1935e126bf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 832 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -1535,6 +1535,8 @@ omit =
homeassistant/components/zwave_me/sensor.py
homeassistant/components/zwave_me/siren.py
homeassistant/components/zwave_me/switch.py
homeassistant/components/electrasmart/climate.py
homeassistant/components/electrasmart/__init__.py
[report]
# Regexes for lines to exclude from consideration

1
.strict-typing
View File

@ -105,6 +105,7 @@ homeassistant.components.dormakaba_dkey.*
homeassistant.components.dsmr.*
homeassistant.components.dunehd.*
homeassistant.components.efergy.*
homeassistant.components.electrasmart.*
homeassistant.components.elgato.*
homeassistant.components.elkm1.*
homeassistant.components.emulated_hue.*

2
CODEOWNERS
View File

@ -309,6 +309,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/egardia/ @jeroenterheerdt
/homeassistant/components/eight_sleep/ @mezz64 @raman325
/tests/components/eight_sleep/ @mezz64 @raman325
/homeassistant/components/electrasmart/ @jafar-atili
/tests/components/electrasmart/ @jafar-atili
/homeassistant/components/elgato/ @frenck
/tests/components/elgato/ @frenck
/homeassistant/components/elkm1/ @gwww @bdraco

46
homeassistant/components/electrasmart/__init__.py Normal file
View File

@ -0,0 +1,46 @@
"""The Electra Air Conditioner integration."""
from __future__ import annotations
from typing import cast
from electrasmart.api import ElectraAPI, ElectraApiError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_IMEI, DOMAIN
PLATFORMS: list[Platform] = [Platform.CLIMATE]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Electra Smart Air Conditioner from a config entry."""
hass.data.setdefault(DOMAIN, {})
entry.async_on_unload(entry.add_update_listener(update_listener))
hass.data[DOMAIN][entry.entry_id] = ElectraAPI(
async_get_clientsession(hass), entry.data[CONF_IMEI], entry.data[CONF_TOKEN]
)
try:
await cast(ElectraAPI, hass.data[DOMAIN][entry.entry_id]).fetch_devices()
except ElectraApiError as exp:
raise ConfigEntryNotReady(f"Error communicating with API: {exp}") from exp
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Update listener."""
await hass.config_entries.async_reload(config_entry.entry_id)

334
homeassistant/components/electrasmart/climate.py Normal file
View File

@ -0,0 +1,334 @@
"""Support for the Electra climate."""
from __future__ import annotations
from datetime import timedelta
import logging
import time
from typing import Any
from electrasmart.api import STATUS_SUCCESS, Attributes, ElectraAPI, ElectraApiError
from electrasmart.device import ElectraAirConditioner, OperationMode
from electrasmart.device.const import MAX_TEMP, MIN_TEMP, Feature
from homeassistant.components.climate import (
FAN_AUTO,
FAN_HIGH,
FAN_LOW,
FAN_MEDIUM,
SWING_BOTH,
SWING_HORIZONTAL,
SWING_OFF,
SWING_VERTICAL,
ClimateEntity,
ClimateEntityFeature,
HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
API_DELAY,
CONSECUTIVE_FAILURE_THRESHOLD,
DOMAIN,
PRESET_NONE,
PRESET_SHABAT,
SCAN_INTERVAL_SEC,
UNAVAILABLE_THRESH_SEC,
)
FAN_ELECTRA_TO_HASS = {
OperationMode.FAN_SPEED_AUTO: FAN_AUTO,
OperationMode.FAN_SPEED_LOW: FAN_LOW,
OperationMode.FAN_SPEED_MED: FAN_MEDIUM,
OperationMode.FAN_SPEED_HIGH: FAN_HIGH,
}
FAN_HASS_TO_ELECTRA = {
FAN_AUTO: OperationMode.FAN_SPEED_AUTO,
FAN_LOW: OperationMode.FAN_SPEED_LOW,
FAN_MEDIUM: OperationMode.FAN_SPEED_MED,
FAN_HIGH: OperationMode.FAN_SPEED_HIGH,
}
HVAC_MODE_ELECTRA_TO_HASS = {
OperationMode.MODE_COOL: HVACMode.COOL,
OperationMode.MODE_HEAT: HVACMode.HEAT,
OperationMode.MODE_FAN: HVACMode.FAN_ONLY,
OperationMode.MODE_DRY: HVACMode.DRY,
OperationMode.MODE_AUTO: HVACMode.AUTO,
}
HVAC_MODE_HASS_TO_ELECTRA = {
HVACMode.COOL: OperationMode.MODE_COOL,
HVACMode.HEAT: OperationMode.MODE_HEAT,
HVACMode.FAN_ONLY: OperationMode.MODE_FAN,
HVACMode.DRY: OperationMode.MODE_DRY,
HVACMode.AUTO: OperationMode.MODE_AUTO,
}
ELECTRA_FAN_MODES = [FAN_AUTO, FAN_HIGH, FAN_MEDIUM, FAN_LOW]
ELECTRA_MODES = [
HVACMode.OFF,
HVACMode.HEAT,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.AUTO,
]
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=SCAN_INTERVAL_SEC)
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Add Electra AC devices."""
api: ElectraAPI = hass.data[DOMAIN][entry.entry_id]
_LOGGER.debug("Discovered %i Electra devices", len(api.devices))
async_add_entities(
(ElectraClimateEntity(device, api) for device in api.devices), True
)
class ElectraClimateEntity(ClimateEntity):
"""Define an Electra climate."""
_attr_fan_modes = ELECTRA_FAN_MODES
_attr_target_temperature_step = 1
_attr_max_temp = MAX_TEMP
_attr_min_temp = MIN_TEMP
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_hvac_modes = ELECTRA_MODES
def __init__(self, device: ElectraAirConditioner, api: ElectraAPI) -> None:
"""Initialize Electra climate entity."""
self._api = api
self._electra_ac_device = device
self._attr_name = device.name
self._attr_unique_id = device.mac
self._attr_supported_features = (
ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.FAN_MODE
)
swing_modes: list = []
if Feature.V_SWING in self._electra_ac_device.features:
swing_modes.append(SWING_VERTICAL)
if Feature.H_SWING in self._electra_ac_device.features:
swing_modes.append(SWING_HORIZONTAL)
if all(elem in [SWING_HORIZONTAL, SWING_VERTICAL] for elem in swing_modes):
swing_modes.append(SWING_BOTH)
if swing_modes:
swing_modes.append(SWING_OFF)
self._attr_swing_modes = swing_modes
self._attr_supported_features |= ClimateEntityFeature.SWING_MODE
self._attr_preset_modes = [
PRESET_NONE,
PRESET_SHABAT,
]
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._electra_ac_device.mac)},
name=self.name,
model=self._electra_ac_device.model,
manufacturer=self._electra_ac_device.manufactor,
)
# This attribute will be used to mark the time we communicated
# a command to the API
self._last_state_update = 0
# count the consecutive update failures, used to print error log
self._consecutive_failures = 0
self._skip_update = True
self._was_available = True
_LOGGER.debug("Added %s Electra AC device", self._attr_name)
@property
def available(self) -> bool:
"""Return True if the AC is available."""
return (
not self._electra_ac_device.is_disconnected(UNAVAILABLE_THRESH_SEC)
and super().available
)
async def async_update(self) -> None:
"""Update Electra device."""
# if we communicated a change to the API in the last API_DELAY seconds,
# then don't receive any updates as the API takes few seconds
# until it start sending it last recent change
if self._last_state_update and int(time.time()) < (
self._last_state_update + API_DELAY
):
_LOGGER.debug("Skipping state update, keeping old values")
return
self._last_state_update = 0
try:
# skip the first update only as we already got the devices with their current state
if self._skip_update:
self._skip_update = False
else:
await self._api.get_last_telemtry(self._electra_ac_device)
if not self.available:
# show the warning once upon state change
if self._was_available:
_LOGGER.warning(
"Electra AC %s (%s) is not available, check its status in the Electra Smart mobile app",
self.name,
self._electra_ac_device.mac,
)
self._was_available = False
return
if not self._was_available:
_LOGGER.info(
"%s (%s) is now available",
self._electra_ac_device.mac,
self.name,
)
self._was_available = True
_LOGGER.debug(
"%s (%s) state updated: %s",
self._electra_ac_device.mac,
self.name,
self._electra_ac_device.__dict__,
)
except ElectraApiError as exp:
self._consecutive_failures += 1
_LOGGER.warning(
"Failed to get %s state: %s (try #%i since last success), keeping old state",
self.name,
exp,
self._consecutive_failures,
)
if self._consecutive_failures >= CONSECUTIVE_FAILURE_THRESHOLD:
raise HomeAssistantError(
f"Failed to get {self.name} state: {exp} for the {self._consecutive_failures} time",
) from ElectraApiError
self._consecutive_failures = 0
self._update_device_attrs()
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set AC fan mode."""
mode = FAN_HASS_TO_ELECTRA[fan_mode]
self._electra_ac_device.set_fan_speed(mode)
await self._async_operate_electra_ac()
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set hvac mode."""
if hvac_mode == HVACMode.OFF:
self._electra_ac_device.turn_off()
else:
self._electra_ac_device.set_mode(HVAC_MODE_HASS_TO_ELECTRA[hvac_mode])
self._electra_ac_device.turn_on()
await self._async_operate_electra_ac()
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
raise ValueError("No target temperature provided")
self._electra_ac_device.set_temperature(temperature)
await self._async_operate_electra_ac()
def _update_device_attrs(self) -> None:
self._attr_fan_mode = FAN_ELECTRA_TO_HASS[
self._electra_ac_device.get_fan_speed()
]
self._attr_current_temperature = (
self._electra_ac_device.get_sensor_temperature()
)
self._attr_target_temperature = self._electra_ac_device.get_temperature()
self._attr_hvac_mode = (
HVACMode.OFF
if not self._electra_ac_device.is_on()
else HVAC_MODE_ELECTRA_TO_HASS[self._electra_ac_device.get_mode()]
)
if (
self._electra_ac_device.is_horizontal_swing()
and self._electra_ac_device.is_vertical_swing()
):
self._attr_swing_mode = SWING_BOTH
elif self._electra_ac_device.is_horizontal_swing():
self._attr_swing_mode = SWING_HORIZONTAL
elif self._electra_ac_device.is_vertical_swing():
self._attr_swing_mode = SWING_VERTICAL
else:
self._attr_swing_mode = SWING_OFF
self._attr_preset_mode = (
PRESET_SHABAT if self._electra_ac_device.get_shabat_mode() else PRESET_NONE
)
async def async_set_swing_mode(self, swing_mode: str) -> None:
"""Set AC swing mdde."""
if swing_mode == SWING_BOTH:
self._electra_ac_device.set_horizontal_swing(True)
self._electra_ac_device.set_vertical_swing(True)
elif swing_mode == SWING_VERTICAL:
self._electra_ac_device.set_horizontal_swing(False)
self._electra_ac_device.set_vertical_swing(True)
elif swing_mode == SWING_HORIZONTAL:
self._electra_ac_device.set_horizontal_swing(True)
self._electra_ac_device.set_vertical_swing(False)
else:
self._electra_ac_device.set_horizontal_swing(False)
self._electra_ac_device.set_vertical_swing(False)
await self._async_operate_electra_ac()
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set Preset mode."""
if preset_mode == PRESET_SHABAT:
self._electra_ac_device.set_shabat_mode(True)
else:
self._electra_ac_device.set_shabat_mode(False)
await self._async_operate_electra_ac()
async def _async_operate_electra_ac(self) -> None:
"""Send HVAC parameters to API."""
try:
resp = await self._api.set_state(self._electra_ac_device)
except ElectraApiError as exp:
raise HomeAssistantError(
f"Error communicating with Electra API: {exp}"
) from exp
if not (
resp[Attributes.STATUS] == STATUS_SUCCESS
and resp[Attributes.DATA][Attributes.RES] == STATUS_SUCCESS
):
self._async_write_ha_state()
raise HomeAssistantError(f"Failed to update {self.name}, error: {resp}")
self._update_device_attrs()
self._last_state_update = int(time.time())
self._async_write_ha_state()

158
homeassistant/components/electrasmart/config_flow.py Normal file
View File

@ -0,0 +1,158 @@
"""Config flow for Electra Air Conditioner integration."""
from __future__ import annotations
import logging
from typing import Any
from electrasmart.api import STATUS_SUCCESS, Attributes, ElectraAPI, ElectraApiError
from electrasmart.api.utils import generate_imei
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_TOKEN
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_IMEI, CONF_OTP, CONF_PHONE_NUMBER, DOMAIN
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Electra Air Conditioner."""
VERSION = 1
def __init__(self) -> None:
"""Device settings."""
self._phone_number: str | None = None
self._description_placeholders = None
self._otp: str | None = None
self._imei: str | None = None
self._token: str | None = None
self._api: ElectraAPI | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if not self._api:
self._api = ElectraAPI(async_get_clientsession(self.hass))
errors: dict[str, Any] = {}
if user_input is None:
return self._show_setup_form(user_input, errors, "user")
return await self._validate_phone_number(user_input)
def _show_setup_form(
self,
user_input: dict[str, str] | None = None,
errors: dict[str, str] | None = None,
step_id: str = "user",
) -> FlowResult:
"""Show the setup form to the user."""
if user_input is None:
user_input = {}
if step_id == "user":
schema = {
vol.Required(
CONF_PHONE_NUMBER, default=user_input.get(CONF_PHONE_NUMBER, "")
): str
}
else:
schema = {vol.Required(CONF_OTP, default=user_input.get(CONF_OTP, "")): str}
return self.async_show_form(
step_id=step_id,
data_schema=vol.Schema(schema),
errors=errors or {},
description_placeholders=self._description_placeholders,
)
async def _validate_phone_number(self, user_input: dict[str, str]) -> FlowResult:
"""Check if config is valid and create entry if so."""
self._phone_number = user_input[CONF_PHONE_NUMBER]
self._imei = generate_imei()
# Check if already configured
if self.unique_id is None:
await self.async_set_unique_id(self._phone_number)
self._abort_if_unique_id_configured()
assert isinstance(self._api, ElectraAPI)
try:
resp = await self._api.generate_new_token(self._phone_number, self._imei)
except ElectraApiError as exp:
_LOGGER.error("Failed to connect to API: %s", exp)
return self._show_setup_form(user_input, {"base": "cannot_connect"}, "user")
if resp[Attributes.STATUS] == STATUS_SUCCESS:
if resp[Attributes.DATA][Attributes.RES] != STATUS_SUCCESS:
return self._show_setup_form(
user_input, {CONF_PHONE_NUMBER: "invalid_phone_number"}, "user"
)
return await self.async_step_one_time_password()
async def _validate_one_time_password(
self, user_input: dict[str, str]
) -> FlowResult:
self._otp = user_input[CONF_OTP]
assert isinstance(self._api, ElectraAPI)
assert isinstance(self._imei, str)
assert isinstance(self._phone_number, str)
assert isinstance(self._otp, str)
try:
resp = await self._api.validate_one_time_password(
self._otp, self._imei, self._phone_number
)
except ElectraApiError as exp:
_LOGGER.error("Failed to connect to API: %s", exp)
return self._show_setup_form(
user_input, {"base": "cannot_connect"}, CONF_OTP
)
if resp[Attributes.DATA][Attributes.RES] == STATUS_SUCCESS:
self._token = resp[Attributes.DATA][Attributes.TOKEN]
data = {
CONF_TOKEN: self._token,
CONF_IMEI: self._imei,
CONF_PHONE_NUMBER: self._phone_number,
}
return self.async_create_entry(title=self._phone_number, data=data)
return self._show_setup_form(user_input, {CONF_OTP: "invalid_auth"}, CONF_OTP)
async def async_step_one_time_password(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> FlowResult:
"""Ask the verification code to the user."""
if errors is None:
errors = {}
if user_input is None:
return await self._show_otp_form(errors)
return await self._validate_one_time_password(user_input)
async def _show_otp_form(
self,
errors: dict[str, str] | None = None,
) -> FlowResult:
"""Show the verification_code form to the user."""
return self.async_show_form(
step_id=CONF_OTP,
data_schema=vol.Schema({vol.Required(CONF_OTP): str}),
errors=errors or {},
)

13
homeassistant/components/electrasmart/const.py Normal file
View File

@ -0,0 +1,13 @@
"""Constants for the Electra Air Conditioner integration."""
DOMAIN = "electrasmart"
CONF_PHONE_NUMBER = "phone_number"
CONF_OTP = "one_time_password"
CONF_IMEI = "imei"
SCAN_INTERVAL_SEC = 30
API_DELAY = 5
CONSECUTIVE_FAILURE_THRESHOLD = 4
UNAVAILABLE_THRESH_SEC = 120
PRESET_NONE = "None"
PRESET_SHABAT = "Shabat"

9
homeassistant/components/electrasmart/manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"domain": "electrasmart",
"name": "Electra Smart",
"codeowners": ["@jafar-atili"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/electrasmart",
"iot_class": "cloud_polling",
"requirements": ["pyelectra==1.2.0"]
}

25
homeassistant/components/electrasmart/strings.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"step": {
"user": {
"data": {
"phone_number": "Phone Number"
}
},
"one_time_password": {
"data": {
"one_time_password": "One Time Password"
}
}
},
"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%]",
"invalid_phone_number": "Either wrong phone number or unregistered user"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

25
homeassistant/components/electrasmart/translations/en.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"abort": {
"already_configured": "Phone number already configured"
},
"error": {
"cannot_connect": "Failed to connect to Electra API",
"invalid_auth": "Wrong one time password key",
"invalid_phone_number": "Either wrong phone number or unregistered user",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"phone_number": "Phone Number (eg. 0501234567)"
}
},
"one_time_password": {
"data": {
"one_time_password": "One Time Password (OTP)"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -113,6 +113,7 @@ FLOWS = {
"edl21",
"efergy",
"eight_sleep",
"electrasmart",
"elgato",
"elkm1",
"elmax",

6
homeassistant/generated/integrations.json
View File

@ -1299,6 +1299,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"electrasmart": {
"name": "Electra Smart",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"elgato": {
"name": "Elgato",
"integrations": {

10
mypy.ini
View File

@ -812,6 +812,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.electrasmart.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.elgato.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt
View File

@ -1607,6 +1607,9 @@ pyefergy==22.1.1
# homeassistant.components.eight_sleep
pyeight==0.3.2
# homeassistant.components.electrasmart
pyelectra==1.2.0
# homeassistant.components.emby
pyemby==1.8

3
requirements_test_all.txt
View File

@ -1177,6 +1177,9 @@ pyefergy==22.1.1
# homeassistant.components.eight_sleep
pyeight==0.3.2
# homeassistant.components.electrasmart
pyelectra==1.2.0
# homeassistant.components.everlights
pyeverlights==0.1.0

1
tests/components/electrasmart/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Electra Air Conditioner integration."""

6
tests/components/electrasmart/fixtures/generate_token_response.json Normal file
View File

@ -0,0 +1,6 @@
{
"id": 99,
"status": 0,
"desc": "None",
"data": { "res": 0, "res_desc": "None" }
}

6
tests/components/electrasmart/fixtures/invalid_otp_response.json Normal file
View File

@ -0,0 +1,6 @@
{
"id": 99,
"status": 1,
"desc": "None",
"data": { "res": 100, "res_desc": "None" }
}

6
tests/components/electrasmart/fixtures/invalid_phone_number_response.json Normal file
View File

@ -0,0 +1,6 @@
{
"id": 99,
"status": 0,
"desc": "None",
"data": { "res": 100, "res_desc": "None" }
}

11
tests/components/electrasmart/fixtures/otp_response.json Normal file
View File

@ -0,0 +1,11 @@
{
"id": 99,
"status": 0,
"desc": "None",
"data": {
"token": "ec7a0db6c1f148ca8c0f48aabb5f8150",
"sid": "bd6f11f947244e5d9612eba89e91112b",
"res": 0,
"res_desc": "None"
}
}

164
tests/components/electrasmart/test_config_flow.py Normal file
View File

@ -0,0 +1,164 @@
"""Test the Electra Smart config flow."""
from json import loads
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.electrasmart.config_flow import ElectraApiError
from homeassistant.components.electrasmart.const import (
CONF_OTP,
CONF_PHONE_NUMBER,
DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import load_fixture
async def test_form(hass: HomeAssistant):
"""Test user config."""
mock_generate_token = loads(load_fixture("generate_token_response.json", DOMAIN))
with patch(
"electrasmart.api.ElectraAPI.generate_new_token",
return_value=mock_generate_token,
):
# test with required
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=None,
)
assert result["step_id"] == "user"
# test with required
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_PHONE_NUMBER: "0521234567"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == CONF_OTP
async def test_one_time_password(hass: HomeAssistant):
"""Test one time password."""
mock_generate_token = loads(load_fixture("generate_token_response.json", DOMAIN))
mock_otp_response = loads(load_fixture("otp_response.json", DOMAIN))
with patch(
"electrasmart.api.ElectraAPI.generate_new_token",
return_value=mock_generate_token,
), patch(
"electrasmart.api.ElectraAPI.validate_one_time_password",
return_value=mock_otp_response,
), patch(
"electrasmart.api.ElectraAPI.fetch_devices", return_value=[]
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_PHONE_NUMBER: "0521234567", CONF_OTP: "1234"},
)
# test with required
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_OTP: "1234"}
)
assert result["type"] == FlowResultType.CREATE_ENTRY
async def test_one_time_password_api_error(hass: HomeAssistant):
"""Test one time password."""
mock_generate_token = loads(load_fixture("generate_token_response.json", DOMAIN))
with patch(
"electrasmart.api.ElectraAPI.generate_new_token",
return_value=mock_generate_token,
), patch(
"electrasmart.api.ElectraAPI.validate_one_time_password",
side_effect=ElectraApiError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_PHONE_NUMBER: "0521234567"},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_OTP: "1234"}
)
assert result["type"] == FlowResultType.FORM
async def test_cannot_connect(hass: HomeAssistant):
"""Test cannot connect."""
with patch(
"electrasmart.api.ElectraAPI.generate_new_token",
side_effect=ElectraApiError,
):
# test with required
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_PHONE_NUMBER: "0521234567"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_invalid_phone_number(hass: HomeAssistant):
"""Test invalid phone number."""
mock_invalid_phone_number_response = loads(
load_fixture("invalid_phone_number_response.json", DOMAIN)
)
with patch(
"electrasmart.api.ElectraAPI.generate_new_token",
return_value=mock_invalid_phone_number_response,
):
# test with required
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_PHONE_NUMBER: "0521234567"},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"phone_number": "invalid_phone_number"}
async def test_invalid_auth(hass: HomeAssistant):
"""Test invalid auth."""
mock_generate_token_response = loads(
load_fixture("generate_token_response.json", DOMAIN)
)
mock_invalid_otp_response = loads(load_fixture("invalid_otp_response.json", DOMAIN))
with patch(
"electrasmart.api.ElectraAPI.generate_new_token",
return_value=mock_generate_token_response,
), patch(
"electrasmart.api.ElectraAPI.validate_one_time_password",
return_value=mock_invalid_otp_response,
):
# test with required
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_PHONE_NUMBER: "0521234567", CONF_OTP: "1234"},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_OTP: "1234"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == CONF_OTP
assert result["errors"] == {CONF_OTP: "invalid_auth"}