mirror of
https://github.com/home-assistant/core
synced 2024-10-01 19:08:29 +00:00
Add Tami4 Integration (#90056)
Co-authored-by: Franck Nijhof <frenck@frenck.nl> Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
parent
91cf719588
commit
5730cb1e85
|
@ -328,6 +328,7 @@ homeassistant.components.synology_dsm.*
|
|||
homeassistant.components.systemmonitor.*
|
||||
homeassistant.components.tag.*
|
||||
homeassistant.components.tailscale.*
|
||||
homeassistant.components.tami4.*
|
||||
homeassistant.components.tautulli.*
|
||||
homeassistant.components.tcp.*
|
||||
homeassistant.components.text.*
|
||||
|
|
|
@ -1265,6 +1265,8 @@ build.json @home-assistant/supervisor
|
|||
/tests/components/tag/ @balloob @dmulcahey
|
||||
/homeassistant/components/tailscale/ @frenck
|
||||
/tests/components/tailscale/ @frenck
|
||||
/homeassistant/components/tami4/ @Guy293
|
||||
/tests/components/tami4/ @Guy293
|
||||
/homeassistant/components/tankerkoenig/ @guillempages @mib1185
|
||||
/tests/components/tankerkoenig/ @guillempages @mib1185
|
||||
/homeassistant/components/tapsaff/ @bazwilliams
|
||||
|
|
46
homeassistant/components/tami4/__init__.py
Normal file
46
homeassistant/components/tami4/__init__.py
Normal file
|
@ -0,0 +1,46 @@
|
|||
"""The Tami4Edge integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from Tami4EdgeAPI import Tami4EdgeAPI, exceptions
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
|
||||
|
||||
from .const import API, CONF_REFRESH_TOKEN, COORDINATOR, DOMAIN
|
||||
from .coordinator import Tami4EdgeWaterQualityCoordinator
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up tami4 from a config entry."""
|
||||
refresh_token = entry.data.get(CONF_REFRESH_TOKEN)
|
||||
|
||||
try:
|
||||
api = await hass.async_add_executor_job(Tami4EdgeAPI, refresh_token)
|
||||
except exceptions.RefreshTokenExpiredException as ex:
|
||||
raise ConfigEntryError("API Refresh token expired") from ex
|
||||
except exceptions.TokenRefreshFailedException as ex:
|
||||
raise ConfigEntryNotReady("Error connecting to API") from ex
|
||||
|
||||
coordinator = Tami4EdgeWaterQualityCoordinator(hass, api)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
|
||||
API: api,
|
||||
COORDINATOR: coordinator,
|
||||
}
|
||||
|
||||
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
|
95
homeassistant/components/tami4/config_flow.py
Normal file
95
homeassistant/components/tami4/config_flow.py
Normal file
|
@ -0,0 +1,95 @@
|
|||
"""Config flow for edge integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from Tami4EdgeAPI import Tami4EdgeAPI, exceptions
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from .const import CONF_PHONE, CONF_REFRESH_TOKEN, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
_STEP_PHONE_NUMBER_SCHEMA = vol.Schema({vol.Required(CONF_PHONE): cv.string})
|
||||
|
||||
_STEP_OTP_CODE_SCHEMA = vol.Schema({vol.Required("otp"): cv.string})
|
||||
_PHONE_MATCHER = re.compile(r"^(\+?972)?0?(?P<number>\d{8,9})$")
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Tami4Edge."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
phone: str
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the otp request step."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
phone = user_input[CONF_PHONE].strip()
|
||||
|
||||
try:
|
||||
if m := _PHONE_MATCHER.match(phone):
|
||||
self.phone = f"+972{m.group('number')}"
|
||||
else:
|
||||
raise InvalidPhoneNumber
|
||||
await self.hass.async_add_executor_job(
|
||||
Tami4EdgeAPI.request_otp, self.phone
|
||||
)
|
||||
except InvalidPhoneNumber:
|
||||
errors["base"] = "invalid_phone"
|
||||
except exceptions.Tami4EdgeAPIException:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return await self.async_step_otp()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=_STEP_PHONE_NUMBER_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
async def async_step_otp(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the otp submission step."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
otp = user_input["otp"]
|
||||
try:
|
||||
refresh_token = await self.hass.async_add_executor_job(
|
||||
Tami4EdgeAPI.submit_otp, self.phone, otp
|
||||
)
|
||||
api = await self.hass.async_add_executor_job(
|
||||
Tami4EdgeAPI, refresh_token
|
||||
)
|
||||
except exceptions.OTPFailedException:
|
||||
errors["base"] = "invalid_auth"
|
||||
except exceptions.Tami4EdgeAPIException:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=api.device.name, data={CONF_REFRESH_TOKEN: refresh_token}
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="otp", data_schema=_STEP_OTP_CODE_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
|
||||
class InvalidPhoneNumber(HomeAssistantError):
|
||||
"""Error to indicate that the phone number is invalid."""
|
6
homeassistant/components/tami4/const.py
Normal file
6
homeassistant/components/tami4/const.py
Normal file
|
@ -0,0 +1,6 @@
|
|||
"""Constants for tami4 component."""
|
||||
DOMAIN = "tami4"
|
||||
CONF_PHONE = "phone"
|
||||
CONF_REFRESH_TOKEN = "refresh_token"
|
||||
API = "api"
|
||||
COORDINATOR = "coordinator"
|
61
homeassistant/components/tami4/coordinator.py
Normal file
61
homeassistant/components/tami4/coordinator.py
Normal file
|
@ -0,0 +1,61 @@
|
|||
"""Water quality coordinator for Tami4Edge."""
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, timedelta
|
||||
import logging
|
||||
|
||||
from Tami4EdgeAPI import Tami4EdgeAPI, exceptions
|
||||
from Tami4EdgeAPI.water_quality import WaterQuality
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FlattenedWaterQuality:
|
||||
"""Flattened WaterQuality dataclass."""
|
||||
|
||||
uv_last_replacement: date
|
||||
uv_upcoming_replacement: date
|
||||
uv_status: str
|
||||
filter_last_replacement: date
|
||||
filter_upcoming_replacement: date
|
||||
filter_status: str
|
||||
filter_litters_passed: float
|
||||
|
||||
def __init__(self, water_quality: WaterQuality) -> None:
|
||||
"""Flatten WaterQuality dataclass."""
|
||||
|
||||
self.uv_last_replacement = water_quality.uv.last_replacement
|
||||
self.uv_upcoming_replacement = water_quality.uv.upcoming_replacement
|
||||
self.uv_status = water_quality.uv.status
|
||||
self.filter_last_replacement = water_quality.filter.last_replacement
|
||||
self.filter_upcoming_replacement = water_quality.filter.upcoming_replacement
|
||||
self.filter_status = water_quality.filter.status
|
||||
self.filter_litters_passed = water_quality.filter.milli_litters_passed / 1000
|
||||
|
||||
|
||||
class Tami4EdgeWaterQualityCoordinator(DataUpdateCoordinator[FlattenedWaterQuality]):
|
||||
"""Tami4Edge water quality coordinator."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, api: Tami4EdgeAPI) -> None:
|
||||
"""Initialize the water quality coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name="Tami4Edge water quality coordinator",
|
||||
update_interval=timedelta(minutes=60),
|
||||
)
|
||||
self._api = api
|
||||
|
||||
async def _async_update_data(self) -> FlattenedWaterQuality:
|
||||
"""Fetch data from the API endpoint."""
|
||||
try:
|
||||
water_quality = await self.hass.async_add_executor_job(
|
||||
self._api.get_water_quality
|
||||
)
|
||||
|
||||
return FlattenedWaterQuality(water_quality)
|
||||
except exceptions.APIRequestFailedException as ex:
|
||||
raise UpdateFailed("Error communicating with API") from ex
|
33
homeassistant/components/tami4/entity.py
Normal file
33
homeassistant/components/tami4/entity.py
Normal file
|
@ -0,0 +1,33 @@
|
|||
"""Base entity for Tami4Edge."""
|
||||
from __future__ import annotations
|
||||
|
||||
from Tami4EdgeAPI import Tami4EdgeAPI
|
||||
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity import Entity, EntityDescription
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
|
||||
class Tami4EdgeBaseEntity(Entity):
|
||||
"""Base class for Tami4Edge entities."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(
|
||||
self, api: Tami4EdgeAPI, entity_description: EntityDescription
|
||||
) -> None:
|
||||
"""Initialize the Tami4Edge."""
|
||||
self._state = None
|
||||
self._api = api
|
||||
device_id = api.device.psn
|
||||
self.entity_description = entity_description
|
||||
self._attr_unique_id = f"{device_id}_{self.entity_description.key}"
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, device_id)},
|
||||
manufacturer="Stratuss",
|
||||
name=api.device.name,
|
||||
model="Tami4",
|
||||
sw_version=api.device.device_firmware,
|
||||
suggested_area="Kitchen",
|
||||
)
|
9
homeassistant/components/tami4/manifest.json
Normal file
9
homeassistant/components/tami4/manifest.json
Normal file
|
@ -0,0 +1,9 @@
|
|||
{
|
||||
"domain": "tami4",
|
||||
"name": "Tami4 Edge / Edge+",
|
||||
"codeowners": ["@Guy293"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/tami4",
|
||||
"iot_class": "cloud_polling",
|
||||
"requirements": ["Tami4EdgeAPI==2.1"]
|
||||
}
|
118
homeassistant/components/tami4/sensor.py
Normal file
118
homeassistant/components/tami4/sensor.py
Normal file
|
@ -0,0 +1,118 @@
|
|||
"""Sensor entities for Tami4Edge."""
|
||||
import logging
|
||||
|
||||
from Tami4EdgeAPI import Tami4EdgeAPI
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import UnitOfVolume
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import API, COORDINATOR, DOMAIN
|
||||
from .coordinator import Tami4EdgeWaterQualityCoordinator
|
||||
from .entity import Tami4EdgeBaseEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
ENTITY_DESCRIPTIONS = [
|
||||
SensorEntityDescription(
|
||||
key="uv_last_replacement",
|
||||
translation_key="uv_last_replacement",
|
||||
icon="mdi:calendar",
|
||||
device_class=SensorDeviceClass.DATE,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="uv_upcoming_replacement",
|
||||
translation_key="uv_upcoming_replacement",
|
||||
icon="mdi:calendar",
|
||||
device_class=SensorDeviceClass.DATE,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="uv_status",
|
||||
translation_key="uv_status",
|
||||
icon="mdi:clipboard-check-multiple",
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="filter_last_replacement",
|
||||
translation_key="filter_last_replacement",
|
||||
icon="mdi:calendar",
|
||||
device_class=SensorDeviceClass.DATE,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="filter_upcoming_replacement",
|
||||
translation_key="filter_upcoming_replacement",
|
||||
icon="mdi:calendar",
|
||||
device_class=SensorDeviceClass.DATE,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="filter_status",
|
||||
translation_key="filter_status",
|
||||
icon="mdi:clipboard-check-multiple",
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="filter_litters_passed",
|
||||
translation_key="filter_litters_passed",
|
||||
icon="mdi:water",
|
||||
state_class=SensorStateClass.TOTAL,
|
||||
device_class=SensorDeviceClass.WATER,
|
||||
native_unit_of_measurement=UnitOfVolume.LITERS,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Perform the setup for Tami4Edge."""
|
||||
data = hass.data[DOMAIN][entry.entry_id]
|
||||
api: Tami4EdgeAPI = data[API]
|
||||
coordinator: Tami4EdgeWaterQualityCoordinator = data[COORDINATOR]
|
||||
|
||||
entities = []
|
||||
for entity_description in ENTITY_DESCRIPTIONS:
|
||||
entities.append(
|
||||
Tami4EdgeSensorEntity(
|
||||
coordinator=coordinator,
|
||||
api=api,
|
||||
entity_description=entity_description,
|
||||
)
|
||||
)
|
||||
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class Tami4EdgeSensorEntity(
|
||||
Tami4EdgeBaseEntity,
|
||||
CoordinatorEntity[Tami4EdgeWaterQualityCoordinator],
|
||||
SensorEntity,
|
||||
):
|
||||
"""Representation of the entity."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: Tami4EdgeWaterQualityCoordinator,
|
||||
api: Tami4EdgeAPI,
|
||||
entity_description: SensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the Tami4Edge sensor entity."""
|
||||
Tami4EdgeBaseEntity.__init__(self, api, entity_description)
|
||||
CoordinatorEntity.__init__(self, coordinator)
|
||||
self._update_attr()
|
||||
|
||||
def _update_attr(self) -> None:
|
||||
self._attr_native_value = getattr(
|
||||
self.coordinator.data, self.entity_description.key
|
||||
)
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle updated data from the coordinator."""
|
||||
self._update_attr()
|
||||
self.async_write_ha_state()
|
54
homeassistant/components/tami4/strings.json
Normal file
54
homeassistant/components/tami4/strings.json
Normal file
|
@ -0,0 +1,54 @@
|
|||
{
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"uv_last_replacement": {
|
||||
"name": "UV last replacement"
|
||||
},
|
||||
"uv_upcoming_replacement": {
|
||||
"name": "UV upcoming replacement"
|
||||
},
|
||||
"uv_status": {
|
||||
"name": "UV status"
|
||||
},
|
||||
"filter_last_replacement": {
|
||||
"name": "Filter last replacement"
|
||||
},
|
||||
"filter_upcoming_replacement": {
|
||||
"name": "Filter upcoming replacement"
|
||||
},
|
||||
"filter_status": {
|
||||
"name": "Filter status"
|
||||
},
|
||||
"filter_litters_passed": {
|
||||
"name": "Filter water passed"
|
||||
}
|
||||
}
|
||||
},
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "SMS Verification",
|
||||
"description": "Enter your phone number (same as what you used to register to the tami4 app)",
|
||||
"data": {
|
||||
"phone": "Phone Number"
|
||||
}
|
||||
},
|
||||
"otp": {
|
||||
"title": "[%key:component::tami4::config::step::user::title%]",
|
||||
"description": "Enter the code you received via SMS",
|
||||
"data": {
|
||||
"otp": "SMS Code"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"invalid_phone": "Invalid phone number, please use the following format: +972xxxxxxxx",
|
||||
"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%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
}
|
||||
}
|
|
@ -470,6 +470,7 @@ FLOWS = {
|
|||
"system_bridge",
|
||||
"tado",
|
||||
"tailscale",
|
||||
"tami4",
|
||||
"tankerkoenig",
|
||||
"tasmota",
|
||||
"tautulli",
|
||||
|
|
|
@ -5629,6 +5629,12 @@
|
|||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"tami4": {
|
||||
"name": "Tami4 Edge / Edge+",
|
||||
"integration_type": "hub",
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"tank_utility": {
|
||||
"name": "Tank Utility",
|
||||
"integration_type": "hub",
|
||||
|
|
10
mypy.ini
10
mypy.ini
|
@ -3042,6 +3042,16 @@ disallow_untyped_defs = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.tami4.*]
|
||||
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.tautulli.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
|
@ -134,6 +134,9 @@ RtmAPI==0.7.2
|
|||
# homeassistant.components.sql
|
||||
SQLAlchemy==2.0.21
|
||||
|
||||
# homeassistant.components.tami4
|
||||
Tami4EdgeAPI==2.1
|
||||
|
||||
# homeassistant.components.travisci
|
||||
TravisPy==0.3.5
|
||||
|
||||
|
|
|
@ -121,6 +121,9 @@ RtmAPI==0.7.2
|
|||
# homeassistant.components.sql
|
||||
SQLAlchemy==2.0.21
|
||||
|
||||
# homeassistant.components.tami4
|
||||
Tami4EdgeAPI==2.1
|
||||
|
||||
# homeassistant.components.onvif
|
||||
WSDiscovery==2.0.0
|
||||
|
||||
|
|
1
tests/components/tami4/__init__.py
Normal file
1
tests/components/tami4/__init__.py
Normal file
|
@ -0,0 +1 @@
|
|||
"""Tests for the Tami4 integration."""
|
125
tests/components/tami4/conftest.py
Normal file
125
tests/components/tami4/conftest.py
Normal file
|
@ -0,0 +1,125 @@
|
|||
"""Common fixutres with default mocks as well as common test helper methods."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from datetime import datetime
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from Tami4EdgeAPI.device import Device
|
||||
from Tami4EdgeAPI.water_quality import UV, Filter, WaterQuality
|
||||
|
||||
from homeassistant.components.tami4.const import CONF_REFRESH_TOKEN, DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def create_config_entry(hass: HomeAssistant) -> MockConfigEntry:
|
||||
"""Create an entry in hass."""
|
||||
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Device name",
|
||||
data={CONF_REFRESH_TOKEN: "refresh_token"},
|
||||
)
|
||||
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_api(mock__get_devices, mock_get_water_quality):
|
||||
"""Fixture to mock all API calls."""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock__get_devices(request):
|
||||
"""Fixture to mock _get_devices which makes a call to the API."""
|
||||
|
||||
side_effect = getattr(request, "param", None)
|
||||
|
||||
device = Device(
|
||||
id=1,
|
||||
name="Drink Water",
|
||||
connected=True,
|
||||
psn="psn",
|
||||
type="type",
|
||||
device_firmware="v1.1",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"Tami4EdgeAPI.Tami4EdgeAPI.Tami4EdgeAPI._get_devices",
|
||||
return_value=[device],
|
||||
side_effect=side_effect,
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_get_water_quality(request):
|
||||
"""Fixture to mock get_water_quality which makes a call to the API."""
|
||||
|
||||
side_effect = getattr(request, "param", None)
|
||||
|
||||
water_quality = WaterQuality(
|
||||
uv=UV(
|
||||
last_replacement=int(datetime.now().timestamp()),
|
||||
upcoming_replacement=int(datetime.now().timestamp()),
|
||||
status="on",
|
||||
),
|
||||
filter=Filter(
|
||||
last_replacement=int(datetime.now().timestamp()),
|
||||
upcoming_replacement=int(datetime.now().timestamp()),
|
||||
status="on",
|
||||
milli_litters_passed=1000,
|
||||
),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"Tami4EdgeAPI.Tami4EdgeAPI.Tami4EdgeAPI.get_water_quality",
|
||||
return_value=water_quality,
|
||||
side_effect=side_effect,
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
|
||||
"""Mock setting up a config entry."""
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tami4.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_request_otp(request):
|
||||
"""Mock request_otp."""
|
||||
|
||||
side_effect = getattr(request, "param", None)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tami4.config_flow.Tami4EdgeAPI.request_otp",
|
||||
return_value=None,
|
||||
side_effect=side_effect,
|
||||
) as mock_request_otp:
|
||||
yield mock_request_otp
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_submit_otp(request):
|
||||
"""Mock submit_otp."""
|
||||
|
||||
side_effect = getattr(request, "param", None)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.tami4.config_flow.Tami4EdgeAPI.submit_otp",
|
||||
return_value="refresh_token",
|
||||
side_effect=side_effect,
|
||||
) as mock_submit_otp:
|
||||
yield mock_submit_otp
|
163
tests/components/tami4/test_config_flow.py
Normal file
163
tests/components/tami4/test_config_flow.py
Normal file
|
@ -0,0 +1,163 @@
|
|||
"""Tests for the Tami4 config flow."""
|
||||
|
||||
import pytest
|
||||
from Tami4EdgeAPI import exceptions
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.tami4.const import CONF_PHONE, DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
|
||||
async def test_step_user_valid_number(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry,
|
||||
mock_request_otp,
|
||||
mock__get_devices,
|
||||
) -> None:
|
||||
"""Test user step with valid phone number."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_PHONE: "+972555555555"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "otp"
|
||||
assert result["errors"] == {}
|
||||
|
||||
|
||||
async def test_step_user_invalid_number(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry,
|
||||
mock_request_otp,
|
||||
mock__get_devices,
|
||||
) -> None:
|
||||
"""Test user step with invalid phone number."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_PHONE: "+275123"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "invalid_phone"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mock_request_otp", "expected_error"),
|
||||
[(Exception, "unknown"), (exceptions.OTPFailedException, "cannot_connect")],
|
||||
indirect=["mock_request_otp"],
|
||||
)
|
||||
async def test_step_user_exception(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry,
|
||||
mock_request_otp,
|
||||
mock__get_devices,
|
||||
expected_error,
|
||||
) -> None:
|
||||
"""Test user step with exception."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_PHONE: "+972555555555"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": expected_error}
|
||||
|
||||
|
||||
async def test_step_otp_valid(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry,
|
||||
mock_request_otp,
|
||||
mock_submit_otp,
|
||||
mock__get_devices,
|
||||
) -> None:
|
||||
"""Test user step with valid phone number."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_PHONE: "+972555555555"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "otp"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"otp": "123456"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Drink Water"
|
||||
assert "refresh_token" in result["data"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mock_submit_otp", "expected_error"),
|
||||
[
|
||||
(Exception, "unknown"),
|
||||
(exceptions.Tami4EdgeAPIException, "cannot_connect"),
|
||||
(exceptions.OTPFailedException, "invalid_auth"),
|
||||
],
|
||||
indirect=["mock_submit_otp"],
|
||||
)
|
||||
async def test_step_otp_exception(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry,
|
||||
mock_request_otp,
|
||||
mock_submit_otp,
|
||||
mock__get_devices,
|
||||
expected_error,
|
||||
) -> None:
|
||||
"""Test user step with valid phone number."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_PHONE: "+972555555555"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "otp"
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={"otp": "123456"},
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "otp"
|
||||
assert result["errors"] == {"base": expected_error}
|
59
tests/components/tami4/test_init.py
Normal file
59
tests/components/tami4/test_init.py
Normal file
|
@ -0,0 +1,59 @@
|
|||
"""Test the Tami4 component."""
|
||||
import pytest
|
||||
from Tami4EdgeAPI import exceptions
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import create_config_entry
|
||||
|
||||
|
||||
async def test_init_success(mock_api, hass: HomeAssistant) -> None:
|
||||
"""Test setup and that we can create the entry."""
|
||||
|
||||
entry = await create_config_entry(hass)
|
||||
assert entry.state == ConfigEntryState.LOADED
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mock_get_water_quality", [exceptions.APIRequestFailedException], indirect=True
|
||||
)
|
||||
async def test_init_with_api_error(mock_api, hass: HomeAssistant) -> None:
|
||||
"""Test init with api error."""
|
||||
|
||||
entry = await create_config_entry(hass)
|
||||
assert entry.state == ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mock__get_devices", "expected_state"),
|
||||
[
|
||||
(
|
||||
exceptions.RefreshTokenExpiredException,
|
||||
ConfigEntryState.SETUP_ERROR,
|
||||
),
|
||||
(
|
||||
exceptions.TokenRefreshFailedException,
|
||||
ConfigEntryState.SETUP_RETRY,
|
||||
),
|
||||
],
|
||||
indirect=["mock__get_devices"],
|
||||
)
|
||||
async def test_init_error_raised(
|
||||
mock_api, hass: HomeAssistant, expected_state: ConfigEntryState
|
||||
) -> None:
|
||||
"""Test init when an error is raised."""
|
||||
|
||||
entry = await create_config_entry(hass)
|
||||
assert entry.state == expected_state
|
||||
|
||||
|
||||
async def test_load_unload(mock_api, hass: HomeAssistant) -> None:
|
||||
"""Config entry can be unloaded."""
|
||||
|
||||
entry = await create_config_entry(hass)
|
||||
|
||||
await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert entry.state is ConfigEntryState.NOT_LOADED
|
Loading…
Reference in a new issue