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

Add Homepluscontrol integration (#46783)

Browse source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
chemaaa 2021-03-25 14:12:31 +01:00 • committed by GitHub
parent 3188f796f9
commit 1b60c8efb8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 1393 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.coveragerc
View file

@ -400,6 +400,9 @@ omit =
homeassistant/components/homematic/climate.py
homeassistant/components/homematic/cover.py
homeassistant/components/homematic/notify.py
homeassistant/components/home_plus_control/api.py
homeassistant/components/home_plus_control/helpers.py
homeassistant/components/home_plus_control/switch.py
homeassistant/components/homeworks/*
homeassistant/components/honeywell/climate.py
homeassistant/components/horizon/media_player.py

1
CODEOWNERS
View file

@ -196,6 +196,7 @@ homeassistant/components/history/* @home-assistant/core
homeassistant/components/hive/* @Rendili @KJonline
homeassistant/components/hlk_sw16/* @jameshilliard
homeassistant/components/home_connect/* @DavidMStraub
homeassistant/components/home_plus_control/* @chemaaa
homeassistant/components/homeassistant/* @home-assistant/core
homeassistant/components/homekit/* @bdraco
homeassistant/components/homekit_controller/* @Jc2k

179
homeassistant/components/home_plus_control/__init__.py Normal file
View file

@ -0,0 +1,179 @@
"""The Legrand Home+ Control integration."""
import asyncio
from datetime import timedelta
import logging
import async_timeout
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
from homeassistant.core import HomeAssistant
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.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 = ["switch"]
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the Legrand Home+ Control component from configuration.yaml."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
return True
# 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, config_entry: ConfigEntry) -> bool:
"""Set up Legrand Home+ Control from a config entry."""
hass_entry_data = hass.data[DOMAIN].setdefault(config_entry.entry_id, {})
# Retrieve the registered implementation
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, config_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, config_entry, implementation
)
# Set of entity unique identifiers of this integration
uids = hass_entry_data[ENTITY_UIDS] = set()
# 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 async_timeout.timeout(10):
module_data = await api.async_get_modules()
except HomePlusControlApiError as err:
raise UpdateFailed(
f"Error communicating with API: {err} [{type(err)}]"
) from err
# 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({(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,
)
return module_data
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=60),
)
hass_entry_data[DATA_COORDINATOR] = coordinator
async def start_platforms():
"""Continue setting up the platforms."""
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_setup(config_entry, platform)
for platform in PLATFORMS
]
)
# Only refresh the coordinator after all platforms are loaded.
await coordinator.async_refresh()
hass.async_create_task(start_platforms())
return True
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload the Legrand Home+ Control config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in 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)
return unload_ok

55
homeassistant/components/home_plus_control/api.py Normal file
View file

@ -0,0 +1,55 @@
"""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
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
)
# 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"]

32
homeassistant/components/home_plus_control/config_flow.py Normal file
View file

@ -0,0 +1,32 @@
"""Config flow for Legrand Home+ Control."""
import logging
from homeassistant import config_entries
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
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
@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 Normal file
View file

@ -0,0 +1,45 @@
"""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 Normal file
View file

@ -0,0 +1,53 @@
"""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 HomeAssitant GUI).
"""
def __init__(
self,
hass: HomeAssistant,
config_data: dict,
):
"""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"

15
homeassistant/components/home_plus_control/manifest.json Normal file
View file

@ -0,0 +1,15 @@
{
"domain": "home_plus_control",
"name": "Legrand Home+ Control",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/home_plus_control",
"requirements": [
"homepluscontrol==0.0.5"
],
"dependencies": [
"http"
],
"codeowners": [
"@chemaaa"
]
}

21
homeassistant/components/home_plus_control/strings.json Normal file
View file

@ -0,0 +1,21 @@
{
"title": "Legrand Home+ Control",
"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%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
}
}

129
homeassistant/components/home_plus_control/switch.py Normal file
View file

@ -0,0 +1,129 @@
"""Legrand Home+ Control Switch Entity Module that uses the HomeAssistant DataUpdateCoordinator."""
from functools import partial
from homeassistant.components.switch import (
DEVICE_CLASS_OUTLET,
DEVICE_CLASS_SWITCH,
SwitchEntity,
)
from homeassistant.core import callback
from homeassistant.helpers import dispatcher
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, config_entry, async_add_entities):
"""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.
"""
def __init__(self, coordinator, idx):
"""Pass coordinator to CoordinatorEntity."""
super().__init__(coordinator)
self.idx = idx
self.module = self.coordinator.data[self.idx]
@property
def name(self):
"""Name of the device."""
return self.module.name
@property
def unique_id(self):
"""ID (unique) of the device."""
return self.idx
@property
def device_info(self):
"""Device information."""
return {
"identifiers": {
# Unique identifiers within the domain
(DOMAIN, self.unique_id)
},
"name": self.name,
"manufacturer": "Legrand",
"model": HW_TYPE.get(self.module.hw_type),
"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 DEVICE_CLASS_OUTLET
return DEVICE_CLASS_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):
"""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):
"""Turn the entity off."""
await self.module.turn_off()
# Update the data
await self.coordinator.async_request_refresh()

15
homeassistant/components/home_plus_control/translations/en.json Normal file
View file

@ -0,0 +1,15 @@
{
"config": {
"abort": {
"authorize_url_timeout": "Timeout generating authorize URL.",
"missing_configuration": "The component is not configured. Please follow the documentation.",
"no_url_available": "No URL available. For information about this error, [check the help section]({docs_url})",
"single_instance_allowed": "Integration is already being configured in another instance. Only one is allowed at any one time.",
"oauth_error": "Error in the authentication flow."
},
"create_entry": {
"default": "Successfully authenticated"
}
},
"title": "Legrand Home+ Control"
}

1
homeassistant/generated/config_flows.py
View file

@ -95,6 +95,7 @@ FLOWS = [
"hive",
"hlk_sw16",
"home_connect",
"home_plus_control",
"homekit",
"homekit_controller",
"homematicip_cloud",

3
requirements_all.txt
View file

@ -771,6 +771,9 @@ homeconnect==0.6.3
# homeassistant.components.homematicip_cloud
homematicip==0.13.1
# homeassistant.components.home_plus_control
homepluscontrol==0.0.5
# homeassistant.components.horizon
horimote==0.4.1

3
requirements_test_all.txt
View file

@ -420,6 +420,9 @@ homeconnect==0.6.3
# homeassistant.components.homematicip_cloud
homematicip==0.13.1
# homeassistant.components.home_plus_control
homepluscontrol==0.0.5
# homeassistant.components.google
# homeassistant.components.remember_the_milk
httplib2==0.19.0

1
tests/components/home_plus_control/__init__.py Normal file
View file

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

106
tests/components/home_plus_control/conftest.py Normal file
View file

@ -0,0 +1,106 @@
"""Test setup and fixtures for component Home+ Control by Legrand."""
from homepluscontrol.homeplusinteractivemodule import HomePlusInteractiveModule
from homepluscontrol.homeplusplant import HomePlusPlant
import pytest
from homeassistant import config_entries
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",
connection_class=config_entries.CONN_CLASS_LOCAL_POLL,
options={},
system_options={"disable_new_entities": False},
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

192
tests/components/home_plus_control/test_config_flow.py Normal file
View file

@ -0,0 +1,192 @@
"""Test the Legrand Home+ Control config flow."""
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.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
from tests.components.home_plus_control.conftest import (
CLIENT_ID,
CLIENT_SECRET,
SUBSCRIPTION_KEY,
)
async def test_full_flow(
hass, aiohttp_client, aioclient_mock, current_request_with_host
):
"""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( # pylint: disable=protected-access
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_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 aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
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.RESULT_TYPE_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, current_request_with_host):
"""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.RESULT_TYPE_ABORT
assert result["reason"] == "already_in_progress"
async def test_abort_if_entry_exists(hass, current_request_with_host):
"""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.RESULT_TYPE_ABORT
assert result["reason"] == "single_instance_allowed"
async def test_abort_if_invalid_token(
hass, aiohttp_client, aioclient_mock, current_request_with_host
):
"""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( # pylint: disable=protected-access
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_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 aiohttp_client(hass.http.app)
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
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.RESULT_TYPE_ABORT
assert result["reason"] == "oauth_error"

75
tests/components/home_plus_control/test_init.py Normal file
View file

@ -0,0 +1,75 @@
"""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 tests.components.home_plus_control.conftest import (
CLIENT_ID,
CLIENT_SECRET,
SUBSCRIPTION_KEY,
)
async def test_loading(hass, mock_config_entry):
"""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 == config_entries.ENTRY_STATE_LOADED
async def test_loading_with_no_config(hass, mock_config_entry):
"""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 == config_entries.ENTRY_STATE_SETUP_ERROR
async def test_unloading(hass, mock_config_entry):
"""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 == config_entries.ENTRY_STATE_LOADED
# We now unload the entry
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state == config_entries.ENTRY_STATE_NOT_LOADED

464
tests/components/home_plus_control/test_switch.py Normal file
View file

@ -0,0 +1,464 @@
"""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 tests.common import async_fire_time_changed
from tests.components.home_plus_control.conftest import (
CLIENT_ID,
CLIENT_SECRET,
SUBSCRIPTION_KEY,
)
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 = hass.helpers.entity_registry.async_get(hass)
device_reg = hass.helpers.device_registry.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 = hass.helpers.entity_registry.async_get(hass)
device_reg = hass.helpers.device_registry.async_get(hass)
device_id = device_reg.async_get_device({(DOMAIN, device_uid)}).id
entity_entries = hass.helpers.entity_registry.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,
mock_config_entry,
mock_modules,
):
"""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,
mock_config_entry,
mock_modules,
):
"""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.timezone.utc) + dt.timedelta(seconds=100)
)
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,
mock_config_entry,
mock_modules,
):
"""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.timezone.utc) + dt.timedelta(seconds=100)
)
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, mock_config_entry, mock_modules):
"""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.timezone.utc) + dt.timedelta(seconds=100)
)
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,
mock_config_entry,
mock_modules,
):
"""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.timezone.utc) + dt.timedelta(seconds=100)
)
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,
mock_config_entry,
mock_modules,
):
"""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 == config_entries.ENTRY_STATE_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,
mock_config_entry,
mock_modules,
):
"""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 == config_entries.ENTRY_STATE_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.timezone.utc) + dt.timedelta(seconds=100)
)
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