Move switcher_kis services to entity services (#45204)

This commit is contained in:
Shay Levy 2021-01-16 19:18:40 +02:00 committed by GitHub
parent 562d30319b
commit 8151721fbc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 163 additions and 161 deletions

View File

@ -4,35 +4,22 @@ from datetime import datetime, timedelta
import logging import logging
from typing import Dict, Optional from typing import Dict, Optional
from aioswitcher.api import SwitcherV2Api
from aioswitcher.bridge import SwitcherV2Bridge from aioswitcher.bridge import SwitcherV2Bridge
from aioswitcher.consts import COMMAND_ON
import voluptuous as vol import voluptuous as vol
from homeassistant.auth.permissions.const import POLICY_EDIT
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, EVENT_HOMEASSISTANT_STOP from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import callback, split_entity_id from homeassistant.core import callback
from homeassistant.exceptions import Unauthorized, UnknownUser
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_listen_platform, async_load_platform from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ( from homeassistant.helpers.typing import EventType, HomeAssistantType
ContextType,
DiscoveryInfoType,
EventType,
HomeAssistantType,
ServiceCallType,
)
from homeassistant.loader import bind_hass
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DOMAIN = "switcher_kis" DOMAIN = "switcher_kis"
CONF_AUTO_OFF = "auto_off"
CONF_TIMER_MINUTES = "timer_minutes"
CONF_DEVICE_ID = "device_id" CONF_DEVICE_ID = "device_id"
CONF_DEVICE_PASSWORD = "device_password" CONF_DEVICE_PASSWORD = "device_password"
CONF_PHONE_ID = "phone_id" CONF_PHONE_ID = "phone_id"
@ -58,39 +45,6 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA, extra=vol.ALLOW_EXTRA,
) )
SERVICE_SET_AUTO_OFF_NAME = "set_auto_off"
SERVICE_SET_AUTO_OFF_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(CONF_AUTO_OFF): cv.time_period_str,
}
)
SERVICE_TURN_ON_WITH_TIMER_NAME = "turn_on_with_timer"
SERVICE_TURN_ON_WITH_TIMER_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(CONF_TIMER_MINUTES): vol.All(
cv.positive_int, vol.Range(min=1, max=90)
),
}
)
@bind_hass
async def _validate_edit_permission(
hass: HomeAssistantType, context: ContextType, entity_id: str
) -> None:
"""Use for validating user control permissions."""
splited = split_entity_id(entity_id)
if splited[0] != SWITCH_DOMAIN or not splited[1].startswith(DOMAIN):
raise Unauthorized(context=context, entity_id=entity_id, permission=POLICY_EDIT)
user = await hass.auth.async_get_user(context.user_id)
if user is None:
raise UnknownUser(context=context, entity_id=entity_id, permission=POLICY_EDIT)
if not user.permissions.check_entity(entity_id, POLICY_EDIT):
raise Unauthorized(context=context, entity_id=entity_id, permission=POLICY_EDIT)
async def async_setup(hass: HomeAssistantType, config: Dict) -> bool: async def async_setup(hass: HomeAssistantType, config: Dict) -> bool:
"""Set up the switcher component.""" """Set up the switcher component."""
@ -117,53 +71,6 @@ async def async_setup(hass: HomeAssistantType, config: Dict) -> bool:
return False return False
hass.data[DOMAIN] = {DATA_DEVICE: device_data} hass.data[DOMAIN] = {DATA_DEVICE: device_data}
async def async_switch_platform_discovered(
platform: str, discovery_info: DiscoveryInfoType
) -> None:
"""Use for registering services after switch platform is discovered."""
if platform != DOMAIN:
return
async def async_set_auto_off_service(service: ServiceCallType) -> None:
"""Use for handling setting device auto-off service calls."""
await _validate_edit_permission(
hass, service.context, service.data[ATTR_ENTITY_ID]
)
async with SwitcherV2Api(
hass.loop, device_data.ip_addr, phone_id, device_id, device_password
) as swapi:
await swapi.set_auto_shutdown(service.data[CONF_AUTO_OFF])
async def async_turn_on_with_timer_service(service: ServiceCallType) -> None:
"""Use for handling turning device on with a timer service calls."""
await _validate_edit_permission(
hass, service.context, service.data[ATTR_ENTITY_ID]
)
async with SwitcherV2Api(
hass.loop, device_data.ip_addr, phone_id, device_id, device_password
) as swapi:
await swapi.control_device(COMMAND_ON, service.data[CONF_TIMER_MINUTES])
hass.services.async_register(
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
async_set_auto_off_service,
schema=SERVICE_SET_AUTO_OFF_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_TURN_ON_WITH_TIMER_NAME,
async_turn_on_with_timer_service,
schema=SERVICE_TURN_ON_WITH_TIMER_SCHEMA,
)
async_listen_platform(hass, SWITCH_DOMAIN, async_switch_platform_discovered)
hass.async_create_task(async_load_platform(hass, SWITCH_DOMAIN, DOMAIN, {}, config)) hass.async_create_task(async_load_platform(hass, SWITCH_DOMAIN, DOMAIN, {}, config))
@callback @callback

