jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 06:07:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to eight_sleep (#71095)

Browse Source 
* Add config flow to eight_sleep

* simplify tests

* Remove extra file

* remove unused import

* fix redundant code

* Update homeassistant/components/eight_sleep/__init__.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* incorporate feedback

* Review comments

* remove typing from tests

* Fix based on changes

* Fix requirements

* Remove stale comment

* Fix tests

* Reverse the flow and force the config entry to reconnect

* Review comments

* Abort if import flow fails

* Split import and user logic

* Fix error

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Raman Gupta 2022-06-11 02:16:46 -04:00 committed by GitHub
parent 63b51f566d
commit dc48791864
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 432 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.coveragerc
View File

@ -262,7 +262,9 @@ omit =
homeassistant/components/eddystone_temperature/sensor.py
homeassistant/components/edimax/switch.py
homeassistant/components/egardia/*
homeassistant/components/eight_sleep/*
homeassistant/components/eight_sleep/__init__.py
homeassistant/components/eight_sleep/binary_sensor.py
homeassistant/components/eight_sleep/sensor.py
homeassistant/components/eliqonline/sensor.py
homeassistant/components/elkm1/__init__.py
homeassistant/components/elkm1/alarm_control_panel.py

1
CODEOWNERS
View File

@ -273,6 +273,7 @@ build.json @home-assistant/supervisor
/tests/components/efergy/ @tkdrob
/homeassistant/components/egardia/ @jeroenterheerdt
/homeassistant/components/eight_sleep/ @mezz64 @raman325
/tests/components/eight_sleep/ @mezz64 @raman325
/homeassistant/components/elgato/ @frenck
/tests/components/elgato/ @frenck
/homeassistant/components/elkm1/ @gwww @bdraco

181
homeassistant/components/eight_sleep/__init__.py
View File

@ -1,34 +1,38 @@
"""Support for Eight smart mattress covers and mattresses."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from pyeight.eight import EightSleep
from pyeight.exceptions import RequestError
from pyeight.user import EightUser
import voluptuous as vol
from homeassistant.const import ATTR_ENTITY_ID, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import discovery
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
ATTR_HW_VERSION,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_SW_VERSION,
CONF_PASSWORD,
CONF_USERNAME,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.device_registry import async_get
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.typing import UNDEFINED, ConfigType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import (
ATTR_HEAT_DURATION,
ATTR_TARGET_HEAT,
DATA_API,
DATA_HEAT,
DATA_USER,
DOMAIN,
NAME_MAP,
SERVICE_HEAT_SET,
)
from .const import DOMAIN, NAME_MAP
_LOGGER = logging.getLogger(__name__)
@ -37,17 +41,6 @@ PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
HEAT_SCAN_INTERVAL = timedelta(seconds=60)
USER_SCAN_INTERVAL = timedelta(seconds=300)
VALID_TARGET_HEAT = vol.All(vol.Coerce(int), vol.Clamp(min=-100, max=100))
VALID_DURATION = vol.All(vol.Coerce(int), vol.Clamp(min=0, max=28800))
SERVICE_EIGHT_SCHEMA = vol.Schema(
{
ATTR_ENTITY_ID: cv.entity_ids,
ATTR_TARGET_HEAT: VALID_TARGET_HEAT,
ATTR_HEAT_DURATION: VALID_DURATION,
}
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
@ -61,6 +54,15 @@ CONFIG_SCHEMA = vol.Schema(
)
@dataclass
class EightSleepConfigEntryData:
"""Data used for all entities for a given config entry."""
api: EightSleep
heat_coordinator: DataUpdateCoordinator
user_coordinator: DataUpdateCoordinator
def _get_device_unique_id(eight: EightSleep, user_obj: EightUser | None = None) -> str:
"""Get the device's unique ID."""
unique_id = eight.device_id
@ -71,23 +73,36 @@ def _get_device_unique_id(eight: EightSleep, user_obj: EightUser | None = None)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Eight Sleep component."""
"""Old set up method for the Eight Sleep component."""
if DOMAIN in config:
_LOGGER.warning(
"Your Eight Sleep configuration has been imported into the UI; "
"please remove it from configuration.yaml as support for it "
"will be removed in a future release"
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config[DOMAIN]
)
)
if DOMAIN not in config:
return True
return True
conf = config[DOMAIN]
user = conf[CONF_USERNAME]
password = conf[CONF_PASSWORD]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the Eight Sleep config entry."""
eight = EightSleep(
user, password, hass.config.time_zone, async_get_clientsession(hass)
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
hass.config.time_zone,
async_get_clientsession(hass),
)
hass.data.setdefault(DOMAIN, {})
# Authenticate, build sensors
success = await eight.start()
try:
success = await eight.start()
except RequestError as err:
raise ConfigEntryNotReady from err
if not success:
# Authentication failed, cannot continue
return False
@ -113,47 +128,60 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
# No users, cannot continue
return False
hass.data[DOMAIN] = {
DATA_API: eight,
DATA_HEAT: heat_coordinator,
DATA_USER: user_coordinator,
dev_reg = async_get(hass)
assert eight.device_data
device_data = {
ATTR_MANUFACTURER: "Eight Sleep",
ATTR_MODEL: eight.device_data.get("modelString", UNDEFINED),
ATTR_HW_VERSION: eight.device_data.get("sensorInfo", {}).get(
"hwRevision", UNDEFINED
),
ATTR_SW_VERSION: eight.device_data.get("firmwareVersion", UNDEFINED),
}
for platform in PLATFORMS:
hass.async_create_task(
discovery.async_load_platform(hass, platform, DOMAIN, {}, config)
dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, _get_device_unique_id(eight))},
name=f"{entry.data[CONF_USERNAME]}'s Eight Sleep",
**device_data,
)
for user in eight.users.values():
assert user.user_profile
dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, _get_device_unique_id(eight, user))},
name=f"{user.user_profile['firstName']}'s Eight Sleep Side",
via_device=(DOMAIN, _get_device_unique_id(eight)),
**device_data,
)
async def async_service_handler(service: ServiceCall) -> None:
"""Handle eight sleep service calls."""
params = service.data.copy()
sensor = params.pop(ATTR_ENTITY_ID, None)
target = params.pop(ATTR_TARGET_HEAT, None)
duration = params.pop(ATTR_HEAT_DURATION, 0)
for sens in sensor:
side = sens.split("_")[1]
user_id = eight.fetch_user_id(side)
assert user_id
usr_obj = eight.users[user_id]
await usr_obj.set_heating_level(target, duration)
await heat_coordinator.async_request_refresh()
# Register services
hass.services.async_register(
DOMAIN, SERVICE_HEAT_SET, async_service_handler, schema=SERVICE_EIGHT_SCHEMA
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = EightSleepConfigEntryData(
eight, heat_coordinator, user_coordinator
)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
# stop the API before unloading everything
config_entry_data: EightSleepConfigEntryData = hass.data[DOMAIN][entry.entry_id]
await config_entry_data.api.stop()
hass.data[DOMAIN].pop(entry.entry_id)
if not hass.data[DOMAIN]:
hass.data.pop(DOMAIN)
return unload_ok
class EightSleepBaseEntity(CoordinatorEntity[DataUpdateCoordinator]):
"""The base Eight Sleep entity class."""
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
user_id: str | None,
@ -161,6 +189,7 @@ class EightSleepBaseEntity(CoordinatorEntity[DataUpdateCoordinator]):
) -> None:
"""Initialize the data object."""
super().__init__(coordinator)
self._config_entry = entry
self._eight = eight
self._user_id = user_id
self._sensor = sensor
@ -170,9 +199,25 @@ class EightSleepBaseEntity(CoordinatorEntity[DataUpdateCoordinator]):
mapped_name = NAME_MAP.get(sensor, sensor.replace("_", " ").title())
if self._user_obj is not None:
mapped_name = f"{self._user_obj.side.title()} {mapped_name}"
assert self._user_obj.user_profile
name = f"{self._user_obj.user_profile['firstName']}'s {mapped_name}"
self._attr_name = name
else:
self._attr_name = f"Eight Sleep {mapped_name}"
unique_id = f"{_get_device_unique_id(eight, self._user_obj)}.{sensor}"
self._attr_unique_id = unique_id
identifiers = {(DOMAIN, _get_device_unique_id(eight, self._user_obj))}
self._attr_device_info = DeviceInfo(identifiers=identifiers)
self._attr_name = f"Eight {mapped_name}"
self._attr_unique_id = (
f"{_get_device_unique_id(eight, self._user_obj)}.{sensor}"
)
async def async_heat_set(self, target: int, duration: int) -> None:
"""Handle eight sleep service calls."""
if self._user_obj is None:
raise HomeAssistantError(
"This entity does not support the heat set service."
)
await self._user_obj.set_heating_level(target, duration)
config_entry_data: EightSleepConfigEntryData = self.hass.data[DOMAIN][
self._config_entry.entry_id
]
await config_entry_data.heat_coordinator.async_request_refresh()

38
homeassistant/components/eight_sleep/binary_sensor.py
View File

@ -9,37 +9,30 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import EightSleepBaseEntity
from .const import DATA_API, DATA_HEAT, DOMAIN
from . import EightSleepBaseEntity, EightSleepConfigEntryData
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
BINARY_SENSORS = ["bed_presence"]
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the eight sleep binary sensor."""
if discovery_info is None:
return
eight: EightSleep = hass.data[DOMAIN][DATA_API]
heat_coordinator: DataUpdateCoordinator = hass.data[DOMAIN][DATA_HEAT]
entities = []
for user in eight.users.values():
entities.append(
EightHeatSensor(heat_coordinator, eight, user.user_id, "bed_presence")
)
async_add_entities(entities)
config_entry_data: EightSleepConfigEntryData = hass.data[DOMAIN][entry.entry_id]
eight = config_entry_data.api
heat_coordinator = config_entry_data.heat_coordinator
async_add_entities(
EightHeatSensor(entry, heat_coordinator, eight, user.user_id, binary_sensor)
for user in eight.users.values()
for binary_sensor in BINARY_SENSORS
)
class EightHeatSensor(EightSleepBaseEntity, BinarySensorEntity):
@ -49,13 +42,14 @@ class EightHeatSensor(EightSleepBaseEntity, BinarySensorEntity):
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
user_id: str | None,
sensor: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, eight, user_id, sensor)
super().__init__(entry, coordinator, eight, user_id, sensor)
assert self._user_obj
_LOGGER.debug(
"Presence Sensor: %s, Side: %s, User: %s",

90
homeassistant/components/eight_sleep/config_flow.py Normal file
View File

@ -0,0 +1,90 @@
"""Config flow for Eight Sleep integration."""
from __future__ import annotations
import logging
from typing import Any
from pyeight.eight import EightSleep
from pyeight.exceptions import RequestError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): TextSelector(
TextSelectorConfig(type=TextSelectorType.EMAIL)
),
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Eight Sleep."""
VERSION = 1
async def _validate_data(self, config: dict[str, str]) -> str | None:
"""Validate input data and return any error."""
await self.async_set_unique_id(config[CONF_USERNAME].lower())
self._abort_if_unique_id_configured()
eight = EightSleep(
config[CONF_USERNAME],
config[CONF_PASSWORD],
self.hass.config.time_zone,
client_session=async_get_clientsession(self.hass),
)
try:
await eight.fetch_token()
except RequestError as err:
return str(err)
return None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
if (err := await self._validate_data(user_input)) is not None:
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors={"base": "cannot_connect"},
description_placeholders={"error": err},
)
return self.async_create_entry(title=user_input[CONF_USERNAME], data=user_input)
async def async_step_import(self, import_config: dict) -> FlowResult:
"""Handle import."""
if (err := await self._validate_data(import_config)) is not None:
_LOGGER.error("Unable to import configuration.yaml configuration: %s", err)
return self.async_abort(
reason="cannot_connect", description_placeholders={"error": err}
)
return self.async_create_entry(
title=import_config[CONF_USERNAME], data=import_config
)

7
homeassistant/components/eight_sleep/const.py
View File

@ -1,7 +1,4 @@
"""Eight Sleep constants."""
DATA_HEAT = "heat"
DATA_USER = "user"
DATA_API = "api"
DOMAIN = "eight_sleep"
HEAT_ENTITY = "heat"
@ -15,5 +12,5 @@ NAME_MAP = {
SERVICE_HEAT_SET = "heat_set"
ATTR_TARGET_HEAT = "target"
ATTR_HEAT_DURATION = "duration"
ATTR_TARGET = "target"
ATTR_DURATION = "duration"

3
homeassistant/components/eight_sleep/manifest.json
View File

@ -5,5 +5,6 @@
"requirements": ["pyeight==0.3.0"],
"codeowners": ["@mezz64", "@raman325"],
"iot_class": "cloud_polling",
"loggers": ["pyeight"]
"loggers": ["pyeight"],
"config_flow": true
}

73
homeassistant/components/eight_sleep/sensor.py
View File

@ -5,16 +5,17 @@ import logging
from typing import Any
from pyeight.eight import EightSleep
import voluptuous as vol
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, TEMP_CELSIUS
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers import entity_platform as ep
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import EightSleepBaseEntity
from .const import DATA_API, DATA_HEAT, DATA_USER, DOMAIN
from . import EightSleepBaseEntity, EightSleepConfigEntryData
from .const import ATTR_DURATION, ATTR_TARGET, DOMAIN, SERVICE_HEAT_SET
ATTR_ROOM_TEMP = "Room Temperature"
ATTR_AVG_ROOM_TEMP = "Average Room Temperature"
@ -53,37 +54,50 @@ EIGHT_USER_SENSORS = [
EIGHT_HEAT_SENSORS = ["bed_state"]
EIGHT_ROOM_SENSORS = ["room_temperature"]
VALID_TARGET_HEAT = vol.All(vol.Coerce(int), vol.Clamp(min=-100, max=100))
VALID_DURATION = vol.All(vol.Coerce(int), vol.Clamp(min=0, max=28800))
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
SERVICE_EIGHT_SCHEMA = {
ATTR_TARGET: VALID_TARGET_HEAT,
ATTR_DURATION: VALID_DURATION,
}
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: ep.AddEntitiesCallback
) -> None:
"""Set up the eight sleep sensors."""
if discovery_info is None:
return
eight: EightSleep = hass.data[DOMAIN][DATA_API]
heat_coordinator: DataUpdateCoordinator = hass.data[DOMAIN][DATA_HEAT]
user_coordinator: DataUpdateCoordinator = hass.data[DOMAIN][DATA_USER]
config_entry_data: EightSleepConfigEntryData = hass.data[DOMAIN][entry.entry_id]
eight = config_entry_data.api
heat_coordinator = config_entry_data.heat_coordinator
user_coordinator = config_entry_data.user_coordinator
all_sensors: list[SensorEntity] = []
for obj in eight.users.values():
for sensor in EIGHT_USER_SENSORS:
all_sensors.append(
EightUserSensor(user_coordinator, eight, obj.user_id, sensor)
)
for sensor in EIGHT_HEAT_SENSORS:
all_sensors.append(
EightHeatSensor(heat_coordinator, eight, obj.user_id, sensor)
)
for sensor in EIGHT_ROOM_SENSORS:
all_sensors.append(EightRoomSensor(user_coordinator, eight, sensor))
all_sensors.extend(
EightUserSensor(entry, user_coordinator, eight, obj.user_id, sensor)
for sensor in EIGHT_USER_SENSORS
)
all_sensors.extend(
EightHeatSensor(entry, heat_coordinator, eight, obj.user_id, sensor)
for sensor in EIGHT_HEAT_SENSORS
)
all_sensors.extend(
EightRoomSensor(entry, user_coordinator, eight, sensor)
for sensor in EIGHT_ROOM_SENSORS
)
async_add_entities(all_sensors)
platform = ep.async_get_current_platform()
platform.async_register_entity_service(
SERVICE_HEAT_SET,
SERVICE_EIGHT_SCHEMA,
"async_heat_set",
)
class EightHeatSensor(EightSleepBaseEntity, SensorEntity):
"""Representation of an eight sleep heat-based sensor."""
@ -92,13 +106,14 @@ class EightHeatSensor(EightSleepBaseEntity, SensorEntity):
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
user_id: str,
sensor: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, eight, user_id, sensor)
super().__init__(entry, coordinator, eight, user_id, sensor)
assert self._user_obj
_LOGGER.debug(
@ -147,13 +162,14 @@ class EightUserSensor(EightSleepBaseEntity, SensorEntity):
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
user_id: str,
sensor: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, eight, user_id, sensor)
super().__init__(entry, coordinator, eight, user_id, sensor)
assert self._user_obj
if self._sensor == "bed_temperature":
@ -260,12 +276,13 @@ class EightRoomSensor(EightSleepBaseEntity, SensorEntity):
def __init__(
self,
entry,
coordinator: DataUpdateCoordinator,
eight: EightSleep,
sensor: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, eight, None, sensor)
super().__init__(entry, coordinator, eight, None, sensor)
@property
def native_value(self) -> int | float | None:

12
homeassistant/components/eight_sleep/services.yaml
View File

@ -1,6 +1,10 @@
heat_set:
name: Heat set
description: Set heating/cooling level for eight sleep.
target:
entity:
integration: eight_sleep
domain: sensor
fields:
duration:
name: Duration
@ -11,14 +15,6 @@ heat_set:
min: 0
max: 28800
unit_of_measurement: seconds
entity_id:
name: Entity
description: Entity id of the bed state to adjust.
required: true
selector:
entity:
integration: eight_sleep
domain: sensor
target:
name: Target
description: Target cooling/heating level from -100 to 100.

19
homeassistant/components/eight_sleep/strings.json Normal file
View File

@ -0,0 +1,19 @@
{
"config": {
"step": {
"user": {
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "Cannot connect to Eight Sleep cloud: {error}"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "Cannot connect to Eight Sleep cloud: {error}"
}
}
}

18
homeassistant/components/eight_sleep/translations/en.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect: {error}"
},
"step": {
"user": {
"data": {
"password": "Password",
"username": "Username"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -87,6 +87,7 @@ FLOWS = {
"ecobee",
"econet",
"efergy",
"eight_sleep",
"elgato",
"elkm1",
"elmax",

3
requirements_test_all.txt
View File

@ -979,6 +979,9 @@ pyeconet==0.1.15
# homeassistant.components.efergy
pyefergy==22.1.1
# homeassistant.components.eight_sleep
pyeight==0.3.0
# homeassistant.components.everlights
pyeverlights==0.1.0

1
tests/components/eight_sleep/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Eight Sleep integration."""

29
tests/components/eight_sleep/conftest.py Normal file
View File

@ -0,0 +1,29 @@
"""Fixtures for Eight Sleep."""
from unittest.mock import patch
from pyeight.exceptions import RequestError
import pytest
@pytest.fixture(name="bypass", autouse=True)
def bypass_fixture():
"""Bypasses things that slow te tests down or block them from testing the behavior."""
with patch(
"homeassistant.components.eight_sleep.config_flow.EightSleep.fetch_token",
), patch(
"homeassistant.components.eight_sleep.config_flow.EightSleep.at_exit",
), patch(
"homeassistant.components.eight_sleep.async_setup_entry",
return_value=True,
):
yield
@pytest.fixture(name="token_error")
def token_error_fixture():
"""Simulate error when fetching token."""
with patch(
"homeassistant.components.eight_sleep.config_flow.EightSleep.fetch_token",
side_effect=RequestError,
):
yield

85
tests/components/eight_sleep/test_config_flow.py Normal file
View File

@ -0,0 +1,85 @@
"""Test the Eight Sleep config flow."""
from homeassistant import config_entries
from homeassistant.components.eight_sleep.const import DOMAIN
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
async def test_form(hass) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "test-username",
"password": "test-password",
},
)
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "test-username"
assert result2["data"] == {
"username": "test-username",
"password": "test-password",
}
async def test_form_invalid_auth(hass, token_error) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"username": "bad-username",
"password": "bad-password",
},
)
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_import(hass) -> None:
"""Test import works."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"username": "test-username",
"password": "test-password",
},
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "test-username"
assert result["data"] == {
"username": "test-username",
"password": "test-password",
}
async def test_import_invalid_auth(hass, token_error) -> None:
"""Test we handle invalid auth on import."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"username": "bad-username",
"password": "bad-password",
},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"