Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove home_plus_control and mark as virtual integration supported by Netatmo (#107587) 

* Mark home_plus_control a virtual integration using Netatmo

* Apply code review suggestion

Co-authored-by: Robert Resch <robert@resch.dev>

---------

Co-authored-by: Robert Resch <robert@resch.dev>
pull/108813/head
Jan-Philipp Benecke 2024-01-23 16:18:03 +01:00 committed by GitHub
parent f3b1f47d34
commit 13887793a7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 5 additions and 1418 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -515,8 +515,6 @@ omit =
homeassistant/components/home_connect/light.py
homeassistant/components/home_connect/sensor.py
homeassistant/components/home_connect/switch.py
homeassistant/components/home_plus_control/api.py
homeassistant/components/home_plus_control/switch.py
homeassistant/components/homematic/__init__.py
homeassistant/components/homematic/binary_sensor.py
homeassistant/components/homematic/climate.py

2
CODEOWNERS
View File

@ -543,8 +543,6 @@ build.json @home-assistant/supervisor
/tests/components/holiday/ @jrieger @gjohansson-ST
/homeassistant/components/home_connect/ @DavidMStraub
/tests/components/home_connect/ @DavidMStraub
/homeassistant/components/home_plus_control/ @chemaaa
/tests/components/home_plus_control/ @chemaaa
/homeassistant/components/homeassistant/ @home-assistant/core
/tests/components/homeassistant/ @home-assistant/core
/homeassistant/components/homeassistant_alerts/ @home-assistant/core

209
homeassistant/components/home_plus_control/__init__.py
View File

@ -1,208 +1 @@
"""The Legrand Home+ Control integration."""
import asyncio
from datetime import timedelta
import logging
from homepluscontrol.homeplusapi import HomePlusControlApiError
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import (
config_entry_oauth2_flow,
config_validation as cv,
dispatcher,
)
from homeassistant.helpers.device_registry import async_get as async_get_device_registry
from homeassistant.helpers.issue_registry import (
IssueSeverity,
async_create_issue,
async_delete_issue,
)
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from . import config_flow, helpers
from .api import HomePlusControlAsyncApi
from .const import (
API,
CONF_SUBSCRIPTION_KEY,
DATA_COORDINATOR,
DISPATCHER_REMOVERS,
DOMAIN,
ENTITY_UIDS,
SIGNAL_ADD_ENTITIES,
)
# Configuration schema for component in configuration.yaml
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(CONF_CLIENT_ID): cv.string,
vol.Required(CONF_CLIENT_SECRET): cv.string,
vol.Required(CONF_SUBSCRIPTION_KEY): cv.string,
}
)
},
extra=vol.ALLOW_EXTRA,
)
# The Legrand Home+ Control platform is currently limited to "switch" entities
PLATFORMS = [Platform.SWITCH]
_LOGGER = logging.getLogger(__name__)
_ISSUE_MOVE_TO_NETATMO = "move_to_netatmo"
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Legrand Home+ Control component from configuration.yaml."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
async_create_issue(
hass,
DOMAIN,
_ISSUE_MOVE_TO_NETATMO,
is_fixable=False,
is_persistent=False,
breaks_in_ha_version="2023.12.0", # Netatmo decided to shutdown the api in december
severity=IssueSeverity.WARNING,
translation_key=_ISSUE_MOVE_TO_NETATMO,
translation_placeholders={
"url": "https://www.home-assistant.io/integrations/netatmo/"
},
)
# Register the implementation from the config information
config_flow.HomePlusControlFlowHandler.async_register_implementation(
hass,
helpers.HomePlusControlOAuth2Implementation(hass, config[DOMAIN]),
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Legrand Home+ Control from a config entry."""
hass_entry_data = hass.data[DOMAIN].setdefault(entry.entry_id, {})
async_create_issue(
hass,
DOMAIN,
_ISSUE_MOVE_TO_NETATMO,
is_fixable=False,
is_persistent=False,
breaks_in_ha_version="2023.12.0", # Netatmo decided to shutdown the api in december
severity=IssueSeverity.WARNING,
translation_key=_ISSUE_MOVE_TO_NETATMO,
translation_placeholders={
"url": "https://www.home-assistant.io/integrations/netatmo/"
},
)
# Retrieve the registered implementation
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
# Using an aiohttp-based API lib, so rely on async framework
# Add the API object to the domain's data in HA
api = hass_entry_data[API] = HomePlusControlAsyncApi(hass, entry, implementation)
# Set of entity unique identifiers of this integration
uids: set[str] = set()
hass_entry_data[ENTITY_UIDS] = uids
# Integration dispatchers
hass_entry_data[DISPATCHER_REMOVERS] = []
device_registry = async_get_device_registry(hass)
# Register the Data Coordinator with the integration
async def async_update_data():
"""Fetch data from API endpoint.
This is the place to pre-process the data to lookup tables
so entities can quickly look up their data.
"""
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
async with asyncio.timeout(10):
return await api.async_get_modules()
except HomePlusControlApiError as err:
raise UpdateFailed(
f"Error communicating with API: {err} [{type(err)}]"
) from err
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
# Name of the data. For logging purposes.
name="home_plus_control_module",
update_method=async_update_data,
# Polling interval. Will only be polled if there are subscribers.
update_interval=timedelta(seconds=300),
)
hass_entry_data[DATA_COORDINATOR] = coordinator
@callback
def _async_update_entities():
"""Process entities and add or remove them based after an update."""
if not (module_data := coordinator.data):
return
# Remove obsolete entities from Home Assistant
entity_uids_to_remove = uids - set(module_data)
for uid in entity_uids_to_remove:
uids.remove(uid)
device = device_registry.async_get_device(identifiers={(DOMAIN, uid)})
device_registry.async_remove_device(device.id)
# Send out signal for new entity addition to Home Assistant
new_entity_uids = set(module_data) - uids
if new_entity_uids:
uids.update(new_entity_uids)
dispatcher.async_dispatcher_send(
hass,
SIGNAL_ADD_ENTITIES,
new_entity_uids,
coordinator,
)
entry.async_on_unload(coordinator.async_add_listener(_async_update_entities))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# Only refresh the coordinator after all platforms are loaded.
await coordinator.async_refresh()
return True
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload the Legrand Home+ Control config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(
config_entry, PLATFORMS
)
if unload_ok:
# Unsubscribe the config_entry signal dispatcher connections
dispatcher_removers = hass.data[DOMAIN][config_entry.entry_id].pop(
"dispatcher_removers"
)
for remover in dispatcher_removers:
remover()
# And finally unload the domain config entry data
hass.data[DOMAIN].pop(config_entry.entry_id)
async_delete_issue(hass, DOMAIN, _ISSUE_MOVE_TO_NETATMO)
return unload_ok
"""Virtual integration: Legrand Home+ Control."""

58
homeassistant/components/home_plus_control/api.py
View File

@ -1,58 +0,0 @@
"""API for Legrand Home+ Control bound to Home Assistant OAuth."""
from homepluscontrol.homeplusapi import HomePlusControlAPI
from homeassistant import config_entries, core
from homeassistant.helpers import aiohttp_client, config_entry_oauth2_flow
from .const import DEFAULT_UPDATE_INTERVALS
from .helpers import HomePlusControlOAuth2Implementation
class HomePlusControlAsyncApi(HomePlusControlAPI):
"""Legrand Home+ Control object that interacts with the OAuth2-based API of the provider.
This API is bound the HomeAssistant Config Entry that corresponds to this component.
Attributes:.
hass (HomeAssistant): HomeAssistant core object.
config_entry (ConfigEntry): ConfigEntry object that configures this API.
implementation (AbstractOAuth2Implementation): OAuth2 implementation that handles AA and
token refresh.
_oauth_session (OAuth2Session): OAuth2Session object within implementation.
"""
def __init__(
self,
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
) -> None:
"""Initialize the HomePlusControlAsyncApi object.
Initialize the authenticated API for the Legrand Home+ Control component.
Args:.
hass (HomeAssistant): HomeAssistant core object.
config_entry (ConfigEntry): ConfigEntry object that configures this API.
implementation (AbstractOAuth2Implementation): OAuth2 implementation that handles AA
and token refresh.
"""
self._oauth_session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, implementation
)
assert isinstance(implementation, HomePlusControlOAuth2Implementation)
# Create the API authenticated client - external library
super().__init__(
subscription_key=implementation.subscription_key,
oauth_client=aiohttp_client.async_get_clientsession(hass),
update_intervals=DEFAULT_UPDATE_INTERVALS,
)
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
if not self._oauth_session.valid_token:
await self._oauth_session.async_ensure_token_valid()
return self._oauth_session.token["access_token"]

30
homeassistant/components/home_plus_control/config_flow.py
View File

@ -1,30 +0,0 @@
"""Config flow for Legrand Home+ Control."""
import logging
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN
class HomePlusControlFlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle Home+ Control OAuth2 authentication."""
DOMAIN = DOMAIN
# Pick the Cloud Poll class
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
async def async_step_user(self, user_input=None):
"""Handle a flow start initiated by the user."""
await self.async_set_unique_id(DOMAIN)
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
return await super().async_step_user(user_input)

45
homeassistant/components/home_plus_control/const.py
View File

@ -1,45 +0,0 @@
"""Constants for the Legrand Home+ Control integration."""
API = "api"
CONF_SUBSCRIPTION_KEY = "subscription_key"
CONF_PLANT_UPDATE_INTERVAL = "plant_update_interval"
CONF_PLANT_TOPOLOGY_UPDATE_INTERVAL = "plant_topology_update_interval"
CONF_MODULE_STATUS_UPDATE_INTERVAL = "module_status_update_interval"
DATA_COORDINATOR = "coordinator"
DOMAIN = "home_plus_control"
ENTITY_UIDS = "entity_unique_ids"
DISPATCHER_REMOVERS = "dispatcher_removers"
# Legrand Model Identifiers - https://developer.legrand.com/documentation/product-cluster-list/#
HW_TYPE = {
"NLC": "NLC - Cable Outlet",
"NLF": "NLF - On-Off Dimmer Switch w/o Neutral",
"NLP": "NLP - Socket (Connected) Outlet",
"NLPM": "NLPM - Mobile Socket Outlet",
"NLM": "NLM - Micromodule Switch",
"NLV": "NLV - Shutter Switch with Neutral",
"NLLV": "NLLV - Shutter Switch with Level Control",
"NLL": "NLL - On-Off Toggle Switch with Neutral",
"NLT": "NLT - Remote Switch",
"NLD": "NLD - Double Gangs On-Off Remote Switch",
}
# Legrand OAuth2 URIs
OAUTH2_AUTHORIZE = "https://partners-login.eliotbylegrand.com/authorize"
OAUTH2_TOKEN = "https://partners-login.eliotbylegrand.com/token"
# The Legrand Home+ Control API has very limited request quotas - at the time of writing, it is
# limited to 500 calls per day (resets at 00:00) - so we want to keep updates to a minimum.
DEFAULT_UPDATE_INTERVALS = {
# Seconds between API checks for plant information updates. This is expected to change very
# little over time because a user's plants (homes) should rarely change.
CONF_PLANT_UPDATE_INTERVAL: 7200, # 120 minutes
# Seconds between API checks for plant topology updates. This is expected to change little
# over time because the modules in the user's plant should be relatively stable.
CONF_PLANT_TOPOLOGY_UPDATE_INTERVAL: 3600, # 60 minutes
# Seconds between API checks for module status updates. This can change frequently so we
# check often
CONF_MODULE_STATUS_UPDATE_INTERVAL: 300, # 5 minutes
}
SIGNAL_ADD_ENTITIES = "home_plus_control_add_entities_signal"

53
homeassistant/components/home_plus_control/helpers.py
View File

@ -1,53 +0,0 @@
"""Helper classes and functions for the Legrand Home+ Control integration."""
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from .const import CONF_SUBSCRIPTION_KEY, DOMAIN, OAUTH2_AUTHORIZE, OAUTH2_TOKEN
class HomePlusControlOAuth2Implementation(
config_entry_oauth2_flow.LocalOAuth2Implementation
):
"""OAuth2 implementation that extends the HomeAssistant local implementation.
It provides the name of the integration and adds support for the subscription key.
Attributes:
hass (HomeAssistant): HomeAssistant core object.
client_id (str): Client identifier assigned by the API provider when registering an app.
client_secret (str): Client secret assigned by the API provider when registering an app.
subscription_key (str): Subscription key obtained from the API provider.
authorize_url (str): Authorization URL initiate authentication flow.
token_url (str): URL to retrieve access/refresh tokens.
name (str): Name of the implementation (appears in the HomeAssistant GUI).
"""
def __init__(
self,
hass: HomeAssistant,
config_data: dict,
) -> None:
"""HomePlusControlOAuth2Implementation Constructor.
Initialize the authentication implementation for the Legrand Home+ Control API.
Args:
hass (HomeAssistant): HomeAssistant core object.
config_data (dict): Configuration data that complies with the config Schema
of this component.
"""
super().__init__(
hass=hass,
domain=DOMAIN,
client_id=config_data[CONF_CLIENT_ID],
client_secret=config_data[CONF_CLIENT_SECRET],
authorize_url=OAUTH2_AUTHORIZE,
token_url=OAUTH2_TOKEN,
)
self.subscription_key = config_data[CONF_SUBSCRIPTION_KEY]
@property
def name(self) -> str:
"""Name of the implementation."""
return "Home+ Control"

9
homeassistant/components/home_plus_control/manifest.json
View File

@ -1,11 +1,6 @@
{
"domain": "home_plus_control",
"name": "Legrand Home+ Control",
"codeowners": ["@chemaaa"],
"config_flow": true,
"dependencies": ["auth"],
"documentation": "https://www.home-assistant.io/integrations/home_plus_control",
"iot_class": "cloud_polling",
"loggers": ["homepluscontrol"],
"requirements": ["homepluscontrol==0.0.5"]
"integration_type": "virtual",
"supported_by": "netatmo"
}

30
homeassistant/components/home_plus_control/strings.json
View File

@ -1,30 +0,0 @@
{
"config": {
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
}
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
},
"issues": {
"move_to_netatmo": {
"title": "Legrand Home+ Control deprecation",
"description": "Home Assistant has been informed that the platform the Legrand Home+ Control integration is using, will be shutting down upcoming December.\n\nOnce that happens, it means this integration is no longer functional. We advise you to remove this integration and switch to the [Netatmo]({url}) integration, which provides a replacement for controlling your Legrand Home+ Control devices."
}
}
}