View File

@ -15,5 +15,5 @@ turn_on_with_timer:
description: "Name of the entity id associated with the integration, used for permission validation." description: "Name of the entity id associated with the integration, used for permission validation."
example: "switch.switcher_kis_boiler" example: "switch.switcher_kis_boiler"
timer_minutes: timer_minutes:
description: 'Minutes to turn on (valid range from 1 to 90)' description: 'Minutes to turn on (valid range from 1 to 150)'
example: '30' example: '30'

View File

@ -1,7 +1,8 @@
"""Home Assistant Switcher Component Switch platform.""" """Home Assistant Switcher Component Switch platform."""
from typing import TYPE_CHECKING, Callable, Dict from typing import Callable, Dict
from aioswitcher.api import SwitcherV2Api from aioswitcher.api import SwitcherV2Api
from aioswitcher.api.messages import SwitcherV2ControlResponseMSG
from aioswitcher.consts import ( from aioswitcher.consts import (
COMMAND_OFF, COMMAND_OFF,
COMMAND_ON, COMMAND_ON,
@ -9,10 +10,13 @@ from aioswitcher.consts import (
STATE_ON as SWITCHER_STATE_ON, STATE_ON as SWITCHER_STATE_ON,
WAITING_TEXT, WAITING_TEXT,
) )
from aioswitcher.devices import SwitcherV2Device
import voluptuous as vol
from homeassistant.components.switch import ATTR_CURRENT_POWER_W, SwitchEntity from homeassistant.components.switch import ATTR_CURRENT_POWER_W, SwitchEntity
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType, ServiceCallType
from . import ( from . import (
ATTR_AUTO_OFF_SET, ATTR_AUTO_OFF_SET,
@ -23,11 +27,8 @@ from . import (
SIGNAL_SWITCHER_DEVICE_UPDATE, SIGNAL_SWITCHER_DEVICE_UPDATE,
) )
# pylint: disable=ungrouped-imports CONF_AUTO_OFF = "auto_off"
if TYPE_CHECKING: CONF_TIMER_MINUTES = "timer_minutes"
from aioswitcher.api.messages import SwitcherV2ControlResponseMSG
from aioswitcher.devices import SwitcherV2Device
DEVICE_PROPERTIES_TO_HA_ATTRIBUTES = { DEVICE_PROPERTIES_TO_HA_ATTRIBUTES = {
"power_consumption": ATTR_CURRENT_POWER_W, "power_consumption": ATTR_CURRENT_POWER_W,
@ -36,6 +37,18 @@ DEVICE_PROPERTIES_TO_HA_ATTRIBUTES = {
"auto_off_set": ATTR_AUTO_OFF_SET, "auto_off_set": ATTR_AUTO_OFF_SET,
} }
SERVICE_SET_AUTO_OFF_NAME = "set_auto_off"
SERVICE_SET_AUTO_OFF_SCHEMA = {
vol.Required(CONF_AUTO_OFF): cv.time_period_str,
}
SERVICE_TURN_ON_WITH_TIMER_NAME = "turn_on_with_timer"
SERVICE_TURN_ON_WITH_TIMER_SCHEMA = {
vol.Required(CONF_TIMER_MINUTES): vol.All(
cv.positive_int, vol.Range(min=1, max=150)
),
}
async def async_setup_platform( async def async_setup_platform(
hass: HomeAssistantType, hass: HomeAssistantType,
@ -46,13 +59,57 @@ async def async_setup_platform(
"""Set up the switcher platform for the switch component.""" """Set up the switcher platform for the switch component."""
if discovery_info is None: if discovery_info is None:
return return
async def async_set_auto_off_service(entity, service_call: ServiceCallType) -> None:
"""Use for handling setting device auto-off service calls."""
async with SwitcherV2Api(
hass.loop,
device_data.ip_addr,
device_data.phone_id,
device_data.device_id,
device_data.device_password,
) as swapi:
await swapi.set_auto_shutdown(service_call.data[CONF_AUTO_OFF])
async def async_turn_on_with_timer_service(
entity, service_call: ServiceCallType
) -> None:
"""Use for handling turning device on with a timer service calls."""
async with SwitcherV2Api(
hass.loop,
device_data.ip_addr,
device_data.phone_id,
device_data.device_id,
device_data.device_password,
) as swapi:
await swapi.control_device(
COMMAND_ON, service_call.data[CONF_TIMER_MINUTES]
)
device_data = hass.data[DOMAIN][DATA_DEVICE]
async_add_entities([SwitcherControl(hass.data[DOMAIN][DATA_DEVICE])]) async_add_entities([SwitcherControl(hass.data[DOMAIN][DATA_DEVICE])])
platform = entity_platform.current_platform.get()
platform.async_register_entity_service(
SERVICE_SET_AUTO_OFF_NAME,
SERVICE_SET_AUTO_OFF_SCHEMA,
async_set_auto_off_service,
)
platform.async_register_entity_service(
SERVICE_TURN_ON_WITH_TIMER_NAME,
SERVICE_TURN_ON_WITH_TIMER_SCHEMA,
async_turn_on_with_timer_service,
)
class SwitcherControl(SwitchEntity): class SwitcherControl(SwitchEntity):
"""Home Assistant switch entity.""" """Home Assistant switch entity."""
def __init__(self, device_data: "SwitcherV2Device") -> None: def __init__(self, device_data: SwitcherV2Device) -> None:
"""Initialize the entity.""" """Initialize the entity."""
self._self_initiated = False self._self_initiated = False
self._device_data = device_data self._device_data = device_data
@ -111,7 +168,7 @@ class SwitcherControl(SwitchEntity):
) )
) )
async def async_update_data(self, device_data: "SwitcherV2Device") -> None: async def async_update_data(self, device_data: SwitcherV2Device) -> None:
"""Update the entity data.""" """Update the entity data."""
if device_data: if device_data:
if self._self_initiated: if self._self_initiated:
@ -132,7 +189,7 @@ class SwitcherControl(SwitchEntity):
async def _control_device(self, send_on: bool) -> None: async def _control_device(self, send_on: bool) -> None:
"""Turn the entity on or off.""" """Turn the entity on or off."""
response: "SwitcherV2ControlResponseMSG" = None response: SwitcherV2ControlResponseMSG = None
async with SwitcherV2Api( async with SwitcherV2Api(
self.hass.loop, self.hass.loop,
self._device_data.ip_addr, self._device_data.ip_addr,

View File

@ -11,6 +11,7 @@ from .consts import (
DUMMY_AUTO_OFF_SET, DUMMY_AUTO_OFF_SET,
DUMMY_DEVICE_ID, DUMMY_DEVICE_ID,
DUMMY_DEVICE_NAME, DUMMY_DEVICE_NAME,
DUMMY_DEVICE_PASSWORD,
DUMMY_DEVICE_STATE, DUMMY_DEVICE_STATE,
DUMMY_ELECTRIC_CURRENT, DUMMY_ELECTRIC_CURRENT,
DUMMY_IP_ADDRESS, DUMMY_IP_ADDRESS,
@ -79,6 +80,11 @@ class MockSwitcherV2Device:
"""Return the phone id.""" """Return the phone id."""
return DUMMY_PHONE_ID return DUMMY_PHONE_ID
@property
def device_password(self) -> str:
"""Return the device password."""
return DUMMY_DEVICE_PASSWORD
@property @property
def last_data_update(self) -> datetime: def last_data_update(self) -> datetime:
"""Return the timestamp of the last update.""" """Return the timestamp of the last update."""
@ -169,10 +175,11 @@ def mock_api_fixture() -> Generator[AsyncMock, Any, None]:
patchers = [ patchers = [
patch( patch(
"homeassistant.components.switcher_kis.SwitcherV2Api.connect", new=mock_api "homeassistant.components.switcher_kis.switch.SwitcherV2Api.connect",
new=mock_api,
), ),
patch( patch(
"homeassistant.components.switcher_kis.SwitcherV2Api.disconnect", "homeassistant.components.switcher_kis.switch.SwitcherV2Api.disconnect",
new=mock_api, new=mock_api,
), ),
] ]

View File

@ -8,6 +8,7 @@ from homeassistant.components.switcher_kis import (
) )
DUMMY_AUTO_OFF_SET = "01:30:00" DUMMY_AUTO_OFF_SET = "01:30:00"
DUMMY_TIMER_MINUTES_SET = "90"
DUMMY_DEVICE_ID = "a123bc" DUMMY_DEVICE_ID = "a123bc"
DUMMY_DEVICE_NAME = "Device Name" DUMMY_DEVICE_NAME = "Device Name"
DUMMY_DEVICE_PASSWORD = "12345678" DUMMY_DEVICE_PASSWORD = "12345678"
@ -22,7 +23,7 @@ DUMMY_POWER_CONSUMPTION = 2780
DUMMY_REMAINING_TIME = "01:29:32" DUMMY_REMAINING_TIME = "01:29:32"
# Adjust if any modification were made to DUMMY_DEVICE_NAME # Adjust if any modification were made to DUMMY_DEVICE_NAME
SWITCH_ENTITY_ID = "switch.switcher_kis_device_name" SWITCH_ENTITY_ID = "switch.device_name"
MANDATORY_CONFIGURATION = { MANDATORY_CONFIGURATION = {
DOMAIN: { DOMAIN: {

View File

@ -1,21 +1,27 @@
"""Test cases for the switcher_kis component.""" """Test cases for the switcher_kis component."""
from datetime import timedelta from datetime import timedelta
from typing import TYPE_CHECKING, Any, Generator from typing import Any, Generator
from unittest.mock import patch
from aioswitcher.consts import COMMAND_ON
from aioswitcher.devices import SwitcherV2Device
from pytest import raises from pytest import raises
from homeassistant.components.switcher_kis import ( from homeassistant.components.switcher_kis import (
CONF_AUTO_OFF,
DATA_DEVICE, DATA_DEVICE,
DOMAIN, DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
SERVICE_SET_AUTO_OFF_SCHEMA,
SIGNAL_SWITCHER_DEVICE_UPDATE, SIGNAL_SWITCHER_DEVICE_UPDATE,
) )
from homeassistant.components.switcher_kis.switch import (
CONF_AUTO_OFF,
CONF_TIMER_MINUTES,
SERVICE_SET_AUTO_OFF_NAME,
SERVICE_TURN_ON_WITH_TIMER_NAME,
)
from homeassistant.const import CONF_ENTITY_ID from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import Context, callback from homeassistant.core import Context, callback
from homeassistant.exceptions import Unauthorized, UnknownUser from homeassistant.exceptions import UnknownUser
from homeassistant.helpers.config_validation import time_period_str
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
@ -32,16 +38,12 @@ from .consts import (
DUMMY_PHONE_ID, DUMMY_PHONE_ID,
DUMMY_POWER_CONSUMPTION, DUMMY_POWER_CONSUMPTION,
DUMMY_REMAINING_TIME, DUMMY_REMAINING_TIME,
DUMMY_TIMER_MINUTES_SET,
MANDATORY_CONFIGURATION, MANDATORY_CONFIGURATION,
SWITCH_ENTITY_ID, SWITCH_ENTITY_ID,
) )
from tests.common import async_fire_time_changed, async_mock_service from tests.common import MockUser, async_fire_time_changed
if TYPE_CHECKING:
from aioswitcher.devices import SwitcherV2Device
from tests.common import MockUser
async def test_failed_config( async def test_failed_config(
@ -83,8 +85,7 @@ async def test_set_auto_off_service(
hass: HomeAssistantType, hass: HomeAssistantType,
mock_bridge: Generator[None, Any, None], mock_bridge: Generator[None, Any, None],
mock_api: Generator[None, Any, None], mock_api: Generator[None, Any, None],
hass_owner_user: "MockUser", hass_owner_user: MockUser,
hass_read_only_user: "MockUser",
) -> None: ) -> None:
"""Test the set_auto_off service.""" """Test the set_auto_off service."""
assert await async_setup_component(hass, DOMAIN, MANDATORY_CONFIGURATION) assert await async_setup_component(hass, DOMAIN, MANDATORY_CONFIGURATION)
@ -101,31 +102,6 @@ async def test_set_auto_off_service(
context=Context(user_id=hass_owner_user.id), context=Context(user_id=hass_owner_user.id),
) )
with raises(Unauthorized) as unauthorized_read_only_exc:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
{CONF_ENTITY_ID: SWITCH_ENTITY_ID, CONF_AUTO_OFF: DUMMY_AUTO_OFF_SET},
blocking=True,
context=Context(user_id=hass_read_only_user.id),
)
assert unauthorized_read_only_exc.type is Unauthorized
with raises(Unauthorized) as unauthorized_wrong_entity_exc:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
{
CONF_ENTITY_ID: "light.not_related_entity",
CONF_AUTO_OFF: DUMMY_AUTO_OFF_SET,
},
blocking=True,
context=Context(user_id=hass_owner_user.id),
)
assert unauthorized_wrong_entity_exc.type is Unauthorized
with raises(UnknownUser) as unknown_user_exc: with raises(UnknownUser) as unknown_user_exc:
await hass.services.async_call( await hass.services.async_call(
DOMAIN, DOMAIN,
@ -137,20 +113,74 @@ async def test_set_auto_off_service(
assert unknown_user_exc.type is UnknownUser assert unknown_user_exc.type is UnknownUser
service_calls = async_mock_service( with patch(
hass, DOMAIN, SERVICE_SET_AUTO_OFF_NAME, SERVICE_SET_AUTO_OFF_SCHEMA "homeassistant.components.switcher_kis.switch.SwitcherV2Api.set_auto_shutdown"
) ) as mock_set_auto_shutdown:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
{CONF_ENTITY_ID: SWITCH_ENTITY_ID, CONF_AUTO_OFF: DUMMY_AUTO_OFF_SET},
)
await hass.services.async_call( await hass.async_block_till_done()
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME, mock_set_auto_shutdown.assert_called_once_with(
{CONF_ENTITY_ID: SWITCH_ENTITY_ID, CONF_AUTO_OFF: DUMMY_AUTO_OFF_SET}, time_period_str(DUMMY_AUTO_OFF_SET)
) )
async def test_turn_on_with_timer_service(
hass: HomeAssistantType,
mock_bridge: Generator[None, Any, None],
mock_api: Generator[None, Any, None],
hass_owner_user: MockUser,
) -> None:
"""Test the set_auto_off service."""
assert await async_setup_component(hass, DOMAIN, MANDATORY_CONFIGURATION)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(service_calls) == 1 assert hass.services.has_service(DOMAIN, SERVICE_TURN_ON_WITH_TIMER_NAME)
assert str(service_calls[0].data[CONF_AUTO_OFF]) == DUMMY_AUTO_OFF_SET.lstrip("0")
await hass.services.async_call(
DOMAIN,
SERVICE_TURN_ON_WITH_TIMER_NAME,
{CONF_ENTITY_ID: SWITCH_ENTITY_ID, CONF_TIMER_MINUTES: DUMMY_TIMER_MINUTES_SET},
blocking=True,
context=Context(user_id=hass_owner_user.id),
)
with raises(UnknownUser) as unknown_user_exc:
await hass.services.async_call(
DOMAIN,
SERVICE_TURN_ON_WITH_TIMER_NAME,
{
CONF_ENTITY_ID: SWITCH_ENTITY_ID,
CONF_TIMER_MINUTES: DUMMY_TIMER_MINUTES_SET,
},
blocking=True,
context=Context(user_id="not_real_user"),
)
assert unknown_user_exc.type is UnknownUser
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherV2Api.control_device"
) as mock_control_device:
await hass.services.async_call(
DOMAIN,
SERVICE_TURN_ON_WITH_TIMER_NAME,
{
CONF_ENTITY_ID: SWITCH_ENTITY_ID,
CONF_TIMER_MINUTES: DUMMY_TIMER_MINUTES_SET,
},
)
await hass.async_block_till_done()
mock_control_device.assert_called_once_with(
COMMAND_ON, int(DUMMY_TIMER_MINUTES_SET)
)
async def test_signal_dispatcher( async def test_signal_dispatcher(
@ -162,7 +192,7 @@ async def test_signal_dispatcher(
await hass.async_block_till_done() await hass.async_block_till_done()
@callback @callback
def verify_update_data(device: "SwitcherV2Device") -> None: def verify_update_data(device: SwitcherV2Device) -> None:
"""Use as callback for signal dispatcher.""" """Use as callback for signal dispatcher."""
pass pass