evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Convert Anova to cloud push (#109508)

Browse source 
* current state

* finish refactor

* Apply suggestions from code review

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* address MR comments

* Change to sensor setup to be listener based.

* remove assert for websocket handler

* added assert for log

* remove mixin

* fix linting

* fix merge change

* Add clarifying comment

* Apply suggestions from code review

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Address MR comments

* bump version and fix typing check

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
Co-authored-by: Erik Montnemery <erik@montnemery.com>
This commit is contained in:
Luke Lashley 2024-05-08 08:53:44 -04:00 • committed by GitHub
parent de62e205dd
commit 22bc11f397
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 387 additions and 304 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
homeassistant/components/anova/__init__.py
View file

@ -3,18 +3,25 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from anova_wifi import AnovaApi, AnovaPrecisionCooker, InvalidLogin, NoDevicesFound
from anova_wifi import (
AnovaApi,
APCWifiDevice,
InvalidLogin,
NoDevicesFound,
WebsocketFailure,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICES, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client
from .const import DOMAIN
from .coordinator import AnovaCoordinator
from .models import AnovaData
from .util import serialize_device_list
PLATFORMS = [Platform.SENSOR]
@ -36,36 +43,27 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
return False
assert api.jwt
api.existing_devices = [
AnovaPrecisionCooker(
aiohttp_client.async_get_clientsession(hass),
device[0],
device[1],
api.jwt,
)
for device in entry.data[CONF_DEVICES]
]
try:
new_devices = await api.get_devices()
except NoDevicesFound:
# get_devices raises an exception if no devices are online
new_devices = []
devices = api.existing_devices
if new_devices:
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
CONF_DEVICES: serialize_device_list(devices),
},
)
await api.create_websocket()
except NoDevicesFound as err:
# Can later setup successfully and spawn a repair.
raise ConfigEntryNotReady(
"No devices were found on the websocket, perhaps you don't have any devices on this account?"
) from err
except WebsocketFailure as err:
raise ConfigEntryNotReady("Failed connecting to the websocket.") from err
# Create a coordinator per device, if the device is offline, no data will be on the
# websocket, and the coordinator should auto mark as unavailable. But as long as
# the websocket successfully connected, config entry should setup.
devices: list[APCWifiDevice] = []
if TYPE_CHECKING:
# api.websocket_handler can't be None after successfully creating the
# websocket client
assert api.websocket_handler is not None
devices = list(api.websocket_handler.devices.values())
coordinators = [AnovaCoordinator(hass, device) for device in devices]
for coordinator in coordinators:
await coordinator.async_config_entry_first_refresh()
firmware_version = coordinator.data.sensor.firmware_version
coordinator.async_setup(str(firmware_version))
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = AnovaData(
api_jwt=api.jwt, precision_cookers=devices, coordinators=coordinators
api_jwt=api.jwt, coordinators=coordinators, api=api
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@ -74,6 +72,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
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)
anova_data: AnovaData = hass.data[DOMAIN].pop(entry.entry_id)
# Disconnect from WS
await anova_data.api.disconnect_websocket()
return unload_ok

15
homeassistant/components/anova/config_flow.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
from anova_wifi import AnovaApi, InvalidLogin, NoDevicesFound
from anova_wifi import AnovaApi, InvalidLogin
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
@ -10,7 +10,6 @@ from homeassistant.const import CONF_DEVICES, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .util import serialize_device_list
class AnovaConfligFlow(ConfigFlow, domain=DOMAIN):
@ -33,22 +32,18 @@ class AnovaConfligFlow(ConfigFlow, domain=DOMAIN):
self._abort_if_unique_id_configured()
try:
await api.authenticate()
devices = await api.get_devices()
except InvalidLogin:
errors["base"] = "invalid_auth"
except NoDevicesFound:
errors["base"] = "no_devices_found"
except Exception: # noqa: BLE001
errors["base"] = "unknown"
else:
# We store device list in config flow in order to persist found devices on restart, as the Anova api get_devices does not return any devices that are offline.
device_list = serialize_device_list(devices)
return self.async_create_entry(
title="Anova",
data={
CONF_USERNAME: api.username,
CONF_PASSWORD: api.password,
CONF_DEVICES: device_list,
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
# this can be removed in a migration to 1.2 in 2024.11
CONF_DEVICES: [],
},
)

