will/owlet
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactoring, added tests

Browse Source 
###Fix
# In light of submitting this as a pull request to the core of HA there have been some refactoring changes to comply with HA's style requirements
#Sensor names now moved to strings file to allow for translations
#Coordinator now properly handles multiple devices
#Spelling of signal strength sensor corrected

###Feature
#Tests added
This commit is contained in:
RyanClark123
2023-05-30 13:30:59 +01:00
parent c6f37493ca
commit 26f81c14bf
21 changed files with 4202 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/owlet/__init__.py
+31 -26
View File
@@ -1,6 +1,7 @@
"""The Owlet Smart Sock integration."""
from __future__ import annotations
import asyncio
import logging
from pyowletapi.api import OwletAPI
@@ -13,11 +14,7 @@ from pyowletapi.exceptions import (
)
from pyowletapi.sock import Sock
from homeassistant.config_entries import (
ConfigEntry,
ConfigEntryAuthFailed,
ConfigEntryNotReady,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_TOKEN,
CONF_REGION,
@@ -26,8 +23,8 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceEntry
from .const import CONF_OWLET_EXPIRY, CONF_OWLET_REFRESH, DOMAIN, SUPPORTED_VERSIONS
from .coordinator import OwletCoordinator
@@ -50,23 +47,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
try:
token = await owlet_api.authenticate()
if token:
if token := await owlet_api.authenticate():
hass.config_entries.async_update_entry(entry, data={**entry.data, **token})
devices = await owlet_api.get_devices(SUPPORTED_VERSIONS)
if devices["tokens"]:
hass.config_entries.async_update_entry(
entry, data={**entry.data, **devices["tokens"]}
)
socks = {
device["device"]["dsn"]: Sock(owlet_api, device["device"])
for device in devices["response"]
}
except (OwletAuthenticationError, OwletEmailError, OwletPasswordError) as err:
_LOGGER.error("Credentials no longer valid, please setup owlet again")
raise ConfigEntryAuthFailed(
@@ -78,14 +63,34 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
f"Error connecting to {entry.data[CONF_USERNAME]}"
) from err
coordinators = [
OwletCoordinator(hass, sock, entry.options.get(CONF_SCAN_INTERVAL))
for sock in socks.values()
]
except OwletDevicesError:
_LOGGER.error("No owlet devices found to set up")
return False
for coordinator in coordinators:
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = coordinator
if devices["tokens"]:
hass.config_entries.async_update_entry(
entry, data={**entry.data, **devices["tokens"]}
)
socks = {
device["device"]["dsn"]: Sock(owlet_api, device["device"])
for device in devices["response"]
}
scan_interval = entry.options.get(CONF_SCAN_INTERVAL)
coordinators = {
serial: OwletCoordinator(hass, sock, scan_interval)
for (serial, sock) in socks.items()
}
await asyncio.gather(
*(
coordinator.async_config_entry_first_refresh()
for coordinator in list(coordinators.values())
)
)
hass.data[DOMAIN][entry.entry_id] = coordinators
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
custom_components/owlet/binary_sensor.py
+10 -10
View File
@@ -34,61 +34,61 @@ class OwletBinarySensorEntityDescription(
SENSORS: tuple[OwletBinarySensorEntityDescription, ...] = (
OwletBinarySensorEntityDescription(
key="charging",
name="Charging",
translation_key="charging",
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
element="charging",
),
OwletBinarySensorEntityDescription(
key="highhr",
name="High heart rate alert",
translation_key="high_hr_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="high_heart_rate_alert",
),
OwletBinarySensorEntityDescription(
key="lowhr",
name="Low Heart Rate Alert",
translation_key="low_hr_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="low_heart_rate_alert",
),
OwletBinarySensorEntityDescription(
key="higho2",
name="High oxygen alert",
translation_key="high_ox_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="high_oxygen_alert",
),
OwletBinarySensorEntityDescription(
key="lowo2",
name="Low oxygen alert",
translation_key="low_ox_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="low_oxygen_alert",
),
OwletBinarySensorEntityDescription(
key="lowbattery",
name="Low Battery alert",
translation_key="low_batt_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="low_battery_alert",
),
OwletBinarySensorEntityDescription(
key="lostpower",
name="Lost power alert",
translation_key="lost_pwr_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="lost_power_alert",
),
OwletBinarySensorEntityDescription(
key="sockdisconnected",
name="Sock disconnected alert",
translation_key="sock_discon_alrt",
device_class=BinarySensorDeviceClass.SOUND,
element="sock_disconnected",
),
OwletBinarySensorEntityDescription(
key="sock_off",
name="Sock off",
translation_key="sock_off",
device_class=BinarySensorDeviceClass.POWER,
element="sock_off",
),
OwletBinarySensorEntityDescription(
key="awake",
name="Awake",
translation_key="awake",
element="sleep_state",
icon="mdi:sleep",
),
custom_components/owlet/config_flow.py
+32 -25
View File
@@ -1,11 +1,17 @@
"""Config flow for Owlet Smart Sock integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from pyowletapi.api import OwletAPI
from pyowletapi.exceptions import OwletDevicesError, OwletEmailError, OwletPasswordError
from pyowletapi.exceptions import (
OwletCredentialsError,
OwletDevicesError,
OwletEmailError,
OwletPasswordError,
)
from pyowletapi.sock import Sock
import voluptuous as vol
@@ -41,6 +47,7 @@ class OwletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self) -> None:
"""Initialise config flow."""
self._entry: ConfigEntry
self._region: str
self._username: str
@@ -70,29 +77,31 @@ class OwletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
try:
token = await owlet_api.authenticate()
try:
await owlet_api.validate_authentication()
return self.async_create_entry(
title=self._username,
data={
CONF_REGION: self._region,
CONF_USERNAME: self._username,
CONF_API_TOKEN: token[CONF_API_TOKEN],
CONF_OWLET_EXPIRY: token[CONF_OWLET_EXPIRY],
CONF_OWLET_REFRESH: token[CONF_OWLET_REFRESH],
},
options={CONF_SCAN_INTERVAL: POLLING_INTERVAL},
)
except OwletDevicesError:
errors["base"] = "no_devices"
await owlet_api.validate_authentication()
except OwletDevicesError:
errors["base"] = "no_devices"
except OwletEmailError:
errors["base"] = "invalid_email"
except OwletPasswordError:
errors["base"] = "invalid_password"
except OwletCredentialsError:
errors["base"] = "invalid_credentials"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=self._username,
data={
CONF_REGION: self._region,
CONF_USERNAME: self._username,
CONF_API_TOKEN: token[CONF_API_TOKEN],
CONF_OWLET_EXPIRY: token[CONF_OWLET_EXPIRY],
CONF_OWLET_REFRESH: token[CONF_OWLET_REFRESH],
},
options={CONF_SCAN_INTERVAL: POLLING_INTERVAL},
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
@@ -104,10 +113,8 @@ class OwletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
async def async_step_reauth(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle reauth"""
async def async_step_reauth(self, user_input: Mapping[str, Any]) -> FlowResult:
"""Handle reauth."""
self.reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
@@ -116,7 +123,7 @@ class OwletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that reauth is required"""
"""Dialog that informs the user that reauth is required."""
assert self.reauth_entry is not None
errors: dict[str, str] = {}
@@ -152,16 +159,16 @@ class OwletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a options flow for owlet"""
"""Handle a options flow for owlet."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialise options flow"""
"""Initialise options flow."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle options flow"""
"""Handle options flow."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
@@ -178,4 +185,4 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
class InvalidAuth(exceptions.HomeAssistantError):
"""Error to indiciate there is invalud auth"""
"""Error to indicate there is invalid auth."""
custom_components/owlet/const.py
+1 -1
View File
@@ -8,4 +8,4 @@ CONF_OWLET_REFRESH = "refresh"
SUPPORTED_VERSIONS = [3]
POLLING_INTERVAL = 5
MANUFACTURER = "Owlet Baby Care"
SLEEP_STATES = {0: "Unknown", 1: "Awake", 8: "Light Sleep", 15: "Deep Sleep"}
SLEEP_STATES = {0: "unknown", 1: "awake", 8: "light_sleep", 15: "deep_sleep"}
custom_components/owlet/coordinator.py
+9 -14
View File
@@ -11,14 +11,12 @@ from pyowletapi.exceptions import (
)
from pyowletapi.sock import Sock
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
ConfigEntryAuthFailed,
DataUpdateCoordinator,
UpdateFailed,
)
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, MANUFACTURER
@@ -28,7 +26,7 @@ _LOGGER = logging.getLogger(__name__)
class OwletCoordinator(DataUpdateCoordinator):
"""Coordinator is responsible for querying the device at a specified route."""
def __init__(self, hass: HomeAssistant, sock: Sock, interval: int) -> None:
def __init__(self, hass: HomeAssistant, sock: Sock, interval) -> None:
"""Initialise a custom coordinator."""
super().__init__(
hass,
@@ -37,18 +35,15 @@ class OwletCoordinator(DataUpdateCoordinator):
update_interval=timedelta(seconds=interval),
)
assert self.config_entry is not None
self._device_unique_id = sock.serial
self._model = sock.model
self._sw_version = sock.sw_version
self._hw_version = sock.version
self.config_entry: ConfigEntry
self.sock = sock
self.device_info = DeviceInfo(
identifiers={(DOMAIN, self._device_unique_id)},
identifiers={(DOMAIN, sock.serial)},
name="Owlet Baby Care Sock",
manufacturer=MANUFACTURER,
model=self._model,
sw_version=self._sw_version,
hw_version=self._hw_version,
model=sock.model,
sw_version=sock.sw_version,
hw_version=sock.version,
)
async def _async_update_data(self) -> None:
custom_components/owlet/manifest.json
+2 -3
View File
@@ -3,9 +3,8 @@
"name": "Owlet Smart Sock",
"codeowners": ["@ryanbdclark"],
"config_flow": true,
"dependencies": [],
"documentation": "https://www.home-assistant.io/integrations/owlet",
"homekit": {},
"iot_class": "cloud_polling",
"requirements": ["pyowletapi==2023.5.28"]
"requirements": ["pyowletapi==2023.5.30"],
"version":"2023.5.6"
}
custom_components/owlet/sensor.py
+28 -19
View File
@@ -27,7 +27,7 @@ from .entity import OwletBaseEntity
@dataclass
class OwletSensorEntityDescriptionMixin:
"""Owlet sensor description mix in"""
"""Owlet sensor description mix in."""
element: str
@@ -42,7 +42,7 @@ class OwletSensorEntityDescription(
SENSORS: tuple[OwletSensorEntityDescription, ...] = (
OwletSensorEntityDescription(
key="batterypercentage",
name="Battery",
translation_key="batterypercent",
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
state_class=SensorStateClass.MEASUREMENT,
@@ -50,7 +50,7 @@ SENSORS: tuple[OwletSensorEntityDescription, ...] = (
),
OwletSensorEntityDescription(
key="oxygensaturation",
name="O2 Saturation",
translation_key="o2saturation",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
element="oxygen_saturation",
@@ -58,7 +58,7 @@ SENSORS: tuple[OwletSensorEntityDescription, ...] = (
),
OwletSensorEntityDescription(
key="oxygensaturation10a",
name="O2 Saturation 10 Minute Average",
translation_key="o2saturation10a",
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
element="oxygen_10_av",
@@ -66,7 +66,7 @@ SENSORS: tuple[OwletSensorEntityDescription, ...] = (
),
OwletSensorEntityDescription(
key="heartrate",
name="Heart rate",
translation_key="heartrate",
native_unit_of_measurement="bpm",
state_class=SensorStateClass.MEASUREMENT,
element="heart_rate",
@@ -74,7 +74,7 @@ SENSORS: tuple[OwletSensorEntityDescription, ...] = (
),
OwletSensorEntityDescription(
key="batteryminutes",
name="Battery Remaining",
translation_key="batterymin",
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=SensorDeviceClass.DURATION,
state_class=SensorStateClass.MEASUREMENT,
@@ -82,7 +82,7 @@ SENSORS: tuple[OwletSensorEntityDescription, ...] = (
),
OwletSensorEntityDescription(
key="signalstrength",
name="Signal Strength",
translation_key="signalstrength",
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
state_class=SensorStateClass.MEASUREMENT,
@@ -90,7 +90,7 @@ SENSORS: tuple[OwletSensorEntityDescription, ...] = (
),
OwletSensorEntityDescription(
key="skintemp",
name="Skin Temperature",
translation_key="skintemp",
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
@@ -106,12 +106,19 @@ async def async_setup_entry(
) -> None:
"""Set up the owlet sensors from config entry."""
coordinator: OwletCoordinator = hass.data[DOMAIN][config_entry.entry_id]
coordinators: list[OwletCoordinator] = list(
hass.data[DOMAIN][config_entry.entry_id].values()
)
entities = [OwletSensor(coordinator, sensor) for sensor in SENSORS]
entities.append(OwletSleepStateSensor(coordinator))
async_add_entities(
OwletSensor(coordinator, sensor)
for coordinator in coordinators
for sensor in SENSORS
)
async_add_entities(entities)
async_add_entities(
OwletSleepStateSensor(coordinator) for coordinator in coordinators
)
class OwletSensor(OwletBaseEntity, SensorEntity):
@@ -124,12 +131,14 @@ class OwletSensor(OwletBaseEntity, SensorEntity):
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = sensor_description
self._attr_unique_id = f"{self.sock.serial}-{self.entity_description.name}"
self.entity_description: OwletSensorEntityDescription = sensor_description
self._attr_unique_id = (
f"{self.sock.serial}-{self.entity_description.translation_key}"
)
@property
def native_value(self) -> StateType:
"""Return sensor value"""
"""Return sensor value."""
if (
self.entity_description.element
@@ -158,20 +167,20 @@ class OwletSleepStateSensor(OwletBaseEntity, SensorEntity):
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._attr_unique_id = f"{self.sock.serial}-Sleep State"
self._attr_unique_id = f"{self.sock.serial}-sleepstate"
self._attr_icon = "mdi:sleep"
self._attr_device_class = SensorDeviceClass.ENUM
self._attr_translation_key = "sleepstate"
self._attr_name = "Sleep State"
@property
def native_value(self) -> str:
"""Return sensor value"""
"""Return sensor value."""
if self.sock.properties["charging"]:
return None
return "unknown"
return SLEEP_STATES[self.sock.properties["sleep_state"]]
@property
def options(self) -> list[str]:
"""Set options for sleep state."""
return list(SLEEP_STATES.values())
custom_components/owlet/strings.json
+78 -10
View File
@@ -1,17 +1,17 @@
{
"config": {
"step": {
"user":{
"user": {
"title": "Enter login details",
"data":{
"data": {
"region": "Region",
"username": "Email",
"password": "Password"
}
},
"reauth_confirm":{
"reauth_confirm": {
"title": "Reauthentiaction required for Owlet",
"data":{
"data": {
"password": "Password"
}
}
@@ -20,20 +20,88 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_email": "Entered email address is incorrect",
"invalid_password": "Entered password is incorrect",
"unknown": "[%key:common::config_flow::error::unknown%]"
"invalid_credentials": "Entered credentials are incorrect",
"unknown": "Unknown error occured"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "Device already configured",
"reauth_successful": "Reauthentication successful"
}
},
"options": {
"step": {
"init":{
"title":"Configure options for Owlet",
"data":{
"init": {
"title": "Configure options for Owlet",
"data": {
"pollinterval": "Polling interval in seconds, min 10"
}
}
}
},
"entity": {
"binary_sensor":{
"charging": {
"name": "Charging"
},
"high_hr_alrt":{
"name": "High Heart Rate Alert"
},
"low_hr_alrt":{
"name":"Low Heart Rate Alert"
},
"high_ox_alrt":{
"name":"High Oxygen Alert"
},
"low_ox_alrt":{
"name":"Low Oxygen Alert"
},
"low_batt_alrt":{
"name": "Low Battery Alert"
},
"lost_pwr_alrt":{
"name": "Lost Power Alert"
},
"sock_discon_alrt":{
"name": "Sock Diconnected Alert"
},
"sock_off":{
"name":"sock_off"
},
"awake":{
"name":"Awake"
}
},
"sensor": {
"batterypercent": {
"name": "Battery Percentage"
},
"o2saturation": {
"name": "O2 Saturation"
},
"o2saturation10a": {
"name": "O2 Saturation 10 Minute Average"
},
"heartrate": {
"name": "Heart Rate"
},
"batterymin": {
"name": "Battery Remaining"
},
"signalstrength": {
"name": "Signal Strength"
},
"skintemp": {
"name": "Skin Temperature"
},
"sleepstate": {
"name": "Sleep State",
"state": {
"unknown": "Unknown",
"awake": "Awake",
"light_sleep": "Light Sleep",
"deep_sleep": "Deep Sleep"
}
}
}
}
}
}
custom_components/owlet/translations/en.json
+78 -10
View File
@@ -1,17 +1,17 @@
{
"config": {
"step": {
"user":{
"user": {
"title": "Enter login details",
"data":{
"data": {
"region": "Region",
"username": "Email",
"password": "Password"
}
},
"reauth_confirm":{
"reauth_confirm": {
"title": "Reauthentiaction required for Owlet",
"data":{
"data": {
"password": "Password"
}
}
@@ -20,20 +20,88 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_email": "Entered email address is incorrect",
"invalid_password": "Entered password is incorrect",
"unknown": "[%key:common::config_flow::error::unknown%]"
"invalid_credentials": "Entered credentials are incorrect",
"unknown": "Unknown error occured"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "Device already configured",
"reauth_successful": "Reauthentication successful"
}
},
"options": {
"step": {
"init":{
"title":"Configure options for Owlet",
"data":{
"init": {
"title": "Configure options for Owlet",
"data": {
"pollinterval": "Polling interval in seconds, min 10"
}
}
}
},
"entity": {
"binary_sensor":{
"charging": {
"name": "Charging"
},
"high_hr_alrt":{
"name": "High Heart Rate Alert"
},
"low_hr_alrt":{
"name":"Low Heart Rate Alert"
},
"high_ox_alrt":{
"name":"High Oxygen Alert"
},
"low_ox_alrt":{
"name":"Low Oxygen Alert"
},
"low_batt_alrt":{
"name": "Low Battery Alert"
},
"lost_pwr_alrt":{
"name": "Lost Power Alert"
},
"sock_discon_alrt":{
"name": "Sock Diconnected Alert"
},
"sock_off":{
"name":"sock_off"
},
"awake":{
"name":"Awake"
}
},
"sensor": {
"batterypercent": {
"name": "Battery Percentage"
},
"o2saturation": {
"name": "O2 Saturation"
},
"o2saturation10a": {
"name": "O2 Saturation 10 Minute Average"
},
"heartrate": {
"name": "Heart Rate"
},
"batterymin": {
"name": "Battery Remaining"
},
"signalstrength": {
"name": "Signal Strength"
},
"skintemp": {
"name": "Skin Temperature"
},
"sleepstate": {
"name": "Sleep State",
"state": {
"unknown": "Unknown",
"awake": "Awake",
"light_sleep": "Light Sleep",
"deep_sleep": "Deep Sleep"
}
}
}
}
}
}
custom_components/owlet/translations/uk.json
+78 -10
View File
@@ -1,17 +1,17 @@
{
"config": {
"step": {
"user":{
"user": {
"title": "Enter login details",
"data":{
"data": {
"region": "Region",
"username": "Email",
"password": "Password"
}
},
"reauth_confirm":{
"reauth_confirm": {
"title": "Reauthentiaction required for Owlet",
"data":{
"data": {
"password": "Password"
}
}
@@ -20,20 +20,88 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_email": "Entered email address is incorrect",
"invalid_password": "Entered password is incorrect",
"unknown": "[%key:common::config_flow::error::unknown%]"
"invalid_credentials": "Entered credentials are incorrect",
"unknown": "Unknown error occured"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "Device already configured",
"reauth_successful": "Reauthentication successful"
}
},
"options": {
"step": {
"init":{
"title":"Configure options for Owlet",
"data":{
"init": {
"title": "Configure options for Owlet",
"data": {
"pollinterval": "Polling interval in seconds, min 10"
}
}
}
},
"entity": {
"binary_sensor":{
"charging": {
"name": "Charging"
},
"high_hr_alrt":{
"name": "High Heart Rate Alert"
},
"low_hr_alrt":{
"name":"Low Heart Rate Alert"
},
"high_ox_alrt":{
"name":"High Oxygen Alert"
},
"low_ox_alrt":{
"name":"Low Oxygen Alert"
},
"low_batt_alrt":{
"name": "Low Battery Alert"
},
"lost_pwr_alrt":{
"name": "Lost Power Alert"
},
"sock_discon_alrt":{
"name": "Sock Diconnected Alert"
},
"sock_off":{
"name":"Sock Off"
},
"awake":{
"name":"Awake"
}
},
"sensor": {
"batterypercent": {
"name": "Battery Percentage"
},
"o2saturation": {
"name": "O2 Saturation"
},
"o2saturation10a": {
"name": "O2 Saturation 10 Minute Average"
},
"heartrate": {
"name": "Heart Rate"
},
"batterymin": {
"name": "Battery Remaining"
},
"signalstrength": {
"name": "Signal Strength"
},
"skintemp": {
"name": "Skin Temperature"
},
"sleepstate": {
"name": "Sleep State",
"state": {
"unknown": "Unknown",
"awake": "Awake",
"light_sleep": "Light Sleep",
"deep_sleep": "Deep Sleep"
}
}
}
}
}
}
tests/__init__.py
+60
View File
@@ -0,0 +1,60 @@
"""Tests for the Owlet integration."""
from __future__ import annotations
import json
from unittest.mock import patch
from homeassistant.components.owlet.const import (
CONF_OWLET_EXPIRY,
CONF_OWLET_REFRESH,
DOMAIN,
POLLING_INTERVAL,
)
from homeassistant.const import (
CONF_API_TOKEN,
CONF_REGION,
CONF_SCAN_INTERVAL,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_fixture
async def async_init_integration(
hass: HomeAssistant,
skip_setup: bool = False,
properties_fixture: str = "update_properties_charging.json",
devices_fixture: str = "get_devices.json",
) -> MockConfigEntry:
"""Set up integration entry."""
entry = MockConfigEntry(
domain=DOMAIN,
title="sample@gmail.com",
unique_id="sample@gmail.com",
data={
CONF_REGION: "europe",
CONF_USERNAME: "sample@gmail.com",
CONF_API_TOKEN: "api_token",
CONF_OWLET_EXPIRY: 100,
CONF_OWLET_REFRESH: "refresh_token",
},
options={CONF_SCAN_INTERVAL: POLLING_INTERVAL},
)
entry.add_to_hass(hass)
if not skip_setup:
with patch(
"homeassistant.components.owlet.OwletAPI.get_properties",
return_value=json.loads(load_fixture(properties_fixture, "owlet")),
), patch(
"homeassistant.components.owlet.OwletAPI.authenticate", return_value=None
), patch(
"homeassistant.components.owlet.OwletAPI.get_devices",
return_value=json.loads(load_fixture(devices_fixture, "owlet")),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
tests/const.py
+14
View File
@@ -0,0 +1,14 @@
"""Constants used for Owlet tests."""
AUTH_RETURN = {
"api_token": "api_token",
"expiry": 100,
"refresh": "refresh_token",
}
CONF_INPUT = {
"region": "europe",
"username": "sample@gmail.com",
"password": "sample",
}
tests/fixtures/get_devices.json
Vendored
+32
View File
@@ -0,0 +1,32 @@
{
"response": [
{
"device": {
"product_name": "Owlet Baby Monitors",
"model": "AY001MTL1",
"dsn": "SERIAL_NUMBER",
"oem_model": "SS3-OBL-EU",
"sw_version": "bc 2.9.7-beta 05/01/19 16:15:18 ID beb2858",
"template_id": 48775,
"mac": "MAC",
"unique_hardware_id": "None",
"hwsig": "123-123-123",
"lan_ip": "192.0.0.1",
"connected_at": "2023-05-23T01:21:29Z",
"key": 123456,
"lan_enabled": false,
"connection_priority": [],
"has_properties": true,
"product_class": "None",
"connection_status": "Online",
"lat": "0.0",
"lng": "0.0",
"locality": "QW12",
"device_type": "Wifi",
"dealer": "None",
"manuf_model": "LBEE5PA1LD-222"
}
}
],
"tokens": ""
}
tests/fixtures/get_devices_with_tokens.json
Vendored
+36
View File
@@ -0,0 +1,36 @@
{
"response": [
{
"device": {
"product_name": "Owlet Baby Monitors",
"model": "AY001MTL1",
"dsn": "SERIAL_NUMBER",
"oem_model": "SS3-OBL-EU",
"sw_version": "bc 2.9.7-beta 05/01/19 16:15:18 ID beb2858",
"template_id": 48775,
"mac": "MAC",
"unique_hardware_id": "None",
"hwsig": "123-123-123",
"lan_ip": "192.0.0.1",
"connected_at": "2023-05-23T01:21:29Z",
"key": 123456,
"lan_enabled": false,
"connection_priority": [],
"has_properties": true,
"product_class": "None",
"connection_status": "Online",
"lat": "0.0",
"lng": "0.0",
"locality": "QW12",
"device_type": "Wifi",
"dealer": "None",
"manuf_model": "LBEE5PA1LD-222"
}
}
],
"tokens": {
"api_token": "new_api_token",
"expiry": 200,
"refresh": "new_refresh_token"
}
}
tests/fixtures/update_properties_asleep.json
Vendored
+1051
View File
File diff suppressed because it is too large Load Diff
tests/fixtures/update_properties_awake.json
Vendored
+1051
View File
File diff suppressed because it is too large Load Diff
tests/fixtures/update_properties_charging.json
Vendored
+1055
View File
File diff suppressed because it is too large Load Diff
tests/test_config_flow.py
+237
View File
@@ -0,0 +1,237 @@
"""Test Owlet config flow."""
from __future__ import annotations
from unittest.mock import patch
from pyowletapi.exceptions import (
OwletCredentialsError,
OwletDevicesError,
OwletEmailError,
OwletPasswordError,
)
from homeassistant import config_entries
from homeassistant.components.owlet.const import DOMAIN, POLLING_INTERVAL
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import CONF_PASSWORD, CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import async_init_integration
from .const import AUTH_RETURN, CONF_INPUT
async def test_form(hass: HomeAssistant) -> None:
"""Test that the form is served with no input."""
# await async_init_integration(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
return_value=AUTH_RETURN,
), patch(
"homeassistant.components.owlet.config_flow.OwletAPI.validate_authentication"
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "sample@gmail.com"
assert result["data"] == {
"region": "europe",
"username": "sample@gmail.com",
"api_token": "api_token",
"expiry": 100,
"refresh": "refresh_token",
}
assert result["options"] == {"scan_interval": POLLING_INTERVAL}
async def test_flow_wrong_password(hass: HomeAssistant) -> None:
"""Test incorrect login throwing error."""
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
side_effect=OwletPasswordError(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "invalid_password"}
async def test_flow_wrong_email(hass: HomeAssistant) -> None:
"""Test incorrect login throwing error."""
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
side_effect=OwletEmailError(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "invalid_email"}
async def test_flow_credentials_error(hass: HomeAssistant) -> None:
"""Test incorrect login throwing error."""
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
side_effect=OwletCredentialsError(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "invalid_credentials"}
async def test_flow_unknown_error(hass: HomeAssistant) -> None:
"""Test unknown error throwing error."""
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
side_effect=Exception(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
async def test_flow_no_devices(hass: HomeAssistant) -> None:
"""Test unknown error throwing error."""
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate"
), patch(
"homeassistant.components.owlet.config_flow.OwletAPI.validate_authentication",
side_effect=OwletDevicesError(),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "no_devices"}
async def test_reauth_success(hass: HomeAssistant) -> None:
"""Test reauth form."""
entry = await async_init_integration(hass, skip_setup=True)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH, "entry_id": entry.entry_id}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
return_value=AUTH_RETURN,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PASSWORD: "sample"},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
await hass.config_entries.async_unload(entry.entry_id)
async def test_reauth_invalid_password(hass: HomeAssistant) -> None:
"""Test reauth with invalid password errir."""
entry = await async_init_integration(hass, skip_setup=True)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH, "entry_id": entry.entry_id}
)
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
side_effect=OwletPasswordError(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PASSWORD: "sample"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "invalid_password"}
async def test_reauth_unknown_error(hass: HomeAssistant) -> None:
"""Test reauthing with an unknown error."""
entry = await async_init_integration(hass, skip_setup=True)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH, "entry_id": entry.entry_id}
)
with patch(
"homeassistant.components.owlet.config_flow.OwletAPI.authenticate",
side_effect=Exception(),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PASSWORD: "sample"}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test that the form is served with no input."""
entry = await async_init_integration(hass, skip_setup=True)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_SCAN_INTERVAL: 10},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == ""
assert result["data"] == {CONF_SCAN_INTERVAL: 10}
tests/test_coordinator.py
+74
View File
@@ -0,0 +1,74 @@
"""Test owlet coordinator."""
from __future__ import annotations
import json
from unittest.mock import patch
from pyowletapi.exceptions import OwletAuthenticationError, OwletConnectionError
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from . import async_init_integration
from tests.common import load_fixture
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def test_coordinator_auth_error(hass: HomeAssistant) -> None:
"""Test coordinator setup authentication error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.Sock.update_properties",
side_effect=OwletAuthenticationError(),
), patch(
"homeassistant.components.owlet.OwletAPI.authenticate", return_value=None
), patch(
"homeassistant.components.owlet.OwletAPI.get_devices",
return_value=json.loads(load_fixture("get_devices.json", "owlet")),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_RETRY
async def test_coordinator_connection_error(hass: HomeAssistant) -> None:
"""Test coordinator setup connection error error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.Sock.update_properties",
side_effect=OwletConnectionError(),
), patch(
"homeassistant.components.owlet.OwletAPI.authenticate", return_value=None
), patch(
"homeassistant.components.owlet.OwletAPI.get_devices",
return_value=json.loads(load_fixture("get_devices.json", "owlet")),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_RETRY
async def test_coordinator_error(hass: HomeAssistant) -> None:
"""Test coordinator setup generic error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.Sock.update_properties",
side_effect=Exception(),
), patch(
"homeassistant.components.owlet.OwletAPI.authenticate", return_value=None
), patch(
"homeassistant.components.owlet.OwletAPI.get_devices",
return_value=json.loads(load_fixture("get_devices.json", "owlet")),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_RETRY
tests/test_init.py
+136
View File
@@ -0,0 +1,136 @@
"""Test Owlet init."""
from __future__ import annotations
from unittest.mock import patch
from pyowletapi.exceptions import (
OwletAuthenticationError,
OwletConnectionError,
OwletDevicesError,
OwletError,
)
from homeassistant.components.owlet.const import (
CONF_OWLET_EXPIRY,
CONF_OWLET_REFRESH,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_API_TOKEN, CONF_REGION, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import async_init_integration
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def test_async_setup_entry(hass: HomeAssistant) -> None:
"""Test setting up entry."""
entry = await async_init_integration(hass)
assert entry.state == ConfigEntryState.LOADED
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=entry.entry_id, identifiers={(DOMAIN, "SERIAL_NUMBER")}
)
assert device_entry.name == "Owlet Baby Care Sock"
entity_registry = er.async_get(hass)
entities = er.async_entries_for_device(entity_registry, device_entry.id)
assert len(entities) == 8
await entry.async_unload(hass)
assert entry.state == ConfigEntryState.NOT_LOADED
async def test_async_setup_entry_new_tokens(hass: HomeAssistant) -> None:
"""Test setting up entry and getting new tokens."""
entry = await async_init_integration(
hass, devices_fixture="get_devices_with_tokens.json"
)
assert entry.data == {
CONF_REGION: "europe",
CONF_USERNAME: "sample@gmail.com",
CONF_API_TOKEN: "new_api_token",
CONF_OWLET_EXPIRY: 200,
CONF_OWLET_REFRESH: "new_refresh_token",
}
assert entry.state == ConfigEntryState.LOADED
await entry.async_unload(hass)
assert entry.state == ConfigEntryState.NOT_LOADED
async def test_async_setup_entry_auth_error(hass: HomeAssistant) -> None:
"""Test setting up entry with auth error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.OwletAPI.authenticate",
side_effect=OwletAuthenticationError(),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
await entry.async_unload(hass)
async def test_async_setup_entry_connection_error(hass: HomeAssistant) -> None:
"""Test setting up entry with connection error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.OwletAPI.authenticate",
side_effect=OwletConnectionError(),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_RETRY
await entry.async_unload(hass)
async def test_async_setup_entry_devices_error(hass: HomeAssistant) -> None:
"""Test setting up entry with device error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.OwletAPI.authenticate", return_value=None
), patch(
"homeassistant.components.owlet.OwletAPI.get_devices",
side_effect=OwletDevicesError(),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
await entry.async_unload(hass)
async def test_async_setup_entry_error(hass: HomeAssistant) -> None:
"""Test setting up entry with unknown error."""
entry = await async_init_integration(hass, skip_setup=True)
with patch(
"homeassistant.components.owlet.OwletAPI.authenticate",
side_effect=OwletError(),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
await entry.async_unload(hass)
tests/test_sensor.py
+109
View File
@@ -0,0 +1,109 @@
"""Test Owlet Sensor."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from . import async_init_integration
async def test_sensors_asleep(hass: HomeAssistant) -> None:
"""Test sensor values."""
await async_init_integration(
hass, properties_fixture="update_properties_asleep.json"
)
assert len(hass.states.async_all("sensor")) == 8
assert (
hass.states.get("sensor.owlet_baby_care_sock_battery_percentage").state
== "50.0"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_battery_remaining").state
== "400.0"
)
assert hass.states.get("sensor.owlet_baby_care_sock_heart_rate").state == "97.0"
assert hass.states.get("sensor.owlet_baby_care_sock_o2_saturation").state == "99.0"
assert (
hass.states.get(
"sensor.owlet_baby_care_sock_o2_saturation_10_minute_average"
).state
== "97.0"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_signal_strength").state == "30.0"
)
assert hass.states.get("sensor.owlet_baby_care_sock_skin_temperature").state == "34"
assert (
hass.states.get("sensor.owlet_baby_care_sock_sleep_state").state
== "light_sleep"
)
async def test_sensors_awake(hass: HomeAssistant) -> None:
"""Test sensor values."""
await async_init_integration(
hass, properties_fixture="update_properties_awake.json"
)
assert len(hass.states.async_all("sensor")) == 8
assert (
hass.states.get("sensor.owlet_baby_care_sock_battery_percentage").state
== "80.0"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_battery_remaining").state
== "600.0"
)
assert hass.states.get("sensor.owlet_baby_care_sock_heart_rate").state == "110.0"
assert hass.states.get("sensor.owlet_baby_care_sock_o2_saturation").state == "98.0"
assert (
hass.states.get(
"sensor.owlet_baby_care_sock_o2_saturation_10_minute_average"
).state
== "98.0"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_signal_strength").state == "34.0"
)
assert hass.states.get("sensor.owlet_baby_care_sock_skin_temperature").state == "35"
assert hass.states.get("sensor.owlet_baby_care_sock_sleep_state").state == "awake"
async def test_sensors_charging(hass: HomeAssistant) -> None:
"""Test sensor values."""
await async_init_integration(
hass, properties_fixture="update_properties_charging.json"
)
assert len(hass.states.async_all("sensor")) == 8
assert (
hass.states.get("sensor.owlet_baby_care_sock_battery_percentage").state
== "100.0"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_battery_remaining").state
== "unknown"
)
assert hass.states.get("sensor.owlet_baby_care_sock_heart_rate").state == "unknown"
assert (
hass.states.get("sensor.owlet_baby_care_sock_o2_saturation").state == "unknown"
)
assert (
hass.states.get(
"sensor.owlet_baby_care_sock_o2_saturation_10_minute_average"
).state
== "unknown"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_signal_strength").state == "34.0"
)
assert (
hass.states.get("sensor.owlet_baby_care_sock_skin_temperature").state
== "unknown"
)
assert hass.states.get("sensor.owlet_baby_care_sock_sleep_state").state == "unknown"