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

Make Netatmo use async pyatmo (#49717)

Browse source 
* Split initialization from data retrival

* Await class initialization

* Async camera

* More async

* Remove stale code

* Clean up

* Update tests

* Fix test

* Improve error handling

* Bump pyatmo version to 5.0.0

* Add tests

* Add cloudhook test

* Increase coverage

* Add test with no camera devices

* Add test for ApiError

* Add test for timeout

* Clean up

* Catch pyatmo ApiError

* Fix PublicData

* Fix media source bug

* Increase coverage for light

* Test webhook with delayed start

* Increase coverage

* Clean up leftover data classes

* Make nonprivate

* Review comments

* Clean up stale code

* Increase cov

* Clean up code

* Code clean up

* Revert delay

* Update homeassistant/components/netatmo/climate.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/netatmo/sensor.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Address comment

* Raise cov

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Tobias Sauerwein 2021-05-20 14:59:19 +02:00 • committed by GitHub
parent e06a2a53c4
commit ceec871340
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 846 additions and 476 deletions
Download patch file Download diff file Expand all files Collapse all files

23
homeassistant/components/netatmo/__init__.py
View file

@ -19,7 +19,11 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import CoreState, HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow, config_validation as cv
from homeassistant.helpers import (
aiohttp_client,
config_entry_oauth2_flow,
config_validation as cv,
)
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
@ -102,8 +106,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
if not entry.unique_id:
hass.config_entries.async_update_entry(entry, unique_id=DOMAIN)
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
hass.data[DOMAIN][entry.entry_id] = {
AUTH: api.ConfigEntryNetatmoAuth(hass, entry, implementation)
AUTH: api.AsyncConfigEntryNetatmoAuth(
aiohttp_client.async_get_clientsession(hass), session
)
}
data_handler = NetatmoDataHandler(hass, entry)
@ -122,6 +129,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
{"type": "None", "data": {"push_type": "webhook_deactivation"}},
)
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
await hass.data[DOMAIN][entry.entry_id][AUTH].async_dropwebhook()
async def register_webhook(event):
if CONF_WEBHOOK_ID not in entry.data:
@ -175,11 +183,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
handle_event,
)
activation_timeout = async_call_later(hass, 10, unregister_webhook)
activation_timeout = async_call_later(hass, 30, unregister_webhook)
await hass.async_add_executor_job(
hass.data[DOMAIN][entry.entry_id][AUTH].addwebhook, webhook_url
)
await hass.data[DOMAIN][entry.entry_id][AUTH].async_addwebhook(webhook_url)
_LOGGER.info("Register Netatmo webhook: %s", webhook_url)
except pyatmo.ApiError as err:
_LOGGER.error("Error during webhook registration - %s", err)
@ -202,9 +208,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
if CONF_WEBHOOK_ID in entry.data:
await hass.async_add_executor_job(
hass.data[DOMAIN][entry.entry_id][AUTH].dropwebhook
)
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
await hass.data[DOMAIN][entry.entry_id][AUTH].async_dropwebhook()
_LOGGER.info("Unregister Netatmo webhook")
await hass.data[DOMAIN][entry.entry_id][DATA_HANDLER].async_cleanup()

34
homeassistant/components/netatmo/api.py
View file

@ -1,34 +1,24 @@
"""API for Netatmo bound to HASS OAuth."""
from asyncio import run_coroutine_threadsafe
from aiohttp import ClientSession
import pyatmo
from homeassistant import config_entries, core
from homeassistant.helpers import config_entry_oauth2_flow
class ConfigEntryNetatmoAuth(pyatmo.auth.NetatmoOAuth2):
class AsyncConfigEntryNetatmoAuth(pyatmo.auth.AbstractAsyncAuth):
"""Provide Netatmo authentication tied to an OAuth2 based config entry."""
def __init__(
self,
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
websession: ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session,
) -> None:
"""Initialize Netatmo Auth."""
self.hass = hass
self.session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, implementation
)
super().__init__(token=self.session.token)
"""Initialize the auth."""
super().__init__(websession)
self._oauth_session = oauth_session
def refresh_tokens(
self,
) -> dict:
"""Refresh and return new Netatmo tokens using Home Assistant OAuth2 session."""
run_coroutine_threadsafe(
self.session.async_ensure_token_valid(), self.hass.loop
).result()
return self.session.token
async def async_get_access_token(self):
"""Return a valid access token for Netatmo API."""
if not self._oauth_session.valid_token:
await self._oauth_session.async_ensure_token_valid()
return self._oauth_session.token["access_token"]

126
homeassistant/components/netatmo/camera.py
View file