34
homeassistant/components/anova/coordinator.py
View file

@ -1,14 +1,13 @@
"""Support for Anova Coordinators."""
from asyncio import timeout
from datetime import timedelta
import logging
from anova_wifi import AnovaOffline, AnovaPrecisionCooker, APCUpdate
from anova_wifi import APCUpdate, APCWifiDevice
from homeassistant.core import HomeAssistant, callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
@ -18,37 +17,24 @@ _LOGGER = logging.getLogger(__name__)
class AnovaCoordinator(DataUpdateCoordinator[APCUpdate]):
"""Anova custom coordinator."""
def __init__(
self,
hass: HomeAssistant,
anova_device: AnovaPrecisionCooker,
) -> None:
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant, anova_device: APCWifiDevice) -> None:
"""Set up Anova Coordinator."""
super().__init__(
hass,
name="Anova Precision Cooker",
logger=_LOGGER,
update_interval=timedelta(seconds=30),
)
assert self.config_entry is not None
self.device_unique_id = anova_device.device_key
self.device_unique_id = anova_device.cooker_id
self.anova_device = anova_device
self.anova_device.set_update_listener(self.async_set_updated_data)
self.device_info: DeviceInfo | None = None
@callback
def async_setup(self, firmware_version: str) -> None:
"""Set the firmware version info."""
self.device_info = DeviceInfo(
identifiers={(DOMAIN, self.device_unique_id)},
name="Anova Precision Cooker",
manufacturer="Anova",
model="Precision Cooker",
sw_version=firmware_version,
)
async def _async_update_data(self) -> APCUpdate:
try:
async with timeout(5):
return await self.anova_device.update()
except AnovaOffline as err:
raise UpdateFailed(err) from err
self.sensor_data_set: bool = False

5
homeassistant/components/anova/entity.py
View file

@ -19,6 +19,11 @@ class AnovaEntity(CoordinatorEntity[AnovaCoordinator], Entity):
self.device = coordinator.anova_device
self._attr_device_info = coordinator.device_info
@property
def available(self) -> bool:
"""Return if entity is available."""
return self.coordinator.data is not None and super().available
class AnovaDescriptionEntity(AnovaEntity):
"""Defines an Anova entity that uses a description."""

4
homeassistant/components/anova/manifest.json
View file

@ -4,7 +4,7 @@
"codeowners": ["@Lash-L"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/anova",
"iot_class": "cloud_polling",
"iot_class": "cloud_push",
"loggers": ["anova_wifi"],
"requirements": ["anova-wifi==0.10.0"]
"requirements": ["anova-wifi==0.12.0"]
}

4
homeassistant/components/anova/models.py
View file

@ -2,7 +2,7 @@
from dataclasses import dataclass
from anova_wifi import AnovaPrecisionCooker
from anova_wifi import AnovaApi
from .coordinator import AnovaCoordinator
@ -12,5 +12,5 @@ class AnovaData:
"""Data for the Anova integration."""
api_jwt: str
precision_cookers: list[AnovaPrecisionCooker]
coordinators: list[AnovaCoordinator]
api: AnovaApi