131
homeassistant/components/home_plus_control/switch.py
View File

@ -1,131 +0,0 @@
"""Legrand Home+ Control Switch Entity Module that uses the HomeAssistant DataUpdateCoordinator."""
from functools import partial
from typing import Any
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import dispatcher
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DISPATCHER_REMOVERS, DOMAIN, HW_TYPE, SIGNAL_ADD_ENTITIES
@callback
def add_switch_entities(new_unique_ids, coordinator, add_entities):
"""Add switch entities to the platform.
Args:
new_unique_ids (set): Unique identifiers of entities to be added to Home Assistant.
coordinator (DataUpdateCoordinator): Data coordinator of this platform.
add_entities (function): Method called to add entities to Home Assistant.
"""
new_entities = []
for uid in new_unique_ids:
new_ent = HomeControlSwitchEntity(coordinator, uid)
new_entities.append(new_ent)
add_entities(new_entities)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Legrand Home+ Control Switch platform in HomeAssistant.
Args:
hass (HomeAssistant): HomeAssistant core object.
config_entry (ConfigEntry): ConfigEntry object that configures this platform.
async_add_entities (function): Function called to add entities of this platform.
"""
partial_add_switch_entities = partial(
add_switch_entities, add_entities=async_add_entities
)
# Connect the dispatcher for the switch platform
hass.data[DOMAIN][config_entry.entry_id][DISPATCHER_REMOVERS].append(
dispatcher.async_dispatcher_connect(
hass, SIGNAL_ADD_ENTITIES, partial_add_switch_entities
)
)
class HomeControlSwitchEntity(CoordinatorEntity, SwitchEntity):
"""Entity that represents a Legrand Home+ Control switch.
It extends the HomeAssistant-provided classes of the CoordinatorEntity and the SwitchEntity.
The CoordinatorEntity class provides:
should_poll
async_update
async_added_to_hass
The SwitchEntity class provides the functionality of a ToggleEntity and additional power
consumption methods and state attributes.
"""
_attr_has_entity_name = True
_attr_name = None
def __init__(self, coordinator, idx):
"""Pass coordinator to CoordinatorEntity."""
super().__init__(coordinator)
self.idx = idx
self.module = self.coordinator.data[self.idx]
@property
def unique_id(self):
"""ID (unique) of the device."""
return self.idx
@property
def device_info(self) -> DeviceInfo:
"""Device information."""
return DeviceInfo(
identifiers={
# Unique identifiers within the domain
(DOMAIN, self.unique_id)
},
manufacturer="Legrand",
model=HW_TYPE.get(self.module.hw_type),
name=self.module.name,
sw_version=self.module.fw,
)
@property
def device_class(self):
"""Return the class of this device, from component DEVICE_CLASSES."""
if self.module.device == "plug":
return SwitchDeviceClass.OUTLET
return SwitchDeviceClass.SWITCH
@property
def available(self) -> bool:
"""Return if entity is available.
This is the case when the coordinator is able to update the data successfully
AND the switch entity is reachable.
This method overrides the one of the CoordinatorEntity
"""
return self.coordinator.last_update_success and self.module.reachable
@property
def is_on(self):
"""Return entity state."""
return self.module.status == "on"
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the light on."""
# Do the turning on.
await self.module.turn_on()
# Update the data
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
await self.module.turn_off()
# Update the data
await self.coordinator.async_request_refresh()