@ -1,8 +1,8 @@
"""Support for the Netatmo cameras."""
import logging
import aiohttp
import pyatmo
import requests
import voluptuous as vol
from homeassistant.components.camera import SUPPORT_STREAM, Camera
@ -46,58 +46,40 @@ async def async_setup_entry(hass, entry, async_add_entities):
_LOGGER.info(
"Cameras are currently not supported with this authentication method"
)
return
data_handler = hass.data[DOMAIN][entry.entry_id][DATA_HANDLER]
await data_handler.register_data_class(
CAMERA_DATA_CLASS_NAME, CAMERA_DATA_CLASS_NAME, None
)
data_class = data_handler.data.get(CAMERA_DATA_CLASS_NAME)
if CAMERA_DATA_CLASS_NAME not in data_handler.data:
if not data_class or not data_class.raw_data:
raise PlatformNotReady
async def get_entities():
"""Retrieve Netatmo entities."""
all_cameras = []
for home in data_class.cameras.values():
for camera in home.values():
all_cameras.append(camera)
if not data_handler.data.get(CAMERA_DATA_CLASS_NAME):
return []
entities = [
NetatmoCamera(
data_handler,
camera["id"],
camera["type"],
camera["home_id"],
DEFAULT_QUALITY,
)
for camera in all_cameras
]
data_class = data_handler.data[CAMERA_DATA_CLASS_NAME]
for person_id, person_data in data_handler.data[
CAMERA_DATA_CLASS_NAME
].persons.items():
hass.data[DOMAIN][DATA_PERSONS][person_id] = person_data.get(ATTR_PSEUDO)
entities = []
try:
all_cameras = []
for home in data_class.cameras.values():
for camera in home.values():
all_cameras.append(camera)
for camera in all_cameras:
_LOGGER.debug("Adding camera %s %s", camera["id"], camera["name"])
entities.append(
NetatmoCamera(
data_handler,
camera["id"],
camera["type"],
camera["home_id"],
DEFAULT_QUALITY,
)
)
for person_id, person_data in data_handler.data[
CAMERA_DATA_CLASS_NAME
].persons.items():
hass.data[DOMAIN][DATA_PERSONS][person_id] = person_data.get(
ATTR_PSEUDO
)
except pyatmo.NoDevice:
_LOGGER.debug("No cameras found")
return entities
async_add_entities(await get_entities(), True)
await data_handler.unregister_data_class(CAMERA_DATA_CLASS_NAME, None)
_LOGGER.debug("Adding cameras %s", entities)
async_add_entities(entities, True)
platform = entity_platform.async_get_current_platform()
@ -188,33 +170,17 @@ class NetatmoCamera(NetatmoBase, Camera):
self.async_write_ha_state()
return
def camera_image(self):
async def async_camera_image(self):
"""Return a still image response from the camera."""
try:
if self._localurl:
response = requests.get(
f"{self._localurl}/live/snapshot_720.jpg", timeout=10
)
elif self._vpnurl:
response = requests.get(
f"{self._vpnurl}/live/snapshot_720.jpg",
timeout=10,
verify=True,
)
else:
_LOGGER.error("Welcome/Presence VPN URL is None")
(self._vpnurl, self._localurl) = self._data.camera_urls(
camera_id=self._id
)
return None
except requests.exceptions.RequestException as error:
_LOGGER.info("Welcome/Presence URL changed: %s", error)
self._data.update_camera_urls(camera_id=self._id)
(self._vpnurl, self._localurl) = self._data.camera_urls(camera_id=self._id)
return None
return response.content
return await self._data.async_get_live_snapshot(camera_id=self._id)
except (
aiohttp.ClientPayloadError,
pyatmo.exceptions.ApiError,
aiohttp.ContentTypeError,
) as err:
_LOGGER.debug("Could not fetch live camera image (%s)", err)
return None
@property
def extra_state_attributes(self):
@ -255,15 +221,17 @@ class NetatmoCamera(NetatmoBase, Camera):
"""Return true if on."""
return self.is_streaming
def turn_off(self):
async def async_turn_off(self):
"""Turn off camera."""
self._data.set_state(
await self._data.async_set_state(
home_id=self._home_id, camera_id=self._id, monitoring="off"
)
def turn_on(self):
async def async_turn_on(self):
"""Turn on camera."""
self._data.set_state(home_id=self._home_id, camera_id=self._id, monitoring="on")
await self._data.async_set_state(
home_id=self._home_id, camera_id=self._id, monitoring="on"
)
async def stream_source(self):
"""Return the stream source."""
@ -312,7 +280,7 @@ class NetatmoCamera(NetatmoBase, Camera):
] = f"{self._vpnurl}/vod/{event['video_id']}/files/{self._quality}/index.m3u8"
return events
def _service_set_persons_home(self, **kwargs):
async def _service_set_persons_home(self, **kwargs):
"""Service to change current home schedule."""
persons = kwargs.get(ATTR_PERSONS)
person_ids = []
@ -321,10 +289,12 @@ class NetatmoCamera(NetatmoBase, Camera):
if data.get("pseudo") == person:
person_ids.append(pid)
self._data.set_persons_home(person_ids=person_ids, home_id=self._home_id)
await self._data.async_set_persons_home(
person_ids=person_ids, home_id=self._home_id
)
_LOGGER.debug("Set %s as at home", persons)
def _service_set_person_away(self, **kwargs):
async def _service_set_person_away(self, **kwargs):
"""Service to mark a person as away or set the home as empty."""
person = kwargs.get(ATTR_PERSON)
person_id = None
@ -333,25 +303,25 @@ class NetatmoCamera(NetatmoBase, Camera):
if data.get("pseudo") == person:
person_id = pid
if person_id is not None:
self._data.set_persons_away(
if person_id:
await self._data.async_set_persons_away(
person_id=person_id,
home_id=self._home_id,
)
_LOGGER.debug("Set %s as away", person)
else:
self._data.set_persons_away(
await self._data.async_set_persons_away(
person_id=person_id,
home_id=self._home_id,
)
_LOGGER.debug("Set home as empty")
def _service_set_camera_light(self, **kwargs):
async def _service_set_camera_light(self, **kwargs):
"""Service to set light mode."""
mode = kwargs.get(ATTR_CAMERA_LIGHT_MODE)
_LOGGER.debug("Turn %s camera light for '%s'", mode, self._name)
self._data.set_state(
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,
floodlight=mode,

118
homeassistant/components/netatmo/climate.py
View file

@ -116,47 +116,39 @@ async def async_setup_entry(hass, entry, async_add_entities):
)
home_data = data_handler.data.get(HOMEDATA_DATA_CLASS_NAME)
if not home_data or home_data.raw_data == {}:
raise PlatformNotReady
if HOMEDATA_DATA_CLASS_NAME not in data_handler.data:
raise PlatformNotReady
async def get_entities():
"""Retrieve Netatmo entities."""
entities = []
entities = []
for home_id in get_all_home_ids(home_data):
for room_id in home_data.rooms[home_id]:
signal_name = f"{HOMESTATUS_DATA_CLASS_NAME}-{home_id}"
await data_handler.register_data_class(
HOMESTATUS_DATA_CLASS_NAME, signal_name, None, home_id=home_id
)
home_status = data_handler.data.get(signal_name)
if home_status and room_id in home_status.rooms:
entities.append(NetatmoThermostat(data_handler, home_id, room_id))
for home_id in get_all_home_ids(home_data):
_LOGGER.debug("Setting up home %s", home_id)
for room_id in home_data.rooms[home_id].keys():
room_name = home_data.rooms[home_id][room_id]["name"]
_LOGGER.debug("Setting up room %s (%s)", room_name, room_id)
signal_name = f"{HOMESTATUS_DATA_CLASS_NAME}-{home_id}"
await data_handler.register_data_class(
HOMESTATUS_DATA_CLASS_NAME, signal_name, None, home_id=home_id
)
home_status = data_handler.data.get(signal_name)
if home_status and room_id in home_status.rooms:
entities.append(NetatmoThermostat(data_handler, home_id, room_id))
hass.data[DOMAIN][DATA_SCHEDULES][home_id] = {
schedule_id: schedule_data.get("name")
for schedule_id, schedule_data in (
data_handler.data[HOMEDATA_DATA_CLASS_NAME]
.schedules[home_id]
.items()
)
}
hass.data[DOMAIN][DATA_HOMES] = {
home_id: home_data.get("name")
for home_id, home_data in (
data_handler.data[HOMEDATA_DATA_CLASS_NAME].homes.items()
hass.data[DOMAIN][DATA_SCHEDULES][home_id] = {
schedule_id: schedule_data.get("name")
for schedule_id, schedule_data in (
data_handler.data[HOMEDATA_DATA_CLASS_NAME].schedules[home_id].items()
)
}
return entities
hass.data[DOMAIN][DATA_HOMES] = {
home_id: home_data.get("name")
for home_id, home_data in (
data_handler.data[HOMEDATA_DATA_CLASS_NAME].homes.items()
)
}
async_add_entities(await get_entities(), True)
await data_handler.unregister_data_class(HOMEDATA_DATA_CLASS_NAME, None)
_LOGGER.debug("Adding climate devices %s", entities)
async_add_entities(entities, True)
platform = entity_platform.async_get_current_platform()
@ -164,7 +156,7 @@ async def async_setup_entry(hass, entry, async_add_entities):
platform.async_register_entity_service(
SERVICE_SET_SCHEDULE,
{vol.Required(ATTR_SCHEDULE_NAME): cv.string},
"_service_set_schedule",
"_async_service_set_schedule",
)
@ -205,7 +197,6 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
self._model = NA_THERM
break
self._state = None
self._device_name = self._data.rooms[home_id][room_id]["name"]
self._name = f"{MANUFACTURER} {self._device_name}"
self._current_temperature = None
@ -357,24 +348,24 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return CURRENT_HVAC_HEAT
return CURRENT_HVAC_IDLE
def set_hvac_mode(self, hvac_mode: str) -> None:
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
"""Set new target hvac mode."""
if hvac_mode == HVAC_MODE_OFF:
self.turn_off()
await self.async_turn_off()
elif hvac_mode == HVAC_MODE_AUTO:
if self.hvac_mode == HVAC_MODE_OFF:
self.turn_on()
self.set_preset_mode(PRESET_SCHEDULE)
await self.async_turn_on()
await self.async_set_preset_mode(PRESET_SCHEDULE)
elif hvac_mode == HVAC_MODE_HEAT:
self.set_preset_mode(PRESET_BOOST)
await self.async_set_preset_mode(PRESET_BOOST)
def set_preset_mode(self, preset_mode: str) -> None:
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new preset mode."""
if self.hvac_mode == HVAC_MODE_OFF:
self.turn_on()
await self.async_turn_on()
if self.target_temperature == 0:
self._home_status.set_room_thermpoint(
await self._home_status.async_set_room_thermpoint(
self._id,
STATE_NETATMO_HOME,
)
@ -384,14 +375,14 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
and self._model == NA_VALVE
and self.hvac_mode == HVAC_MODE_HEAT
):
self._home_status.set_room_thermpoint(
await self._home_status.async_set_room_thermpoint(
self._id,
STATE_NETATMO_HOME,
)
elif (
preset_mode in [PRESET_BOOST, STATE_NETATMO_MAX] and self._model == NA_VALVE
):
self._home_status.set_room_thermpoint(
await self._home_status.async_set_room_thermpoint(
self._id,
STATE_NETATMO_MANUAL,
DEFAULT_MAX_TEMP,
@ -400,13 +391,15 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
preset_mode in [PRESET_BOOST, STATE_NETATMO_MAX]
and self.hvac_mode == HVAC_MODE_HEAT
):
self._home_status.set_room_thermpoint(self._id, STATE_NETATMO_HOME)
await self._home_status.async_set_room_thermpoint(
self._id, STATE_NETATMO_HOME
)
elif preset_mode in [PRESET_BOOST, STATE_NETATMO_MAX]:
self._home_status.set_room_thermpoint(
await self._home_status.async_set_room_thermpoint(
self._id, PRESET_MAP_NETATMO[preset_mode]
)
elif preset_mode in [PRESET_SCHEDULE, PRESET_FROST_GUARD, PRESET_AWAY]:
self._home_status.set_thermmode(PRESET_MAP_NETATMO[preset_mode])
await self._home_status.async_set_thermmode(PRESET_MAP_NETATMO[preset_mode])
else:
_LOGGER.error("Preset mode '%s' not available", preset_mode)
@ -422,12 +415,14 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
"""Return a list of available preset modes."""
return SUPPORT_PRESET
def set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs):
"""Set new target temperature for 2 hours."""
temp = kwargs.get(ATTR_TEMPERATURE)
if temp is None:
return
self._home_status.set_room_thermpoint(self._id, STATE_NETATMO_MANUAL, temp)
await self._home_status.async_set_room_thermpoint(
self._id, STATE_NETATMO_MANUAL, temp
)
self.async_write_ha_state()
@ -449,21 +444,23 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return attr
def turn_off(self):
async def async_turn_off(self):
"""Turn the entity off."""
if self._model == NA_VALVE:
self._home_status.set_room_thermpoint(
await self._home_status.async_set_room_thermpoint(
self._id,
STATE_NETATMO_MANUAL,
DEFAULT_MIN_TEMP,
)
elif self.hvac_mode != HVAC_MODE_OFF:
self._home_status.set_room_thermpoint(self._id, STATE_NETATMO_OFF)
await self._home_status.async_set_room_thermpoint(
self._id, STATE_NETATMO_OFF
)
self.async_write_ha_state()
def turn_on(self):
async def async_turn_on(self):
"""Turn the entity on."""
self._home_status.set_room_thermpoint(self._id, STATE_NETATMO_HOME)
await self._home_status.async_set_room_thermpoint(self._id, STATE_NETATMO_HOME)
self.async_write_ha_state()
@property
@ -475,6 +472,11 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
def async_update_callback(self):
"""Update the entity's state."""
self._home_status = self.data_handler.data[self._home_status_class]
if self._home_status is None:
if self.available:
self._connected = False
return
self._room_status = self._home_status.rooms.get(self._id)
self._room_data = self._data.rooms.get(self._home_id, {}).get(self._id)
@ -570,7 +572,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return {}
def _service_set_schedule(self, **kwargs):
async def _async_service_set_schedule(self, **kwargs):
schedule_name = kwargs.get(ATTR_SCHEDULE_NAME)
schedule_id = None
for sid, name in self.hass.data[DOMAIN][DATA_SCHEDULES][self._home_id].items():
@ -581,7 +583,9 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
_LOGGER.error("%s is not a valid schedule", kwargs.get(ATTR_SCHEDULE_NAME))
return
self._data.switch_home_schedule(home_id=self._home_id, schedule_id=schedule_id)
await self._data.async_switch_home_schedule(
home_id=self._home_id, schedule_id=schedule_id
)
_LOGGER.debug(
"Setting %s schedule to %s (%s)",
self._home_id,

89
homeassistant/components/netatmo/data_handler.py
View file

@ -1,9 +1,9 @@
"""The Netatmo data handler."""
from __future__ import annotations
import asyncio
from collections import deque
from datetime import timedelta
from functools import partial
from itertools import islice
import logging
from time import time
@ -19,22 +19,22 @@ from .const import AUTH, DOMAIN, MANUFACTURER
_LOGGER = logging.getLogger(__name__)
CAMERA_DATA_CLASS_NAME = "CameraData"
WEATHERSTATION_DATA_CLASS_NAME = "WeatherStationData"
HOMECOACH_DATA_CLASS_NAME = "HomeCoachData"
HOMEDATA_DATA_CLASS_NAME = "HomeData"
HOMESTATUS_DATA_CLASS_NAME = "HomeStatus"
PUBLICDATA_DATA_CLASS_NAME = "PublicData"
CAMERA_DATA_CLASS_NAME = "AsyncCameraData"
WEATHERSTATION_DATA_CLASS_NAME = "AsyncWeatherStationData"
HOMECOACH_DATA_CLASS_NAME = "AsyncHomeCoachData"
HOMEDATA_DATA_CLASS_NAME = "AsyncHomeData"
HOMESTATUS_DATA_CLASS_NAME = "AsyncHomeStatus"
PUBLICDATA_DATA_CLASS_NAME = "AsyncPublicData"
NEXT_SCAN = "next_scan"
DATA_CLASSES = {
WEATHERSTATION_DATA_CLASS_NAME: pyatmo.WeatherStationData,
HOMECOACH_DATA_CLASS_NAME: pyatmo.HomeCoachData,
CAMERA_DATA_CLASS_NAME: pyatmo.CameraData,
HOMEDATA_DATA_CLASS_NAME: pyatmo.HomeData,
HOMESTATUS_DATA_CLASS_NAME: pyatmo.HomeStatus,
PUBLICDATA_DATA_CLASS_NAME: pyatmo.PublicData,
WEATHERSTATION_DATA_CLASS_NAME: pyatmo.AsyncWeatherStationData,
HOMECOACH_DATA_CLASS_NAME: pyatmo.AsyncHomeCoachData,
CAMERA_DATA_CLASS_NAME: pyatmo.AsyncCameraData,
HOMEDATA_DATA_CLASS_NAME: pyatmo.AsyncHomeData,
HOMESTATUS_DATA_CLASS_NAME: pyatmo.AsyncHomeStatus,
PUBLICDATA_DATA_CLASS_NAME: pyatmo.AsyncPublicData,
}
BATCH_SIZE = 3
@ -57,7 +57,7 @@ class NetatmoDataHandler:
self.hass = hass
self._auth = hass.data[DOMAIN][entry.entry_id][AUTH]
self.listeners: list[CALLBACK_TYPE] = []
self._data_classes: dict = {}
self.data_classes: dict = {}
self.data = {}
self._queue = deque()
self._webhook: bool = False
@ -87,21 +87,19 @@ class NetatmoDataHandler:
for data_class in islice(self._queue, 0, BATCH_SIZE):
if data_class[NEXT_SCAN] > time():
continue
self._data_classes[data_class["name"]][NEXT_SCAN] = (
self.data_classes[data_class["name"]][NEXT_SCAN] = (
time() + data_class["interval"]
)
await self.async_fetch_data(
data_class["class"], data_class["name"], **data_class["kwargs"]
)
await self.async_fetch_data(data_class["name"])
self._queue.rotate(BATCH_SIZE)
@callback
def async_force_update(self, data_class_entry):
"""Prioritize data retrieval for given data class entry."""
self._data_classes[data_class_entry][NEXT_SCAN] = time()
self._queue.rotate(-(self._queue.index(self._data_classes[data_class_entry])))
self.data_classes[data_class_entry][NEXT_SCAN] = time()
self._queue.rotate(-(self._queue.index(self.data_classes[data_class_entry])))
async def async_cleanup(self):
"""Clean up the Netatmo data handler."""
@ -122,19 +120,10 @@ class NetatmoDataHandler:
_LOGGER.debug("%s camera reconnected", MANUFACTURER)
self.async_force_update(CAMERA_DATA_CLASS_NAME)
async def async_fetch_data(self, data_class, data_class_entry, **kwargs):
async def async_fetch_data(self, data_class_entry):
"""Fetch data and notify."""
try:
self.data[data_class_entry] = await self.hass.async_add_executor_job(
partial(data_class, **kwargs),
self._auth,
)
for update_callback in self._data_classes[data_class_entry][
"subscriptions"
]:
if update_callback:
update_callback()
await self.data[data_class_entry].async_update()
except pyatmo.NoDevice as err:
_LOGGER.debug(err)
@ -143,42 +132,46 @@ class NetatmoDataHandler:
except pyatmo.ApiError as err:
_LOGGER.debug(err)
except asyncio.TimeoutError as err:
_LOGGER.debug(err)
return
for update_callback in self.data_classes[data_class_entry]["subscriptions"]:
if update_callback:
update_callback()
async def register_data_class(
self, data_class_name, data_class_entry, update_callback, **kwargs
):
"""Register data class."""
if data_class_entry in self._data_classes:
self._data_classes[data_class_entry]["subscriptions"].append(
update_callback
)
if data_class_entry in self.data_classes:
self.data_classes[data_class_entry]["subscriptions"].append(update_callback)
return
self._data_classes[data_class_entry] = {
"class": DATA_CLASSES[data_class_name],
self.data_classes[data_class_entry] = {
"name": data_class_entry,
"interval": DEFAULT_INTERVALS[data_class_name],
NEXT_SCAN: time() + DEFAULT_INTERVALS[data_class_name],
"kwargs": kwargs,
"subscriptions": [update_callback],
}
await self.async_fetch_data(
DATA_CLASSES[data_class_name], data_class_entry, **kwargs
self.data[data_class_entry] = DATA_CLASSES[data_class_name](
self._auth, **kwargs
)
self._queue.append(self._data_classes[data_class_entry])
await self.async_fetch_data(data_class_entry)
self._queue.append(self.data_classes[data_class_entry])
_LOGGER.debug("Data class %s added", data_class_entry)
async def unregister_data_class(self, data_class_entry, update_callback):
"""Unregister data class."""
if update_callback not in self._data_classes[data_class_entry]["subscriptions"]:
return
self.data_classes[data_class_entry]["subscriptions"].remove(update_callback)
self._data_classes[data_class_entry]["subscriptions"].remove(update_callback)
if not self._data_classes[data_class_entry].get("subscriptions"):
self._queue.remove(self._data_classes[data_class_entry])
self._data_classes.pop(data_class_entry)
if not self.data_classes[data_class_entry].get("subscriptions"):
self._queue.remove(self.data_classes[data_class_entry])
self.data_classes.pop(data_class_entry)
self.data.pop(data_class_entry)
_LOGGER.debug("Data class %s removed", data_class_entry)
@property

58
homeassistant/components/netatmo/light.py
View file

@ -1,8 +1,6 @@
"""Support for the Netatmo camera lights."""
import logging
import pyatmo
from homeassistant.components.light import LightEntity
from homeassistant.core import callback
from homeassistant.exceptions import PlatformNotReady
@ -34,41 +32,29 @@ async def async_setup_entry(hass, entry, async_add_entities):
await data_handler.register_data_class(
CAMERA_DATA_CLASS_NAME, CAMERA_DATA_CLASS_NAME, None
)
data_class = data_handler.data.get(CAMERA_DATA_CLASS_NAME)
if CAMERA_DATA_CLASS_NAME not in data_handler.data:
if not data_class or data_class.raw_data == {}:
raise PlatformNotReady
async def get_entities():
"""Retrieve Netatmo entities."""
all_cameras = []
for home in data_handler.data[CAMERA_DATA_CLASS_NAME].cameras.values():
for camera in home.values():
all_cameras.append(camera)
entities = []
all_cameras = []
entities = [
NetatmoLight(
data_handler,
camera["id"],
camera["type"],
camera["home_id"],
)
for camera in all_cameras
if camera["type"] == "NOC"
]
try:
for home in data_handler.data[CAMERA_DATA_CLASS_NAME].cameras.values():
for camera in home.values():
all_cameras.append(camera)
except pyatmo.NoDevice:
_LOGGER.debug("No cameras found")
for camera in all_cameras:
if camera["type"] == "NOC":
_LOGGER.debug("Adding camera light %s %s", camera["id"], camera["name"])
entities.append(
NetatmoLight(
data_handler,
camera["id"],
camera["type"],
camera["home_id"],
)
)
return entities
async_add_entities(await get_entities(), True)
await data_handler.unregister_data_class(CAMERA_DATA_CLASS_NAME, None)
_LOGGER.debug("Adding camera lights %s", entities)
async_add_entities(entities, True)
class NetatmoLight(NetatmoBase, LightEntity):
@ -136,19 +122,19 @@ class NetatmoLight(NetatmoBase, LightEntity):
"""Return true if light is on."""
return self._is_on
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
"""Turn camera floodlight on."""
_LOGGER.debug("Turn camera '%s' on", self._name)
self._data.set_state(
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,
floodlight="on",
)
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs):
"""Turn camera floodlight into auto mode."""
_LOGGER.debug("Turn camera '%s' to auto mode", self._name)
self._data.set_state(
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,
floodlight="auto",

2
homeassistant/components/netatmo/manifest.json
View file

@ -3,7 +3,7 @@
"name": "Netatmo",
"documentation": "https://www.home-assistant.io/integrations/netatmo",
"requirements": [
"pyatmo==4.2.3"
"pyatmo==5.0.1"
],
"after_dependencies": [
"cloud",

2
homeassistant/components/netatmo/media_source.py
View file

@ -159,7 +159,7 @@ def async_parse_identifier(
item: MediaSourceItem,
) -> tuple[str, str, int | None]:
"""Parse identifier."""
if not item.identifier:
if "/" not in item.identifier:
return "events", "", None
source, path = item.identifier.lstrip("/").split("/", 1)

9
homeassistant/components/netatmo/netatmo_entity_base.py
View file

@ -7,7 +7,7 @@ from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.helpers.entity import Entity
from .const import DATA_DEVICE_IDS, DOMAIN, MANUFACTURER, MODELS, SIGNAL_NAME
from .data_handler import NetatmoDataHandler
from .data_handler import PUBLICDATA_DATA_CLASS_NAME, NetatmoDataHandler
_LOGGER = logging.getLogger(__name__)
@ -29,7 +29,6 @@ class NetatmoBase(Entity):
async def async_added_to_hass(self) -> None:
"""Entity created."""
_LOGGER.debug("New client %s", self.entity_id)
for data_class in self._data_classes:
signal_name = data_class[SIGNAL_NAME]
@ -41,7 +40,7 @@ class NetatmoBase(Entity):
home_id=data_class["home_id"],
)
elif data_class["name"] == "PublicData":
elif data_class["name"] == PUBLICDATA_DATA_CLASS_NAME:
await self.data_handler.register_data_class(
data_class["name"],
signal_name,
@ -57,7 +56,9 @@ class NetatmoBase(Entity):
data_class["name"], signal_name, self.async_update_callback
)
await self.data_handler.unregister_data_class(signal_name, None)
for sub in self.data_handler.data_classes[signal_name].get("subscriptions"):
if sub is None:
await self.data_handler.unregister_data_class(signal_name, None)
registry = await self.hass.helpers.device_registry.async_get_registry()
device = registry.async_get_device({(DOMAIN, self._id)}, set())

35
homeassistant/components/netatmo/sensor.py
View file

@ -132,18 +132,8 @@ async def async_setup_entry(hass, entry, async_add_entities):
"""Set up the Netatmo weather and homecoach platform."""
data_handler = hass.data[DOMAIN][entry.entry_id][DATA_HANDLER]
await data_handler.register_data_class(
WEATHERSTATION_DATA_CLASS_NAME, WEATHERSTATION_DATA_CLASS_NAME, None
)
await data_handler.register_data_class(
HOMECOACH_DATA_CLASS_NAME, HOMECOACH_DATA_CLASS_NAME, None
)
async def find_entities(data_class_name):
"""Find all entities."""
if data_class_name not in data_handler.data:
raise PlatformNotReady
all_module_infos = {}
data = data_handler.data
@ -167,11 +157,6 @@ async def async_setup_entry(hass, entry, async_add_entities):
_LOGGER.debug("Skipping module %s", module.get("module_name"))
continue
_LOGGER.debug(
"Adding module %s %s",
module.get("module_name"),
module.get("_id"),
)
conditions = [
c.lower()
for c in data_class.get_monitored_conditions(module_id=module["_id"])
@ -188,14 +173,19 @@ async def async_setup_entry(hass, entry, async_add_entities):
NetatmoSensor(data_handler, data_class_name, module, condition)
)
await data_handler.unregister_data_class(data_class_name, None)
_LOGGER.debug("Adding weather sensors %s", entities)
return entities
for data_class_name in [
WEATHERSTATION_DATA_CLASS_NAME,
HOMECOACH_DATA_CLASS_NAME,
]:
await data_handler.register_data_class(data_class_name, data_class_name, None)
data_class = data_handler.data.get(data_class_name)
if not data_class or not data_class.raw_data:
raise PlatformNotReady
async_add_entities(await find_entities(data_class_name), True)
device_registry = await hass.helpers.device_registry.async_get_registry()
@ -410,6 +400,8 @@ class NetatmoSensor(NetatmoBase, SensorEntity):
self._state = None
return
self.async_write_ha_state()
def fix_angle(angle: int) -> int:
"""Fix angle when value is negative."""
@ -615,13 +607,6 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
@callback
def async_update_callback(self):
"""Update the entity's state."""
if self._data is None:
if self._state is None:
return
_LOGGER.warning("No data from update")
self._state = None
return
data = None
if self.type == "temperature":
@ -655,3 +640,5 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
self._state = round(sum(values) / len(values), 1)
elif self._mode == "max":
self._state = max(values)
self.async_write_ha_state()

2
requirements_all.txt
View file

@ -1295,7 +1295,7 @@ pyarlo==0.2.4
pyatag==0.3.5.3
# homeassistant.components.netatmo
pyatmo==4.2.3
pyatmo==5.0.1
# homeassistant.components.atome
pyatome==0.1.1

2
requirements_test_all.txt
View file

@ -720,7 +720,7 @@ pyarlo==0.2.4
pyatag==0.3.5.3
# homeassistant.components.netatmo
pyatmo==4.2.3
pyatmo==5.0.1
# homeassistant.components.apple_tv
pyatv==0.7.7

25
tests/components/netatmo/common.py
View file

@ -1,5 +1,7 @@
"""Common methods used across tests for Netatmo."""
from contextlib import contextmanager
import json
from unittest.mock import patch
from homeassistant.components.webhook import async_handle_webhook
from homeassistant.util.aiohttp import MockRequest
@ -35,13 +37,19 @@ FAKE_WEBHOOK_ACTIVATION = {
"push_type": "webhook_activation",
}
DEFAULT_PLATFORMS = ["camera", "climate", "light", "sensor"]
def fake_post_request(**args):
async def fake_post_request(*args, **kwargs):
"""Return fake data."""
if "url" not in args:
if "url" not in kwargs:
return "{}"
endpoint = args["url"].split("/")[-1]
endpoint = kwargs["url"].split("/")[-1]
if endpoint in "snapshot_720.jpg":
return b"test stream image bytes"
if endpoint in [
"setpersonsaway",
"setpersonshome",
@ -55,7 +63,7 @@ def fake_post_request(**args):
return json.loads(load_fixture(f"netatmo/{endpoint}.json"))
def fake_post_request_no_data(**args):
async def fake_post_request_no_data(*args, **kwargs):
"""Fake error during requesting backend data."""
return "{}"
@ -68,3 +76,12 @@ async def simulate_webhook(hass, webhook_id, response):
)
await async_handle_webhook(hass, webhook_id, request)
await hass.async_block_till_done()
@contextmanager
def selected_platforms(platforms):
"""Restrict loaded platforms to list given."""
with patch("homeassistant.components.netatmo.PLATFORMS", platforms), patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch("homeassistant.components.webhook.async_generate_url"):
yield

91
tests/components/netatmo/conftest.py
View file

@ -1,17 +1,16 @@
"""Provide common Netatmo fixtures."""
from contextlib import contextmanager
from time import time
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pytest
from .common import ALL_SCOPES, TEST_TIME, fake_post_request, fake_post_request_no_data
from .common import ALL_SCOPES, fake_post_request
from tests.common import MockConfigEntry
@pytest.fixture(name="config_entry")
async def mock_config_entry_fixture(hass):
def mock_config_entry_fixture(hass):
"""Mock a config entry."""
mock_entry = MockConfigEntry(
domain="netatmo",
@ -54,81 +53,13 @@ async def mock_config_entry_fixture(hass):
return mock_entry
@contextmanager
def selected_platforms(platforms=["camera", "climate", "light", "sensor"]):
@pytest.fixture
def netatmo_auth():
"""Restrict loaded platforms to list given."""
with patch("homeassistant.components.netatmo.PLATFORMS", platforms), patch(
"homeassistant.components.netatmo.api.ConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.post_request.side_effect = fake_post_request
yield
@pytest.fixture(name="entry")
async def mock_entry_fixture(hass, config_entry):
"""Mock setup of all platforms."""
with selected_platforms():
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.fixture(name="sensor_entry")
async def mock_sensor_entry_fixture(hass, config_entry):
"""Mock setup of sensor platform."""
with patch("time.time", return_value=TEST_TIME), selected_platforms(["sensor"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
yield config_entry
@pytest.fixture(name="camera_entry")
async def mock_camera_entry_fixture(hass, config_entry):
"""Mock setup of camera platform."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.fixture(name="light_entry")
async def mock_light_entry_fixture(hass, config_entry):
"""Mock setup of light platform."""
with selected_platforms(["light"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.fixture(name="climate_entry")
async def mock_climate_entry_fixture(hass, config_entry):
"""Mock setup of climate platform."""
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.fixture(name="entry_error")
async def mock_entry_error_fixture(hass, config_entry):
"""Mock erroneous setup of platforms."""
with patch(
"homeassistant.components.netatmo.api.ConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.post_request.side_effect = fake_post_request_no_data
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
yield config_entry
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth:
mock_auth.return_value.async_post_request.side_effect = fake_post_request
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
yield

194
tests/components/netatmo/test_camera.py
View file

@ -1,6 +1,9 @@
"""The tests for Netatmo camera."""
from datetime import timedelta
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pyatmo
import pytest
from homeassistant.components import camera
from homeassistant.components.camera import STATE_STREAMING
@ -13,14 +16,19 @@ from homeassistant.components.netatmo.const import (
from homeassistant.const import CONF_WEBHOOK_ID
from homeassistant.util import dt
from .common import fake_post_request, simulate_webhook
from .common import fake_post_request, selected_platforms, simulate_webhook
from tests.common import async_capture_events, async_fire_time_changed
async def test_setup_component_with_webhook(hass, camera_entry):
async def test_setup_component_with_webhook(hass, config_entry, netatmo_auth):
"""Test setup with webhook."""
webhook_id = camera_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
await hass.async_block_till_done()
camera_entity_indoor = "camera.netatmo_hall"
@ -58,7 +66,7 @@ async def test_setup_component_with_webhook(hass, camera_entry):
}
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(camera_entity_indoor).state == "streaming"
assert hass.states.get(camera_entity_outdoor).state == "streaming"
assert hass.states.get(camera_entity_outdoor).attributes["light_state"] == "on"
response = {
@ -84,12 +92,39 @@ async def test_setup_component_with_webhook(hass, camera_entry):
assert hass.states.get(camera_entity_indoor).state == "streaming"
assert hass.states.get(camera_entity_outdoor).attributes["light_state"] == "auto"
with patch("pyatmo.camera.AsyncCameraData.async_set_state") as mock_set_state:
await hass.services.async_call(
"camera", "turn_off", service_data={"entity_id": "camera.netatmo_hall"}
)
await hass.async_block_till_done()
mock_set_state.assert_called_once_with(
home_id="91763b24c43d3e344f424e8b",
camera_id="12:34:56:00:f1:62",
monitoring="off",
)
with patch("pyatmo.camera.AsyncCameraData.async_set_state") as mock_set_state:
await hass.services.async_call(
"camera", "turn_on", service_data={"entity_id": "camera.netatmo_hall"}
)
await hass.async_block_till_done()
mock_set_state.assert_called_once_with(
home_id="91763b24c43d3e344f424e8b",
camera_id="12:34:56:00:f1:62",
monitoring="on",
)
IMAGE_BYTES_FROM_STREAM = b"test stream image bytes"
async def test_camera_image_local(hass, camera_entry, requests_mock):
async def test_camera_image_local(hass, config_entry, requests_mock, netatmo_auth):
"""Test retrieval or local camera image."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done()
uri = "http://192.168.0.123/678460a0d47e5618699fb31169e2b47d"
@ -111,8 +146,13 @@ async def test_camera_image_local(hass, camera_entry, requests_mock):
assert image.content == IMAGE_BYTES_FROM_STREAM
async def test_camera_image_vpn(hass, camera_entry, requests_mock):
async def test_camera_image_vpn(hass, config_entry, requests_mock, netatmo_auth):
"""Test retrieval of remote camera image."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done()
uri = (
@ -137,8 +177,13 @@ async def test_camera_image_vpn(hass, camera_entry, requests_mock):
assert image.content == IMAGE_BYTES_FROM_STREAM
async def test_service_set_person_away(hass, camera_entry):
async def test_service_set_person_away(hass, config_entry, netatmo_auth):
"""Test service to set person as away."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done()
data = {
@ -146,7 +191,9 @@ async def test_service_set_person_away(hass, camera_entry):
"person": "Richard Doe",
}
with patch("pyatmo.camera.CameraData.set_persons_away") as mock_set_persons_away:
with patch(
"pyatmo.camera.AsyncCameraData.async_set_persons_away"
) as mock_set_persons_away:
await hass.services.async_call(
"netatmo", SERVICE_SET_PERSON_AWAY, service_data=data
)
@ -160,7 +207,9 @@ async def test_service_set_person_away(hass, camera_entry):
"entity_id": "camera.netatmo_hall",
}
with patch("pyatmo.camera.CameraData.set_persons_away") as mock_set_persons_away:
with patch(
"pyatmo.camera.AsyncCameraData.async_set_persons_away"
) as mock_set_persons_away:
await hass.services.async_call(
"netatmo", SERVICE_SET_PERSON_AWAY, service_data=data
)
@ -171,8 +220,13 @@ async def test_service_set_person_away(hass, camera_entry):
)
async def test_service_set_persons_home(hass, camera_entry):
async def test_service_set_persons_home(hass, config_entry, netatmo_auth):
"""Test service to set persons as home."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done()
data = {
@ -180,7 +234,9 @@ async def test_service_set_persons_home(hass, camera_entry):
"persons": "John Doe",
}
with patch("pyatmo.camera.CameraData.set_persons_home") as mock_set_persons_home:
with patch(
"pyatmo.camera.AsyncCameraData.async_set_persons_home"
) as mock_set_persons_home:
await hass.services.async_call(
"netatmo", SERVICE_SET_PERSONS_HOME, service_data=data
)
@ -191,8 +247,13 @@ async def test_service_set_persons_home(hass, camera_entry):
)
async def test_service_set_camera_light(hass, camera_entry):
async def test_service_set_camera_light(hass, config_entry, netatmo_auth):
"""Test service to set the outdoor camera light mode."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done()
data = {
@ -200,7 +261,7 @@ async def test_service_set_camera_light(hass, camera_entry):
"camera_light_mode": "on",
}
with patch("pyatmo.camera.CameraData.set_state") as mock_set_state:
with patch("pyatmo.camera.AsyncCameraData.async_set_state") as mock_set_state:
await hass.services.async_call(
"netatmo", SERVICE_SET_CAMERA_LIGHT, service_data=data
)
@ -214,16 +275,26 @@ async def test_service_set_camera_light(hass, camera_entry):
async def test_camera_reconnect_webhook(hass, config_entry):
"""Test webhook event on camera reconnect."""
fake_post_hits = 0
async def fake_post(*args, **kwargs):
"""Fake error during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return await fake_post_request(*args, **kwargs)
with patch(
"homeassistant.components.netatmo.api.ConfigEntryNetatmoAuth.post_request"
) as mock_post, patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", ["camera"]
), patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook:
mock_post.side_effect = fake_post_request
mock_auth.return_value.async_post_request.side_effect = fake_post
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
mock_webhook.return_value = "https://example.com"
await hass.config_entries.async_setup(config_entry.entry_id)
@ -238,8 +309,9 @@ async def test_camera_reconnect_webhook(hass, config_entry):
await simulate_webhook(hass, webhook_id, response)
await hass.async_block_till_done()
mock_post.assert_called()
mock_post.reset_mock()
assert fake_post_hits == 5
calls = fake_post_hits
# Fake camera reconnect
response = {
@ -253,11 +325,16 @@ async def test_camera_reconnect_webhook(hass, config_entry):
dt.utcnow() + timedelta(seconds=60),
)
await hass.async_block_till_done()
mock_post.assert_called()
assert fake_post_hits > calls
async def test_webhook_person_event(hass, camera_entry):
async def test_webhook_person_event(hass, config_entry, netatmo_auth):
"""Test that person events are handled."""
with selected_platforms(["camera"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
test_netatmo_event = async_capture_events(hass, NETATMO_EVENT)
assert not test_netatmo_event
@ -282,7 +359,80 @@ async def test_webhook_person_event(hass, camera_entry):
"push_type": "NACamera-person",
}
webhook_id = camera_entry.data[CONF_WEBHOOK_ID]
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
await simulate_webhook(hass, webhook_id, fake_webhook_event)
assert test_netatmo_event
async def test_setup_component_no_devices(hass, config_entry):
"""Test setup with no devices."""
fake_post_hits = 0
async def fake_post_no_data(*args, **kwargs):
"""Fake error during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return "{}"
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", ["camera"]
), patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.async_post_request.side_effect = fake_post_no_data
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert fake_post_hits == 1
async def test_camera_image_raises_exception(hass, config_entry, requests_mock):
"""Test setup with no devices."""
fake_post_hits = 0
async def fake_post(*args, **kwargs):
"""Return fake data."""
nonlocal fake_post_hits
fake_post_hits += 1
if "url" not in kwargs:
return "{}"
endpoint = kwargs["url"].split("/")[-1]
if "snapshot_720.jpg" in endpoint:
raise pyatmo.exceptions.ApiError()
return await fake_post_request(*args, **kwargs)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", ["camera"]
), patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.async_post_request.side_effect = fake_post
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
camera_entity_indoor = "camera.netatmo_hall"
with pytest.raises(Exception) as excinfo:
await camera.async_get_image(hass, camera_entity_indoor)
assert excinfo.value.args == ("Unable to get image",)
assert fake_post_hits == 6

116
tests/components/netatmo/test_climate.py
View file

@ -26,12 +26,17 @@ from homeassistant.components.netatmo.const import (
)
from homeassistant.const import ATTR_ENTITY_ID, ATTR_TEMPERATURE, CONF_WEBHOOK_ID
from .common import simulate_webhook
from .common import selected_platforms, simulate_webhook
async def test_webhook_event_handling_thermostats(hass, climate_entry):
async def test_webhook_event_handling_thermostats(hass, config_entry, netatmo_auth):
"""Test service and webhook event handling with thermostats."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_livingroom = "climate.netatmo_livingroom"
assert hass.states.get(climate_entity_livingroom).state == "auto"
@ -199,9 +204,16 @@ async def test_webhook_event_handling_thermostats(hass, climate_entry):
)
async def test_service_preset_mode_frost_guard_thermostat(hass, climate_entry):
async def test_service_preset_mode_frost_guard_thermostat(
hass, config_entry, netatmo_auth
):
"""Test service with frost guard preset for thermostats."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_livingroom = "climate.netatmo_livingroom"
assert hass.states.get(climate_entity_livingroom).state == "auto"
@ -267,9 +279,14 @@ async def test_service_preset_mode_frost_guard_thermostat(hass, climate_entry):
)
async def test_service_preset_modes_thermostat(hass, climate_entry):
async def test_service_preset_modes_thermostat(hass, config_entry, netatmo_auth):
"""Test service with preset modes for thermostats."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_livingroom = "climate.netatmo_livingroom"
assert hass.states.get(climate_entity_livingroom).state == "auto"
@ -341,10 +358,15 @@ async def test_service_preset_modes_thermostat(hass, climate_entry):
assert hass.states.get(climate_entity_livingroom).attributes["temperature"] == 30
async def test_webhook_event_handling_no_data(hass, climate_entry):
async def test_webhook_event_handling_no_data(hass, config_entry, netatmo_auth):
"""Test service and webhook event handling with erroneous data."""
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# Test webhook without home entry
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
response = {
"push_type": "home_event_changed",
@ -385,14 +407,19 @@ async def test_webhook_event_handling_no_data(hass, climate_entry):
await simulate_webhook(hass, webhook_id, response)
async def test_service_schedule_thermostats(hass, climate_entry, caplog):
async def test_service_schedule_thermostats(hass, config_entry, caplog, netatmo_auth):
"""Test service for selecting Netatmo schedule with thermostats."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_livingroom = "climate.netatmo_livingroom"
# Test setting a valid schedule
with patch(
"pyatmo.thermostat.HomeData.switch_home_schedule"
"pyatmo.thermostat.AsyncHomeData.async_switch_home_schedule"
) as mock_switch_home_schedule:
await hass.services.async_call(
"netatmo",
@ -421,7 +448,7 @@ async def test_service_schedule_thermostats(hass, climate_entry, caplog):
# Test setting an invalid schedule
with patch(
"pyatmo.thermostat.HomeData.switch_home_schedule"
"pyatmo.thermostat.AsyncHomeData.async_switch_home_schedule"
) as mock_switch_home_schedule:
await hass.services.async_call(
"netatmo",
@ -435,9 +462,16 @@ async def test_service_schedule_thermostats(hass, climate_entry, caplog):
assert "summer is not a valid schedule" in caplog.text
async def test_service_preset_mode_already_boost_valves(hass, climate_entry):
async def test_service_preset_mode_already_boost_valves(
hass, config_entry, netatmo_auth
):
"""Test service with boost preset for valves when already in boost mode."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_entrada = "climate.netatmo_entrada"
assert hass.states.get(climate_entity_entrada).state == "auto"
@ -508,9 +542,14 @@ async def test_service_preset_mode_already_boost_valves(hass, climate_entry):
assert hass.states.get(climate_entity_entrada).attributes["temperature"] == 30
async def test_service_preset_mode_boost_valves(hass, climate_entry):
async def test_service_preset_mode_boost_valves(hass, config_entry, netatmo_auth):
"""Test service with boost preset for valves."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_entrada = "climate.netatmo_entrada"
# Test service setting the preset mode to "boost"
@ -553,8 +592,13 @@ async def test_service_preset_mode_boost_valves(hass, climate_entry):
assert hass.states.get(climate_entity_entrada).attributes["temperature"] == 30
async def test_service_preset_mode_invalid(hass, climate_entry, caplog):
async def test_service_preset_mode_invalid(hass, config_entry, caplog, netatmo_auth):
"""Test service with invalid preset."""
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_PRESET_MODE,
@ -566,9 +610,14 @@ async def test_service_preset_mode_invalid(hass, climate_entry, caplog):
assert "Preset mode 'invalid' not available" in caplog.text
async def test_valves_service_turn_off(hass, climate_entry):
async def test_valves_service_turn_off(hass, config_entry, netatmo_auth):
"""Test service turn off for valves."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_entrada = "climate.netatmo_entrada"
# Test turning valve off
@ -606,9 +655,14 @@ async def test_valves_service_turn_off(hass, climate_entry):
assert hass.states.get(climate_entity_entrada).state == "off"
async def test_valves_service_turn_on(hass, climate_entry):
async def test_valves_service_turn_on(hass, config_entry, netatmo_auth):
"""Test service turn on for valves."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_entrada = "climate.netatmo_entrada"
# Test turning valve on
@ -661,9 +715,14 @@ async def test_get_all_home_ids():
assert climate.get_all_home_ids(home_data) == expected
async def test_webhook_home_id_mismatch(hass, climate_entry):
async def test_webhook_home_id_mismatch(hass, config_entry, netatmo_auth):
"""Test service turn on for valves."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_entrada = "climate.netatmo_entrada"
assert hass.states.get(climate_entity_entrada).state == "auto"
@ -694,9 +753,14 @@ async def test_webhook_home_id_mismatch(hass, climate_entry):
assert hass.states.get(climate_entity_entrada).state == "auto"
async def test_webhook_set_point(hass, climate_entry):
async def test_webhook_set_point(hass, config_entry, netatmo_auth):
"""Test service turn on for valves."""
webhook_id = climate_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["climate"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
climate_entity_entrada = "climate.netatmo_entrada"
# Fake backend response for valve being turned on

283
tests/components/netatmo/test_init.py
View file

@ -1,15 +1,26 @@
"""The tests for Netatmo component."""
import asyncio
from datetime import timedelta
from time import time
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pyatmo
from homeassistant import config_entries
from homeassistant.components.netatmo import DOMAIN
from homeassistant.const import CONF_WEBHOOK_ID
from homeassistant.core import CoreState
from homeassistant.setup import async_setup_component
from homeassistant.util import dt
from .common import FAKE_WEBHOOK_ACTIVATION, fake_post_request, simulate_webhook
from .common import (
FAKE_WEBHOOK_ACTIVATION,
fake_post_request,
selected_platforms,
simulate_webhook,
)
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.cloud import mock_cloud
# Fake webhook thermostat mode change to "Max"
@ -57,13 +68,15 @@ async def test_setup_component(hass):
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.ConfigEntryNetatmoAuth"
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook:
mock_auth.return_value.post_request.side_effect = fake_post_request
mock_auth.return_value.async_post_request.side_effect = fake_post_request
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
assert await async_setup_component(hass, "netatmo", {})
await hass.async_block_till_done()
@ -86,38 +99,54 @@ async def test_setup_component(hass):
async def test_setup_component_with_config(hass, config_entry):
"""Test setup of the netatmo component with dev account."""
fake_post_hits = 0
async def fake_post(*args, **kwargs):
"""Fake error during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return await fake_post_request(*args, **kwargs)
with patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook, patch(
"pyatmo.auth.NetatmoOAuth2.post_request"
) as fake_post_requests, patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", ["sensor"]
):
mock_auth.return_value.async_post_request.side_effect = fake_post
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
assert await async_setup_component(
hass, "netatmo", {"netatmo": {"client_id": "123", "client_secret": "abc"}}
)
await hass.async_block_till_done()
fake_post_requests.assert_called()
assert fake_post_hits == 3
mock_impl.assert_called_once()
mock_webhook.assert_called_once()
assert config_entry.state == config_entries.ENTRY_STATE_LOADED
assert hass.config_entries.async_entries(DOMAIN)
assert len(hass.states.async_all()) > 0
assert hass.config_entries.async_entries(DOMAIN)
assert len(hass.states.async_all()) > 0
async def test_setup_component_with_webhook(hass, entry):
async def test_setup_component_with_webhook(hass, config_entry, netatmo_auth):
"""Test setup and teardown of the netatmo component with webhook registration."""
webhook_id = entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["camera", "climate", "light", "sensor"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
assert len(hass.states.async_all()) > 0
webhook_id = entry.data[CONF_WEBHOOK_ID]
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
# Assert webhook is established successfully
@ -134,36 +163,30 @@ async def test_setup_component_with_webhook(hass, entry):
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
async def test_setup_without_https(hass, config_entry):
async def test_setup_without_https(hass, config_entry, caplog):
"""Test if set up with cloud link and without https."""
hass.config.components.add("cloud")
with patch(
"homeassistant.helpers.network.get_url",
return_value="https://example.nabu.casa",
return_value="http://example.nabu.casa",
), patch(
"homeassistant.components.netatmo.api.ConfigEntryNetatmoAuth"
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook:
mock_auth.return_value.post_request.side_effect = fake_post_request
mock_webhook.return_value = "https://example.com"
) as mock_async_generate_url:
mock_auth.return_value.async_post_request.side_effect = fake_post_request
mock_async_generate_url.return_value = "http://example.com"
assert await async_setup_component(
hass, "netatmo", {"netatmo": {"client_id": "123", "client_secret": "abc"}}
)
await hass.async_block_till_done()
await hass.async_block_till_done()
mock_auth.assert_called_once()
mock_async_generate_url.assert_called_once()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
# Assert webhook is established successfully
climate_entity_livingroom = "climate.netatmo_livingroom"
assert hass.states.get(climate_entity_livingroom).state == "auto"
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK)
await hass.async_block_till_done()
assert hass.states.get(climate_entity_livingroom).state == "heat"
assert "https and port 443 is required to register the webhook" in caplog.text
async def test_setup_with_cloud(hass, config_entry):
@ -181,7 +204,7 @@ async def test_setup_with_cloud(hass, config_entry):
) as fake_create_cloudhook, patch(
"homeassistant.components.cloud.async_delete_cloudhook"
) as fake_delete_cloudhook, patch(
"homeassistant.components.netatmo.api.ConfigEntryNetatmoAuth"
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", []
), patch(
@ -189,7 +212,7 @@ async def test_setup_with_cloud(hass, config_entry):
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.post_request.side_effect = fake_post_request
mock_auth.return_value.async_post_request.side_effect = fake_post_request
assert await async_setup_component(
hass, "netatmo", {"netatmo": {"client_id": "123", "client_secret": "abc"}}
)
@ -210,3 +233,199 @@ async def test_setup_with_cloud(hass, config_entry):
await hass.async_block_till_done()
assert not hass.config_entries.async_entries(DOMAIN)
async def test_setup_with_cloudhook(hass):
"""Test if set up with active cloud subscription and cloud hook."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"cloudhook_url": "https://hooks.nabu.casa/ABCD",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
},
},
)
config_entry.add_to_hass(hass)
await mock_cloud(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.cloud.async_is_logged_in", return_value=True
), patch(
"homeassistant.components.cloud.async_active_subscription", return_value=True
), patch(
"homeassistant.components.cloud.async_create_cloudhook",
return_value="https://hooks.nabu.casa/ABCD",
) as fake_create_cloudhook, patch(
"homeassistant.components.cloud.async_delete_cloudhook"
) as fake_delete_cloudhook, patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", []
), patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.async_post_request.side_effect = fake_post_request
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
assert await async_setup_component(hass, "netatmo", {})
assert hass.components.cloud.async_active_subscription() is True
assert (
hass.config_entries.async_entries("netatmo")[0].data["cloudhook_url"]
== "https://hooks.nabu.casa/ABCD"
)
await hass.async_block_till_done()
assert hass.config_entries.async_entries(DOMAIN)
fake_create_cloudhook.assert_not_called()
for config_entry in hass.config_entries.async_entries("netatmo"):
await hass.config_entries.async_remove(config_entry.entry_id)
fake_delete_cloudhook.assert_called_once()
await hass.async_block_till_done()
assert not hass.config_entries.async_entries(DOMAIN)
async def test_setup_component_api_error(hass):
"""Test error on setup of the netatmo component."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
},
},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.async_post_request.side_effect = (
pyatmo.exceptions.ApiError()
)
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
assert await async_setup_component(hass, "netatmo", {})
await hass.async_block_till_done()
mock_auth.assert_called_once()
mock_impl.assert_called_once()
async def test_setup_component_api_timeout(hass):
"""Test timeout on setup of the netatmo component."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
},
},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.async_post_request.side_effect = (
asyncio.exceptions.TimeoutError()
)
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
assert await async_setup_component(hass, "netatmo", {})
await hass.async_block_till_done()
mock_auth.assert_called_once()
mock_impl.assert_called_once()
async def test_setup_component_with_delay(hass, config_entry):
"""Test setup of the netatmo component with delayed startup."""
hass.state = CoreState.not_running
with patch(
"pyatmo.AbstractAsyncAuth.async_addwebhook", side_effect=AsyncMock()
) as mock_addwebhook, patch(
"pyatmo.AbstractAsyncAuth.async_dropwebhook", side_effect=AsyncMock()
) as mock_dropwebhook, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook, patch(
"pyatmo.AbstractAsyncAuth.async_post_request", side_effect=fake_post_request
) as mock_post_request, patch(
"homeassistant.components.netatmo.PLATFORMS", ["light"]
):
assert await async_setup_component(
hass, "netatmo", {"netatmo": {"client_id": "123", "client_secret": "abc"}}
)
await hass.async_block_till_done()
assert mock_post_request.call_count == 5
mock_impl.assert_called_once()
mock_webhook.assert_not_called()
await hass.async_start()
await hass.async_block_till_done()
mock_webhook.assert_called_once()
# Fake webhook activation
await simulate_webhook(
hass, config_entry.data[CONF_WEBHOOK_ID], FAKE_WEBHOOK_ACTIVATION
)
await hass.async_block_till_done()
mock_addwebhook.assert_called_once()
mock_dropwebhook.assert_not_awaited()
async_fire_time_changed(
hass,
dt.utcnow() + timedelta(seconds=60),
)
await hass.async_block_till_done()
assert hass.config_entries.async_entries(DOMAIN)
assert len(hass.states.async_all()) > 0
await hass.async_stop()
mock_dropwebhook.assert_called_once()

58
tests/components/netatmo/test_light.py
View file

@ -1,19 +1,25 @@
"""The tests for Netatmo light."""
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from homeassistant.components.light import (
DOMAIN as LIGHT_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.components.netatmo import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, CONF_WEBHOOK_ID
from .common import FAKE_WEBHOOK_ACTIVATION, simulate_webhook
from .common import FAKE_WEBHOOK_ACTIVATION, selected_platforms, simulate_webhook
async def test_light_setup_and_services(hass, light_entry):
async def test_light_setup_and_services(hass, config_entry, netatmo_auth):
"""Test setup and services."""
webhook_id = light_entry.data[CONF_WEBHOOK_ID]
with selected_platforms(["light"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
# Fake webhook activation
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
@ -45,7 +51,7 @@ async def test_light_setup_and_services(hass, light_entry):
assert hass.states.get(light_entity).state == "on"
# Test turning light off
with patch("pyatmo.camera.CameraData.set_state") as mock_set_state:
with patch("pyatmo.camera.AsyncCameraData.async_set_state") as mock_set_state:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
@ -60,7 +66,7 @@ async def test_light_setup_and_services(hass, light_entry):
)
# Test turning light on
with patch("pyatmo.camera.CameraData.set_state") as mock_set_state:
with patch("pyatmo.camera.AsyncCameraData.async_set_state") as mock_set_state:
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
@ -73,3 +79,43 @@ async def test_light_setup_and_services(hass, light_entry):
camera_id="12:34:56:00:a5:a4",
floodlight="on",
)
async def test_setup_component_no_devices(hass, config_entry):
"""Test setup with no devices."""
fake_post_hits = 0
async def fake_post_request_no_data(*args, **kwargs):
"""Fake error during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return "{}"
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth, patch(
"homeassistant.components.netatmo.PLATFORMS", ["light"]
), patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
), patch(
"homeassistant.components.webhook.async_generate_url"
):
mock_auth.return_value.async_post_request.side_effect = (
fake_post_request_no_data
)
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# Fake webhook activation
await simulate_webhook(
hass, config_entry.data[CONF_WEBHOOK_ID], FAKE_WEBHOOK_ACTIVATION
)
await hass.async_block_till_done()
assert fake_post_hits == 1
assert hass.config_entries.async_entries(DOMAIN)
assert len(hass.states.async_all()) == 0

10
tests/components/netatmo/test_media_source.py
View file

@ -51,6 +51,16 @@ async def test_async_browse_media(hass):
)
assert str(excinfo.value) == "Unknown source directory."
# Test invalid base
with pytest.raises(ValueError) as excinfo:
await media_source.async_browse_media(hass, f"{const.URI_SCHEME}{DOMAIN}/")
assert str(excinfo.value) == "Invalid media source URI"
# Test successful listing
media = await media_source.async_browse_media(
hass, f"{const.URI_SCHEME}{DOMAIN}/events"
)
# Test successful listing
media = await media_source.async_browse_media(
hass, f"{const.URI_SCHEME}{DOMAIN}/events/"

42
tests/components/netatmo/test_sensor.py
View file

@ -1,23 +1,22 @@
"""The tests for the Netatmo sensor platform."""
from datetime import timedelta
from unittest.mock import patch
import pytest
from homeassistant.components.netatmo import sensor
from homeassistant.components.netatmo.sensor import MODULE_TYPE_WIND
from homeassistant.config_entries import RELOAD_AFTER_UPDATE_DELAY
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt
from .common import TEST_TIME
from .conftest import selected_platforms
from tests.common import async_fire_time_changed
from .common import TEST_TIME, selected_platforms
async def test_weather_sensor(hass, sensor_entry):
async def test_weather_sensor(hass, config_entry, netatmo_auth):
"""Test weather sensor setup."""
with patch("time.time", return_value=TEST_TIME), selected_platforms(["sensor"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
prefix = "sensor.netatmo_mystation_"
assert hass.states.get(f"{prefix}temperature").state == "24.6"
@ -26,8 +25,15 @@ async def test_weather_sensor(hass, sensor_entry):
assert hass.states.get(f"{prefix}pressure").state == "1017.3"
async def test_public_weather_sensor(hass, sensor_entry):
async def test_public_weather_sensor(hass, config_entry, netatmo_auth):
"""Test public weather sensor setup."""
with patch("time.time", return_value=TEST_TIME), selected_platforms(["sensor"]):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all()) > 0
prefix = "sensor.netatmo_home_max_"
assert hass.states.get(f"{prefix}temperature").state == "27.4"
@ -40,7 +46,6 @@ async def test_public_weather_sensor(hass, sensor_entry):
assert hass.states.get(f"{prefix}humidity").state == "63.2"
assert hass.states.get(f"{prefix}pressure").state == "1010.3"
assert len(hass.states.async_all()) > 0
entities_before_change = len(hass.states.async_all())
valid_option = {
@ -53,7 +58,7 @@ async def test_public_weather_sensor(hass, sensor_entry):
"mode": "max",
}
result = await hass.config_entries.options.async_init(sensor_entry.entry_id)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={"new_area": "Home avg"}
)
@ -63,18 +68,11 @@ async def test_public_weather_sensor(hass, sensor_entry):
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
async_fire_time_changed(
hass,
dt.utcnow() + timedelta(seconds=RELOAD_AFTER_UPDATE_DELAY + 1),
)
await hass.async_block_till_done()
assert hass.states.get(f"{prefix}temperature").state == "27.4"
assert hass.states.get(f"{prefix}humidity").state == "76"
assert hass.states.get(f"{prefix}pressure").state == "1014.4"
await hass.async_block_till_done()
assert len(hass.states.async_all()) == entities_before_change
assert hass.states.get(f"{prefix}temperature").state == "27.4"
@pytest.mark.parametrize(
@ -213,7 +211,9 @@ async def test_fix_angle(angle, expected):
),
],
)
async def test_weather_sensor_enabling(hass, config_entry, uid, name, expected):
async def test_weather_sensor_enabling(
hass, config_entry, uid, name, expected, netatmo_auth
):
"""Test enabling of by default disabled sensors."""
with patch("time.time", return_value=TEST_TIME), selected_platforms(["sensor"]):
states_before = len(hass.states.async_all())

3
tests/fixtures/netatmo/homestatus.json vendored
View file

@ -27,7 +27,6 @@
"type": "NATherm1",
"firmware_revision": 65,
"rf_strength": 58,
"battery_level": 3793,
"boiler_valve_comfort_boost": false,
"boiler_status": false,
"anticipating": false,
@ -40,7 +39,6 @@
"type": "NRV",
"firmware_revision": 79,
"rf_strength": 51,
"battery_level": 3025,
"bridge": "12:34:56:00:fa:d0",
"battery_state": "full"
},
@ -50,7 +48,6 @@
"type": "NRV",
"firmware_revision": 79,
"rf_strength": 59,
"battery_level": 2329,
"bridge": "12:34:56:00:fa:d0",
"battery_state": "full"
}