57
homeassistant/components/anova/sensor.py
View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from anova_wifi import APCUpdateSensor
from anova_wifi import AnovaMode, AnovaState, APCUpdateSensor
from homeassistant import config_entries
from homeassistant.components.sensor import (
@ -20,25 +20,19 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import DOMAIN
from .coordinator import AnovaCoordinator
from .entity import AnovaDescriptionEntity
from .models import AnovaData
@dataclass(frozen=True)
class AnovaSensorEntityDescriptionMixin:
"""Describes the mixin variables for anova sensors."""
value_fn: Callable[[APCUpdateSensor], float | int | str]
@dataclass(frozen=True)
class AnovaSensorEntityDescription(
SensorEntityDescription, AnovaSensorEntityDescriptionMixin
):
@dataclass(frozen=True, kw_only=True)
class AnovaSensorEntityDescription(SensorEntityDescription):
"""Describes a Anova sensor."""
value_fn: Callable[[APCUpdateSensor], StateType]
SENSOR_DESCRIPTIONS: list[SensorEntityDescription] = [
SENSOR_DESCRIPTIONS: list[AnovaSensorEntityDescription] = [
AnovaSensorEntityDescription(
key="cook_time",
state_class=SensorStateClass.TOTAL_INCREASING,
@ -50,11 +44,15 @@ SENSOR_DESCRIPTIONS: list[SensorEntityDescription] = [
AnovaSensorEntityDescription(
key="state",
translation_key="state",
device_class=SensorDeviceClass.ENUM,
options=[state.name for state in AnovaState],
value_fn=lambda data: data.state,
),
AnovaSensorEntityDescription(
key="mode",
translation_key="mode",
device_class=SensorDeviceClass.ENUM,
options=[mode.name for mode in AnovaMode],
value_fn=lambda data: data.mode,
),
AnovaSensorEntityDescription(
@ -106,11 +104,34 @@ async def async_setup_entry(
) -> None:
"""Set up Anova device."""
anova_data: AnovaData = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
AnovaSensor(coordinator, description)
for coordinator in anova_data.coordinators
for description in SENSOR_DESCRIPTIONS
)
for coordinator in anova_data.coordinators:
setup_coordinator(coordinator, async_add_entities)
def setup_coordinator(
coordinator: AnovaCoordinator,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up an individual Anova Coordinator."""
def _async_sensor_listener() -> None:
"""Listen for new sensor data and add sensors if they did not exist."""
if not coordinator.sensor_data_set:
valid_entities: set[AnovaSensor] = set()
for description in SENSOR_DESCRIPTIONS:
if description.value_fn(coordinator.data.sensor) is not None:
valid_entities.add(AnovaSensor(coordinator, description))
async_add_entities(valid_entities)
coordinator.sensor_data_set = True
if coordinator.data is not None:
_async_sensor_listener()
# It is possible that we don't have any data, but the device exists,
# i.e. slow network, offline device, etc.
# We want to set up sensors after the fact as we don't know what sensors
# are valid until runtime.
coordinator.async_add_listener(_async_sensor_listener)
class AnovaSensor(AnovaDescriptionEntity, SensorEntity):

28
homeassistant/components/anova/strings.json
View file

@ -11,13 +11,9 @@
"description": "[%key:common::config_flow::description::confirm_setup%]"
}
},
"abort": {
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"no_devices_found": "No devices were found. Make sure you have at least one Anova device online."
"unknown": "[%key:common::config_flow::error::unknown%]"
}
},
"entity": {
@ -26,10 +22,28 @@
"name": "Cook time"
},
"state": {
"name": "State"
"name": "State",
"state": {
"preheating": "Preheating",
"cooking": "Cooking",
"maintaining": "Maintaining",
"timer_expired": "Timer expired",
"set_timer": "Set timer",
"no_state": "No state"
}
},
"mode": {
"name": "[%key:common::config_flow::data::mode%]"
"name": "[%key:common::config_flow::data::mode%]",
"state": {
"startup": "Startup",
"idle": "[%key:common::state::idle%]",
"cook": "Cooking",
"low_water": "Low water",
"ota": "Ota",
"provisioning": "Provisioning",
"high_temp": "High temperature",
"device_failure": "Device failure"
}
},
"target_temperature": {
"name": "Target temperature"

8
homeassistant/components/anova/util.py
View file

@ -1,8 +0,0 @@
"""Anova utilities."""
from anova_wifi import AnovaPrecisionCooker
def serialize_device_list(devices: list[AnovaPrecisionCooker]) -> list[tuple[str, str]]:
"""Turn the device list into a serializable list that can be reconstructed."""
return [(device.device_key, device.type) for device in devices]

2
homeassistant/generated/integrations.json
View file

@ -302,7 +302,7 @@
"name": "Anova",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
"iot_class": "cloud_push"
},
"anthemav": {
"name": "Anthem A/V Receivers",

2
requirements_all.txt
View file

@ -440,7 +440,7 @@ androidtvremote2==0.0.15
anel-pwrctrl-homeassistant==0.0.1.dev2
# homeassistant.components.anova
anova-wifi==0.10.0
anova-wifi==0.12.0
# homeassistant.components.anthemav
anthemav==1.4.1

2
requirements_test_all.txt
View file

@ -404,7 +404,7 @@ androidtv[async]==0.0.73
androidtvremote2==0.0.15
# homeassistant.components.anova
anova-wifi==0.10.0
anova-wifi==0.12.0
# homeassistant.components.anthemav
anthemav==1.4.1

21
tests/components/anova/__init__.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
from unittest.mock import patch
from anova_wifi import AnovaPrecisionCooker, APCUpdate, APCUpdateBinary, APCUpdateSensor
from anova_wifi import APCUpdate, APCUpdateBinary, APCUpdateSensor
from homeassistant.components.anova.const import DOMAIN
from homeassistant.config_entries import ConfigEntry
@ -21,7 +21,7 @@ ONLINE_UPDATE = APCUpdate(
sensor=APCUpdateSensor(
0, "Low water", "No state", 23.33, 0, "2.2.0", 20.87, 21.79, 21.33
),
binary_sensor=APCUpdateBinary(False, False, False, True, False, True, False),
binary_sensor=APCUpdateBinary(False, False, False, True, False, True, False, False),
)
@ -33,9 +33,9 @@ def create_entry(hass: HomeAssistant, device_id: str = DEVICE_UNIQUE_ID) -> Conf
data={
CONF_USERNAME: "sample@gmail.com",
CONF_PASSWORD: "sample",
"devices": [(device_id, "type_sample")],
},
unique_id="sample@gmail.com",
version=1,
)
entry.add_to_hass(hass)
return entry
@ -44,23 +44,10 @@ def create_entry(hass: HomeAssistant, device_id: str = DEVICE_UNIQUE_ID) -> Conf
async def async_init_integration(
hass: HomeAssistant,
skip_setup: bool = False,
error: str | None = None,
) -> ConfigEntry:
"""Set up the Anova integration in Home Assistant."""
with (
patch(
"homeassistant.components.anova.coordinator.AnovaPrecisionCooker.update"
) as update_patch,
patch("homeassistant.components.anova.AnovaApi.authenticate"),
patch(
"homeassistant.components.anova.AnovaApi.get_devices",
) as device_patch,
):
update_patch.return_value = ONLINE_UPDATE
device_patch.return_value = [
AnovaPrecisionCooker(None, DEVICE_UNIQUE_ID, "type_sample", None)
]
with patch("homeassistant.components.anova.AnovaApi.authenticate"):
entry = create_entry(hass)
if not skip_setup:

247
tests/components/anova/conftest.py
View file

@ -1,13 +1,176 @@
"""Common fixtures for Anova."""
import asyncio
from dataclasses import dataclass
import json
from typing import Any
from unittest.mock import AsyncMock, patch
from anova_wifi import AnovaApi, AnovaPrecisionCooker, InvalidLogin, NoDevicesFound
from aiohttp import ClientSession
from anova_wifi import (
AnovaApi,
AnovaWebsocketHandler,
InvalidLogin,
NoDevicesFound,
WebsocketFailure,
)
import pytest
from homeassistant.core import HomeAssistant
from . import DEVICE_UNIQUE_ID
DUMMY_ID = "anova_id"
@dataclass
class MockedanovaWebsocketMessage:
"""Mock the websocket message for Anova."""
input_data: dict[str, Any]
data: str = ""
def __post_init__(self) -> None:
"""Set up data after creation."""
self.data = json.dumps(self.input_data)
class MockedAnovaWebsocketStream:
"""Mock the websocket stream for Anova."""
def __init__(self, messages: list[MockedanovaWebsocketMessage]) -> None:
"""Initialize a Anova Websocket Stream that can be manipulated for tests."""
self.messages = messages
def __aiter__(self) -> "MockedAnovaWebsocketStream":
"""Handle async iteration."""
return self
async def __anext__(self) -> MockedanovaWebsocketMessage:
"""Get the next message in the websocket stream."""
if self.messages:
return self.messages.pop(0)
raise StopAsyncIteration
def clear(self) -> None:
"""Clear the Websocket stream."""
self.messages.clear()
class MockedAnovaWebsocketHandler(AnovaWebsocketHandler):
"""Mock the Anova websocket handler."""
def __init__(
self,
firebase_jwt: str,
jwt: str,
session: ClientSession,
connect_messages: list[MockedanovaWebsocketMessage],
post_connect_messages: list[MockedanovaWebsocketMessage],
) -> None:
"""Initialize the websocket handler with whatever messages you want."""
super().__init__(firebase_jwt, jwt, session)
self.connect_messages = connect_messages
self.post_connect_messages = post_connect_messages
async def connect(self) -> None:
"""Create a future for the message listener."""
self.ws = MockedAnovaWebsocketStream(self.connect_messages)
await self.message_listener()
self.ws = MockedAnovaWebsocketStream(self.post_connect_messages)
self.fut = asyncio.ensure_future(self.message_listener())
def anova_api_mock(
connect_messages: list[MockedanovaWebsocketMessage] | None = None,
post_connect_messages: list[MockedanovaWebsocketMessage] | None = None,
) -> AsyncMock:
"""Mock the api for Anova."""
api_mock = AsyncMock()
async def authenticate_side_effect() -> None:
api_mock.jwt = "my_test_jwt"
api_mock._firebase_jwt = "my_test_firebase_jwt"
async def create_websocket_side_effect() -> None:
api_mock.websocket_handler = MockedAnovaWebsocketHandler(
firebase_jwt=api_mock._firebase_jwt,
jwt=api_mock.jwt,
session=AsyncMock(),
connect_messages=connect_messages
if connect_messages is not None
else [
MockedanovaWebsocketMessage(
{
"command": "EVENT_APC_WIFI_LIST",
"payload": [
{
"cookerId": DUMMY_ID,
"type": "a5",
"pairedAt": "2023-08-12T02:33:20.917716Z",
"name": "Anova Precision Cooker",
}
],
}
),
],
post_connect_messages=post_connect_messages
if post_connect_messages is not None
else [
MockedanovaWebsocketMessage(
{
"command": "EVENT_APC_STATE",
"payload": {
"cookerId": DUMMY_ID,
"state": {
"boot-id": "8620610049456548422",
"job": {
"cook-time-seconds": 0,
"id": "8759286e3125b0c547",
"mode": "IDLE",
"ota-url": "",
"target-temperature": 54.72,
"temperature-unit": "F",
},
"job-status": {
"cook-time-remaining": 0,
"job-start-systick": 599679,
"provisioning-pairing-code": 7514,
"state": "",
"state-change-systick": 599679,
},
"pin-info": {
"device-safe": 0,
"water-leak": 0,
"water-level-critical": 0,
"water-temp-too-high": 0,
},
"system-info": {
"class": "A5",
"firmware-version": "2.2.0",
"type": "RA2L1-128",
},
"system-info-details": {
"firmware-version-raw": "VM178_A_02.02.00_MKE15-128",
"systick": 607026,
"version-string": "VM171_A_02.02.00 RA2L1-128",
},
"temperature-info": {
"heater-temperature": 22.37,
"triac-temperature": 36.04,
"water-temperature": 18.33,
},
},
},
}
),
],
)
await api_mock.websocket_handler.connect()
if not api_mock.websocket_handler.devices:
raise NoDevicesFound("No devices were found on the websocket.")
api_mock.authenticate.side_effect = authenticate_side_effect
api_mock.create_websocket.side_effect = create_websocket_side_effect
return api_mock
@pytest.fixture
@ -15,23 +178,14 @@ async def anova_api(
hass: HomeAssistant,
) -> AnovaApi:
"""Mock the api for Anova."""
api_mock = AsyncMock()
api_mock = anova_api_mock()
new_device = AnovaPrecisionCooker(None, DEVICE_UNIQUE_ID, "type_sample", None)
async def authenticate_side_effect():
api_mock.jwt = "my_test_jwt"
async def get_devices_side_effect():
if not api_mock.existing_devices:
api_mock.existing_devices = []
api_mock.existing_devices = [*api_mock.existing_devices, new_device]
return [new_device]
api_mock.authenticate.side_effect = authenticate_side_effect
api_mock.get_devices.side_effect = get_devices_side_effect
with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
with (
patch("homeassistant.components.anova.AnovaApi", return_value=api_mock),
patch(
"homeassistant.components.anova.config_flow.AnovaApi", return_value=api_mock
),
):
api = AnovaApi(
None,
"sample@gmail.com",
@ -45,18 +199,14 @@ async def anova_api_no_devices(
hass: HomeAssistant,
) -> AnovaApi:
"""Mock the api for Anova with no online devices."""
api_mock = AsyncMock()
api_mock = anova_api_mock(connect_messages=[], post_connect_messages=[])
async def authenticate_side_effect():
api_mock.jwt = "my_test_jwt"
async def get_devices_side_effect():
raise NoDevicesFound
api_mock.authenticate.side_effect = authenticate_side_effect
api_mock.get_devices.side_effect = get_devices_side_effect
with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
with (
patch("homeassistant.components.anova.AnovaApi", return_value=api_mock),
patch(
"homeassistant.components.anova.config_flow.AnovaApi", return_value=api_mock
),
):
api = AnovaApi(
None,
"sample@gmail.com",
@ -70,7 +220,7 @@ async def anova_api_wrong_login(
hass: HomeAssistant,
) -> AnovaApi:
"""Mock the api for Anova with a wrong login."""
api_mock = AsyncMock()
api_mock = anova_api_mock()
async def authenticate_side_effect():
raise InvalidLogin
@ -84,3 +234,40 @@ async def anova_api_wrong_login(
"sample",
)
yield api
@pytest.fixture
async def anova_api_no_data(
hass: HomeAssistant,
) -> AnovaApi:
"""Mock the api for Anova with a wrong login."""
api_mock = anova_api_mock(post_connect_messages=[])
with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
api = AnovaApi(
None,
"sample@gmail.com",
"sample",
)
yield api
@pytest.fixture
async def anova_api_websocket_failure(
hass: HomeAssistant,
) -> AnovaApi:
"""Mock the api for Anova with a websocket failure."""
api_mock = anova_api_mock()
async def create_websocket_side_effect():
raise WebsocketFailure
api_mock.create_websocket.side_effect = create_websocket_side_effect
with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
api = AnovaApi(
None,
"sample@gmail.com",
"sample",
)
yield api

107
tests/components/anova/test_config_flow.py
View file

@ -2,83 +2,33 @@
from unittest.mock import patch
from anova_wifi import AnovaPrecisionCooker, InvalidLogin, NoDevicesFound
from anova_wifi import AnovaApi, InvalidLogin
from homeassistant import config_entries
from homeassistant.components.anova.const import DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.const import CONF_DEVICES, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import CONF_INPUT, DEVICE_UNIQUE_ID, create_entry
from . import CONF_INPUT
async def test_flow_user(
hass: HomeAssistant,
) -> None:
async def test_flow_user(hass: HomeAssistant, anova_api: AnovaApi) -> None:
"""Test user initialized flow."""
with (
patch(
"homeassistant.components.anova.config_flow.AnovaApi.authenticate",
) as auth_patch,
patch("homeassistant.components.anova.AnovaApi.get_devices") as device_patch,
patch("homeassistant.components.anova.AnovaApi.authenticate"),
patch(
"homeassistant.components.anova.config_flow.AnovaApi.get_devices"
) as config_flow_device_patch,
):
auth_patch.return_value = True
device_patch.return_value = [
AnovaPrecisionCooker(None, DEVICE_UNIQUE_ID, "type_sample", None)
]
config_flow_device_patch.return_value = [
AnovaPrecisionCooker(None, DEVICE_UNIQUE_ID, "type_sample", None)
]
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_USERNAME: "sample@gmail.com",
CONF_PASSWORD: "sample",
"devices": [(DEVICE_UNIQUE_ID, "type_sample")],
}
async def test_flow_user_already_configured(hass: HomeAssistant) -> None:
"""Test user initialized flow with duplicate device."""
with (
patch(
"homeassistant.components.anova.config_flow.AnovaApi.authenticate",
) as auth_patch,
patch("homeassistant.components.anova.AnovaApi.get_devices") as device_patch,
patch(
"homeassistant.components.anova.config_flow.AnovaApi.get_devices"
) as config_flow_device_patch,
):
auth_patch.return_value = True
device_patch.return_value = [
AnovaPrecisionCooker(None, DEVICE_UNIQUE_ID, "type_sample", None)
]
config_flow_device_patch.return_value = [
AnovaPrecisionCooker(None, DEVICE_UNIQUE_ID, "type_sample", None)
]
create_entry(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_USERNAME: "sample@gmail.com",
CONF_PASSWORD: "sample",
CONF_DEVICES: [],
}
async def test_flow_wrong_login(hass: HomeAssistant) -> None:
@ -115,24 +65,3 @@ async def test_flow_unknown_error(hass: HomeAssistant) -> None:
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
async def test_flow_no_devices(hass: HomeAssistant) -> None:
"""Test unknown error throwing error."""
with (
patch("homeassistant.components.anova.config_flow.AnovaApi.authenticate"),
patch(
"homeassistant.components.anova.config_flow.AnovaApi.get_devices",
side_effect=NoDevicesFound(),
),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_devices_found"}

57
tests/components/anova/test_init.py
View file

@ -1,15 +1,12 @@
"""Test init for Anova."""
from unittest.mock import patch
from anova_wifi import AnovaApi
from homeassistant.components.anova import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from . import ONLINE_UPDATE, async_init_integration, create_entry
from . import async_init_integration, create_entry
async def test_async_setup_entry(hass: HomeAssistant, anova_api: AnovaApi) -> None:
@ -17,8 +14,7 @@ async def test_async_setup_entry(hass: HomeAssistant, anova_api: AnovaApi) -> No
await async_init_integration(hass)
state = hass.states.get("sensor.anova_precision_cooker_mode")
assert state is not None
assert state.state != STATE_UNAVAILABLE
assert state.state == "Low water"
assert state.state == "idle"
async def test_wrong_login(
@ -30,37 +26,6 @@ async def test_wrong_login(
assert entry.state is ConfigEntryState.SETUP_ERROR
async def test_new_devices(hass: HomeAssistant, anova_api: AnovaApi) -> None:
"""Test for if we find a new device on init."""
entry = create_entry(hass, "test_device_2")
with patch(
"homeassistant.components.anova.coordinator.AnovaPrecisionCooker.update"
) as update_patch:
update_patch.return_value = ONLINE_UPDATE
assert len(entry.data["devices"]) == 1
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(entry.data["devices"]) == 2
async def test_device_cached_but_offline(
hass: HomeAssistant, anova_api_no_devices: AnovaApi
) -> None:
"""Test if we have previously seen a device, but it was offline on startup."""
entry = create_entry(hass)
with patch(
"homeassistant.components.anova.coordinator.AnovaPrecisionCooker.update"
) as update_patch:
update_patch.return_value = ONLINE_UPDATE
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(entry.data["devices"]) == 1
state = hass.states.get("sensor.anova_precision_cooker_mode")
assert state is not None
assert state.state == "Low water"
async def test_unload_entry(hass: HomeAssistant, anova_api: AnovaApi) -> None:
"""Test successful unload of entry."""
entry = await async_init_integration(hass)
@ -72,3 +37,21 @@ async def test_unload_entry(hass: HomeAssistant, anova_api: AnovaApi) -> None:
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.NOT_LOADED
async def test_no_devices_found(
hass: HomeAssistant,
anova_api_no_devices: AnovaApi,
) -> None:
"""Test when there don't seem to be any devices on the account."""
entry = await async_init_integration(hass)
assert entry.state is ConfigEntryState.SETUP_RETRY
async def test_websocket_failure(
hass: HomeAssistant,
anova_api_websocket_failure: AnovaApi,
) -> None:
"""Test that we successfully handle a websocket failure on setup."""
entry = await async_init_integration(hass)
assert entry.state is ConfigEntryState.SETUP_RETRY

35
tests/components/anova/test_sensor.py
View file

@ -1,19 +1,13 @@
"""Test the Anova sensors."""
from datetime import timedelta
import logging
from unittest.mock import patch
from anova_wifi import AnovaApi, AnovaOffline
from anova_wifi import AnovaApi
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import async_init_integration
from tests.common import async_fire_time_changed
LOGGER = logging.getLogger(__name__)
@ -28,34 +22,25 @@ async def test_sensors(hass: HomeAssistant, anova_api: AnovaApi) -> None:
assert hass.states.get("sensor.anova_precision_cooker_cook_time").state == "0"
assert (
hass.states.get("sensor.anova_precision_cooker_heater_temperature").state
== "20.87"
== "22.37"
)
assert hass.states.get("sensor.anova_precision_cooker_mode").state == "Low water"
assert hass.states.get("sensor.anova_precision_cooker_state").state == "No state"
assert hass.states.get("sensor.anova_precision_cooker_mode").state == "idle"
assert hass.states.get("sensor.anova_precision_cooker_state").state == "no_state"
assert (
hass.states.get("sensor.anova_precision_cooker_target_temperature").state
== "23.33"
== "54.72"
)
assert (
hass.states.get("sensor.anova_precision_cooker_water_temperature").state
== "21.33"
== "18.33"
)
assert (
hass.states.get("sensor.anova_precision_cooker_triac_temperature").state
== "21.79"
== "36.04"
)
async def test_update_failed(hass: HomeAssistant, anova_api: AnovaApi) -> None:
"""Test updating data after the coordinator has been set up, but anova is offline."""
async def test_no_data_sensors(hass: HomeAssistant, anova_api_no_data: AnovaApi):
"""Test that if we have no data for the device, and we have not set it up previously, It is not immediately set up."""
await async_init_integration(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.anova.AnovaPrecisionCooker.update",
side_effect=AnovaOffline(),
):
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=61))
await hass.async_block_till_done()
state = hass.states.get("sensor.anova_precision_cooker_water_temperature")
assert state.state == STATE_UNAVAILABLE
assert hass.states.get("sensor.anova_precision_cooker_triac_temperature") is None