1
homeassistant/generated/config_flows.py
View File

@ -214,7 +214,6 @@ FLOWS = {
"hlk_sw16",
"holiday",
"home_connect",
"home_plus_control",
"homeassistant_sky_connect",
"homekit",
"homekit_controller",

5
homeassistant/generated/integrations.json
View File

@ -2494,9 +2494,8 @@
},
"home_plus_control": {
"name": "Legrand Home+ Control",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
"integration_type": "virtual",
"supported_by": "netatmo"
},
"homematic": {
"name": "Homematic",

3
requirements_all.txt
View File

@ -1064,9 +1064,6 @@ homeconnect==0.7.2
# homeassistant.components.homematicip_cloud
homematicip==1.0.16
# homeassistant.components.home_plus_control
homepluscontrol==0.0.5
# homeassistant.components.horizon
horimote==0.4.1

3
requirements_test_all.txt
View File

@ -857,9 +857,6 @@ homeconnect==0.7.2
# homeassistant.components.homematicip_cloud
homematicip==1.0.16
# homeassistant.components.home_plus_control
homepluscontrol==0.0.5
# homeassistant.components.remember_the_milk
httplib2==0.20.4

1
tests/components/home_plus_control/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the Legrand Home+ Control integration."""

103
tests/components/home_plus_control/conftest.py
View File

@ -1,103 +0,0 @@
"""Test setup and fixtures for component Home+ Control by Legrand."""
from homepluscontrol.homeplusinteractivemodule import HomePlusInteractiveModule
from homepluscontrol.homeplusplant import HomePlusPlant
import pytest
from homeassistant.components.home_plus_control.const import DOMAIN
from tests.common import MockConfigEntry
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
SUBSCRIPTION_KEY = "12345678901234567890123456789012"
@pytest.fixture
def mock_config_entry():
"""Return a fake config entry.
This is a minimal entry to setup the integration and to ensure that the
OAuth access token will not expire.
"""
return MockConfigEntry(
domain=DOMAIN,
title="Home+ Control",
data={
"auth_implementation": "home_plus_control",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 9999999999,
"expires_at": 9999999999.99999999,
"expires_on": 9999999999,
},
},
source="test",
options={},
unique_id=DOMAIN,
entry_id="home_plus_control_entry_id",
)
@pytest.fixture
def mock_modules():
"""Return the full set of mock modules."""
plant = HomePlusPlant(
id="123456789009876543210", name="My Home", country="ES", oauth_client=None
)
modules = {
"0000000987654321fedcba": HomePlusInteractiveModule(
plant,
id="0000000987654321fedcba",
name="Kitchen Wall Outlet",
hw_type="NLP",
device="plug",
fw="42",
reachable=True,
),
"0000000887654321fedcba": HomePlusInteractiveModule(
plant,
id="0000000887654321fedcba",
name="Bedroom Wall Outlet",
hw_type="NLP",
device="light",
fw="42",
reachable=True,
),
"0000000787654321fedcba": HomePlusInteractiveModule(
plant,
id="0000000787654321fedcba",
name="Living Room Ceiling Light",
hw_type="NLF",
device="light",
fw="46",
reachable=True,
),
"0000000687654321fedcba": HomePlusInteractiveModule(
plant,
id="0000000687654321fedcba",
name="Dining Room Ceiling Light",
hw_type="NLF",
device="light",
fw="46",
reachable=True,
),
"0000000587654321fedcba": HomePlusInteractiveModule(
plant,
id="0000000587654321fedcba",
name="Dining Room Wall Outlet",
hw_type="NLP",
device="plug",
fw="42",
reachable=True,
),
}
# Set lights off and plugs on
for mod_stat in modules.values():
mod_stat.status = "on"
if mod_stat.device == "light":
mod_stat.status = "off"
return modules

203
tests/components/home_plus_control/test_config_flow.py
View File

@ -1,203 +0,0 @@
"""Test the Legrand Home+ Control config flow."""
from http import HTTPStatus
from unittest.mock import patch
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.home_plus_control.const import (
CONF_SUBSCRIPTION_KEY,
DOMAIN,
OAUTH2_AUTHORIZE,
OAUTH2_TOKEN,
)
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from .conftest import CLIENT_ID, CLIENT_SECRET, SUBSCRIPTION_KEY
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
current_request_with_host: None,
) -> None:
"""Check full flow."""
assert await setup.async_setup_component(
hass,
"home_plus_control",
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
result = await hass.config_entries.flow.async_init(
"home_plus_control", context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["type"] == data_entry_flow.FlowResultType.EXTERNAL_STEP
assert result["step_id"] == "auth"
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
)
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(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.home_plus_control.async_setup_entry",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Home+ Control"
config_data = result["data"]
assert config_data["token"]["refresh_token"] == "mock-refresh-token"
assert config_data["token"]["access_token"] == "mock-access-token"
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
async def test_abort_if_entry_in_progress(
hass: HomeAssistant, current_request_with_host: None
) -> None:
"""Check flow abort when an entry is already in progress."""
assert await setup.async_setup_component(
hass,
"home_plus_control",
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
# Start one flow
result = await hass.config_entries.flow.async_init(
"home_plus_control", context={"source": config_entries.SOURCE_USER}
)
# Attempt to start another flow
result = await hass.config_entries.flow.async_init(
"home_plus_control", context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_in_progress"
async def test_abort_if_entry_exists(
hass: HomeAssistant, current_request_with_host: None
) -> None:
"""Check flow abort when an entry already exists."""
existing_entry = MockConfigEntry(domain=DOMAIN)
existing_entry.add_to_hass(hass)
assert await setup.async_setup_component(
hass,
"home_plus_control",
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
"http": {},
},
)
result = await hass.config_entries.flow.async_init(
"home_plus_control", context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
async def test_abort_if_invalid_token(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
current_request_with_host: None,
) -> None:
"""Check flow abort when the token has an invalid value."""
assert await setup.async_setup_component(
hass,
"home_plus_control",
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
result = await hass.config_entries.flow.async_init(
"home_plus_control", context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["type"] == data_entry_flow.FlowResultType.EXTERNAL_STEP
assert result["step_id"] == "auth"
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
)
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(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": "non-integer",
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "oauth_error"

72
tests/components/home_plus_control/test_init.py
View File

@ -1,72 +0,0 @@
"""Test the Legrand Home+ Control integration."""
from unittest.mock import patch
from homeassistant import config_entries, setup
from homeassistant.components.home_plus_control.const import (
CONF_SUBSCRIPTION_KEY,
DOMAIN,
)
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.core import HomeAssistant
from .conftest import CLIENT_ID, CLIENT_SECRET, SUBSCRIPTION_KEY
async def test_loading(hass: HomeAssistant, mock_config_entry) -> None:
"""Test component loading."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value={},
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
assert mock_config_entry.state is config_entries.ConfigEntryState.LOADED
async def test_loading_with_no_config(hass: HomeAssistant, mock_config_entry) -> None:
"""Test component loading failure when it has not configuration."""
mock_config_entry.add_to_hass(hass)
await setup.async_setup_component(hass, DOMAIN, {})
# Component setup fails because the oauth2 implementation could not be registered
assert mock_config_entry.state is config_entries.ConfigEntryState.SETUP_ERROR
async def test_unloading(hass: HomeAssistant, mock_config_entry) -> None:
"""Test component unloading."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value={},
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
assert mock_config_entry.state is config_entries.ConfigEntryState.LOADED
# We now unload the entry
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state is config_entries.ConfigEntryState.NOT_LOADED

463
tests/components/home_plus_control/test_switch.py
View File

@ -1,463 +0,0 @@
"""Test the Legrand Home+ Control switch platform."""
import datetime as dt
from unittest.mock import patch
from homepluscontrol.homeplusapi import HomePlusControlApiError
from homeassistant import config_entries, setup
from homeassistant.components.home_plus_control.const import (
CONF_SUBSCRIPTION_KEY,
DOMAIN,
)
from homeassistant.const import (
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from .conftest import CLIENT_ID, CLIENT_SECRET, SUBSCRIPTION_KEY
from tests.common import async_fire_time_changed
def entity_assertions(
hass,
num_exp_entities,
num_exp_devices=None,
expected_entities=None,
expected_devices=None,
):
"""Assert number of entities and devices."""
entity_reg = er.async_get(hass)
device_reg = dr.async_get(hass)
if num_exp_devices is None:
num_exp_devices = num_exp_entities
assert len(entity_reg.entities) == num_exp_entities
assert len(device_reg.devices) == num_exp_devices
if expected_entities is not None:
for exp_entity_id, present in expected_entities.items():
assert bool(entity_reg.async_get(exp_entity_id)) == present
if expected_devices is not None:
for exp_device_id, present in expected_devices.items():
assert bool(device_reg.async_get(exp_device_id)) == present
def one_entity_state(hass, device_uid):
"""Assert the presence of an entity and return its state."""
entity_reg = er.async_get(hass)
device_reg = dr.async_get(hass)
device_id = device_reg.async_get_device(identifiers={(DOMAIN, device_uid)}).id
entity_entries = er.async_entries_for_device(entity_reg, device_id)
assert len(entity_entries) == 1
entity_entry = entity_entries[0]
return hass.states.get(entity_entry.entity_id).state
async def test_plant_update(
hass: HomeAssistant,
mock_config_entry,
mock_modules,
) -> None:
"""Test entity and device loading."""
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Check the entities and devices
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
async def test_plant_topology_reduction_change(
hass: HomeAssistant,
mock_config_entry,
mock_modules,
) -> None:
"""Test an entity leaving the plant topology."""
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Check the entities and devices - 5 mock entities
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
# Now we refresh the topology with one entity less
mock_modules.pop("0000000987654321fedcba")
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
async_fire_time_changed(
hass, dt.datetime.now(dt.UTC) + dt.timedelta(seconds=400)
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Check for plant, topology and module status - this time only 4 left
entity_assertions(
hass,
num_exp_entities=4,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": False,
},
)
async def test_plant_topology_increase_change(
hass: HomeAssistant,
mock_config_entry,
mock_modules,
) -> None:
"""Test an entity entering the plant topology."""
# Remove one module initially
new_module = mock_modules.pop("0000000987654321fedcba")
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Check the entities and devices - we have 4 entities to start with
entity_assertions(
hass,
num_exp_entities=4,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": False,
},
)
# Now we refresh the topology with one entity more
mock_modules["0000000987654321fedcba"] = new_module
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
async_fire_time_changed(
hass, dt.datetime.now(dt.UTC) + dt.timedelta(seconds=400)
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
async def test_module_status_unavailable(
hass: HomeAssistant, mock_config_entry, mock_modules
) -> None:
"""Test a module becoming unreachable in the plant."""
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Check the entities and devices - 5 mock entities
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
# Confirm the availability of this particular entity
test_entity_uid = "0000000987654321fedcba"
test_entity_state = one_entity_state(hass, test_entity_uid)
assert test_entity_state == STATE_ON
# Now we refresh the topology with the module being unreachable
mock_modules["0000000987654321fedcba"].reachable = False
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
async_fire_time_changed(
hass, dt.datetime.now(dt.UTC) + dt.timedelta(seconds=400)
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Assert the devices and entities
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
await hass.async_block_till_done()
# The entity is present, but not available
test_entity_state = one_entity_state(hass, test_entity_uid)
assert test_entity_state == STATE_UNAVAILABLE
async def test_module_status_available(
hass: HomeAssistant,
mock_config_entry,
mock_modules,
) -> None:
"""Test a module becoming reachable in the plant."""
# Set the module initially unreachable
mock_modules["0000000987654321fedcba"].reachable = False
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Assert the devices and entities
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
# This particular entity is not available
test_entity_uid = "0000000987654321fedcba"
test_entity_state = one_entity_state(hass, test_entity_uid)
assert test_entity_state == STATE_UNAVAILABLE
# Now we refresh the topology with the module being reachable
mock_modules["0000000987654321fedcba"].reachable = True
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
async_fire_time_changed(
hass, dt.datetime.now(dt.UTC) + dt.timedelta(seconds=400)
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Assert the devices and entities remain the same
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
# Now the entity is available
test_entity_uid = "0000000987654321fedcba"
test_entity_state = one_entity_state(hass, test_entity_uid)
assert test_entity_state == STATE_ON
async def test_initial_api_error(
hass: HomeAssistant,
mock_config_entry,
mock_modules,
) -> None:
"""Test an API error on initial call."""
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
side_effect=HomePlusControlApiError,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# The component has been loaded
assert mock_config_entry.state is config_entries.ConfigEntryState.LOADED
# Check the entities and devices - None have been configured
entity_assertions(hass, num_exp_entities=0)
async def test_update_with_api_error(
hass: HomeAssistant,
mock_config_entry,
mock_modules,
) -> None:
"""Test an API timeout when updating the module data."""
# Load the entry
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
) as mock_check:
await setup.async_setup_component(
hass,
DOMAIN,
{
"home_plus_control": {
CONF_CLIENT_ID: CLIENT_ID,
CONF_CLIENT_SECRET: CLIENT_SECRET,
CONF_SUBSCRIPTION_KEY: SUBSCRIPTION_KEY,
},
},
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# The component has been loaded
assert mock_config_entry.state is config_entries.ConfigEntryState.LOADED
# Check the entities and devices - all entities should be there
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
for test_entity_uid in mock_modules:
test_entity_state = one_entity_state(hass, test_entity_uid)
assert test_entity_state in (STATE_ON, STATE_OFF)
# Attempt to update the data, but API update fails
with patch(
"homeassistant.components.home_plus_control.api.HomePlusControlAsyncApi.async_get_modules",
return_value=mock_modules,
side_effect=HomePlusControlApiError,
) as mock_check:
async_fire_time_changed(
hass, dt.datetime.now(dt.UTC) + dt.timedelta(seconds=400)
)
await hass.async_block_till_done()
assert len(mock_check.mock_calls) == 1
# Assert the devices and entities - all should still be present
entity_assertions(
hass,
num_exp_entities=5,
expected_entities={
"switch.dining_room_wall_outlet": True,
"switch.kitchen_wall_outlet": True,
},
)
# This entity has not returned a status, so appears as unavailable
for test_entity_uid in mock_modules:
test_entity_state = one_entity_state(hass, test_entity_uid)
assert test_entity_state == STATE_UNAVAILABLE