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

Use latest withings_api module (#27817)

Browse source 
* Using latest winthings_api module.
Drastically reduced complexity of tests.

* Removing import source.

* Fixing test requirements.

* Using requests_mock instead of responses module.

* Updating file formatting.

* Removing unused method.

* Adding support for new OAuth2 config flow.

* Addressing PR feedback.
Removing unecessary base_url from config, this is a potential breaking change.

* Addressing PR feedback.
This commit is contained in:
Robert Van Gorkom 2019-10-24 09:41:04 -07:00 • committed by Paulus Schoutsen
parent fc09702cc3
commit 15bedd8f27
16 changed files with 998 additions and 1572 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

75
homeassistant/components/withings/__init__.py
View file

@ -4,10 +4,11 @@ Support for the Withings API.
For more details about this platform, please refer to the documentation at
"""
import voluptuous as vol
from withings_api import WithingsAuth
from homeassistant.config_entries import ConfigEntry, SOURCE_IMPORT, SOURCE_USER
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import config_validation as cv, config_entry_oauth2_flow
from . import config_flow, const
from .common import _LOGGER, get_data_manager, NotAuthenticatedError
@ -22,7 +23,6 @@ CONFIG_SCHEMA = vol.Schema(
vol.Required(const.CLIENT_SECRET): vol.All(
cv.string, vol.Length(min=1)
),
vol.Optional(const.BASE_URL): cv.url,
vol.Required(const.PROFILES): vol.All(
cv.ensure_list,
vol.Unique(),
@ -36,50 +36,65 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass: HomeAssistantType, config: ConfigType):
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
"""Set up the Withings component."""
conf = config.get(DOMAIN)
conf = config.get(DOMAIN, {})
if not conf:
return True
hass.data[DOMAIN] = {const.CONFIG: conf}
base_url = conf.get(const.BASE_URL, hass.config.api.base_url).rstrip("/")
hass.http.register_view(config_flow.WithingsAuthCallbackView)
config_flow.register_flow_implementation(
config_flow.WithingsFlowHandler.async_register_implementation(
hass,
conf[const.CLIENT_ID],
conf[const.CLIENT_SECRET],
base_url,
conf[const.PROFILES],
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data={}
)
config_entry_oauth2_flow.LocalOAuth2Implementation(
hass,
const.DOMAIN,
conf[const.CLIENT_ID],
conf[const.CLIENT_SECRET],
f"{WithingsAuth.URL}/oauth2_user/authorize2",
f"{WithingsAuth.URL}/oauth2/token",
),
)
return True
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Set up Withings from a config entry."""
data_manager = get_data_manager(hass, entry)
# Upgrading existing token information to hass managed tokens.
if "auth_implementation" not in entry.data:
_LOGGER.debug("Upgrading existing config entry")
data = entry.data
creds = data.get(const.CREDENTIALS, {})
hass.config_entries.async_update_entry(
entry,
data={
"auth_implementation": const.DOMAIN,
"implementation": const.DOMAIN,
"profile": data.get("profile"),
"token": {
"access_token": creds.get("access_token"),
"refresh_token": creds.get("refresh_token"),
"expires_at": int(creds.get("token_expiry")),
"type": creds.get("token_type"),
"userid": creds.get("userid") or creds.get("user_id"),
},
},
)
implementation = await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
data_manager = get_data_manager(hass, entry, implementation)
_LOGGER.debug("Confirming we're authenticated")
try:
await data_manager.check_authenticated()
except NotAuthenticatedError:
# Trigger new config flow.
hass.async_create_task(
hass.config_entries.flow.async_init(
const.DOMAIN,
context={"source": SOURCE_USER, const.PROFILE: data_manager.profile},
data={},
)
_LOGGER.error(
"Withings auth tokens exired for profile %s, remove and re-add the integration",
data_manager.profile,
)
return False
@ -90,6 +105,6 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
return True
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry):
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Unload Withings config entry."""
return await hass.config_entries.async_forward_entry_unload(entry, "sensor")

191
homeassistant/components/withings/common.py
View file

@ -1,23 +1,36 @@
"""Common code for Withings."""
import datetime
from functools import partial
import logging
import re
import time
from typing import Any, Dict
import withings_api as withings
from oauthlib.oauth2.rfc6749.errors import MissingTokenError
from requests_oauthlib import TokenUpdated
from asyncio import run_coroutine_threadsafe
import requests
from withings_api import (
AbstractWithingsApi,
SleepGetResponse,
MeasureGetMeasResponse,
SleepGetSummaryResponse,
)
from withings_api.common import UnauthorizedException, AuthFailedException
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, PlatformNotReady
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.helpers.config_entry_oauth2_flow import (
AbstractOAuth2Implementation,
OAuth2Session,
)
from homeassistant.util import dt, slugify
from . import const
_LOGGER = logging.getLogger(const.LOG_NAMESPACE)
NOT_AUTHENTICATED_ERROR = re.compile(
".*(Error Code (100|101|102|200|401)|Missing access token parameter).*",
# ".*(Error Code (100|101|102|200|401)|Missing access token parameter).*",
"^401,.*",
re.IGNORECASE,
)
@ -37,40 +50,82 @@ class ServiceError(HomeAssistantError):
class ThrottleData:
"""Throttle data."""
def __init__(self, interval: int, data):
def __init__(self, interval: int, data: Any):
"""Constructor."""
self._time = int(time.time())
self._interval = interval
self._data = data
@property
def time(self):
def time(self) -> int:
"""Get time created."""
return self._time
@property
def interval(self):
def interval(self) -> int:
"""Get interval."""
return self._interval
@property
def data(self):
def data(self) -> Any:
"""Get data."""
return self._data
def is_expired(self):
def is_expired(self) -> bool:
"""Is this data expired."""
return int(time.time()) - self.time > self.interval
class ConfigEntryWithingsApi(AbstractWithingsApi):
"""Withing API that uses HA resources."""
def __init__(
self,
hass: HomeAssistant,
config_entry: ConfigEntry,
implementation: AbstractOAuth2Implementation,
):
"""Initialize object."""
self._hass = hass
self._config_entry = config_entry
self._implementation = implementation
self.session = OAuth2Session(hass, config_entry, implementation)
def _request(
self, path: str, params: Dict[str, Any], method: str = "GET"
) -> Dict[str, Any]:
return run_coroutine_threadsafe(
self.async_do_request(path, params, method), self._hass.loop
).result()
async def async_do_request(
self, path: str, params: Dict[str, Any], method: str = "GET"
) -> Dict[str, Any]:
"""Perform an async request."""
await self.session.async_ensure_token_valid()
response = await self._hass.async_add_executor_job(
partial(
requests.request,
method,
"%s/%s" % (self.URL, path),
params=params,
headers={
"Authorization": "Bearer %s"
% self._config_entry.data["token"]["access_token"]
},
)
)
return response.json()
class WithingsDataManager:
"""A class representing an Withings cloud service connection."""
service_available = None
def __init__(
self, hass: HomeAssistantType, profile: str, api: withings.WithingsApi
):
def __init__(self, hass: HomeAssistant, profile: str, api: ConfigEntryWithingsApi):
"""Constructor."""
self._hass = hass
self._api = api
@ -95,27 +150,27 @@ class WithingsDataManager:
return self._slug
@property
def api(self):
def api(self) -> ConfigEntryWithingsApi:
"""Get the api object."""
return self._api
@property
def measures(self):
def measures(self) -> MeasureGetMeasResponse:
"""Get the current measures data."""
return self._measures
@property
def sleep(self):
def sleep(self) -> SleepGetResponse:
"""Get the current sleep data."""
return self._sleep
@property
def sleep_summary(self):
def sleep_summary(self) -> SleepGetSummaryResponse:
"""Get the current sleep summary data."""
return self._sleep_summary
@staticmethod
def get_throttle_interval():
def get_throttle_interval() -> int:
"""Get the throttle interval."""
return const.THROTTLE_INTERVAL
@ -128,22 +183,26 @@ class WithingsDataManager:
self.throttle_data[domain] = throttle_data
@staticmethod
def print_service_unavailable():
def print_service_unavailable() -> bool:
"""Print the service is unavailable (once) to the log."""
if WithingsDataManager.service_available is not False:
_LOGGER.error("Looks like the service is not available at the moment")
WithingsDataManager.service_available = False
return True
return False
@staticmethod
def print_service_available():
def print_service_available() -> bool:
"""Print the service is available (once) to to the log."""
if WithingsDataManager.service_available is not True:
_LOGGER.info("Looks like the service is available again")
WithingsDataManager.service_available = True
return True
async def call(self, function, is_first_call=True, throttle_domain=None):
return False
async def call(self, function, throttle_domain=None) -> Any:
"""Call an api method and handle the result."""
throttle_data = self.get_throttle_data(throttle_domain)
@ -167,21 +226,12 @@ class WithingsDataManager:
WithingsDataManager.print_service_available()
return result
except TokenUpdated:
WithingsDataManager.print_service_available()
if not is_first_call:
raise ServiceError(
"Stuck in a token update loop. This should never happen"
)
_LOGGER.info("Token updated, re-running call.")
return await self.call(function, False, throttle_domain)
except MissingTokenError as ex:
raise NotAuthenticatedError(ex)
except Exception as ex: # pylint: disable=broad-except
# Service error, probably not authenticated.
# Withings api encountered error.
if isinstance(ex, (UnauthorizedException, AuthFailedException)):
raise NotAuthenticatedError(ex)
# Oauth2 config flow failed to authenticate.
if NOT_AUTHENTICATED_ERROR.match(str(ex)):
raise NotAuthenticatedError(ex)
@ -189,37 +239,37 @@ class WithingsDataManager:
WithingsDataManager.print_service_unavailable()
raise PlatformNotReady(ex)
async def check_authenticated(self):
async def check_authenticated(self) -> bool:
"""Check if the user is authenticated."""
def function():
return self._api.request("user", "getdevice", version="v2")
return bool(self._api.user_get_device())
return await self.call(function)
async def update_measures(self):
async def update_measures(self) -> MeasureGetMeasResponse:
"""Update the measures data."""
def function():
return self._api.get_measures()
return self._api.measure_get_meas()
self._measures = await self.call(function, throttle_domain="update_measures")
return self._measures
async def update_sleep(self):
async def update_sleep(self) -> SleepGetResponse:
"""Update the sleep data."""
end_date = int(time.time())
start_date = end_date - (6 * 60 * 60)
def function():
return self._api.get_sleep(startdate=start_date, enddate=end_date)
return self._api.sleep_get(startdate=start_date, enddate=end_date)
self._sleep = await self.call(function, throttle_domain="update_sleep")
return self._sleep
async def update_sleep_summary(self):
async def update_sleep_summary(self) -> SleepGetSummaryResponse:
"""Update the sleep summary data."""
now = dt.utcnow()
yesterday = now - datetime.timedelta(days=1)
@ -240,7 +290,7 @@ class WithingsDataManager:
)
def function():
return self._api.get_sleep_summary(lastupdate=yesterday_noon.timestamp())
return self._api.sleep_get_summary(lastupdate=yesterday_noon)
self._sleep_summary = await self.call(
function, throttle_domain="update_sleep_summary"
@ -250,36 +300,16 @@ class WithingsDataManager:
def create_withings_data_manager(
hass: HomeAssistantType, entry: ConfigEntry
hass: HomeAssistant,
config_entry: ConfigEntry,
implementation: AbstractOAuth2Implementation,
) -> WithingsDataManager:
"""Set up the sensor config entry."""
entry_creds = entry.data.get(const.CREDENTIALS) or {}
profile = entry.data[const.PROFILE]
credentials = withings.WithingsCredentials(
entry_creds.get("access_token"),
entry_creds.get("token_expiry"),
entry_creds.get("token_type"),
entry_creds.get("refresh_token"),
entry_creds.get("user_id"),
entry_creds.get("client_id"),
entry_creds.get("consumer_secret"),
)
def credentials_saver(credentials_param):
_LOGGER.debug("Saving updated credentials of type %s", type(credentials_param))
# Sanitizing the data as sometimes a WithingsCredentials object
# is passed through from the API.
cred_data = credentials_param
if not isinstance(credentials_param, dict):
cred_data = credentials_param.__dict__
entry.data[const.CREDENTIALS] = cred_data
hass.config_entries.async_update_entry(entry, data={**entry.data})
profile = config_entry.data.get(const.PROFILE)
_LOGGER.debug("Creating withings api instance")
api = withings.WithingsApi(
credentials, refresh_cb=(lambda token: credentials_saver(api.credentials))
api = ConfigEntryWithingsApi(
hass=hass, config_entry=config_entry, implementation=implementation
)
_LOGGER.debug("Creating withings data manager for profile: %s", profile)
@ -287,24 +317,25 @@ def create_withings_data_manager(
def get_data_manager(
hass: HomeAssistantType, entry: ConfigEntry
hass: HomeAssistant,
entry: ConfigEntry,
implementation: AbstractOAuth2Implementation,
) -> WithingsDataManager:
"""Get a data manager for a config entry.
If the data manager doesn't exist yet, it will be
created and cached for later use.
"""
profile = entry.data.get(const.PROFILE)
entry_id = entry.entry_id
if not hass.data.get(const.DOMAIN):
hass.data[const.DOMAIN] = {}
hass.data[const.DOMAIN] = hass.data.get(const.DOMAIN, {})
if not hass.data[const.DOMAIN].get(const.DATA_MANAGER):
hass.data[const.DOMAIN][const.DATA_MANAGER] = {}
domain_dict = hass.data[const.DOMAIN]
domain_dict[const.DATA_MANAGER] = domain_dict.get(const.DATA_MANAGER, {})
if not hass.data[const.DOMAIN][const.DATA_MANAGER].get(profile):
hass.data[const.DOMAIN][const.DATA_MANAGER][
profile
] = create_withings_data_manager(hass, entry)
dm_dict = domain_dict[const.DATA_MANAGER]
dm_dict[entry_id] = dm_dict.get(entry_id) or create_withings_data_manager(
hass, entry, implementation
)
return hass.data[const.DOMAIN][const.DATA_MANAGER][profile]
return dm_dict[entry_id]

210
homeassistant/components/withings/config_flow.py
View file

@ -1,192 +1,64 @@
"""Config flow for Withings."""
from collections import OrderedDict
import logging
from typing import Optional
import aiohttp
import withings_api as withings
import voluptuous as vol
from withings_api.common import AuthScope
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.http import HomeAssistantView
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from . import const
DATA_FLOW_IMPL = "withings_flow_implementation"
from homeassistant import config_entries
from homeassistant.components.withings import const
from homeassistant.helpers import config_entry_oauth2_flow
_LOGGER = logging.getLogger(__name__)
@callback
def register_flow_implementation(hass, client_id, client_secret, base_url, profiles):
"""Register a flow implementation.
hass: Home assistant object.
client_id: Client id.
client_secret: Client secret.
base_url: Base url of home assistant instance.
profiles: The profiles to work with.
"""
if DATA_FLOW_IMPL not in hass.data:
hass.data[DATA_FLOW_IMPL] = OrderedDict()
hass.data[DATA_FLOW_IMPL] = {
const.CLIENT_ID: client_id,
const.CLIENT_SECRET: client_secret,
const.BASE_URL: base_url,
const.PROFILES: profiles,
}
@config_entries.HANDLERS.register(const.DOMAIN)
class WithingsFlowHandler(config_entries.ConfigFlow):
class WithingsFlowHandler(config_entry_oauth2_flow.AbstractOAuth2FlowHandler):
"""Handle a config flow."""
VERSION = 1
DOMAIN = const.DOMAIN
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
_current_data = None
def __init__(self):
"""Initialize flow."""
self.flow_profile = None
self.data = None
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
def async_profile_config_entry(self, profile: str) -> Optional[ConfigEntry]:
"""Get a profile config entry."""
entries = self.hass.config_entries.async_entries(const.DOMAIN)
for entry in entries:
if entry.data.get(const.PROFILE) == profile:
return entry
@property
def extra_authorize_data(self) -> dict:
"""Extra data that needs to be appended to the authorize url."""
return {
"scope": ",".join(
[
AuthScope.USER_INFO.value,
AuthScope.USER_METRICS.value,
AuthScope.USER_ACTIVITY.value,
]
)
}
return None
async def async_oauth_create_entry(self, data: dict) -> dict:
"""Override the create entry so user can select a profile."""
self._current_data = data
return await self.async_step_profile(data)
def get_auth_client(self, profile: str):
"""Get a new auth client."""
flow = self.hass.data[DATA_FLOW_IMPL]
client_id = flow[const.CLIENT_ID]
client_secret = flow[const.CLIENT_SECRET]
base_url = flow[const.BASE_URL].rstrip("/")
async def async_step_profile(self, data: dict) -> dict:
"""Prompt the user to select a user profile."""
profile = data.get(const.PROFILE)
callback_uri = "{}/{}?flow_id={}&profile={}".format(
base_url.rstrip("/"),
const.AUTH_CALLBACK_PATH.lstrip("/"),
self.flow_id,
profile,
)
return withings.WithingsAuth(
client_id,
client_secret,
callback_uri,
scope=",".join(["user.info", "user.metrics", "user.activity"]),
)
async def async_step_import(self, user_input=None):
"""Create user step."""
return await self.async_step_user(user_input)
async def async_step_user(self, user_input=None):
"""Create an entry for selecting a profile."""
flow = self.hass.data.get(DATA_FLOW_IMPL)
if not flow:
return self.async_abort(reason="no_flows")
if user_input:
return await self.async_step_auth(user_input)
if profile:
new_data = {**self._current_data, **{const.PROFILE: profile}}
self._current_data = None
return await self.async_step_finish(new_data)
profiles = self.hass.data[const.DOMAIN][const.CONFIG][const.PROFILES]
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(const.PROFILE): vol.In(flow.get(const.PROFILES))}
),
step_id="profile",
data_schema=vol.Schema({vol.Required(const.PROFILE): vol.In(profiles)}),
)
async def async_step_auth(self, user_input=None):
"""Create an entry for auth."""
if user_input.get(const.CODE):
self.data = user_input
return self.async_external_step_done(next_step_id="finish")
async def async_step_finish(self, data: dict) -> dict:
"""Finish the flow."""
self._current_data = None
profile = user_input.get(const.PROFILE)
auth_client = self.get_auth_client(profile)
url = auth_client.get_authorize_url()
return self.async_external_step(step_id="auth", url=url)
async def async_step_finish(self, user_input=None):
"""Received code for authentication."""
data = user_input or self.data or {}
_LOGGER.debug(
"Should close all flows below %s",
self.hass.config_entries.flow.async_progress(),
)
profile = data[const.PROFILE]
code = data[const.CODE]
return await self._async_create_session(profile, code)
async def _async_create_session(self, profile, code):
"""Create withings session and entries."""
auth_client = self.get_auth_client(profile)
_LOGGER.debug("Requesting credentials with code: %s.", code)
credentials = auth_client.get_credentials(code)
return self.async_create_entry(
title=profile,
data={const.PROFILE: profile, const.CREDENTIALS: credentials.__dict__},
)
class WithingsAuthCallbackView(HomeAssistantView):
"""Withings Authorization Callback View."""
requires_auth = False
url = const.AUTH_CALLBACK_PATH
name = const.AUTH_CALLBACK_NAME
def __init__(self):
"""Constructor."""
async def get(self, request):
"""Receive authorization code."""
hass = request.app["hass"]
code = request.query.get("code")
profile = request.query.get("profile")
flow_id = request.query.get("flow_id")
if not flow_id:
return aiohttp.web_response.Response(
status=400, text="'flow_id' argument not provided in url."
)
if not profile:
return aiohttp.web_response.Response(
status=400, text="'profile' argument not provided in url."
)
if not code:
return aiohttp.web_response.Response(
status=400, text="'code' argument not provided in url."
)
try:
await hass.config_entries.flow.async_configure(
flow_id, {const.PROFILE: profile, const.CODE: code}
)
return aiohttp.web_response.Response(
status=200,
headers={"content-type": "text/html"},
text="<script>window.close()</script>",
)
except data_entry_flow.UnknownFlow:
return aiohttp.web_response.Response(status=400, text="Unknown flow")
return self.async_create_entry(title=data[const.PROFILE], data=data)

35
homeassistant/components/withings/const.py
View file

@ -19,6 +19,7 @@ AUTH_CALLBACK_PATH = "/api/withings/authorize"
AUTH_CALLBACK_NAME = "withings:authorize"
THROTTLE_INTERVAL = 60
SCAN_INTERVAL = 60
STATE_UNKNOWN = const.STATE_UNKNOWN
STATE_AWAKE = "awake"
@ -26,40 +27,6 @@ STATE_DEEP = "deep"
STATE_LIGHT = "light"
STATE_REM = "rem"
MEASURE_TYPE_BODY_TEMP = 71
MEASURE_TYPE_BONE_MASS = 88
MEASURE_TYPE_DIASTOLIC_BP = 9
MEASURE_TYPE_FAT_MASS = 8
MEASURE_TYPE_FAT_MASS_FREE = 5
MEASURE_TYPE_FAT_RATIO = 6
MEASURE_TYPE_HEART_PULSE = 11
MEASURE_TYPE_HEIGHT = 4
MEASURE_TYPE_HYDRATION = 77
MEASURE_TYPE_MUSCLE_MASS = 76
MEASURE_TYPE_PWV = 91
MEASURE_TYPE_SKIN_TEMP = 73
MEASURE_TYPE_SLEEP_DEEP_DURATION = "deepsleepduration"
MEASURE_TYPE_SLEEP_HEART_RATE_AVERAGE = "hr_average"
MEASURE_TYPE_SLEEP_HEART_RATE_MAX = "hr_max"
MEASURE_TYPE_SLEEP_HEART_RATE_MIN = "hr_min"
MEASURE_TYPE_SLEEP_LIGHT_DURATION = "lightsleepduration"
MEASURE_TYPE_SLEEP_REM_DURATION = "remsleepduration"
MEASURE_TYPE_SLEEP_RESPIRATORY_RATE_AVERAGE = "rr_average"
MEASURE_TYPE_SLEEP_RESPIRATORY_RATE_MAX = "rr_max"
MEASURE_TYPE_SLEEP_RESPIRATORY_RATE_MIN = "rr_min"
MEASURE_TYPE_SLEEP_STATE_AWAKE = 0
MEASURE_TYPE_SLEEP_STATE_DEEP = 2
MEASURE_TYPE_SLEEP_STATE_LIGHT = 1
MEASURE_TYPE_SLEEP_STATE_REM = 3
MEASURE_TYPE_SLEEP_TOSLEEP_DURATION = "durationtosleep"
MEASURE_TYPE_SLEEP_TOWAKEUP_DURATION = "durationtowakeup"
MEASURE_TYPE_SLEEP_WAKEUP_DURATION = "wakeupduration"
MEASURE_TYPE_SLEEP_WAKUP_COUNT = "wakeupcount"
MEASURE_TYPE_SPO2 = 54
MEASURE_TYPE_SYSTOLIC_BP = 10
MEASURE_TYPE_TEMP = 12
MEASURE_TYPE_WEIGHT = 1
MEAS_BODY_TEMP_C = "body_temperature_c"
MEAS_BONE_MASS_KG = "bone_mass_kg"
MEAS_DIASTOLIC_MMHG = "diastolic_blood_pressure_mmhg"

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

@ -4,7 +4,7 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/withings",
"requirements": [
"withings-api==2.0.0b8"
"withings-api==2.1.2"
],
"dependencies": [
"api",

262
homeassistant/components/withings/sensor.py
View file

@ -1,10 +1,22 @@
"""Sensors flow for Withings."""
import typing as types
from typing import Callable, List, Union
from withings_api.common import (
MeasureType,
GetSleepSummaryField,
MeasureGetMeasResponse,
SleepGetResponse,
SleepGetSummaryResponse,
get_measure_value,
MeasureGroupAttribs,
SleepState,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import slugify
from homeassistant.helpers import config_entry_oauth2_flow
from . import const
from .common import _LOGGER, WithingsDataManager, get_data_manager
@ -16,57 +28,22 @@ PARALLEL_UPDATES = 1
async def async_setup_entry(
hass: HomeAssistantType,
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: types.Callable[[types.List[Entity], bool], None],
):
async_add_entities: Callable[[List[Entity], bool], None],
) -> None:
"""Set up the sensor config entry."""
data_manager = get_data_manager(hass, entry)
entities = create_sensor_entities(data_manager)
implementation = await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
data_manager = get_data_manager(hass, entry, implementation)
user_id = entry.data["token"]["userid"]
entities = create_sensor_entities(data_manager, user_id)
async_add_entities(entities, True)
def get_measures():
"""Get all the measures.
This function exists to be easily mockable so we can test
one measure at a time. This becomes necessary when integration
testing throttle functionality in the data manager.
"""
return list(WITHINGS_MEASUREMENTS_MAP)
def create_sensor_entities(data_manager: WithingsDataManager):
"""Create sensor entities."""
entities = []
measures = get_measures()
for attribute in WITHINGS_ATTRIBUTES:
if attribute.measurement not in measures:
_LOGGER.debug(
"Skipping measurement %s as it is not in the"
"list of measurements to use",
attribute.measurement,
)
continue
_LOGGER.debug(
"Creating entity for measurement: %s, measure_type: %s,"
"friendly_name: %s, unit_of_measurement: %s",
attribute.measurement,
attribute.measure_type,
attribute.friendly_name,
attribute.unit_of_measurement,
)
entity = WithingsHealthSensor(data_manager, attribute)
entities.append(entity)
return entities
class WithingsAttribute:
"""Base class for modeling withing data."""
@ -107,104 +84,104 @@ class WithingsSleepSummaryAttribute(WithingsAttribute):
WITHINGS_ATTRIBUTES = [
WithingsMeasureAttribute(
const.MEAS_WEIGHT_KG,
const.MEASURE_TYPE_WEIGHT,
MeasureType.WEIGHT,
"Weight",
const.UOM_MASS_KG,
"mdi:weight-kilogram",
),
WithingsMeasureAttribute(
const.MEAS_FAT_MASS_KG,
const.MEASURE_TYPE_FAT_MASS,
MeasureType.FAT_MASS_WEIGHT,
"Fat Mass",
const.UOM_MASS_KG,
"mdi:weight-kilogram",
),
WithingsMeasureAttribute(
const.MEAS_FAT_FREE_MASS_KG,
const.MEASURE_TYPE_FAT_MASS_FREE,
MeasureType.FAT_FREE_MASS,
"Fat Free Mass",
const.UOM_MASS_KG,
"mdi:weight-kilogram",
),
WithingsMeasureAttribute(
const.MEAS_MUSCLE_MASS_KG,
const.MEASURE_TYPE_MUSCLE_MASS,
MeasureType.MUSCLE_MASS,
"Muscle Mass",
const.UOM_MASS_KG,
"mdi:weight-kilogram",
),
WithingsMeasureAttribute(
const.MEAS_BONE_MASS_KG,
const.MEASURE_TYPE_BONE_MASS,
MeasureType.BONE_MASS,
"Bone Mass",
const.UOM_MASS_KG,
"mdi:weight-kilogram",
),
WithingsMeasureAttribute(
const.MEAS_HEIGHT_M,
const.MEASURE_TYPE_HEIGHT,
MeasureType.HEIGHT,
"Height",
const.UOM_LENGTH_M,
"mdi:ruler",
),
WithingsMeasureAttribute(
const.MEAS_TEMP_C,
const.MEASURE_TYPE_TEMP,
MeasureType.TEMPERATURE,
"Temperature",
const.UOM_TEMP_C,
"mdi:thermometer",
),
WithingsMeasureAttribute(
const.MEAS_BODY_TEMP_C,
const.MEASURE_TYPE_BODY_TEMP,
MeasureType.BODY_TEMPERATURE,
"Body Temperature",
const.UOM_TEMP_C,
"mdi:thermometer",
),
WithingsMeasureAttribute(
const.MEAS_SKIN_TEMP_C,
const.MEASURE_TYPE_SKIN_TEMP,
MeasureType.SKIN_TEMPERATURE,
"Skin Temperature",
const.UOM_TEMP_C,
"mdi:thermometer",
),
WithingsMeasureAttribute(
const.MEAS_FAT_RATIO_PCT,
const.MEASURE_TYPE_FAT_RATIO,
MeasureType.FAT_RATIO,
"Fat Ratio",
const.UOM_PERCENT,
None,
),
WithingsMeasureAttribute(
const.MEAS_DIASTOLIC_MMHG,
const.MEASURE_TYPE_DIASTOLIC_BP,
MeasureType.DIASTOLIC_BLOOD_PRESSURE,
"Diastolic Blood Pressure",
const.UOM_MMHG,
None,
),
WithingsMeasureAttribute(
const.MEAS_SYSTOLIC_MMGH,
const.MEASURE_TYPE_SYSTOLIC_BP,
MeasureType.SYSTOLIC_BLOOD_PRESSURE,
"Systolic Blood Pressure",
const.UOM_MMHG,
None,
),
WithingsMeasureAttribute(
const.MEAS_HEART_PULSE_BPM,
const.MEASURE_TYPE_HEART_PULSE,
MeasureType.HEART_RATE,
"Heart Pulse",
const.UOM_BEATS_PER_MINUTE,
"mdi:heart-pulse",
),
WithingsMeasureAttribute(
const.MEAS_SPO2_PCT, const.MEASURE_TYPE_SPO2, "SP02", const.UOM_PERCENT, None
const.MEAS_SPO2_PCT, MeasureType.SP02, "SP02", const.UOM_PERCENT, None
),
WithingsMeasureAttribute(
const.MEAS_HYDRATION, const.MEASURE_TYPE_HYDRATION, "Hydration", "", "mdi:water"
const.MEAS_HYDRATION, MeasureType.HYDRATION, "Hydration", "", "mdi:water"
),
WithingsMeasureAttribute(
const.MEAS_PWV,
const.MEASURE_TYPE_PWV,
MeasureType.PULSE_WAVE_VELOCITY,
"Pulse Wave Velocity",
const.UOM_METERS_PER_SECOND,
None,
@ -214,91 +191,91 @@ WITHINGS_ATTRIBUTES = [
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_WAKEUP_DURATION_SECONDS,
const.MEASURE_TYPE_SLEEP_WAKEUP_DURATION,
GetSleepSummaryField.WAKEUP_DURATION.value,
"Wakeup time",
const.UOM_SECONDS,
"mdi:sleep-off",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_LIGHT_DURATION_SECONDS,
const.MEASURE_TYPE_SLEEP_LIGHT_DURATION,
GetSleepSummaryField.LIGHT_SLEEP_DURATION.value,
"Light sleep",
const.UOM_SECONDS,
"mdi:sleep",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_DEEP_DURATION_SECONDS,
const.MEASURE_TYPE_SLEEP_DEEP_DURATION,
GetSleepSummaryField.DEEP_SLEEP_DURATION.value,
"Deep sleep",
const.UOM_SECONDS,
"mdi:sleep",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_REM_DURATION_SECONDS,
const.MEASURE_TYPE_SLEEP_REM_DURATION,
GetSleepSummaryField.REM_SLEEP_DURATION.value,
"REM sleep",
const.UOM_SECONDS,
"mdi:sleep",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_WAKEUP_COUNT,
const.MEASURE_TYPE_SLEEP_WAKUP_COUNT,
GetSleepSummaryField.WAKEUP_COUNT.value,
"Wakeup count",
const.UOM_FREQUENCY,
"mdi:sleep-off",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_TOSLEEP_DURATION_SECONDS,
const.MEASURE_TYPE_SLEEP_TOSLEEP_DURATION,
GetSleepSummaryField.DURATION_TO_SLEEP.value,
"Time to sleep",
const.UOM_SECONDS,
"mdi:sleep",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_TOWAKEUP_DURATION_SECONDS,
const.MEASURE_TYPE_SLEEP_TOWAKEUP_DURATION,
GetSleepSummaryField.DURATION_TO_WAKEUP.value,
"Time to wakeup",
const.UOM_SECONDS,
"mdi:sleep-off",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_HEART_RATE_AVERAGE,
const.MEASURE_TYPE_SLEEP_HEART_RATE_AVERAGE,
GetSleepSummaryField.HR_AVERAGE.value,
"Average heart rate",
const.UOM_BEATS_PER_MINUTE,
"mdi:heart-pulse",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_HEART_RATE_MIN,
const.MEASURE_TYPE_SLEEP_HEART_RATE_MIN,
GetSleepSummaryField.HR_MIN.value,
"Minimum heart rate",
const.UOM_BEATS_PER_MINUTE,
"mdi:heart-pulse",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_HEART_RATE_MAX,
const.MEASURE_TYPE_SLEEP_HEART_RATE_MAX,
GetSleepSummaryField.HR_MAX.value,
"Maximum heart rate",
const.UOM_BEATS_PER_MINUTE,
"mdi:heart-pulse",
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_RESPIRATORY_RATE_AVERAGE,
const.MEASURE_TYPE_SLEEP_RESPIRATORY_RATE_AVERAGE,
GetSleepSummaryField.RR_AVERAGE.value,
"Average respiratory rate",
const.UOM_BREATHS_PER_MINUTE,
None,
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_RESPIRATORY_RATE_MIN,
const.MEASURE_TYPE_SLEEP_RESPIRATORY_RATE_MIN,
GetSleepSummaryField.RR_MIN.value,
"Minimum respiratory rate",
const.UOM_BREATHS_PER_MINUTE,
None,
),
WithingsSleepSummaryAttribute(
const.MEAS_SLEEP_RESPIRATORY_RATE_MAX,
const.MEASURE_TYPE_SLEEP_RESPIRATORY_RATE_MAX,
GetSleepSummaryField.RR_MAX.value,
"Maximum respiratory rate",
const.UOM_BREATHS_PER_MINUTE,
None,
@ -312,7 +289,10 @@ class WithingsHealthSensor(Entity):
"""Implementation of a Withings sensor."""
def __init__(
self, data_manager: WithingsDataManager, attribute: WithingsAttribute
self,
data_manager: WithingsDataManager,
attribute: WithingsAttribute,
user_id: str,
) -> None:
"""Initialize the Withings sensor."""
self._data_manager = data_manager
@ -320,7 +300,7 @@ class WithingsHealthSensor(Entity):
self._state = None
self._slug = self._data_manager.slug
self._user_id = self._data_manager.api.get_credentials().user_id
self._user_id = user_id
@property
def name(self) -> str:
@ -335,7 +315,7 @@ class WithingsHealthSensor(Entity):
)
@property
def state(self):
def state(self) -> Union[str, int, float, None]:
"""Return the state of the sensor."""
return self._state
@ -350,7 +330,7 @@ class WithingsHealthSensor(Entity):
return self._attribute.icon
@property
def device_state_attributes(self):
def device_state_attributes(self) -> None:
"""Get withings attributes."""
return self._attribute.__dict__
@ -378,71 +358,45 @@ class WithingsHealthSensor(Entity):
await self._data_manager.update_sleep_summary()
await self.async_update_sleep_summary(self._data_manager.sleep_summary)
async def async_update_measure(self, data) -> None:
async def async_update_measure(self, data: MeasureGetMeasResponse) -> None:
"""Update the measures data."""
if data is None:
_LOGGER.error("Provided data is None. Setting state to %s", None)
self._state = None
return
measure_type = self._attribute.measure_type
_LOGGER.debug(
"Finding the unambiguous measure group with measure_type: %s", measure_type
)
measure_groups = [
g
for g in data
if (not g.is_ambiguous() and g.get_measure(measure_type) is not None)
]
if not measure_groups:
_LOGGER.debug("No measure groups found, setting state to %s", None)
value = get_measure_value(data, measure_type, MeasureGroupAttribs.UNAMBIGUOUS)
if value is None:
_LOGGER.debug("Could not find a value, setting state to %s", None)
self._state = None
return
_LOGGER.debug(
"Sorting list of %s measure groups by date created (DESC)",
len(measure_groups),
)
measure_groups.sort(key=(lambda g: g.created), reverse=True)
self._state = round(value, 2)
self._state = round(measure_groups[0].get_measure(measure_type), 4)
async def async_update_sleep_state(self, data) -> None:
async def async_update_sleep_state(self, data: SleepGetResponse) -> None:
"""Update the sleep state data."""
if data is None:
_LOGGER.error("Provided data is None. Setting state to %s", None)
self._state = None
return
if not data.series:
_LOGGER.debug("No sleep data, setting state to %s", None)
self._state = None
return
series = sorted(data.series, key=lambda o: o.enddate, reverse=True)
serie = data.series[len(data.series) - 1]
state = None
if serie.state == SleepState.AWAKE:
state = const.STATE_AWAKE
elif serie.state == SleepState.LIGHT:
state = const.STATE_LIGHT
elif serie.state == SleepState.DEEP:
state = const.STATE_DEEP
elif serie.state == SleepState.REM:
state = const.STATE_REM
serie = series[0]
self._state = state
if serie.state == const.MEASURE_TYPE_SLEEP_STATE_AWAKE:
self._state = const.STATE_AWAKE
elif serie.state == const.MEASURE_TYPE_SLEEP_STATE_LIGHT:
self._state = const.STATE_LIGHT
elif serie.state == const.MEASURE_TYPE_SLEEP_STATE_DEEP:
self._state = const.STATE_DEEP
elif serie.state == const.MEASURE_TYPE_SLEEP_STATE_REM:
self._state = const.STATE_REM
else:
self._state = None
async def async_update_sleep_summary(self, data) -> None:
async def async_update_sleep_summary(self, data: SleepGetSummaryResponse) -> None:
"""Update the sleep summary data."""
if data is None:
_LOGGER.error("Provided data is None. Setting state to %s", None)
self._state = None
return
if not data.series:
_LOGGER.debug("Sleep data has no series, setting state to %s", None)
self._state = None
@ -454,7 +408,59 @@ class WithingsHealthSensor(Entity):
_LOGGER.debug("Determining total value for: %s", measurement)
total = 0
for serie in data.series:
if hasattr(serie, measure_type):
total += getattr(serie, measure_type)
data = serie.data
value = 0
if measure_type == GetSleepSummaryField.REM_SLEEP_DURATION.value:
value = data.remsleepduration
elif measure_type == GetSleepSummaryField.WAKEUP_DURATION.value:
value = data.wakeupduration
elif measure_type == GetSleepSummaryField.LIGHT_SLEEP_DURATION.value:
value = data.lightsleepduration
elif measure_type == GetSleepSummaryField.DEEP_SLEEP_DURATION.value:
value = data.deepsleepduration
elif measure_type == GetSleepSummaryField.WAKEUP_COUNT.value:
value = data.wakeupcount
elif measure_type == GetSleepSummaryField.DURATION_TO_SLEEP.value:
value = data.durationtosleep
elif measure_type == GetSleepSummaryField.DURATION_TO_WAKEUP.value:
value = data.durationtowakeup
elif measure_type == GetSleepSummaryField.HR_AVERAGE.value:
value = data.hr_average
elif measure_type == GetSleepSummaryField.HR_MIN.value:
value = data.hr_min
elif measure_type == GetSleepSummaryField.HR_MAX.value:
value = data.hr_max
elif measure_type == GetSleepSummaryField.RR_AVERAGE.value:
value = data.rr_average
elif measure_type == GetSleepSummaryField.RR_MIN.value:
value = data.rr_min
elif measure_type == GetSleepSummaryField.RR_MAX.value:
value = data.rr_max
# Sometimes a None is provided for value, default to 0.
total += value or 0
self._state = round(total, 4)
def create_sensor_entities(
data_manager: WithingsDataManager, user_id: str
) -> List[WithingsHealthSensor]:
"""Create sensor entities."""
entities = []
for attribute in WITHINGS_ATTRIBUTES:
_LOGGER.debug(
"Creating entity for measurement: %s, measure_type: %s,"
"friendly_name: %s, unit_of_measurement: %s",
attribute.measurement,
attribute.measure_type,
attribute.friendly_name,
attribute.unit_of_measurement,
)
entity = WithingsHealthSensor(data_manager, attribute, user_id)
entities.append(entity)
return entities

10
homeassistant/components/withings/strings.json
View file

@ -2,19 +2,13 @@
"config": {
"title": "Withings",
"step": {
"user": {
"profile": {
"title": "User Profile.",
"description": "Select a user profile to which you want Home Assistant to map with a Withings profile. On the withings page, be sure to select the same user or data will not be labeled correctly.",
"description": "Which profile did you select on the Withings website? It's important the profiles match, otherwise data will be mis-labeled.",
"data": {
"profile": "Profile"
}
}
},
"create_entry": {
"default": "Successfully authenticated with Withings for the selected profile."
},
"abort": {
"no_flows": "You need to configure Withings before being able to authenticate with it. Please read the documentation."
}
}
}

2
requirements_all.txt
View file

@ -1984,7 +1984,7 @@ websockets==6.0
wirelesstagpy==0.4.0
# homeassistant.components.withings
withings-api==2.0.0b8
withings-api==2.1.2
# homeassistant.components.wunderlist
wunderpy2==0.1.6

1
requirements_test.txt
View file

@ -20,3 +20,4 @@ pytest-sugar==0.9.2
pytest-timeout==1.3.3
pytest==5.2.1
requests_mock==1.7.0
responses==0.10.6

3
requirements_test_all.txt
View file

@ -21,6 +21,7 @@ pytest-sugar==0.9.2
pytest-timeout==1.3.3
pytest==5.2.1
requests_mock==1.7.0
responses==0.10.6
# homeassistant.components.homekit
@ -629,7 +630,7 @@ watchdog==0.8.3
websockets==6.0
# homeassistant.components.withings
withings-api==2.0.0b8
withings-api==2.1.2
# homeassistant.components.bluesound
# homeassistant.components.startca

534
tests/components/withings/common.py
View file

@ -1,213 +1,383 @@
"""Common data for for the withings component tests."""
import re
import time
from typing import List
import withings_api as withings
import requests_mock
from withings_api import AbstractWithingsApi
from withings_api.common import (
MeasureGetMeasGroupAttrib,
MeasureGetMeasGroupCategory,
MeasureType,
SleepModel,
SleepState,
)
from homeassistant import data_entry_flow
import homeassistant.components.api as api
import homeassistant.components.http as http
import homeassistant.components.withings.const as const
from homeassistant.config import async_process_ha_core_config
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_UNIT_SYSTEM, CONF_UNIT_SYSTEM_METRIC
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.setup import async_setup_component
from homeassistant.util import slugify
def new_sleep_data(model, series):
"""Create simple dict to simulate api data."""
return {"series": series, "model": model}
def get_entity_id(measure, profile) -> str:
"""Get an entity id for a measure and profile."""
return "sensor.{}_{}_{}".format(const.DOMAIN, measure, slugify(profile))
def new_sleep_data_serie(startdate, enddate, state):
"""Create simple dict to simulate api data."""
return {"startdate": startdate, "enddate": enddate, "state": state}
def assert_state_equals(
hass: HomeAssistant, profile: str, measure: str, expected
) -> None:
"""Assert the state of a withings sensor."""
entity_id = get_entity_id(measure, profile)
state_obj = hass.states.get(entity_id)
assert state_obj, "Expected entity {} to exist but it did not".format(entity_id)
assert state_obj.state == str(
expected
), "Expected {} but was {} for measure {}, {}".format(
expected, state_obj.state, measure, entity_id
)
def new_sleep_summary(timezone, model, startdate, enddate, date, modified, data):
"""Create simple dict to simulate api data."""
return {
"timezone": timezone,
"model": model,
"startdate": startdate,
"enddate": enddate,
"date": date,
"modified": modified,
"data": data,
async def setup_hass(hass: HomeAssistant) -> dict:
"""Configure home assistant."""
profiles = ["Person0", "Person1", "Person2", "Person3", "Person4"]
hass_config = {
"homeassistant": {CONF_UNIT_SYSTEM: CONF_UNIT_SYSTEM_METRIC},
api.DOMAIN: {"base_url": "http://localhost/"},
http.DOMAIN: {"server_port": 8080},
const.DOMAIN: {
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.PROFILES: profiles,
},
}
await async_process_ha_core_config(hass, hass_config.get("homeassistant"))
assert await async_setup_component(hass, http.DOMAIN, hass_config)
assert await async_setup_component(hass, api.DOMAIN, hass_config)
assert await async_setup_component(hass, const.DOMAIN, hass_config)
await hass.async_block_till_done()
def new_sleep_summary_detail(
wakeupduration,
lightsleepduration,
deepsleepduration,
remsleepduration,
wakeupcount,
durationtosleep,
durationtowakeup,
hr_average,
hr_min,
hr_max,
rr_average,
rr_min,
rr_max,
):
"""Create simple dict to simulate api data."""
return {
"wakeupduration": wakeupduration,
"lightsleepduration": lightsleepduration,
"deepsleepduration": deepsleepduration,
"remsleepduration": remsleepduration,
"wakeupcount": wakeupcount,
"durationtosleep": durationtosleep,
"durationtowakeup": durationtowakeup,
"hr_average": hr_average,
"hr_min": hr_min,
"hr_max": hr_max,
"rr_average": rr_average,
"rr_min": rr_min,
"rr_max": rr_max,
}
return hass_config
def new_measure_group(
grpid, attrib, date, created, category, deviceid, more, offset, measures
):
"""Create simple dict to simulate api data."""
return {
"grpid": grpid,
"attrib": attrib,
"date": date,
"created": created,
"category": category,
"deviceid": deviceid,
"measures": measures,
"more": more,
"offset": offset,
"comment": "blah", # deprecated
}
async def configure_integration(
hass: HomeAssistant,
aiohttp_client,
aioclient_mock,
profiles: List[str],
profile_index: int,
get_device_response: dict,
getmeasures_response: dict,
get_sleep_response: dict,
get_sleep_summary_response: dict,
) -> None:
"""Configure the integration for a specific profile."""
selected_profile = profiles[profile_index]
def new_measure(type_str, value, unit):
"""Create simple dict to simulate api data."""
return {
"value": value,
"type": type_str,
"unit": unit,
"algo": -1, # deprecated
"fm": -1, # deprecated
"fw": -1, # deprecated
}
def withings_sleep_response(states):
"""Create a sleep response based on states."""
data = []
for state in states:
data.append(
new_sleep_data_serie(
"2019-02-01 0{}:00:00".format(str(len(data))),
"2019-02-01 0{}:00:00".format(str(len(data) + 1)),
state,
)
with requests_mock.mock() as rqmck:
rqmck.get(
re.compile(AbstractWithingsApi.URL + "/v2/user?.*action=getdevice(&.*|$)"),
status_code=200,
json=get_device_response,
)
return withings.WithingsSleep(new_sleep_data("aa", data))
rqmck.get(
re.compile(AbstractWithingsApi.URL + "/v2/sleep?.*action=get(&.*|$)"),
status_code=200,
json=get_sleep_response,
)
rqmck.get(
re.compile(
AbstractWithingsApi.URL + "/v2/sleep?.*action=getsummary(&.*|$)"
),
status_code=200,
json=get_sleep_summary_response,
)
rqmck.get(
re.compile(AbstractWithingsApi.URL + "/measure?.*action=getmeas(&.*|$)"),
status_code=200,
json=getmeasures_response,
)
# Get the withings config flow.
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": SOURCE_USER}
)
assert result
# pylint: disable=protected-access
state = config_entry_oauth2_flow._encode_jwt(
hass, {"flow_id": result["flow_id"]}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
assert result["url"] == (
"https://account.withings.com/oauth2_user/authorize2?"
"response_type=code&client_id=my_client_id&"
"redirect_uri=http://127.0.0.1:8080/auth/external/callback&"
f"state={state}"
"&scope=user.info,user.metrics,user.activity"
)
# Simulate user being redirected from withings site.
client = await aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
"https://account.withings.com/oauth2/token",
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"userid": "myuserid",
},
)
# Present user with a list of profiles to choose from.
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result.get("type") == "form"
assert result.get("step_id") == "profile"
assert result.get("data_schema").schema["profile"].container == profiles
# Select the user profile.
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {const.PROFILE: selected_profile}
)
# Finish the config flow by calling it again.
assert result.get("type") == "create_entry"
assert result.get("result")
config_data = result.get("result").data
assert config_data.get(const.PROFILE) == profiles[profile_index]
assert config_data.get("auth_implementation") == const.DOMAIN
assert config_data.get("token")
# Ensure all the flows are complete.
flows = hass.config_entries.flow.async_progress()
assert not flows
# Wait for remaining tasks to complete.
await hass.async_block_till_done()
WITHINGS_MEASURES_RESPONSE = withings.WithingsMeasures(
{
"updatetime": "",
"timezone": "",
WITHINGS_GET_DEVICE_RESPONSE_EMPTY = {"status": 0, "body": {"devices": []}}
WITHINGS_GET_DEVICE_RESPONSE = {
"status": 0,
"body": {
"devices": [
{
"type": "type1",
"model": "model1",
"battery": "battery1",
"deviceid": "deviceid1",
"timezone": "UTC",
}
]
},
}
WITHINGS_MEASURES_RESPONSE_EMPTY = {
"status": 0,
"body": {"updatetime": "2019-08-01", "timezone": "UTC", "measuregrps": []},
}
WITHINGS_MEASURES_RESPONSE = {
"status": 0,
"body": {
"updatetime": "2019-08-01",
"timezone": "UTC",
"measuregrps": [
# Un-ambiguous groups.
new_measure_group(
1,
0,
time.time(),
time.time(),
1,
"DEV_ID",
False,
0,
[
new_measure(const.MEASURE_TYPE_WEIGHT, 70, 0),
new_measure(const.MEASURE_TYPE_FAT_MASS, 5, 0),
new_measure(const.MEASURE_TYPE_FAT_MASS_FREE, 60, 0),
new_measure(const.MEASURE_TYPE_MUSCLE_MASS, 50, 0),
new_measure(const.MEASURE_TYPE_BONE_MASS, 10, 0),
new_measure(const.MEASURE_TYPE_HEIGHT, 2, 0),
new_measure(const.MEASURE_TYPE_TEMP, 40, 0),
new_measure(const.MEASURE_TYPE_BODY_TEMP, 35, 0),
new_measure(const.MEASURE_TYPE_SKIN_TEMP, 20, 0),
new_measure(const.MEASURE_TYPE_FAT_RATIO, 70, -3),
new_measure(const.MEASURE_TYPE_DIASTOLIC_BP, 70, 0),
new_measure(const.MEASURE_TYPE_SYSTOLIC_BP, 100, 0),
new_measure(const.MEASURE_TYPE_HEART_PULSE, 60, 0),
new_measure(const.MEASURE_TYPE_SPO2, 95, -2),
new_measure(const.MEASURE_TYPE_HYDRATION, 95, -2),
new_measure(const.MEASURE_TYPE_PWV, 100, 0),
{
"grpid": 1,
"attrib": MeasureGetMeasGroupAttrib.DEVICE_ENTRY_FOR_USER.real,
"date": time.time(),
"created": time.time(),
"category": MeasureGetMeasGroupCategory.REAL.real,
"deviceid": "DEV_ID",
"more": False,
"offset": 0,
"measures": [
{"type": MeasureType.WEIGHT, "value": 70, "unit": 0},
{"type": MeasureType.FAT_MASS_WEIGHT, "value": 5, "unit": 0},
{"type": MeasureType.FAT_FREE_MASS, "value": 60, "unit": 0},
{"type": MeasureType.MUSCLE_MASS, "value": 50, "unit": 0},
{"type": MeasureType.BONE_MASS, "value": 10, "unit": 0},
{"type": MeasureType.HEIGHT, "value": 2, "unit": 0},
{"type": MeasureType.TEMPERATURE, "value": 40, "unit": 0},
{"type": MeasureType.BODY_TEMPERATURE, "value": 40, "unit": 0},
{"type": MeasureType.SKIN_TEMPERATURE, "value": 20, "unit": 0},
{"type": MeasureType.FAT_RATIO, "value": 70, "unit": -3},
{
"type": MeasureType.DIASTOLIC_BLOOD_PRESSURE,
"value": 70,
"unit": 0,
},
{
"type": MeasureType.SYSTOLIC_BLOOD_PRESSURE,
"value": 100,
"unit": 0,
},
{"type": MeasureType.HEART_RATE, "value": 60, "unit": 0},
{"type": MeasureType.SP02, "value": 95, "unit": -2},
{"type": MeasureType.HYDRATION, "value": 95, "unit": -2},
{"type": MeasureType.PULSE_WAVE_VELOCITY, "value": 100, "unit": 0},
],
),
},
# Ambiguous groups (we ignore these)
new_measure_group(
1,
1,
time.time(),
time.time(),
1,
"DEV_ID",
False,
0,
[
new_measure(const.MEASURE_TYPE_WEIGHT, 71, 0),
new_measure(const.MEASURE_TYPE_FAT_MASS, 4, 0),
new_measure(const.MEASURE_TYPE_MUSCLE_MASS, 51, 0),
new_measure(const.MEASURE_TYPE_BONE_MASS, 11, 0),
new_measure(const.MEASURE_TYPE_HEIGHT, 201, 0),
new_measure(const.MEASURE_TYPE_TEMP, 41, 0),
new_measure(const.MEASURE_TYPE_BODY_TEMP, 34, 0),
new_measure(const.MEASURE_TYPE_SKIN_TEMP, 21, 0),
new_measure(const.MEASURE_TYPE_FAT_RATIO, 71, -3),
new_measure(const.MEASURE_TYPE_DIASTOLIC_BP, 71, 0),
new_measure(const.MEASURE_TYPE_SYSTOLIC_BP, 101, 0),
new_measure(const.MEASURE_TYPE_HEART_PULSE, 61, 0),
new_measure(const.MEASURE_TYPE_SPO2, 98, -2),
new_measure(const.MEASURE_TYPE_HYDRATION, 96, -2),
new_measure(const.MEASURE_TYPE_PWV, 102, 0),
{
"grpid": 1,
"attrib": MeasureGetMeasGroupAttrib.DEVICE_ENTRY_FOR_USER.real,
"date": time.time(),
"created": time.time(),
"category": MeasureGetMeasGroupCategory.REAL.real,
"deviceid": "DEV_ID",
"more": False,
"offset": 0,
"measures": [
{"type": MeasureType.WEIGHT, "value": 71, "unit": 0},
{"type": MeasureType.FAT_MASS_WEIGHT, "value": 4, "unit": 0},
{"type": MeasureType.FAT_FREE_MASS, "value": 40, "unit": 0},
{"type": MeasureType.MUSCLE_MASS, "value": 51, "unit": 0},
{"type": MeasureType.BONE_MASS, "value": 11, "unit": 0},
{"type": MeasureType.HEIGHT, "value": 201, "unit": 0},
{"type": MeasureType.TEMPERATURE, "value": 41, "unit": 0},
{"type": MeasureType.BODY_TEMPERATURE, "value": 34, "unit": 0},
{"type": MeasureType.SKIN_TEMPERATURE, "value": 21, "unit": 0},
{"type": MeasureType.FAT_RATIO, "value": 71, "unit": -3},
{
"type": MeasureType.DIASTOLIC_BLOOD_PRESSURE,
"value": 71,
"unit": 0,
},
{
"type": MeasureType.SYSTOLIC_BLOOD_PRESSURE,
"value": 101,
"unit": 0,
},
{"type": MeasureType.HEART_RATE, "value": 61, "unit": 0},
{"type": MeasureType.SP02, "value": 98, "unit": -2},
{"type": MeasureType.HYDRATION, "value": 96, "unit": -2},
{"type": MeasureType.PULSE_WAVE_VELOCITY, "value": 102, "unit": 0},
],
),
},
],
}
)
},
}
WITHINGS_SLEEP_RESPONSE = withings_sleep_response(
[
const.MEASURE_TYPE_SLEEP_STATE_AWAKE,
const.MEASURE_TYPE_SLEEP_STATE_LIGHT,
const.MEASURE_TYPE_SLEEP_STATE_REM,
const.MEASURE_TYPE_SLEEP_STATE_DEEP,
]
)
WITHINGS_SLEEP_RESPONSE_EMPTY = {
"status": 0,
"body": {"model": SleepModel.TRACKER.real, "series": []},
}
WITHINGS_SLEEP_SUMMARY_RESPONSE = withings.WithingsSleepSummary(
{
WITHINGS_SLEEP_RESPONSE = {
"status": 0,
"body": {
"model": SleepModel.TRACKER.real,
"series": [
new_sleep_summary(
"UTC",
32,
"2019-02-01",
"2019-02-02",
"2019-02-02",
"12345",
new_sleep_summary_detail(
110, 210, 310, 410, 510, 610, 710, 810, 910, 1010, 1110, 1210, 1310
),
),
new_sleep_summary(
"UTC",
32,
"2019-02-01",
"2019-02-02",
"2019-02-02",
"12345",
new_sleep_summary_detail(
210, 310, 410, 510, 610, 710, 810, 910, 1010, 1110, 1210, 1310, 1410
),
),
]
}
)
{
"startdate": "2019-02-01 00:00:00",
"enddate": "2019-02-01 01:00:00",
"state": SleepState.AWAKE.real,
},
{
"startdate": "2019-02-01 01:00:00",
"enddate": "2019-02-01 02:00:00",
"state": SleepState.LIGHT.real,
},
{
"startdate": "2019-02-01 02:00:00",
"enddate": "2019-02-01 03:00:00",
"state": SleepState.REM.real,
},
{
"startdate": "2019-02-01 03:00:00",
"enddate": "2019-02-01 04:00:00",
"state": SleepState.DEEP.real,
},
],
},
}
WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY = {
"status": 0,
"body": {"more": False, "offset": 0, "series": []},
}
WITHINGS_SLEEP_SUMMARY_RESPONSE = {
"status": 0,
"body": {
"more": False,
"offset": 0,
"series": [
{
"timezone": "UTC",
"model": SleepModel.SLEEP_MONITOR.real,
"startdate": "2019-02-01",
"enddate": "2019-02-02",
"date": "2019-02-02",
"modified": 12345,
"data": {
"wakeupduration": 110,
"lightsleepduration": 210,
"deepsleepduration": 310,
"remsleepduration": 410,
"wakeupcount": 510,
"durationtosleep": 610,
"durationtowakeup": 710,
"hr_average": 810,
"hr_min": 910,
"hr_max": 1010,
"rr_average": 1110,
"rr_min": 1210,
"rr_max": 1310,
},
},
{
"timezone": "UTC",
"model": SleepModel.SLEEP_MONITOR.real,
"startdate": "2019-02-01",
"enddate": "2019-02-02",
"date": "2019-02-02",
"modified": 12345,
"data": {
"wakeupduration": 210,
"lightsleepduration": 310,
"deepsleepduration": 410,
"remsleepduration": 510,
"wakeupcount": 610,
"durationtosleep": 710,
"durationtowakeup": 810,
"hr_average": 910,
"hr_min": 1010,
"hr_max": 1110,
"rr_average": 1210,
"rr_min": 1310,
"rr_max": 1410,
},
},
],
},
}

350
tests/components/withings/conftest.py
View file

@ -1,350 +0,0 @@
"""Fixtures for withings tests."""
import time
from typing import Awaitable, Callable, List
import asynctest
import withings_api as withings
import pytest
import homeassistant.components.api as api
import homeassistant.components.http as http
import homeassistant.components.withings.const as const
from homeassistant.components.withings import CONFIG_SCHEMA, DOMAIN
from homeassistant.config import async_process_ha_core_config
from homeassistant.const import CONF_UNIT_SYSTEM, CONF_UNIT_SYSTEM_METRIC
from homeassistant.setup import async_setup_component
from .common import (
WITHINGS_MEASURES_RESPONSE,
WITHINGS_SLEEP_RESPONSE,
WITHINGS_SLEEP_SUMMARY_RESPONSE,
)
class WithingsFactoryConfig:
"""Configuration for withings test fixture."""
PROFILE_1 = "Person 1"
PROFILE_2 = "Person 2"
def __init__(
self,
api_config: dict = None,
http_config: dict = None,
measures: List[str] = None,
unit_system: str = None,
throttle_interval: int = const.THROTTLE_INTERVAL,
withings_request_response="DATA",
withings_measures_response: withings.WithingsMeasures = WITHINGS_MEASURES_RESPONSE,
withings_sleep_response: withings.WithingsSleep = WITHINGS_SLEEP_RESPONSE,
withings_sleep_summary_response: withings.WithingsSleepSummary = WITHINGS_SLEEP_SUMMARY_RESPONSE,
) -> None:
"""Constructor."""
self._throttle_interval = throttle_interval
self._withings_request_response = withings_request_response
self._withings_measures_response = withings_measures_response
self._withings_sleep_response = withings_sleep_response
self._withings_sleep_summary_response = withings_sleep_summary_response
self._withings_config = {
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.PROFILES: [
WithingsFactoryConfig.PROFILE_1,
WithingsFactoryConfig.PROFILE_2,
],
}
self._api_config = api_config or {"base_url": "http://localhost/"}
self._http_config = http_config or {}
self._measures = measures
assert self._withings_config, "withings_config must be set."
assert isinstance(
self._withings_config, dict
), "withings_config must be a dict."
assert isinstance(self._api_config, dict), "api_config must be a dict."
assert isinstance(self._http_config, dict), "http_config must be a dict."
self._hass_config = {
"homeassistant": {CONF_UNIT_SYSTEM: unit_system or CONF_UNIT_SYSTEM_METRIC},
api.DOMAIN: self._api_config,
http.DOMAIN: self._http_config,
DOMAIN: self._withings_config,
}
@property
def withings_config(self):
"""Get withings component config."""
return self._withings_config
@property
def api_config(self):
"""Get api component config."""
return self._api_config
@property
def http_config(self):
"""Get http component config."""
return self._http_config
@property
def measures(self):
"""Get the measures."""
return self._measures
@property
def hass_config(self):
"""Home assistant config."""
return self._hass_config
@property
def throttle_interval(self):
"""Throttle interval."""
return self._throttle_interval
@property
def withings_request_response(self):
"""Request response."""
return self._withings_request_response
@property
def withings_measures_response(self) -> withings.WithingsMeasures:
"""Measures response."""
return self._withings_measures_response
@property
def withings_sleep_response(self) -> withings.WithingsSleep:
"""Sleep response."""
return self._withings_sleep_response
@property
def withings_sleep_summary_response(self) -> withings.WithingsSleepSummary:
"""Sleep summary response."""
return self._withings_sleep_summary_response
class WithingsFactoryData:
"""Data about the configured withing test component."""
def __init__(
self,
hass,
flow_id,
withings_auth_get_credentials_mock,
withings_api_request_mock,
withings_api_get_measures_mock,
withings_api_get_sleep_mock,
withings_api_get_sleep_summary_mock,
data_manager_get_throttle_interval_mock,
):
"""Constructor."""
self._hass = hass
self._flow_id = flow_id
self._withings_auth_get_credentials_mock = withings_auth_get_credentials_mock
self._withings_api_request_mock = withings_api_request_mock
self._withings_api_get_measures_mock = withings_api_get_measures_mock
self._withings_api_get_sleep_mock = withings_api_get_sleep_mock
self._withings_api_get_sleep_summary_mock = withings_api_get_sleep_summary_mock
self._data_manager_get_throttle_interval_mock = (
data_manager_get_throttle_interval_mock
)
@property
def hass(self):
"""Get hass instance."""
return self._hass
@property
def flow_id(self):
"""Get flow id."""
return self._flow_id
@property
def withings_auth_get_credentials_mock(self):
"""Get auth credentials mock."""
return self._withings_auth_get_credentials_mock
@property
def withings_api_request_mock(self):
"""Get request mock."""
return self._withings_api_request_mock
@property
def withings_api_get_measures_mock(self):
"""Get measures mock."""
return self._withings_api_get_measures_mock
@property
def withings_api_get_sleep_mock(self):
"""Get sleep mock."""
return self._withings_api_get_sleep_mock
@property
def withings_api_get_sleep_summary_mock(self):
"""Get sleep summary mock."""
return self._withings_api_get_sleep_summary_mock
@property
def data_manager_get_throttle_interval_mock(self):
"""Get throttle mock."""
return self._data_manager_get_throttle_interval_mock
async def configure_user(self):
"""Present a form with user profiles."""
step = await self.hass.config_entries.flow.async_configure(self.flow_id, None)
assert step["step_id"] == "user"
async def configure_profile(self, profile: str):
"""Select the user profile. Present a form with authorization link."""
print("CONFIG_PROFILE:", profile)
step = await self.hass.config_entries.flow.async_configure(
self.flow_id, {const.PROFILE: profile}
)
assert step["step_id"] == "auth"
async def configure_code(self, profile: str, code: str):
"""Handle authorization code. Create config entries."""
step = await self.hass.config_entries.flow.async_configure(
self.flow_id, {const.PROFILE: profile, const.CODE: code}
)
assert step["type"] == "external_done"
await self.hass.async_block_till_done()
step = await self.hass.config_entries.flow.async_configure(
self.flow_id, {const.PROFILE: profile, const.CODE: code}
)
assert step["type"] == "create_entry"
await self.hass.async_block_till_done()
async def configure_all(self, profile: str, code: str):
"""Configure all flow steps."""
await self.configure_user()
await self.configure_profile(profile)
await self.configure_code(profile, code)
WithingsFactory = Callable[[WithingsFactoryConfig], Awaitable[WithingsFactoryData]]
@pytest.fixture(name="withings_factory")
def withings_factory_fixture(request, hass) -> WithingsFactory:
"""Home assistant platform fixture."""
patches = []
async def factory(config: WithingsFactoryConfig) -> WithingsFactoryData:
CONFIG_SCHEMA(config.hass_config.get(DOMAIN))
await async_process_ha_core_config(
hass, config.hass_config.get("homeassistant")
)
assert await async_setup_component(hass, http.DOMAIN, config.hass_config)
assert await async_setup_component(hass, api.DOMAIN, config.hass_config)
withings_auth_get_credentials_patch = asynctest.patch(
"withings_api.WithingsAuth.get_credentials",
return_value=withings.WithingsCredentials(
access_token="my_access_token",
token_expiry=time.time() + 600,
token_type="my_token_type",
refresh_token="my_refresh_token",
user_id="my_user_id",
client_id=config.withings_config.get(const.CLIENT_ID),
consumer_secret=config.withings_config.get(const.CLIENT_SECRET),
),
)
withings_auth_get_credentials_mock = withings_auth_get_credentials_patch.start()
withings_api_request_patch = asynctest.patch(
"withings_api.WithingsApi.request",
return_value=config.withings_request_response,
)
withings_api_request_mock = withings_api_request_patch.start()
withings_api_get_measures_patch = asynctest.patch(
"withings_api.WithingsApi.get_measures",
return_value=config.withings_measures_response,
)
withings_api_get_measures_mock = withings_api_get_measures_patch.start()
withings_api_get_sleep_patch = asynctest.patch(
"withings_api.WithingsApi.get_sleep",
return_value=config.withings_sleep_response,
)
withings_api_get_sleep_mock = withings_api_get_sleep_patch.start()
withings_api_get_sleep_summary_patch = asynctest.patch(
"withings_api.WithingsApi.get_sleep_summary",
return_value=config.withings_sleep_summary_response,
)
withings_api_get_sleep_summary_mock = (
withings_api_get_sleep_summary_patch.start()
)
data_manager_get_throttle_interval_patch = asynctest.patch(
"homeassistant.components.withings.common.WithingsDataManager"
".get_throttle_interval",
return_value=config.throttle_interval,
)
data_manager_get_throttle_interval_mock = (
data_manager_get_throttle_interval_patch.start()
)
get_measures_patch = asynctest.patch(
"homeassistant.components.withings.sensor.get_measures",
return_value=config.measures,
)
get_measures_patch.start()
patches.extend(
[
withings_auth_get_credentials_patch,
withings_api_request_patch,
withings_api_get_measures_patch,
withings_api_get_sleep_patch,
withings_api_get_sleep_summary_patch,
data_manager_get_throttle_interval_patch,
get_measures_patch,
]
)
# Collect the flow id.
tasks = []
orig_async_create_task = hass.async_create_task
def create_task(*args):
task = orig_async_create_task(*args)
tasks.append(task)
return task
async_create_task_patch = asynctest.patch.object(
hass, "async_create_task", side_effect=create_task
)
with async_create_task_patch:
assert await async_setup_component(hass, DOMAIN, config.hass_config)
await hass.async_block_till_done()
flow_id = tasks[2].result()["flow_id"]
return WithingsFactoryData(
hass,
flow_id,
withings_auth_get_credentials_mock,
withings_api_request_mock,
withings_api_get_measures_mock,
withings_api_get_sleep_mock,
withings_api_get_sleep_summary_mock,
data_manager_get_throttle_interval_mock,
)
def cleanup():
for patch in patches:
patch.stop()
request.addfinalizer(cleanup)
return factory

74
tests/components/withings/test_common.py
View file

@ -1,34 +1,33 @@
"""Tests for the Withings component."""
from asynctest import MagicMock
import withings_api as withings
from oauthlib.oauth2.rfc6749.errors import MissingTokenError
import pytest
from requests_oauthlib import TokenUpdated
import pytest
from withings_api import WithingsApi
from withings_api.common import UnauthorizedException, TimeoutException
from homeassistant.exceptions import PlatformNotReady
from homeassistant.components.withings.common import (
NotAuthenticatedError,
ServiceError,
WithingsDataManager,
)
from homeassistant.exceptions import PlatformNotReady
@pytest.fixture(name="withings_api")
def withings_api_fixture():
def withings_api_fixture() -> WithingsApi:
"""Provide withings api."""
withings_api = withings.WithingsApi.__new__(withings.WithingsApi)
withings_api = WithingsApi.__new__(WithingsApi)
withings_api.get_measures = MagicMock()
withings_api.get_sleep = MagicMock()
return withings_api
@pytest.fixture(name="data_manager")
def data_manager_fixture(hass, withings_api: withings.WithingsApi):
def data_manager_fixture(hass, withings_api: WithingsApi) -> WithingsDataManager:
"""Provide data manager."""
return WithingsDataManager(hass, "My Profile", withings_api)
def test_print_service():
def test_print_service() -> None:
"""Test method."""
# Go from None to True
WithingsDataManager.service_available = None
@ -57,54 +56,27 @@ def test_print_service():
assert not WithingsDataManager.print_service_unavailable()
async def test_data_manager_call(data_manager):
async def test_data_manager_call(data_manager: WithingsDataManager) -> None:
"""Test method."""
# Token refreshed.
def hello_func():
return "HELLO2"
function = MagicMock(side_effect=[TokenUpdated("my_token"), hello_func()])
result = await data_manager.call(function)
assert result == "HELLO2"
assert function.call_count == 2
# Too many token refreshes.
function = MagicMock(
side_effect=[TokenUpdated("my_token"), TokenUpdated("my_token")]
)
try:
result = await data_manager.call(function)
assert False, "This should not have ran."
except ServiceError:
assert True
assert function.call_count == 2
# Not authenticated 1.
test_function = MagicMock(side_effect=MissingTokenError("Error Code 401"))
try:
result = await data_manager.call(test_function)
assert False, "An exception should have been thrown."
except NotAuthenticatedError:
assert True
test_function = MagicMock(side_effect=UnauthorizedException(401))
with pytest.raises(NotAuthenticatedError):
await data_manager.call(test_function)
# Not authenticated 2.
test_function = MagicMock(side_effect=Exception("Error Code 401"))
try:
result = await data_manager.call(test_function)
assert False, "An exception should have been thrown."
except NotAuthenticatedError:
assert True
test_function = MagicMock(side_effect=TimeoutException(522))
with pytest.raises(PlatformNotReady):
await data_manager.call(test_function)
# Service error.
test_function = MagicMock(side_effect=PlatformNotReady())
try:
result = await data_manager.call(test_function)
assert False, "An exception should have been thrown."
except PlatformNotReady:
assert True
with pytest.raises(PlatformNotReady):
await data_manager.call(test_function)
async def test_data_manager_call_throttle_enabled(data_manager):
async def test_data_manager_call_throttle_enabled(
data_manager: WithingsDataManager
) -> None:
"""Test method."""
hello_func = MagicMock(return_value="HELLO2")
@ -117,7 +89,9 @@ async def test_data_manager_call_throttle_enabled(data_manager):
assert hello_func.call_count == 1
async def test_data_manager_call_throttle_disabled(data_manager):
async def test_data_manager_call_throttle_disabled(
data_manager: WithingsDataManager
) -> None:
"""Test method."""
hello_func = MagicMock(return_value="HELLO2")

162
tests/components/withings/test_config_flow.py
View file

@ -1,162 +0,0 @@
"""Tests for the Withings config flow."""
from aiohttp.web_request import BaseRequest
from asynctest import CoroutineMock, MagicMock
import pytest
from homeassistant import data_entry_flow
from homeassistant.components.withings import const
from homeassistant.components.withings.config_flow import (
register_flow_implementation,
WithingsFlowHandler,
WithingsAuthCallbackView,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
@pytest.fixture(name="flow_handler")
def flow_handler_fixture(hass: HomeAssistantType):
"""Provide flow handler."""
flow_handler = WithingsFlowHandler()
flow_handler.hass = hass
return flow_handler
def test_flow_handler_init(flow_handler: WithingsFlowHandler):
"""Test the init of the flow handler."""
assert not flow_handler.flow_profile
def test_flow_handler_async_profile_config_entry(
hass: HomeAssistantType, flow_handler: WithingsFlowHandler
):
"""Test profile config entry."""
config_entries = [
ConfigEntry(
version=1,
domain=const.DOMAIN,
title="AAA",
data={},
source="source",
connection_class="connection_class",
system_options={},
),
ConfigEntry(
version=1,
domain=const.DOMAIN,
title="Person 1",
data={const.PROFILE: "Person 1"},
source="source",
connection_class="connection_class",
system_options={},
),
ConfigEntry(
version=1,
domain=const.DOMAIN,
title="BBB",
data={},
source="source",
connection_class="connection_class",
system_options={},
),
]
hass.config_entries.async_entries = MagicMock(return_value=config_entries)
config_entry = flow_handler.async_profile_config_entry
assert not config_entry("GGGG")
hass.config_entries.async_entries.assert_called_with(const.DOMAIN)
assert not config_entry("CCC")
hass.config_entries.async_entries.assert_called_with(const.DOMAIN)
assert config_entry("Person 1") == config_entries[1]
hass.config_entries.async_entries.assert_called_with(const.DOMAIN)
def test_flow_handler_get_auth_client(
hass: HomeAssistantType, flow_handler: WithingsFlowHandler
):
"""Test creation of an auth client."""
register_flow_implementation(
hass, "my_client_id", "my_client_secret", "http://localhost/", ["Person 1"]
)
client = flow_handler.get_auth_client("Person 1")
assert client.client_id == "my_client_id"
assert client.consumer_secret == "my_client_secret"
assert client.callback_uri.startswith(
"http://localhost/api/withings/authorize?flow_id="
)
assert client.callback_uri.endswith("&profile=Person 1")
assert client.scope == "user.info,user.metrics,user.activity"
async def test_auth_callback_view_get(hass: HomeAssistantType):
"""Test get api path."""
view = WithingsAuthCallbackView()
hass.config_entries.flow.async_configure = CoroutineMock(return_value="AAAA")
request = MagicMock(spec=BaseRequest)
request.app = {"hass": hass}
# No args
request.query = {}
response = await view.get(request)
assert response.status == 400
hass.config_entries.flow.async_configure.assert_not_called()
hass.config_entries.flow.async_configure.reset_mock()
# Checking flow_id
request.query = {"flow_id": "my_flow_id"}
response = await view.get(request)
assert response.status == 400
hass.config_entries.flow.async_configure.assert_not_called()
hass.config_entries.flow.async_configure.reset_mock()
# Checking flow_id and profile
request.query = {"flow_id": "my_flow_id", "profile": "my_profile"}
response = await view.get(request)
assert response.status == 400
hass.config_entries.flow.async_configure.assert_not_called()
hass.config_entries.flow.async_configure.reset_mock()
# Checking flow_id, profile, code
request.query = {
"flow_id": "my_flow_id",
"profile": "my_profile",
"code": "my_code",
}
response = await view.get(request)
assert response.status == 200
hass.config_entries.flow.async_configure.assert_called_with(
"my_flow_id", {const.PROFILE: "my_profile", const.CODE: "my_code"}
)
hass.config_entries.flow.async_configure.reset_mock()
# Exception thrown
hass.config_entries.flow.async_configure = CoroutineMock(
side_effect=data_entry_flow.UnknownFlow()
)
request.query = {
"flow_id": "my_flow_id",
"profile": "my_profile",
"code": "my_code",
}
response = await view.get(request)
assert response.status == 400
hass.config_entries.flow.async_configure.assert_called_with(
"my_flow_id", {const.PROFILE: "my_profile", const.CODE: "my_code"}
)
hass.config_entries.flow.async_configure.reset_mock()
async def test_init_without_config(hass):
"""Try initializin a configg flow without it being configured."""
result = await hass.config_entries.flow.async_init(
"withings", context={"source": "user"}
)
assert result["type"] == "abort"
assert result["reason"] == "no_flows"

349
tests/components/withings/test_init.py
View file

@ -1,29 +1,46 @@
"""Tests for the Withings component."""
import re
import time
from asynctest import MagicMock
import requests_mock
import voluptuous as vol
from withings_api import AbstractWithingsApi
from withings_api.common import SleepModel, SleepState
import homeassistant.components.api as api
import homeassistant.components.http as http
from homeassistant.components.withings import async_setup, const, CONFIG_SCHEMA
from homeassistant.components.withings import (
async_setup,
async_setup_entry,
const,
CONFIG_SCHEMA,
)
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from .conftest import WithingsFactory, WithingsFactoryConfig
BASE_HASS_CONFIG = {
http.DOMAIN: {},
api.DOMAIN: {"base_url": "http://localhost/"},
const.DOMAIN: None,
}
from .common import (
assert_state_equals,
configure_integration,
setup_hass,
WITHINGS_GET_DEVICE_RESPONSE,
WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
WITHINGS_SLEEP_RESPONSE,
WITHINGS_SLEEP_RESPONSE_EMPTY,
WITHINGS_SLEEP_SUMMARY_RESPONSE,
WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
WITHINGS_MEASURES_RESPONSE,
WITHINGS_MEASURES_RESPONSE_EMPTY,
)
def config_schema_validate(withings_config):
def config_schema_validate(withings_config) -> None:
"""Assert a schema config succeeds."""
hass_config = BASE_HASS_CONFIG.copy()
hass_config[const.DOMAIN] = withings_config
hass_config = {http.DOMAIN: {}, const.DOMAIN: withings_config}
return CONFIG_SCHEMA(hass_config)
def config_schema_assert_fail(withings_config):
def config_schema_assert_fail(withings_config) -> None:
"""Assert a schema config will fail."""
try:
config_schema_validate(withings_config)
@ -32,7 +49,7 @@ def config_schema_assert_fail(withings_config):
assert True
def test_config_schema_basic_config():
def test_config_schema_basic_config() -> None:
"""Test schema."""
config_schema_validate(
{
@ -43,7 +60,7 @@ def test_config_schema_basic_config():
)
def test_config_schema_client_id():
def test_config_schema_client_id() -> None:
"""Test schema."""
config_schema_assert_fail(
{
@ -67,7 +84,7 @@ def test_config_schema_client_id():
)
def test_config_schema_client_secret():
def test_config_schema_client_secret() -> None:
"""Test schema."""
config_schema_assert_fail(
{const.CLIENT_ID: "my_client_id", const.PROFILES: ["Person 1"]}
@ -88,7 +105,7 @@ def test_config_schema_client_secret():
)
def test_config_schema_profiles():
def test_config_schema_profiles() -> None:
"""Test schema."""
config_schema_assert_fail(
{const.CLIENT_ID: "my_client_id", const.CLIENT_SECRET: "my_client_secret"}
@ -130,50 +147,7 @@ def test_config_schema_profiles():
)
def test_config_schema_base_url():
"""Test schema."""
config_schema_validate(
{
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.PROFILES: ["Person 1"],
}
)
config_schema_assert_fail(
{
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.BASE_URL: 123,
const.PROFILES: ["Person 1"],
}
)
config_schema_assert_fail(
{
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.BASE_URL: "",
const.PROFILES: ["Person 1"],
}
)
config_schema_assert_fail(
{
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.BASE_URL: "blah blah",
const.PROFILES: ["Person 1"],
}
)
config_schema_validate(
{
const.CLIENT_ID: "my_client_id",
const.CLIENT_SECRET: "my_client_secret",
const.BASE_URL: "https://www.blah.blah.blah/blah/blah",
const.PROFILES: ["Person 1"],
}
)
async def test_async_setup_no_config(hass):
async def test_async_setup_no_config(hass: HomeAssistant) -> None:
"""Test method."""
hass.async_create_task = MagicMock()
@ -182,15 +156,258 @@ async def test_async_setup_no_config(hass):
hass.async_create_task.assert_not_called()
async def test_async_setup_teardown(withings_factory: WithingsFactory, hass):
"""Test method."""
data = await withings_factory(WithingsFactoryConfig(measures=[const.MEAS_TEMP_C]))
async def test_upgrade_token(
hass: HomeAssistant, aiohttp_client, aioclient_mock
) -> None:
"""Test upgrading from old config data format to new one."""
config = await setup_hass(hass)
profiles = config[const.DOMAIN][const.PROFILES]
profile = WithingsFactoryConfig.PROFILE_1
await data.configure_all(profile, "authorization_code")
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=0,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
getmeasures_response=WITHINGS_MEASURES_RESPONSE_EMPTY,
get_sleep_response=WITHINGS_SLEEP_RESPONSE_EMPTY,
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
)
entries = hass.config_entries.async_entries(const.DOMAIN)
assert entries
entry = entries[0]
data = entry.data
token = data.get("token")
hass.config_entries.async_update_entry(
entry,
data={
const.PROFILE: data.get(const.PROFILE),
const.CREDENTIALS: {
"access_token": token.get("access_token"),
"refresh_token": token.get("refresh_token"),
"token_expiry": token.get("expires_at"),
"token_type": token.get("type"),
"userid": token.get("userid"),
"client_id": token.get("my_client_id"),
"consumer_secret": token.get("my_consumer_secret"),
},
},
)
with requests_mock.mock() as rqmck:
rqmck.get(
re.compile(AbstractWithingsApi.URL + "/v2/user?.*action=getdevice(&.*|$)"),
status_code=200,
json=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
)
assert await async_setup_entry(hass, entry)
entries = hass.config_entries.async_entries(const.DOMAIN)
assert entries
data = entries[0].data
assert data.get("auth_implementation") == const.DOMAIN
assert data.get("implementation") == const.DOMAIN
assert data.get(const.PROFILE) == profiles[0]
token = data.get("token")
assert token
assert token.get("access_token") == "mock-access-token"
assert token.get("refresh_token") == "mock-refresh-token"
assert token.get("expires_at") > time.time()
assert token.get("type") == "Bearer"
assert token.get("userid") == "myuserid"
assert not token.get("client_id")
assert not token.get("consumer_secret")
async def test_auth_failure(
hass: HomeAssistant, aiohttp_client, aioclient_mock
) -> None:
"""Test auth failure."""
config = await setup_hass(hass)
profiles = config[const.DOMAIN][const.PROFILES]
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=0,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
getmeasures_response=WITHINGS_MEASURES_RESPONSE_EMPTY,
get_sleep_response=WITHINGS_SLEEP_RESPONSE_EMPTY,
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
)
entries = hass.config_entries.async_entries(const.DOMAIN)
assert entries
entry = entries[0]
hass.config_entries.async_update_entry(
entry, data={**entry.data, **{"new_item": 1}}
)
with requests_mock.mock() as rqmck:
rqmck.get(
re.compile(AbstractWithingsApi.URL + "/v2/user?.*action=getdevice(&.*|$)"),
status_code=200,
json={"status": 401, "body": {}},
)
assert not (await async_setup_entry(hass, entry))
async def test_full_setup(hass: HomeAssistant, aiohttp_client, aioclient_mock) -> None:
"""Test the whole component lifecycle."""
config = await setup_hass(hass)
profiles = config[const.DOMAIN][const.PROFILES]
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=0,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE,
getmeasures_response=WITHINGS_MEASURES_RESPONSE,
get_sleep_response=WITHINGS_SLEEP_RESPONSE,
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE,
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=1,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
getmeasures_response=WITHINGS_MEASURES_RESPONSE_EMPTY,
get_sleep_response=WITHINGS_SLEEP_RESPONSE_EMPTY,
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=2,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
getmeasures_response=WITHINGS_MEASURES_RESPONSE_EMPTY,
get_sleep_response={
"status": 0,
"body": {
"model": SleepModel.TRACKER.real,
"series": [
{
"startdate": "2019-02-01 00:00:00",
"enddate": "2019-02-01 01:00:00",
"state": SleepState.AWAKE.real,
}
],
},
},
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=3,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
getmeasures_response=WITHINGS_MEASURES_RESPONSE_EMPTY,
get_sleep_response={
"status": 0,
"body": {
"model": SleepModel.TRACKER.real,
"series": [
{
"startdate": "2019-02-01 00:00:00",
"enddate": "2019-02-01 01:00:00",
"state": SleepState.LIGHT.real,
}
],
},
},
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
)
await configure_integration(
hass=hass,
aiohttp_client=aiohttp_client,
aioclient_mock=aioclient_mock,
profiles=profiles,
profile_index=4,
get_device_response=WITHINGS_GET_DEVICE_RESPONSE_EMPTY,
getmeasures_response=WITHINGS_MEASURES_RESPONSE_EMPTY,
get_sleep_response={
"status": 0,
"body": {
"model": SleepModel.TRACKER.real,
"series": [
{
"startdate": "2019-02-01 00:00:00",
"enddate": "2019-02-01 01:00:00",
"state": SleepState.REM.real,
}
],
},
},
get_sleep_summary_response=WITHINGS_SLEEP_SUMMARY_RESPONSE_EMPTY,
)
# Test the states of the entities.
expected_states = (
(profiles[0], const.MEAS_WEIGHT_KG, 70.0),
(profiles[0], const.MEAS_FAT_MASS_KG, 5.0),
(profiles[0], const.MEAS_FAT_FREE_MASS_KG, 60.0),
(profiles[0], const.MEAS_MUSCLE_MASS_KG, 50.0),
(profiles[0], const.MEAS_BONE_MASS_KG, 10.0),
(profiles[0], const.MEAS_HEIGHT_M, 2.0),
(profiles[0], const.MEAS_FAT_RATIO_PCT, 0.07),
(profiles[0], const.MEAS_DIASTOLIC_MMHG, 70.0),
(profiles[0], const.MEAS_SYSTOLIC_MMGH, 100.0),
(profiles[0], const.MEAS_HEART_PULSE_BPM, 60.0),
(profiles[0], const.MEAS_SPO2_PCT, 0.95),
(profiles[0], const.MEAS_HYDRATION, 0.95),
(profiles[0], const.MEAS_PWV, 100.0),
(profiles[0], const.MEAS_SLEEP_WAKEUP_DURATION_SECONDS, 320),
(profiles[0], const.MEAS_SLEEP_LIGHT_DURATION_SECONDS, 520),
(profiles[0], const.MEAS_SLEEP_DEEP_DURATION_SECONDS, 720),
(profiles[0], const.MEAS_SLEEP_REM_DURATION_SECONDS, 920),
(profiles[0], const.MEAS_SLEEP_WAKEUP_COUNT, 1120),
(profiles[0], const.MEAS_SLEEP_TOSLEEP_DURATION_SECONDS, 1320),
(profiles[0], const.MEAS_SLEEP_TOWAKEUP_DURATION_SECONDS, 1520),
(profiles[0], const.MEAS_SLEEP_HEART_RATE_AVERAGE, 1720),
(profiles[0], const.MEAS_SLEEP_HEART_RATE_MIN, 1920),
(profiles[0], const.MEAS_SLEEP_HEART_RATE_MAX, 2120),
(profiles[0], const.MEAS_SLEEP_RESPIRATORY_RATE_AVERAGE, 2320),
(profiles[0], const.MEAS_SLEEP_RESPIRATORY_RATE_MIN, 2520),
(profiles[0], const.MEAS_SLEEP_RESPIRATORY_RATE_MAX, 2720),
(profiles[0], const.MEAS_SLEEP_STATE, const.STATE_DEEP),
(profiles[1], const.MEAS_SLEEP_STATE, STATE_UNKNOWN),
(profiles[1], const.MEAS_HYDRATION, STATE_UNKNOWN),
(profiles[2], const.MEAS_SLEEP_STATE, const.STATE_AWAKE),
(profiles[3], const.MEAS_SLEEP_STATE, const.STATE_LIGHT),
(profiles[3], const.MEAS_FAT_FREE_MASS_KG, STATE_UNKNOWN),
(profiles[4], const.MEAS_SLEEP_STATE, const.STATE_REM),
)
for (profile, meas, value) in expected_states:
assert_state_equals(hass, profile, meas, value)
# Tear down setup entries.
entries = hass.config_entries.async_entries(const.DOMAIN)
assert entries
for entry in entries:
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()

310
tests/components/withings/test_sensor.py
View file

@ -1,310 +0,0 @@
"""Tests for the Withings component."""
from unittest.mock import MagicMock, patch
import asynctest
from withings_api import (
WithingsApi,
WithingsMeasures,
WithingsSleep,
WithingsSleepSummary,
)
import pytest
from homeassistant.components.withings import DOMAIN
from homeassistant.components.withings.common import NotAuthenticatedError
import homeassistant.components.withings.const as const
from homeassistant.components.withings.sensor import async_setup_entry
from homeassistant.config_entries import ConfigEntry, SOURCE_USER
from homeassistant.const import STATE_UNKNOWN
from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import slugify
from .common import withings_sleep_response
from .conftest import WithingsFactory, WithingsFactoryConfig
def get_entity_id(measure, profile):
"""Get an entity id for a measure and profile."""
return "sensor.{}_{}_{}".format(DOMAIN, measure, slugify(profile))
def assert_state_equals(hass: HomeAssistantType, profile: str, measure: str, expected):
"""Assert the state of a withings sensor."""
entity_id = get_entity_id(measure, profile)
state_obj = hass.states.get(entity_id)
assert state_obj, "Expected entity {} to exist but it did not".format(entity_id)
assert state_obj.state == str(
expected
), "Expected {} but was {} for measure {}".format(
expected, state_obj.state, measure
)
async def test_health_sensor_properties(withings_factory: WithingsFactory):
"""Test method."""
data = await withings_factory(WithingsFactoryConfig(measures=[const.MEAS_HEIGHT_M]))
await data.configure_all(WithingsFactoryConfig.PROFILE_1, "authorization_code")
state = data.hass.states.get("sensor.withings_height_m_person_1")
state_dict = state.as_dict()
assert state_dict.get("state") == "2"
assert state_dict.get("attributes") == {
"measurement": "height_m",
"measure_type": 4,
"friendly_name": "Withings height_m person_1",
"unit_of_measurement": "m",
"icon": "mdi:ruler",
}
SENSOR_TEST_DATA = [
(const.MEAS_WEIGHT_KG, 70),
(const.MEAS_FAT_MASS_KG, 5),
(const.MEAS_FAT_FREE_MASS_KG, 60),
(const.MEAS_MUSCLE_MASS_KG, 50),
(const.MEAS_BONE_MASS_KG, 10),
(const.MEAS_HEIGHT_M, 2),
(const.MEAS_FAT_RATIO_PCT, 0.07),
(const.MEAS_DIASTOLIC_MMHG, 70),
(const.MEAS_SYSTOLIC_MMGH, 100),
(const.MEAS_HEART_PULSE_BPM, 60),
(const.MEAS_SPO2_PCT, 0.95),
(const.MEAS_HYDRATION, 0.95),
(const.MEAS_PWV, 100),
(const.MEAS_SLEEP_WAKEUP_DURATION_SECONDS, 320),
(const.MEAS_SLEEP_LIGHT_DURATION_SECONDS, 520),
(const.MEAS_SLEEP_DEEP_DURATION_SECONDS, 720),
(const.MEAS_SLEEP_REM_DURATION_SECONDS, 920),
(const.MEAS_SLEEP_WAKEUP_COUNT, 1120),
(const.MEAS_SLEEP_TOSLEEP_DURATION_SECONDS, 1320),
(const.MEAS_SLEEP_TOWAKEUP_DURATION_SECONDS, 1520),
(const.MEAS_SLEEP_HEART_RATE_AVERAGE, 1720),
(const.MEAS_SLEEP_HEART_RATE_MIN, 1920),
(const.MEAS_SLEEP_HEART_RATE_MAX, 2120),
(const.MEAS_SLEEP_RESPIRATORY_RATE_AVERAGE, 2320),
(const.MEAS_SLEEP_RESPIRATORY_RATE_MIN, 2520),
(const.MEAS_SLEEP_RESPIRATORY_RATE_MAX, 2720),
]
@pytest.mark.parametrize("measure,expected", SENSOR_TEST_DATA)
async def test_health_sensor_throttled(
withings_factory: WithingsFactory, measure, expected
):
"""Test method."""
data = await withings_factory(WithingsFactoryConfig(measures=measure))
profile = WithingsFactoryConfig.PROFILE_1
await data.configure_all(profile, "authorization_code")
# Checking initial data.
assert_state_equals(data.hass, profile, measure, expected)
# Encountering a throttled data.
await async_update_entity(data.hass, get_entity_id(measure, profile))
assert_state_equals(data.hass, profile, measure, expected)
NONE_SENSOR_TEST_DATA = [
(const.MEAS_WEIGHT_KG, STATE_UNKNOWN),
(const.MEAS_SLEEP_STATE, STATE_UNKNOWN),
(const.MEAS_SLEEP_RESPIRATORY_RATE_MAX, STATE_UNKNOWN),
]
@pytest.mark.parametrize("measure,expected", NONE_SENSOR_TEST_DATA)
async def test_health_sensor_state_none(
withings_factory: WithingsFactory, measure, expected
):
"""Test method."""
data = await withings_factory(
WithingsFactoryConfig(
measures=measure,
withings_measures_response=None,
withings_sleep_response=None,
withings_sleep_summary_response=None,
)
)
profile = WithingsFactoryConfig.PROFILE_1
await data.configure_all(profile, "authorization_code")
# Checking initial data.
assert_state_equals(data.hass, profile, measure, expected)
# Encountering a throttled data.
await async_update_entity(data.hass, get_entity_id(measure, profile))
assert_state_equals(data.hass, profile, measure, expected)
EMPTY_SENSOR_TEST_DATA = [
(const.MEAS_WEIGHT_KG, STATE_UNKNOWN),
(const.MEAS_SLEEP_STATE, STATE_UNKNOWN),
(const.MEAS_SLEEP_RESPIRATORY_RATE_MAX, STATE_UNKNOWN),
]
@pytest.mark.parametrize("measure,expected", EMPTY_SENSOR_TEST_DATA)
async def test_health_sensor_state_empty(
withings_factory: WithingsFactory, measure, expected
):
"""Test method."""
data = await withings_factory(
WithingsFactoryConfig(
measures=measure,
withings_measures_response=WithingsMeasures({"measuregrps": []}),
withings_sleep_response=WithingsSleep({"series": []}),
withings_sleep_summary_response=WithingsSleepSummary({"series": []}),
)
)
profile = WithingsFactoryConfig.PROFILE_1
await data.configure_all(profile, "authorization_code")
# Checking initial data.
assert_state_equals(data.hass, profile, measure, expected)
# Encountering a throttled data.
await async_update_entity(data.hass, get_entity_id(measure, profile))
assert_state_equals(data.hass, profile, measure, expected)
SLEEP_STATES_TEST_DATA = [
(
const.STATE_AWAKE,
[const.MEASURE_TYPE_SLEEP_STATE_DEEP, const.MEASURE_TYPE_SLEEP_STATE_AWAKE],
),
(
const.STATE_LIGHT,
[const.MEASURE_TYPE_SLEEP_STATE_DEEP, const.MEASURE_TYPE_SLEEP_STATE_LIGHT],
),
(
const.STATE_REM,
[const.MEASURE_TYPE_SLEEP_STATE_DEEP, const.MEASURE_TYPE_SLEEP_STATE_REM],
),
(
const.STATE_DEEP,
[const.MEASURE_TYPE_SLEEP_STATE_LIGHT, const.MEASURE_TYPE_SLEEP_STATE_DEEP],
),
(const.STATE_UNKNOWN, [const.MEASURE_TYPE_SLEEP_STATE_LIGHT, "blah,"]),
]
@pytest.mark.parametrize("expected,sleep_states", SLEEP_STATES_TEST_DATA)
async def test_sleep_state_throttled(
withings_factory: WithingsFactory, expected, sleep_states
):
"""Test method."""
measure = const.MEAS_SLEEP_STATE
data = await withings_factory(
WithingsFactoryConfig(
measures=[measure],
withings_sleep_response=withings_sleep_response(sleep_states),
)
)
profile = WithingsFactoryConfig.PROFILE_1
await data.configure_all(profile, "authorization_code")
# Check initial data.
assert_state_equals(data.hass, profile, measure, expected)
# Encountering a throttled data.
await async_update_entity(data.hass, get_entity_id(measure, profile))
assert_state_equals(data.hass, profile, measure, expected)
async def test_async_setup_check_credentials(
hass: HomeAssistantType, withings_factory: WithingsFactory
):
"""Test method."""
check_creds_patch = asynctest.patch(
"homeassistant.components.withings.common.WithingsDataManager"
".check_authenticated",
side_effect=NotAuthenticatedError(),
)
async_init_patch = asynctest.patch.object(
hass.config_entries.flow,
"async_init",
wraps=hass.config_entries.flow.async_init,
)
with check_creds_patch, async_init_patch as async_init_mock:
data = await withings_factory(
WithingsFactoryConfig(measures=[const.MEAS_HEIGHT_M])
)
profile = WithingsFactoryConfig.PROFILE_1
await data.configure_all(profile, "authorization_code")
async_init_mock.assert_called_with(
const.DOMAIN,
context={"source": SOURCE_USER, const.PROFILE: profile},
data={},
)
async def test_async_setup_entry_credentials_saver(hass: HomeAssistantType):
"""Test method."""
expected_creds = {
"access_token": "my_access_token2",
"refresh_token": "my_refresh_token2",
"token_type": "my_token_type2",
"expires_in": "2",
}
original_withings_api = WithingsApi
withings_api_instance = None
def new_withings_api(*args, **kwargs):
nonlocal withings_api_instance
withings_api_instance = original_withings_api(*args, **kwargs)
withings_api_instance.request = MagicMock()
return withings_api_instance
withings_api_patch = patch("withings_api.WithingsApi", side_effect=new_withings_api)
session_patch = patch("requests_oauthlib.OAuth2Session")
client_patch = patch("oauthlib.oauth2.WebApplicationClient")
update_entry_patch = patch.object(
hass.config_entries,
"async_update_entry",
wraps=hass.config_entries.async_update_entry,
)
with session_patch, client_patch, withings_api_patch, update_entry_patch:
async_add_entities = MagicMock()
hass.config_entries.async_update_entry = MagicMock()
config_entry = ConfigEntry(
version=1,
domain=const.DOMAIN,
title="my title",
data={
const.PROFILE: "Person 1",
const.CREDENTIALS: {
"access_token": "my_access_token",
"refresh_token": "my_refresh_token",
"token_type": "my_token_type",
"token_expiry": "9999999999",
},
},
source="source",
connection_class="conn_class",
system_options={},
)
await async_setup_entry(hass, config_entry, async_add_entities)
withings_api_instance.set_token(expected_creds)
new_creds = config_entry.data[const.CREDENTIALS]
assert new_creds["access_token"] == "my_access_token2"