Refactor LaMetric integration (#76759)
* Refactor LaMetric integration * Use async_setup Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * use async_get_service Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update tests/components/lametric/conftest.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update tests/components/lametric/conftest.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Pass hassconfig * Remove try/catch * Fix passing hassconfig * Use menu Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Franck Nijhof
GitHub
parent
071cae2c0b
commit
71cdc1645b
20 changed files with 1385 additions and 173 deletions
|
|
@ -639,7 +639,8 @@ omit =
|
|||
homeassistant/components/kostal_plenticore/switch.py
|
||||
homeassistant/components/kwb/sensor.py
|
||||
homeassistant/components/lacrosse/sensor.py
|
||||
homeassistant/components/lametric/*
|
||||
homeassistant/components/lametric/__init__.py
|
||||
homeassistant/components/lametric/notify.py
|
||||
homeassistant/components/lannouncer/notify.py
|
||||
homeassistant/components/lastfm/sensor.py
|
||||
homeassistant/components/launch_library/__init__.py
|
||||
|
|
|
|||
|
|
@ -586,6 +586,7 @@ build.json @home-assistant/supervisor
|
|||
/homeassistant/components/lacrosse_view/ @IceBotYT
|
||||
/tests/components/lacrosse_view/ @IceBotYT
|
||||
/homeassistant/components/lametric/ @robbiet480 @frenck
|
||||
/tests/components/lametric/ @robbiet480 @frenck
|
||||
/homeassistant/components/launch_library/ @ludeeus @DurgNomis-drol
|
||||
/tests/components/launch_library/ @ludeeus @DurgNomis-drol
|
||||
/homeassistant/components/laundrify/ @xLarry
|
||||
|
|
|
|||
|
|
@ -1,52 +1,83 @@
|
|||
"""Support for LaMetric time."""
|
||||
from lmnotify import LaMetricManager
|
||||
from demetriek import LaMetricConnectionError, LaMetricDevice
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
|
||||
from homeassistant.components.repairs import IssueSeverity, async_create_issue
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_API_KEY,
|
||||
CONF_CLIENT_ID,
|
||||
CONF_CLIENT_SECRET,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import DOMAIN, LOGGER
|
||||
from .const import DOMAIN
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_CLIENT_ID): cv.string,
|
||||
vol.Required(CONF_CLIENT_SECRET): cv.string,
|
||||
}
|
||||
)
|
||||
},
|
||||
vol.All(
|
||||
cv.deprecated(DOMAIN),
|
||||
{
|
||||
DOMAIN: vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_CLIENT_ID): cv.string,
|
||||
vol.Required(CONF_CLIENT_SECRET): cv.string,
|
||||
}
|
||||
)
|
||||
},
|
||||
),
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
|
||||
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the LaMetricManager."""
|
||||
LOGGER.debug("Setting up LaMetric platform")
|
||||
conf = config[DOMAIN]
|
||||
hlmn = HassLaMetricManager(
|
||||
client_id=conf[CONF_CLIENT_ID], client_secret=conf[CONF_CLIENT_SECRET]
|
||||
)
|
||||
if not (devices := hlmn.manager.get_devices()):
|
||||
LOGGER.error("No LaMetric devices found")
|
||||
return False
|
||||
|
||||
hass.data[DOMAIN] = hlmn
|
||||
for dev in devices:
|
||||
LOGGER.debug("Discovered LaMetric device: %s", dev)
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the LaMetric integration."""
|
||||
hass.data[DOMAIN] = {"hass_config": config}
|
||||
if DOMAIN in config:
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"manual_migration",
|
||||
breaks_in_ha_version="2022.9.0",
|
||||
is_fixable=False,
|
||||
severity=IssueSeverity.ERROR,
|
||||
translation_key="manual_migration",
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class HassLaMetricManager:
|
||||
"""A class that encapsulated requests to the LaMetric manager."""
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up LaMetric from a config entry."""
|
||||
lametric = LaMetricDevice(
|
||||
host=entry.data[CONF_HOST],
|
||||
api_key=entry.data[CONF_API_KEY],
|
||||
session=async_get_clientsession(hass),
|
||||
)
|
||||
|
||||
def __init__(self, client_id: str, client_secret: str) -> None:
|
||||
"""Initialize HassLaMetricManager and connect to LaMetric."""
|
||||
try:
|
||||
device = await lametric.device()
|
||||
except LaMetricConnectionError as ex:
|
||||
raise ConfigEntryNotReady("Cannot connect to LaMetric device") from ex
|
||||
|
||||
LOGGER.debug("Connecting to LaMetric")
|
||||
self.manager = LaMetricManager(client_id, client_secret)
|
||||
self._client_id = client_id
|
||||
self._client_secret = client_secret
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = lametric
|
||||
|
||||
# Set up notify platform, no entry support for notify component yet,
|
||||
# have to use discovery to load platform.
|
||||
hass.async_create_task(
|
||||
discovery.async_load_platform(
|
||||
hass,
|
||||
Platform.NOTIFY,
|
||||
DOMAIN,
|
||||
{CONF_NAME: device.name, "entry_id": entry.entry_id},
|
||||
hass.data[DOMAIN]["hass_config"],
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
|
|
|||
11
homeassistant/components/lametric/application_credentials.py
Normal file
11
homeassistant/components/lametric/application_credentials.py
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
"""Application credentials platform for LaMetric."""
|
||||
from homeassistant.components.application_credentials import AuthorizationServer
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
|
||||
async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
|
||||
"""Return authorization server."""
|
||||
return AuthorizationServer(
|
||||
authorize_url="https://developer.lametric.com/api/v2/oauth2/authorize",
|
||||
token_url="https://developer.lametric.com/api/v2/oauth2/token",
|
||||
)
|
||||
251
homeassistant/components/lametric/config_flow.py
Normal file
251
homeassistant/components/lametric/config_flow.py
Normal file
|
|
@ -0,0 +1,251 @@
|
|||
"""Config flow to configure the LaMetric integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from ipaddress import ip_address
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from demetriek import (
|
||||
CloudDevice,
|
||||
LaMetricCloud,
|
||||
LaMetricConnectionError,
|
||||
LaMetricDevice,
|
||||
Model,
|
||||
Notification,
|
||||
NotificationIconType,
|
||||
NotificationSound,
|
||||
Simple,
|
||||
Sound,
|
||||
)
|
||||
import voluptuous as vol
|
||||
from yarl import URL
|
||||
|
||||
from homeassistant.components.ssdp import (
|
||||
ATTR_UPNP_FRIENDLY_NAME,
|
||||
ATTR_UPNP_SERIAL,
|
||||
SsdpServiceInfo,
|
||||
)
|
||||
from homeassistant.const import CONF_API_KEY, CONF_DEVICE, CONF_HOST, CONF_MAC
|
||||
from homeassistant.data_entry_flow import AbortFlow, FlowResult
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import AbstractOAuth2FlowHandler
|
||||
from homeassistant.helpers.selector import (
|
||||
SelectOptionDict,
|
||||
SelectSelector,
|
||||
SelectSelectorConfig,
|
||||
SelectSelectorMode,
|
||||
TextSelector,
|
||||
TextSelectorConfig,
|
||||
TextSelectorType,
|
||||
)
|
||||
from homeassistant.util.network import is_link_local
|
||||
|
||||
from .const import DOMAIN, LOGGER
|
||||
|
||||
|
||||
class LaMetricFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
|
||||
"""Handle a LaMetric config flow."""
|
||||
|
||||
DOMAIN = DOMAIN
|
||||
VERSION = 1
|
||||
|
||||
devices: dict[str, CloudDevice]
|
||||
discovered_host: str
|
||||
discovered_serial: str
|
||||
discovered: bool = False
|
||||
|
||||
@property
|
||||
def logger(self) -> logging.Logger:
|
||||
"""Return logger."""
|
||||
return LOGGER
|
||||
|
||||
@property
|
||||
def extra_authorize_data(self) -> dict[str, Any]:
|
||||
"""Extra data that needs to be appended to the authorize url."""
|
||||
return {"scope": "basic devices_read"}
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initiated by the user."""
|
||||
return await self.async_step_choice_enter_manual_or_fetch_cloud()
|
||||
|
||||
async def async_step_ssdp(self, discovery_info: SsdpServiceInfo) -> FlowResult:
|
||||
"""Handle a flow initiated by SSDP discovery."""
|
||||
url = URL(discovery_info.ssdp_location or "")
|
||||
if url.host is None or not (
|
||||
serial := discovery_info.upnp.get(ATTR_UPNP_SERIAL)
|
||||
):
|
||||
return self.async_abort(reason="invalid_discovery_info")
|
||||
|
||||
if is_link_local(ip_address(url.host)):
|
||||
return self.async_abort(reason="link_local_address")
|
||||
|
||||
await self.async_set_unique_id(serial)
|
||||
self._abort_if_unique_id_configured(updates={CONF_HOST: url.host})
|
||||
|
||||
self.context.update(
|
||||
{
|
||||
"title_placeholders": {
|
||||
"name": discovery_info.upnp.get(
|
||||
ATTR_UPNP_FRIENDLY_NAME, "LaMetric TIME"
|
||||
),
|
||||
},
|
||||
"configuration_url": "https://developer.lametric.com",
|
||||
}
|
||||
)
|
||||
|
||||
self.discovered = True
|
||||
self.discovered_host = str(url.host)
|
||||
self.discovered_serial = serial
|
||||
return await self.async_step_choice_enter_manual_or_fetch_cloud()
|
||||
|
||||
async def async_step_choice_enter_manual_or_fetch_cloud(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the user's choice of entering the manual credentials or fetching the cloud credentials."""
|
||||
return self.async_show_menu(
|
||||
step_id="choice_enter_manual_or_fetch_cloud",
|
||||
menu_options=["pick_implementation", "manual_entry"],
|
||||
)
|
||||
|
||||
async def async_step_manual_entry(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the user's choice of entering the device manually."""
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
if self.discovered:
|
||||
host = self.discovered_host
|
||||
else:
|
||||
host = user_input[CONF_HOST]
|
||||
|
||||
try:
|
||||
return await self._async_step_create_entry(
|
||||
host, user_input[CONF_API_KEY]
|
||||
)
|
||||
except AbortFlow as ex:
|
||||
raise ex
|
||||
except LaMetricConnectionError as ex:
|
||||
LOGGER.error("Error connecting to LaMetric: %s", ex)
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
LOGGER.exception("Unexpected error occurred")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
# Don't ask for a host if it was discovered
|
||||
schema = {
|
||||
vol.Required(CONF_API_KEY): TextSelector(
|
||||
TextSelectorConfig(type=TextSelectorType.PASSWORD)
|
||||
)
|
||||
}
|
||||
if not self.discovered:
|
||||
schema = {vol.Required(CONF_HOST): TextSelector()} | schema
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="manual_entry",
|
||||
data_schema=vol.Schema(schema),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_cloud_fetch_devices(self, data: dict[str, Any]) -> FlowResult:
|
||||
"""Fetch information about devices from the cloud."""
|
||||
lametric = LaMetricCloud(
|
||||
token=data["token"]["access_token"],
|
||||
session=async_get_clientsession(self.hass),
|
||||
)
|
||||
self.devices = {
|
||||
device.serial_number: device
|
||||
for device in sorted(await lametric.devices(), key=lambda d: d.name)
|
||||
}
|
||||
|
||||
if not self.devices:
|
||||
return self.async_abort(reason="no_devices")
|
||||
|
||||
return await self.async_step_cloud_select_device()
|
||||
|
||||
async def async_step_cloud_select_device(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle device selection from devices offered by the cloud."""
|
||||
if self.discovered:
|
||||
user_input = {CONF_DEVICE: self.discovered_serial}
|
||||
elif len(self.devices) == 1:
|
||||
user_input = {CONF_DEVICE: list(self.devices.values())[0].serial_number}
|
||||
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
device = self.devices[user_input[CONF_DEVICE]]
|
||||
try:
|
||||
return await self._async_step_create_entry(
|
||||
str(device.ip), device.api_key
|
||||
)
|
||||
except AbortFlow as ex:
|
||||
raise ex
|
||||
except LaMetricConnectionError as ex:
|
||||
LOGGER.error("Error connecting to LaMetric: %s", ex)
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
LOGGER.exception("Unexpected error occurred")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="cloud_select_device",
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_DEVICE): SelectSelector(
|
||||
SelectSelectorConfig(
|
||||
mode=SelectSelectorMode.DROPDOWN,
|
||||
options=[
|
||||
SelectOptionDict(
|
||||
value=device.serial_number,
|
||||
label=device.name,
|
||||
)
|
||||
for device in self.devices.values()
|
||||
],
|
||||
)
|
||||
),
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def _async_step_create_entry(self, host: str, api_key: str) -> FlowResult:
|
||||
"""Create entry."""
|
||||
lametric = LaMetricDevice(
|
||||
host=host,
|
||||
api_key=api_key,
|
||||
session=async_get_clientsession(self.hass),
|
||||
)
|
||||
|
||||
device = await lametric.device()
|
||||
|
||||
await self.async_set_unique_id(device.serial_number)
|
||||
self._abort_if_unique_id_configured(
|
||||
updates={CONF_HOST: lametric.host, CONF_API_KEY: lametric.api_key}
|
||||
)
|
||||
|
||||
await lametric.notify(
|
||||
notification=Notification(
|
||||
icon_type=NotificationIconType.INFO,
|
||||
model=Model(
|
||||
cycles=2,
|
||||
frames=[Simple(text="Connected to Home Assistant!", icon=7956)],
|
||||
sound=Sound(id=NotificationSound.WIN),
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
return self.async_create_entry(
|
||||
title=device.name,
|
||||
data={
|
||||
CONF_API_KEY: lametric.api_key,
|
||||
CONF_HOST: lametric.host,
|
||||
CONF_MAC: device.wifi.mac,
|
||||
},
|
||||
)
|
||||
|
||||
# Replace OAuth create entry with a fetch devices step
|
||||
# LaMetric only use OAuth to get device information, but doesn't
|
||||
# use it later on.
|
||||
async_oauth_create_entry = async_step_cloud_fetch_devices
|
||||
|
|
@ -7,10 +7,8 @@ DOMAIN: Final = "lametric"
|
|||
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
|
||||
AVAILABLE_PRIORITIES: Final = ["info", "warning", "critical"]
|
||||
AVAILABLE_ICON_TYPES: Final = ["none", "info", "alert"]
|
||||
|
||||
CONF_CYCLES: Final = "cycles"
|
||||
CONF_ICON_TYPE: Final = "icon_type"
|
||||
CONF_LIFETIME: Final = "lifetime"
|
||||
CONF_PRIORITY: Final = "priority"
|
||||
CONF_ICON_TYPE: Final = "icon_type"
|
||||
CONF_SOUND: Final = "sound"
|
||||
|
|
|
|||
|
|
@ -2,8 +2,15 @@
|
|||
"domain": "lametric",
|
||||
"name": "LaMetric",
|
||||
"documentation": "https://www.home-assistant.io/integrations/lametric",
|
||||
"requirements": ["lmnotify==0.0.4"],
|
||||
"requirements": ["demetriek==0.2.2"],
|
||||
"codeowners": ["@robbiet480", "@frenck"],
|
||||
"iot_class": "cloud_push",
|
||||
"loggers": ["lmnotify"]
|
||||
"iot_class": "local_push",
|
||||
"dependencies": ["application_credentials", "repairs"],
|
||||
"loggers": ["demetriek"],
|
||||
"config_flow": true,
|
||||
"ssdp": [
|
||||
{
|
||||
"deviceType": "urn:schemas-upnp-org:device:LaMetric:1"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,157 +3,70 @@ from __future__ import annotations
|
|||
|
||||
from typing import Any
|
||||
|
||||
from lmnotify import Model, SimpleFrame, Sound
|
||||
from oauthlib.oauth2 import TokenExpiredError
|
||||
from requests.exceptions import ConnectionError as RequestsConnectionError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.notify import (
|
||||
ATTR_DATA,
|
||||
ATTR_TARGET,
|
||||
PLATFORM_SCHEMA,
|
||||
BaseNotificationService,
|
||||
from demetriek import (
|
||||
LaMetricDevice,
|
||||
LaMetricError,
|
||||
Model,
|
||||
Notification,
|
||||
NotificationIconType,
|
||||
NotificationPriority,
|
||||
Simple,
|
||||
Sound,
|
||||
)
|
||||
|
||||
from homeassistant.components.notify import ATTR_DATA, BaseNotificationService
|
||||
from homeassistant.const import CONF_ICON
|
||||
from homeassistant.core import HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import HassLaMetricManager
|
||||
from .const import (
|
||||
AVAILABLE_ICON_TYPES,
|
||||
AVAILABLE_PRIORITIES,
|
||||
CONF_CYCLES,
|
||||
CONF_ICON_TYPE,
|
||||
CONF_LIFETIME,
|
||||
CONF_PRIORITY,
|
||||
DOMAIN,
|
||||
LOGGER,
|
||||
)
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_ICON, default="a7956"): cv.string,
|
||||
vol.Optional(CONF_LIFETIME, default=10): cv.positive_int,
|
||||
vol.Optional(CONF_CYCLES, default=1): cv.positive_int,
|
||||
vol.Optional(CONF_PRIORITY, default="warning"): vol.In(AVAILABLE_PRIORITIES),
|
||||
vol.Optional(CONF_ICON_TYPE, default="info"): vol.In(AVAILABLE_ICON_TYPES),
|
||||
}
|
||||
)
|
||||
from .const import CONF_CYCLES, CONF_ICON_TYPE, CONF_PRIORITY, CONF_SOUND, DOMAIN
|
||||
|
||||
|
||||
def get_service(
|
||||
async def async_get_service(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> LaMetricNotificationService:
|
||||
) -> LaMetricNotificationService | None:
|
||||
"""Get the LaMetric notification service."""
|
||||
return LaMetricNotificationService(
|
||||
hass.data[DOMAIN],
|
||||
config[CONF_ICON],
|
||||
config[CONF_LIFETIME] * 1000,
|
||||
config[CONF_CYCLES],
|
||||
config[CONF_PRIORITY],
|
||||
config[CONF_ICON_TYPE],
|
||||
)
|
||||
if discovery_info is None:
|
||||
return None
|
||||
lametric: LaMetricDevice = hass.data[DOMAIN][discovery_info["entry_id"]]
|
||||
return LaMetricNotificationService(lametric)
|
||||
|
||||
|
||||
class LaMetricNotificationService(BaseNotificationService):
|
||||
"""Implement the notification service for LaMetric."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hasslametricmanager: HassLaMetricManager,
|
||||
icon: str,
|
||||
lifetime: int,
|
||||
cycles: int,
|
||||
priority: str,
|
||||
icon_type: str,
|
||||
) -> None:
|
||||
def __init__(self, lametric: LaMetricDevice) -> None:
|
||||
"""Initialize the service."""
|
||||
self.hasslametricmanager = hasslametricmanager
|
||||
self._icon = icon
|
||||
self._lifetime = lifetime
|
||||
self._cycles = cycles
|
||||
self._priority = priority
|
||||
self._icon_type = icon_type
|
||||
self._devices: list[dict[str, Any]] = []
|
||||
self.lametric = lametric
|
||||
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to some LaMetric device."""
|
||||
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a LaMetric device."""
|
||||
if not (data := kwargs.get(ATTR_DATA)):
|
||||
data = {}
|
||||
|
||||
targets = kwargs.get(ATTR_TARGET)
|
||||
data = kwargs.get(ATTR_DATA)
|
||||
LOGGER.debug("Targets/Data: %s/%s", targets, data)
|
||||
icon = self._icon
|
||||
cycles = self._cycles
|
||||
sound = None
|
||||
priority = self._priority
|
||||
icon_type = self._icon_type
|
||||
if CONF_SOUND in data:
|
||||
sound = Sound(id=data[CONF_SOUND], category=None)
|
||||
|
||||
# Additional data?
|
||||
if data is not None:
|
||||
if "icon" in data:
|
||||
icon = data["icon"]
|
||||
if "sound" in data:
|
||||
try:
|
||||
sound = Sound(category="notifications", sound_id=data["sound"])
|
||||
LOGGER.debug("Adding notification sound %s", data["sound"])
|
||||
except AssertionError:
|
||||
LOGGER.error("Sound ID %s unknown, ignoring", data["sound"])
|
||||
if "cycles" in data:
|
||||
cycles = int(data["cycles"])
|
||||
if "icon_type" in data:
|
||||
if data["icon_type"] in AVAILABLE_ICON_TYPES:
|
||||
icon_type = data["icon_type"]
|
||||
else:
|
||||
LOGGER.warning(
|
||||
"Priority %s invalid, using default %s",
|
||||
data["priority"],
|
||||
priority,
|
||||
notification = Notification(
|
||||
icon_type=NotificationIconType(data.get(CONF_ICON_TYPE, "none")),
|
||||
priority=NotificationPriority(data.get(CONF_PRIORITY, "info")),
|
||||
model=Model(
|
||||
frames=[
|
||||
Simple(
|
||||
icon=data.get(CONF_ICON, "a7956"),
|
||||
text=message,
|
||||
)
|
||||
if "priority" in data:
|
||||
if data["priority"] in AVAILABLE_PRIORITIES:
|
||||
priority = data["priority"]
|
||||
else:
|
||||
LOGGER.warning(
|
||||
"Priority %s invalid, using default %s",
|
||||
data["priority"],
|
||||
priority,
|
||||
)
|
||||
text_frame = SimpleFrame(icon, message)
|
||||
LOGGER.debug(
|
||||
"Icon/Message/Cycles/Lifetime: %s, %s, %d, %d",
|
||||
icon,
|
||||
message,
|
||||
self._cycles,
|
||||
self._lifetime,
|
||||
],
|
||||
cycles=int(data.get(CONF_CYCLES, 1)),
|
||||
sound=sound,
|
||||
),
|
||||
)
|
||||
|
||||
frames = [text_frame]
|
||||
|
||||
model = Model(frames=frames, cycles=cycles, sound=sound)
|
||||
lmn = self.hasslametricmanager.manager
|
||||
try:
|
||||
self._devices = lmn.get_devices()
|
||||
except TokenExpiredError:
|
||||
LOGGER.debug("Token expired, fetching new token")
|
||||
lmn.get_token()
|
||||
self._devices = lmn.get_devices()
|
||||
except RequestsConnectionError:
|
||||
LOGGER.warning(
|
||||
"Problem connecting to LaMetric, using cached devices instead"
|
||||
)
|
||||
for dev in self._devices:
|
||||
if targets is None or dev["name"] in targets:
|
||||
try:
|
||||
lmn.set_device(dev)
|
||||
lmn.send_notification(
|
||||
model,
|
||||
lifetime=self._lifetime,
|
||||
priority=priority,
|
||||
icon_type=icon_type,
|
||||
)
|
||||
LOGGER.debug("Sent notification to LaMetric %s", dev["name"])
|
||||
except OSError:
|
||||
LOGGER.warning("Cannot connect to LaMetric %s", dev["name"])
|
||||
await self.lametric.notify(notification=notification)
|
||||
except LaMetricError as ex:
|
||||
raise HomeAssistantError("Could not send LaMetric notification") from ex
|
||||
|
|
|
|||
50
homeassistant/components/lametric/strings.json
Normal file
50
homeassistant/components/lametric/strings.json
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"choice_enter_manual_or_fetch_cloud": {
|
||||
"description": "A LaMetric device can be set up in Home Assistant in two different ways.\n\nYou can enter all device information and API tokens yourself, or Home Asssistant can import them from your LaMetric.com account.",
|
||||
"menu_options": {
|
||||
"pick_implementation": "Import from LaMetric.com (recommended)",
|
||||
"manual_entry": "Enter manually"
|
||||
}
|
||||
},
|
||||
"pick_implementation": {
|
||||
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
|
||||
},
|
||||
"manual_entry": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"api_key": "[%key:common::config_flow::data::api_key%]"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "The IP address or hostname of your LaMetric TIME on your network.",
|
||||
"api_key": "You can find this API key in [devices page in your LaMetric developer account](https://developer.lametric.com/user/devices)."
|
||||
}
|
||||
},
|
||||
"user_cloud_select_device": {
|
||||
"data": {
|
||||
"device": "Select the LaMetric device to add"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
|
||||
"invalid_discovery_info": "Invalid discovery information received",
|
||||
"link_local_address": "Link local addresses are not supported",
|
||||
"missing_configuration": "The LaMetric integration is not configured. Please follow the documentation.",
|
||||
"no_devices": "The authorized user has no LaMetric devices",
|
||||
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]"
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"manual_migration": {
|
||||
"title": "Manual migration required for LaMetric",
|
||||
"description": "The LaMetric integration has been modernized: It is now configured and set up via the user interface and the communcations are now local.\n\nUnfortunately, there is no automatic migration path possible and thus requires you to re-setup your LaMetric with Home Assistant. Please consult the Home Assistant LaMetric integration documentation on how to set it up.\n\nRemove the old LaMetric YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
|
||||
}
|
||||
}
|
||||
}
|
||||
50
homeassistant/components/lametric/translations/en.json
Normal file
50
homeassistant/components/lametric/translations/en.json
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured",
|
||||
"authorize_url_timeout": "Timeout generating authorize URL.",
|
||||
"invalid_discovery_info": "Invalid discovery information received",
|
||||
"link_local_address": "Link local addresses are not supported",
|
||||
"missing_configuration": "The LaMetric integration is not configured. Please follow the documentation.",
|
||||
"no_devices": "The authorized user has no LaMetric devices",
|
||||
"no_url_available": "No URL available. For information about this error, [check the help section]({docs_url})"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"choice_enter_manual_or_fetch_cloud": {
|
||||
"description": "A LaMetric device can be set up in Home Assistant in two different ways.\n\nYou can enter all device information and API tokens yourself, or Home Asssistant can import them from your LaMetric.com account.",
|
||||
"menu_options": {
|
||||
"manual_entry": "Enter manually",
|
||||
"pick_implementation": "Import from LaMetric.com (recommended)"
|
||||
}
|
||||
},
|
||||
"manual_entry": {
|
||||
"data": {
|
||||
"api_key": "API Key",
|
||||
"host": "Host"
|
||||
},
|
||||
"data_description": {
|
||||
"api_key": "You can find this API key in [devices page in your LaMetric developer account](https://developer.lametric.com/user/devices).",
|
||||
"host": "The IP address or hostname of your LaMetric TIME on your network."
|
||||
}
|
||||
},
|
||||
"pick_implementation": {
|
||||
"title": "Pick Authentication Method"
|
||||
},
|
||||
"user_cloud_select_device": {
|
||||
"data": {
|
||||
"device": "Select the LaMetric device to add"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"manual_migration": {
|
||||
"description": "The LaMetric integration has been modernized: It is now configured and set up via the user interface and the communcations are now local.\n\nUnfortunately, there is no automatic migration path possible and thus requires you to re-setup your LaMetric with Home Assistant. Please consult the Home Assistant LaMetric integration documentation on how to set it up.\n\nRemove the old LaMetric YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue.",
|
||||
"title": "Manual migration required for LaMetric"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -9,6 +9,7 @@ APPLICATION_CREDENTIALS = [
|
|||
"geocaching",
|
||||
"google",
|
||||
"home_connect",
|
||||
"lametric",
|
||||
"lyric",
|
||||
"neato",
|
||||
"nest",
|
||||
|
|
|
|||
|
|
@ -196,6 +196,7 @@ FLOWS = {
|
|||
"kraken",
|
||||
"kulersky",
|
||||
"lacrosse_view",
|
||||
"lametric",
|
||||
"launch_library",
|
||||
"laundrify",
|
||||
"lg_soundbar",
|
||||
|
|
|
|||
|
|
@ -190,6 +190,11 @@ SSDP = {
|
|||
"manufacturer": "konnected.io"
|
||||
}
|
||||
],
|
||||
"lametric": [
|
||||
{
|
||||
"deviceType": "urn:schemas-upnp-org:device:LaMetric:1"
|
||||
}
|
||||
],
|
||||
"nanoleaf": [
|
||||
{
|
||||
"st": "Nanoleaf_aurora:light"
|
||||
|
|
|
|||
|
|
@ -539,6 +539,9 @@ defusedxml==0.7.1
|
|||
# homeassistant.components.deluge
|
||||
deluge-client==1.7.1
|
||||
|
||||
# homeassistant.components.lametric
|
||||
demetriek==0.2.2
|
||||
|
||||
# homeassistant.components.denonavr
|
||||
denonavr==0.10.11
|
||||
|
||||
|
|
@ -979,9 +982,6 @@ limitlessled==1.1.3
|
|||
# homeassistant.components.linode
|
||||
linode-api==4.1.9b1
|
||||
|
||||
# homeassistant.components.lametric
|
||||
lmnotify==0.0.4
|
||||
|
||||
# homeassistant.components.google_maps
|
||||
locationsharinglib==4.1.5
|
||||
|
||||
|
|
|
|||
|
|
@ -410,6 +410,9 @@ defusedxml==0.7.1
|
|||
# homeassistant.components.deluge
|
||||
deluge-client==1.7.1
|
||||
|
||||
# homeassistant.components.lametric
|
||||
demetriek==0.2.2
|
||||
|
||||
# homeassistant.components.denonavr
|
||||
denonavr==0.10.11
|
||||
|
||||
|
|
|
|||
1
tests/components/lametric/__init__.py
Normal file
1
tests/components/lametric/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the LaMetric integration."""
|
||||
81
tests/components/lametric/conftest.py
Normal file
81
tests/components/lametric/conftest.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
"""Fixtures for LaMetric integration tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from demetriek import CloudDevice, Device
|
||||
from pydantic import parse_raw_as
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.application_credentials import (
|
||||
ClientCredential,
|
||||
async_import_client_credential,
|
||||
)
|
||||
from homeassistant.components.lametric.const import DOMAIN
|
||||
from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_MAC
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry, load_fixture
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def setup_credentials(hass: HomeAssistant) -> None:
|
||||
"""Fixture to setup credentials."""
|
||||
assert await async_setup_component(hass, "application_credentials", {})
|
||||
await async_import_client_credential(
|
||||
hass, DOMAIN, ClientCredential("client", "secret"), "credentials"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Return the default mocked config entry."""
|
||||
return MockConfigEntry(
|
||||
title="My LaMetric",
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_HOST: "127.0.0.2",
|
||||
CONF_API_KEY: "mock-from-fixture",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
},
|
||||
unique_id="SA110405124500W00BS9",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
|
||||
"""Mock setting up a config entry."""
|
||||
with patch(
|
||||
"homeassistant.components.lametric.async_setup_entry", return_value=True
|
||||
) as mock_setup:
|
||||
yield mock_setup
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_lametric_config_flow() -> Generator[MagicMock, None, None]:
|
||||
"""Return a mocked LaMetric client."""
|
||||
with patch(
|
||||
"homeassistant.components.lametric.config_flow.LaMetricDevice", autospec=True
|
||||
) as lametric_mock:
|
||||
lametric = lametric_mock.return_value
|
||||
lametric.api_key = "mock-api-key"
|
||||
lametric.host = "127.0.0.1"
|
||||
lametric.device.return_value = Device.parse_raw(
|
||||
load_fixture("device.json", DOMAIN)
|
||||
)
|
||||
yield lametric
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_lametric_cloud_config_flow() -> Generator[MagicMock, None, None]:
|
||||
"""Return a mocked LaMetric Cloud client."""
|
||||
with patch(
|
||||
"homeassistant.components.lametric.config_flow.LaMetricCloud", autospec=True
|
||||
) as lametric_mock:
|
||||
lametric = lametric_mock.return_value
|
||||
lametric.devices.return_value = parse_raw_as(
|
||||
list[CloudDevice], load_fixture("cloud_devices.json", DOMAIN)
|
||||
)
|
||||
yield lametric
|
||||
38
tests/components/lametric/fixtures/cloud_devices.json
Normal file
38
tests/components/lametric/fixtures/cloud_devices.json
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
[
|
||||
{
|
||||
"id": 1,
|
||||
"name": "Frenck's LaMetric",
|
||||
"state": "configured",
|
||||
"serial_number": "SA110405124500W00BS9",
|
||||
"api_key": "mock-api-key",
|
||||
"ipv4_internal": "127.0.0.1",
|
||||
"mac": "AA:BB:CC:DD:EE:FF",
|
||||
"wifi_ssid": "IoT",
|
||||
"created_at": "2015-08-12T15:15:55+00:00",
|
||||
"updated_at": "2016-08-13T18:16:17+00:00"
|
||||
},
|
||||
{
|
||||
"id": 21,
|
||||
"name": "Blackjack",
|
||||
"state": "configured",
|
||||
"serial_number": "SA140100002200W00B21",
|
||||
"api_key": "8adaa0c98278dbb1ecb218d1c3e11f9312317ba474ab3361f80c0bd4f13a6721",
|
||||
"ipv4_internal": "192.168.1.21",
|
||||
"mac": "AA:BB:CC:DD:EE:21",
|
||||
"wifi_ssid": "AllYourBaseAreBelongToUs",
|
||||
"created_at": "2015-03-06T15:15:55+00:00",
|
||||
"updated_at": "2016-06-14T18:27:13+00:00"
|
||||
},
|
||||
{
|
||||
"id": 42,
|
||||
"name": "The Answer",
|
||||
"state": "configured",
|
||||
"serial_number": "SA140100002200W00B42",
|
||||
"api_key": "8adaa0c98278dbb1ecb218d1c3e11f9312317ba474ab3361f80c0bd4f13a6742",
|
||||
"ipv4_internal": "192.168.1.42",
|
||||
"mac": "AA:BB:CC:DD:EE:42",
|
||||
"wifi_ssid": "AllYourBaseAreBelongToUs",
|
||||
"created_at": "2015-03-06T15:15:55+00:00",
|
||||
"updated_at": "2016-06-14T18:27:13+00:00"
|
||||
}
|
||||
]
|
||||
72
tests/components/lametric/fixtures/device.json
Normal file
72
tests/components/lametric/fixtures/device.json
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
{
|
||||
"audio": {
|
||||
"volume": 100,
|
||||
"volume_limit": {
|
||||
"max": 100,
|
||||
"min": 0
|
||||
},
|
||||
"volume_range": {
|
||||
"max": 100,
|
||||
"min": 0
|
||||
}
|
||||
},
|
||||
"bluetooth": {
|
||||
"active": false,
|
||||
"address": "AA:BB:CC:DD:EE:FF",
|
||||
"available": true,
|
||||
"discoverable": true,
|
||||
"low_energy": {
|
||||
"active": true,
|
||||
"advertising": true,
|
||||
"connectable": true
|
||||
},
|
||||
"name": "LM1234",
|
||||
"pairable": true
|
||||
},
|
||||
"display": {
|
||||
"brightness": 100,
|
||||
"brightness_limit": {
|
||||
"max": 100,
|
||||
"min": 2
|
||||
},
|
||||
"brightness_mode": "auto",
|
||||
"brightness_range": {
|
||||
"max": 100,
|
||||
"min": 0
|
||||
},
|
||||
"height": 8,
|
||||
"screensaver": {
|
||||
"enabled": false,
|
||||
"modes": {
|
||||
"time_based": {
|
||||
"enabled": true,
|
||||
"local_start_time": "01:00:39",
|
||||
"start_time": "00:00:39"
|
||||
},
|
||||
"when_dark": {
|
||||
"enabled": false
|
||||
}
|
||||
},
|
||||
"widget": "08b8eac21074f8f7e5a29f2855ba8060"
|
||||
},
|
||||
"type": "mixed",
|
||||
"width": 37
|
||||
},
|
||||
"id": "12345",
|
||||
"mode": "auto",
|
||||
"model": "LM 37X8",
|
||||
"name": "Frenck's LaMetric",
|
||||
"os_version": "2.2.2",
|
||||
"serial_number": "SA110405124500W00BS9",
|
||||
"wifi": {
|
||||
"active": true,
|
||||
"mac": "AA:BB:CC:DD:EE:FF",
|
||||
"available": true,
|
||||
"encryption": "WPA",
|
||||
"ssid": "IoT",
|
||||
"ip": "127.0.0.1",
|
||||
"mode": "dhcp",
|
||||
"netmask": "255.255.255.0",
|
||||
"rssi": 21
|
||||
}
|
||||
}
|
||||
697
tests/components/lametric/test_config_flow.py
Normal file
697
tests/components/lametric/test_config_flow.py
Normal file
|
|
@ -0,0 +1,697 @@
|
|||
"""Tests for the LaMetric config flow."""
|
||||
from collections.abc import Awaitable, Callable
|
||||
from http import HTTPStatus
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from aiohttp.test_utils import TestClient
|
||||
from demetriek import (
|
||||
LaMetricConnectionError,
|
||||
LaMetricConnectionTimeoutError,
|
||||
LaMetricError,
|
||||
)
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.lametric.const import DOMAIN
|
||||
from homeassistant.components.ssdp import (
|
||||
ATTR_UPNP_FRIENDLY_NAME,
|
||||
ATTR_UPNP_SERIAL,
|
||||
SsdpServiceInfo,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
|
||||
from homeassistant.const import CONF_API_KEY, CONF_DEVICE, CONF_HOST, CONF_MAC
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers import config_entry_oauth2_flow
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
|
||||
SSDP_DISCOVERY_INFO = SsdpServiceInfo(
|
||||
ssdp_usn="mock_usn",
|
||||
ssdp_st="mock_st",
|
||||
ssdp_location="http://127.0.0.1:44057/465d585b-1c05-444a-b14e-6ffb875b46a6/device_description.xml",
|
||||
upnp={
|
||||
ATTR_UPNP_FRIENDLY_NAME: "LaMetric Time (LM1245)",
|
||||
ATTR_UPNP_SERIAL: "SA110405124500W00BS9",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def test_full_cloud_import_flow_multiple_devices(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry: MagicMock,
|
||||
mock_lametric_cloud_config_flow: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
) -> None:
|
||||
"""Check a full flow importing from cloud, with multiple devices."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result.get("type") == FlowResultType.MENU
|
||||
assert result.get("step_id") == "choice_enter_manual_or_fetch_cloud"
|
||||
assert result.get("menu_options") == ["pick_implementation", "manual_entry"]
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "pick_implementation"}
|
||||
)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": flow_id,
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.EXTERNAL_STEP
|
||||
assert result2.get("url") == (
|
||||
"https://developer.lametric.com/api/v2/oauth2/authorize"
|
||||
"?response_type=code&client_id=client"
|
||||
"&redirect_uri=https://example.com/auth/external/callback"
|
||||
f"&state={state}"
|
||||
"&scope=basic+devices_read"
|
||||
)
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert resp.headers["content-type"] == "text/html; charset=utf-8"
|
||||
|
||||
aioclient_mock.post(
|
||||
"https://developer.lametric.com/api/v2/oauth2/token",
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
result3 = await hass.config_entries.flow.async_configure(flow_id)
|
||||
|
||||
assert result3.get("type") == FlowResultType.FORM
|
||||
assert result3.get("step_id") == "cloud_select_device"
|
||||
|
||||
result4 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_DEVICE: "SA110405124500W00BS9"}
|
||||
)
|
||||
|
||||
assert result4.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result4.get("title") == "Frenck's LaMetric"
|
||||
assert result4.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result4
|
||||
assert result4["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_full_cloud_import_flow_single_device(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry: MagicMock,
|
||||
mock_lametric_cloud_config_flow: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
) -> None:
|
||||
"""Check a full flow importing from cloud, with a single device."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result.get("type") == FlowResultType.MENU
|
||||
assert result.get("step_id") == "choice_enter_manual_or_fetch_cloud"
|
||||
assert result.get("menu_options") == ["pick_implementation", "manual_entry"]
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "pick_implementation"}
|
||||
)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": flow_id,
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.EXTERNAL_STEP
|
||||
assert result2.get("url") == (
|
||||
"https://developer.lametric.com/api/v2/oauth2/authorize"
|
||||
"?response_type=code&client_id=client"
|
||||
"&redirect_uri=https://example.com/auth/external/callback"
|
||||
f"&state={state}"
|
||||
"&scope=basic+devices_read"
|
||||
)
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert resp.headers["content-type"] == "text/html; charset=utf-8"
|
||||
|
||||
aioclient_mock.post(
|
||||
"https://developer.lametric.com/api/v2/oauth2/token",
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
# Stage a single device
|
||||
# Should skip step that ask for device selection
|
||||
mock_lametric_cloud_config_flow.devices.return_value = [
|
||||
mock_lametric_cloud_config_flow.devices.return_value[0]
|
||||
]
|
||||
result3 = await hass.config_entries.flow.async_configure(flow_id)
|
||||
|
||||
assert result3.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result3.get("title") == "Frenck's LaMetric"
|
||||
assert result3.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result3
|
||||
assert result3["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_full_manual(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
) -> None:
|
||||
"""Check a full flow manual entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result.get("type") == FlowResultType.MENU
|
||||
assert result.get("step_id") == "choice_enter_manual_or_fetch_cloud"
|
||||
assert result.get("menu_options") == ["pick_implementation", "manual_entry"]
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "manual_entry"}
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.FORM
|
||||
assert result2.get("step_id") == "manual_entry"
|
||||
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_HOST: "127.0.0.1", CONF_API_KEY: "mock-api-key"}
|
||||
)
|
||||
|
||||
assert result3.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result3.get("title") == "Frenck's LaMetric"
|
||||
assert result3.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result3
|
||||
assert result3["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_full_ssdp_with_cloud_import(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry: MagicMock,
|
||||
mock_lametric_cloud_config_flow: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
) -> None:
|
||||
"""Check a full flow triggered by SSDP, importing from cloud."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_SSDP}, data=SSDP_DISCOVERY_INFO
|
||||
)
|
||||
|
||||
assert result.get("type") == FlowResultType.MENU
|
||||
assert result.get("step_id") == "choice_enter_manual_or_fetch_cloud"
|
||||
assert result.get("menu_options") == ["pick_implementation", "manual_entry"]
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "pick_implementation"}
|
||||
)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": flow_id,
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.EXTERNAL_STEP
|
||||
assert result2.get("url") == (
|
||||
"https://developer.lametric.com/api/v2/oauth2/authorize"
|
||||
"?response_type=code&client_id=client"
|
||||
"&redirect_uri=https://example.com/auth/external/callback"
|
||||
f"&state={state}"
|
||||
"&scope=basic+devices_read"
|
||||
)
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert resp.headers["content-type"] == "text/html; charset=utf-8"
|
||||
|
||||
aioclient_mock.post(
|
||||
"https://developer.lametric.com/api/v2/oauth2/token",
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
result3 = await hass.config_entries.flow.async_configure(flow_id)
|
||||
|
||||
assert result3.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result3.get("title") == "Frenck's LaMetric"
|
||||
assert result3.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result3
|
||||
assert result3["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_full_ssdp_manual_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
) -> None:
|
||||
"""Check a full flow triggered by SSDP, with manual API key entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_SSDP}, data=SSDP_DISCOVERY_INFO
|
||||
)
|
||||
|
||||
assert result.get("type") == FlowResultType.MENU
|
||||
assert result.get("step_id") == "choice_enter_manual_or_fetch_cloud"
|
||||
assert result.get("menu_options") == ["pick_implementation", "manual_entry"]
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "manual_entry"}
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.FORM
|
||||
assert result2.get("step_id") == "manual_entry"
|
||||
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_API_KEY: "mock-api-key"}
|
||||
)
|
||||
|
||||
assert result3.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result3.get("title") == "Frenck's LaMetric"
|
||||
assert result3.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result3
|
||||
assert result3["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data,reason",
|
||||
[
|
||||
(
|
||||
SsdpServiceInfo(ssdp_usn="mock_usn", ssdp_st="mock_st", upnp={}),
|
||||
"invalid_discovery_info",
|
||||
),
|
||||
(
|
||||
SsdpServiceInfo(
|
||||
ssdp_usn="mock_usn",
|
||||
ssdp_st="mock_st",
|
||||
ssdp_location="http://169.254.0.1:44057/465d585b-1c05-444a-b14e-6ffb875b46a6/device_description.xml",
|
||||
upnp={
|
||||
ATTR_UPNP_SERIAL: "SA110405124500W00BS9",
|
||||
},
|
||||
),
|
||||
"link_local_address",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_ssdp_abort_invalid_discovery(
|
||||
hass: HomeAssistant, data: SsdpServiceInfo, reason: str
|
||||
) -> None:
|
||||
"""Check a full flow triggered by SSDP, with manual API key entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_SSDP}, data=data
|
||||
)
|
||||
assert result.get("type") == FlowResultType.ABORT
|
||||
assert result.get("reason") == reason
|
||||
|
||||
|
||||
async def test_cloud_import_updates_existing_entry(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
current_request_with_host: None,
|
||||
mock_lametric_cloud_config_flow: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test cloud importing existing device updates existing entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "pick_implementation"}
|
||||
)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": flow_id,
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
client = await hass_client_no_auth()
|
||||
await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
aioclient_mock.post(
|
||||
"https://developer.lametric.com/api/v2/oauth2/token",
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
await hass.config_entries.flow.async_configure(flow_id)
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_DEVICE: "SA110405124500W00BS9"}
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.ABORT
|
||||
assert result2.get("reason") == "already_configured"
|
||||
assert mock_config_entry.data == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 0
|
||||
|
||||
|
||||
async def test_manual_updates_existing_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test adding existing device updates existing entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "manual_entry"}
|
||||
)
|
||||
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_HOST: "127.0.0.1", CONF_API_KEY: "mock-api-key"}
|
||||
)
|
||||
|
||||
assert result3.get("type") == FlowResultType.ABORT
|
||||
assert result3.get("reason") == "already_configured"
|
||||
assert mock_config_entry.data == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 0
|
||||
|
||||
|
||||
async def test_discovery_updates_existing_entry(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Test discovery of existing device updates entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_SSDP}, data=SSDP_DISCOVERY_INFO
|
||||
)
|
||||
|
||||
assert result.get("type") == FlowResultType.ABORT
|
||||
assert result.get("reason") == "already_configured"
|
||||
assert mock_config_entry.data == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-from-fixture",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
|
||||
|
||||
async def test_cloud_abort_no_devices(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
current_request_with_host: None,
|
||||
mock_lametric_cloud_config_flow: MagicMock,
|
||||
) -> None:
|
||||
"""Test cloud importing aborts when account has no devices."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "pick_implementation"}
|
||||
)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": flow_id,
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
client = await hass_client_no_auth()
|
||||
await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
aioclient_mock.post(
|
||||
"https://developer.lametric.com/api/v2/oauth2/token",
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
# Stage there are no devices
|
||||
mock_lametric_cloud_config_flow.devices.return_value = []
|
||||
result2 = await hass.config_entries.flow.async_configure(flow_id)
|
||||
|
||||
assert result2.get("type") == FlowResultType.ABORT
|
||||
assert result2.get("reason") == "no_devices"
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"side_effect,reason",
|
||||
[
|
||||
(LaMetricConnectionTimeoutError, "cannot_connect"),
|
||||
(LaMetricConnectionError, "cannot_connect"),
|
||||
(LaMetricError, "unknown"),
|
||||
(RuntimeError, "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_manual_errors(
|
||||
hass: HomeAssistant,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
mock_setup_entry: MagicMock,
|
||||
side_effect: Exception,
|
||||
reason: str,
|
||||
) -> None:
|
||||
"""Test adding existing device updates existing entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "manual_entry"}
|
||||
)
|
||||
|
||||
mock_lametric_config_flow.device.side_effect = side_effect
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_HOST: "127.0.0.1", CONF_API_KEY: "mock-api-key"}
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.FORM
|
||||
assert result2.get("step_id") == "manual_entry"
|
||||
assert result2.get("errors") == {"base": reason}
|
||||
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 0
|
||||
assert len(mock_setup_entry.mock_calls) == 0
|
||||
|
||||
mock_lametric_config_flow.device.side_effect = None
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_HOST: "127.0.0.1", CONF_API_KEY: "mock-api-key"}
|
||||
)
|
||||
|
||||
assert result3.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result3.get("title") == "Frenck's LaMetric"
|
||||
assert result3.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result3
|
||||
assert result3["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 2
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"side_effect,reason",
|
||||
[
|
||||
(LaMetricConnectionTimeoutError, "cannot_connect"),
|
||||
(LaMetricConnectionError, "cannot_connect"),
|
||||
(LaMetricError, "unknown"),
|
||||
(RuntimeError, "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_cloud_errors(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
current_request_with_host: None,
|
||||
mock_setup_entry: MagicMock,
|
||||
mock_lametric_cloud_config_flow: MagicMock,
|
||||
mock_lametric_config_flow: MagicMock,
|
||||
side_effect: Exception,
|
||||
reason: str,
|
||||
) -> None:
|
||||
"""Test adding existing device updates existing entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert "flow_id" in result
|
||||
flow_id = result["flow_id"]
|
||||
|
||||
await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={"next_step_id": "pick_implementation"}
|
||||
)
|
||||
|
||||
# pylint: disable=protected-access
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": flow_id,
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
client = await hass_client_no_auth()
|
||||
await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
aioclient_mock.post(
|
||||
"https://developer.lametric.com/api/v2/oauth2/token",
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
await hass.config_entries.flow.async_configure(flow_id)
|
||||
|
||||
mock_lametric_config_flow.device.side_effect = side_effect
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_DEVICE: "SA110405124500W00BS9"}
|
||||
)
|
||||
|
||||
assert result2.get("type") == FlowResultType.FORM
|
||||
assert result2.get("step_id") == "cloud_select_device"
|
||||
assert result2.get("errors") == {"base": reason}
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 0
|
||||
assert len(mock_setup_entry.mock_calls) == 0
|
||||
|
||||
mock_lametric_config_flow.device.side_effect = None
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
flow_id, user_input={CONF_DEVICE: "SA110405124500W00BS9"}
|
||||
)
|
||||
|
||||
assert result3.get("type") == FlowResultType.CREATE_ENTRY
|
||||
assert result3.get("title") == "Frenck's LaMetric"
|
||||
assert result3.get("data") == {
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_API_KEY: "mock-api-key",
|
||||
CONF_MAC: "AA:BB:CC:DD:EE:FF",
|
||||
}
|
||||
assert "result" in result3
|
||||
assert result3["result"].unique_id == "SA110405124500W00BS9"
|
||||
|
||||
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
|
||||
assert len(mock_lametric_config_flow.device.mock_calls) == 2
|
||||
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
Loading…
Add table
Add a link
Reference in a new issue