mirror of
https://github.com/home-assistant/core.git
synced 2025-09-06 05:11:35 +02:00
Add cover platform for switchbot cloud (#148993)
This commit is contained in:
@@ -29,6 +29,7 @@ PLATFORMS: list[Platform] = [
|
|||||||
Platform.BINARY_SENSOR,
|
Platform.BINARY_SENSOR,
|
||||||
Platform.BUTTON,
|
Platform.BUTTON,
|
||||||
Platform.CLIMATE,
|
Platform.CLIMATE,
|
||||||
|
Platform.COVER,
|
||||||
Platform.FAN,
|
Platform.FAN,
|
||||||
Platform.LIGHT,
|
Platform.LIGHT,
|
||||||
Platform.LOCK,
|
Platform.LOCK,
|
||||||
@@ -47,6 +48,7 @@ class SwitchbotDevices:
|
|||||||
)
|
)
|
||||||
buttons: list[tuple[Device, SwitchBotCoordinator]] = field(default_factory=list)
|
buttons: list[tuple[Device, SwitchBotCoordinator]] = field(default_factory=list)
|
||||||
climates: list[tuple[Remote, SwitchBotCoordinator]] = field(default_factory=list)
|
climates: list[tuple[Remote, SwitchBotCoordinator]] = field(default_factory=list)
|
||||||
|
covers: list[tuple[Device, SwitchBotCoordinator]] = field(default_factory=list)
|
||||||
switches: list[tuple[Device | Remote, SwitchBotCoordinator]] = field(
|
switches: list[tuple[Device | Remote, SwitchBotCoordinator]] = field(
|
||||||
default_factory=list
|
default_factory=list
|
||||||
)
|
)
|
||||||
@@ -192,6 +194,27 @@ async def make_device_data(
|
|||||||
)
|
)
|
||||||
devices_data.fans.append((device, coordinator))
|
devices_data.fans.append((device, coordinator))
|
||||||
devices_data.sensors.append((device, coordinator))
|
devices_data.sensors.append((device, coordinator))
|
||||||
|
if isinstance(device, Device) and device.device_type in [
|
||||||
|
"Curtain",
|
||||||
|
"Curtain3",
|
||||||
|
"Roller Shade",
|
||||||
|
"Blind Tilt",
|
||||||
|
]:
|
||||||
|
coordinator = await coordinator_for_device(
|
||||||
|
hass, entry, api, device, coordinators_by_id
|
||||||
|
)
|
||||||
|
devices_data.covers.append((device, coordinator))
|
||||||
|
devices_data.binary_sensors.append((device, coordinator))
|
||||||
|
devices_data.sensors.append((device, coordinator))
|
||||||
|
|
||||||
|
if isinstance(device, Device) and device.device_type in [
|
||||||
|
"Garage Door Opener",
|
||||||
|
]:
|
||||||
|
coordinator = await coordinator_for_device(
|
||||||
|
hass, entry, api, device, coordinators_by_id
|
||||||
|
)
|
||||||
|
devices_data.covers.append((device, coordinator))
|
||||||
|
devices_data.binary_sensors.append((device, coordinator))
|
||||||
|
|
||||||
if isinstance(device, Device) and device.device_type in [
|
if isinstance(device, Device) and device.device_type in [
|
||||||
"Strip Light",
|
"Strip Light",
|
||||||
|
@@ -60,6 +60,11 @@ BINARY_SENSOR_DESCRIPTIONS_BY_DEVICE_TYPES = {
|
|||||||
CALIBRATION_DESCRIPTION,
|
CALIBRATION_DESCRIPTION,
|
||||||
DOOR_OPEN_DESCRIPTION,
|
DOOR_OPEN_DESCRIPTION,
|
||||||
),
|
),
|
||||||
|
"Curtain": (CALIBRATION_DESCRIPTION,),
|
||||||
|
"Curtain3": (CALIBRATION_DESCRIPTION,),
|
||||||
|
"Roller Shade": (CALIBRATION_DESCRIPTION,),
|
||||||
|
"Blind Tilt": (CALIBRATION_DESCRIPTION,),
|
||||||
|
"Garage Door Opener": (DOOR_OPEN_DESCRIPTION,),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@@ -17,3 +17,5 @@ VACUUM_FAN_SPEED_STRONG = "strong"
|
|||||||
VACUUM_FAN_SPEED_MAX = "max"
|
VACUUM_FAN_SPEED_MAX = "max"
|
||||||
|
|
||||||
AFTER_COMMAND_REFRESH = 5
|
AFTER_COMMAND_REFRESH = 5
|
||||||
|
|
||||||
|
COVER_ENTITY_AFTER_COMMAND_REFRESH = 10
|
||||||
|
233
homeassistant/components/switchbot_cloud/cover.py
Normal file
233
homeassistant/components/switchbot_cloud/cover.py
Normal file
@@ -0,0 +1,233 @@
|
|||||||
|
"""Support for the Switchbot BlindTilt, Curtain, Curtain3, RollerShade as Cover."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from switchbot_api import (
|
||||||
|
BlindTiltCommands,
|
||||||
|
CommonCommands,
|
||||||
|
CurtainCommands,
|
||||||
|
Device,
|
||||||
|
Remote,
|
||||||
|
RollerShadeCommands,
|
||||||
|
SwitchBotAPI,
|
||||||
|
)
|
||||||
|
|
||||||
|
from homeassistant.components.cover import (
|
||||||
|
CoverDeviceClass,
|
||||||
|
CoverEntity,
|
||||||
|
CoverEntityFeature,
|
||||||
|
)
|
||||||
|
from homeassistant.config_entries import ConfigEntry
|
||||||
|
from homeassistant.core import HomeAssistant, callback
|
||||||
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
|
||||||
|
from . import SwitchbotCloudData, SwitchBotCoordinator
|
||||||
|
from .const import COVER_ENTITY_AFTER_COMMAND_REFRESH, DOMAIN
|
||||||
|
from .entity import SwitchBotCloudEntity
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
config: ConfigEntry,
|
||||||
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
"""Set up SwitchBot Cloud entry."""
|
||||||
|
data: SwitchbotCloudData = hass.data[DOMAIN][config.entry_id]
|
||||||
|
async_add_entities(
|
||||||
|
_async_make_entity(data.api, device, coordinator)
|
||||||
|
for device, coordinator in data.devices.covers
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class SwitchBotCloudCover(SwitchBotCloudEntity, CoverEntity):
|
||||||
|
"""Representation of a SwitchBot Cover."""
|
||||||
|
|
||||||
|
_attr_name = None
|
||||||
|
_attr_is_closed: bool | None = None
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
if self.coordinator.data is None:
|
||||||
|
return
|
||||||
|
position: int | None = self.coordinator.data.get("slidePosition")
|
||||||
|
if position is None:
|
||||||
|
return
|
||||||
|
self._attr_current_cover_position = 100 - position
|
||||||
|
self._attr_current_cover_tilt_position = 100 - position
|
||||||
|
self._attr_is_closed = position == 100
|
||||||
|
|
||||||
|
|
||||||
|
class SwitchBotCloudCoverCurtain(SwitchBotCloudCover):
|
||||||
|
"""Representation of a SwitchBot Curtain & Curtain3."""
|
||||||
|
|
||||||
|
_attr_device_class = CoverDeviceClass.CURTAIN
|
||||||
|
_attr_supported_features: CoverEntityFeature = (
|
||||||
|
CoverEntityFeature.OPEN
|
||||||
|
| CoverEntityFeature.CLOSE
|
||||||
|
| CoverEntityFeature.STOP
|
||||||
|
| CoverEntityFeature.SET_POSITION
|
||||||
|
)
|
||||||
|
|
||||||
|
async def async_open_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Open the cover."""
|
||||||
|
await self.send_api_command(CommonCommands.ON)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_close_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Close the cover."""
|
||||||
|
await self.send_api_command(CommonCommands.OFF)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_set_cover_position(self, **kwargs: Any) -> None:
|
||||||
|
"""Move the cover to a specific position."""
|
||||||
|
position: int | None = kwargs.get("position")
|
||||||
|
if position is not None:
|
||||||
|
await self.send_api_command(
|
||||||
|
CurtainCommands.SET_POSITION,
|
||||||
|
parameters=f"{0},ff,{100 - position}",
|
||||||
|
)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_stop_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Stop the cover."""
|
||||||
|
await self.send_api_command(CurtainCommands.PAUSE)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
|
||||||
|
class SwitchBotCloudCoverRollerShade(SwitchBotCloudCover):
|
||||||
|
"""Representation of a SwitchBot RollerShade."""
|
||||||
|
|
||||||
|
_attr_device_class = CoverDeviceClass.SHADE
|
||||||
|
_attr_supported_features: CoverEntityFeature = (
|
||||||
|
CoverEntityFeature.SET_POSITION
|
||||||
|
| CoverEntityFeature.OPEN
|
||||||
|
| CoverEntityFeature.CLOSE
|
||||||
|
)
|
||||||
|
|
||||||
|
async def async_open_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Open the cover."""
|
||||||
|
await self.send_api_command(RollerShadeCommands.SET_POSITION, parameters=str(0))
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_close_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Close the cover."""
|
||||||
|
await self.send_api_command(
|
||||||
|
RollerShadeCommands.SET_POSITION, parameters=str(100)
|
||||||
|
)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_set_cover_position(self, **kwargs: Any) -> None:
|
||||||
|
"""Move the cover to a specific position."""
|
||||||
|
position: int | None = kwargs.get("position")
|
||||||
|
if position is not None:
|
||||||
|
await self.send_api_command(
|
||||||
|
RollerShadeCommands.SET_POSITION, parameters=str(100 - position)
|
||||||
|
)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
|
||||||
|
class SwitchBotCloudCoverBlindTilt(SwitchBotCloudCover):
|
||||||
|
"""Representation of a SwitchBot Blind Tilt."""
|
||||||
|
|
||||||
|
_attr_direction: str | None = None
|
||||||
|
_attr_device_class = CoverDeviceClass.BLIND
|
||||||
|
_attr_supported_features: CoverEntityFeature = (
|
||||||
|
CoverEntityFeature.SET_TILT_POSITION
|
||||||
|
| CoverEntityFeature.OPEN_TILT
|
||||||
|
| CoverEntityFeature.CLOSE_TILT
|
||||||
|
)
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
if self.coordinator.data is None:
|
||||||
|
return
|
||||||
|
position: int | None = self.coordinator.data.get("slidePosition")
|
||||||
|
if position is None:
|
||||||
|
return
|
||||||
|
self._attr_is_closed = position in [0, 100]
|
||||||
|
if position > 50:
|
||||||
|
percent = 100 - ((position - 50) * 2)
|
||||||
|
else:
|
||||||
|
percent = 100 - (50 - position) * 2
|
||||||
|
self._attr_current_cover_position = percent
|
||||||
|
self._attr_current_cover_tilt_position = percent
|
||||||
|
direction = self.coordinator.data.get("direction")
|
||||||
|
self._attr_direction = direction.lower() if direction else None
|
||||||
|
|
||||||
|
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
|
||||||
|
"""Move the cover to a specific position."""
|
||||||
|
percent: int | None = kwargs.get("tilt_position")
|
||||||
|
if percent is not None:
|
||||||
|
await self.send_api_command(
|
||||||
|
BlindTiltCommands.SET_POSITION,
|
||||||
|
parameters=f"{self._attr_direction};{percent}",
|
||||||
|
)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_open_cover_tilt(self, **kwargs: Any) -> None:
|
||||||
|
"""Open the cover."""
|
||||||
|
await self.send_api_command(BlindTiltCommands.FULLY_OPEN)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_close_cover_tilt(self, **kwargs: Any) -> None:
|
||||||
|
"""Close the cover."""
|
||||||
|
if self._attr_direction is not None:
|
||||||
|
if "up" in self._attr_direction:
|
||||||
|
await self.send_api_command(BlindTiltCommands.CLOSE_UP)
|
||||||
|
else:
|
||||||
|
await self.send_api_command(BlindTiltCommands.CLOSE_DOWN)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
|
||||||
|
class SwitchBotCloudCoverGarageDoorOpener(SwitchBotCloudCover):
|
||||||
|
"""Representation of a SwitchBot Garage Door Opener."""
|
||||||
|
|
||||||
|
_attr_device_class = CoverDeviceClass.GARAGE
|
||||||
|
_attr_supported_features: CoverEntityFeature = (
|
||||||
|
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
|
||||||
|
)
|
||||||
|
|
||||||
|
def _set_attributes(self) -> None:
|
||||||
|
if self.coordinator.data is None:
|
||||||
|
return
|
||||||
|
door_status: int | None = self.coordinator.data.get("doorStatus")
|
||||||
|
self._attr_is_closed = None if door_status is None else door_status == 1
|
||||||
|
|
||||||
|
async def async_open_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Open the cover."""
|
||||||
|
await self.send_api_command(CommonCommands.ON)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
async def async_close_cover(self, **kwargs: Any) -> None:
|
||||||
|
"""Close the cover."""
|
||||||
|
await self.send_api_command(CommonCommands.OFF)
|
||||||
|
await asyncio.sleep(COVER_ENTITY_AFTER_COMMAND_REFRESH)
|
||||||
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
|
|
||||||
|
@callback
|
||||||
|
def _async_make_entity(
|
||||||
|
api: SwitchBotAPI, device: Device | Remote, coordinator: SwitchBotCoordinator
|
||||||
|
) -> (
|
||||||
|
SwitchBotCloudCoverBlindTilt
|
||||||
|
| SwitchBotCloudCoverRollerShade
|
||||||
|
| SwitchBotCloudCoverCurtain
|
||||||
|
| SwitchBotCloudCoverGarageDoorOpener
|
||||||
|
):
|
||||||
|
"""Make a SwitchBotCloudCover device."""
|
||||||
|
if device.device_type == "Blind Tilt":
|
||||||
|
return SwitchBotCloudCoverBlindTilt(api, device, coordinator)
|
||||||
|
if device.device_type == "Roller Shade":
|
||||||
|
return SwitchBotCloudCoverRollerShade(api, device, coordinator)
|
||||||
|
if device.device_type == "Garage Door Opener":
|
||||||
|
return SwitchBotCloudCoverGarageDoorOpener(api, device, coordinator)
|
||||||
|
return SwitchBotCloudCoverCurtain(api, device, coordinator)
|
@@ -15,7 +15,7 @@ from homeassistant.core import HomeAssistant
|
|||||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||||
|
|
||||||
from . import SwitchbotCloudData
|
from . import SwitchbotCloudData
|
||||||
from .const import DOMAIN
|
from .const import AFTER_COMMAND_REFRESH, DOMAIN
|
||||||
from .entity import SwitchBotCloudEntity
|
from .entity import SwitchBotCloudEntity
|
||||||
|
|
||||||
|
|
||||||
@@ -88,13 +88,13 @@ class SwitchBotCloudFan(SwitchBotCloudEntity, FanEntity):
|
|||||||
command=BatteryCirculatorFanCommands.SET_WIND_SPEED,
|
command=BatteryCirculatorFanCommands.SET_WIND_SPEED,
|
||||||
parameters=str(self.percentage),
|
parameters=str(self.percentage),
|
||||||
)
|
)
|
||||||
await asyncio.sleep(5)
|
await asyncio.sleep(AFTER_COMMAND_REFRESH)
|
||||||
await self.coordinator.async_request_refresh()
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
async def async_turn_off(self, **kwargs: Any) -> None:
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
||||||
"""Turn off the fan."""
|
"""Turn off the fan."""
|
||||||
await self.send_api_command(CommonCommands.OFF)
|
await self.send_api_command(CommonCommands.OFF)
|
||||||
await asyncio.sleep(5)
|
await asyncio.sleep(AFTER_COMMAND_REFRESH)
|
||||||
await self.coordinator.async_request_refresh()
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
async def async_set_percentage(self, percentage: int) -> None:
|
async def async_set_percentage(self, percentage: int) -> None:
|
||||||
@@ -107,7 +107,7 @@ class SwitchBotCloudFan(SwitchBotCloudEntity, FanEntity):
|
|||||||
command=BatteryCirculatorFanCommands.SET_WIND_SPEED,
|
command=BatteryCirculatorFanCommands.SET_WIND_SPEED,
|
||||||
parameters=str(percentage),
|
parameters=str(percentage),
|
||||||
)
|
)
|
||||||
await asyncio.sleep(5)
|
await asyncio.sleep(AFTER_COMMAND_REFRESH)
|
||||||
await self.coordinator.async_request_refresh()
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
async def async_set_preset_mode(self, preset_mode: str) -> None:
|
async def async_set_preset_mode(self, preset_mode: str) -> None:
|
||||||
@@ -116,5 +116,5 @@ class SwitchBotCloudFan(SwitchBotCloudEntity, FanEntity):
|
|||||||
command=BatteryCirculatorFanCommands.SET_WIND_MODE,
|
command=BatteryCirculatorFanCommands.SET_WIND_MODE,
|
||||||
parameters=preset_mode,
|
parameters=preset_mode,
|
||||||
)
|
)
|
||||||
await asyncio.sleep(5)
|
await asyncio.sleep(AFTER_COMMAND_REFRESH)
|
||||||
await self.coordinator.async_request_refresh()
|
await self.coordinator.async_request_refresh()
|
||||||
|
@@ -139,6 +139,10 @@ SENSOR_DESCRIPTIONS_BY_DEVICE_TYPES = {
|
|||||||
"Smart Lock Lite": (BATTERY_DESCRIPTION,),
|
"Smart Lock Lite": (BATTERY_DESCRIPTION,),
|
||||||
"Smart Lock Pro": (BATTERY_DESCRIPTION,),
|
"Smart Lock Pro": (BATTERY_DESCRIPTION,),
|
||||||
"Smart Lock Ultra": (BATTERY_DESCRIPTION,),
|
"Smart Lock Ultra": (BATTERY_DESCRIPTION,),
|
||||||
|
"Curtain": (BATTERY_DESCRIPTION,),
|
||||||
|
"Curtain3": (BATTERY_DESCRIPTION,),
|
||||||
|
"Roller Shade": (BATTERY_DESCRIPTION,),
|
||||||
|
"Blind Tilt": (BATTERY_DESCRIPTION,),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@@ -39,3 +39,13 @@ def mock_after_command_refresh():
|
|||||||
"homeassistant.components.switchbot_cloud.const.AFTER_COMMAND_REFRESH", 0
|
"homeassistant.components.switchbot_cloud.const.AFTER_COMMAND_REFRESH", 0
|
||||||
):
|
):
|
||||||
yield
|
yield
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="package", autouse=True)
|
||||||
|
def mock_after_command_refresh_for_cover():
|
||||||
|
"""Mock after command refresh."""
|
||||||
|
with patch(
|
||||||
|
"homeassistant.components.switchbot_cloud.const.COVER_ENTITY_AFTER_COMMAND_REFRESH",
|
||||||
|
0,
|
||||||
|
):
|
||||||
|
yield
|
||||||
|
457
tests/components/switchbot_cloud/test_cover.py
Normal file
457
tests/components/switchbot_cloud/test_cover.py
Normal file
@@ -0,0 +1,457 @@
|
|||||||
|
"""Test for the switchbot_cloud Cover."""
|
||||||
|
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from switchbot_api import (
|
||||||
|
BlindTiltCommands,
|
||||||
|
CommonCommands,
|
||||||
|
CurtainCommands,
|
||||||
|
Device,
|
||||||
|
RollerShadeCommands,
|
||||||
|
)
|
||||||
|
|
||||||
|
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN
|
||||||
|
from homeassistant.components.switchbot_cloud import SwitchBotAPI
|
||||||
|
from homeassistant.config_entries import ConfigEntryState
|
||||||
|
from homeassistant.const import (
|
||||||
|
ATTR_ENTITY_ID,
|
||||||
|
SERVICE_CLOSE_COVER,
|
||||||
|
SERVICE_CLOSE_COVER_TILT,
|
||||||
|
SERVICE_OPEN_COVER,
|
||||||
|
SERVICE_OPEN_COVER_TILT,
|
||||||
|
SERVICE_SET_COVER_POSITION,
|
||||||
|
SERVICE_SET_COVER_TILT_POSITION,
|
||||||
|
SERVICE_STOP_COVER,
|
||||||
|
STATE_CLOSED,
|
||||||
|
STATE_OPEN,
|
||||||
|
STATE_UNKNOWN,
|
||||||
|
)
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
|
from . import configure_integration
|
||||||
|
|
||||||
|
|
||||||
|
async def test_cover_set_attributes_normal(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test cover set_attributes normal."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Roller Shade",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
mock_get_status.return_value = {"slidePosition": 100, "direction": "up"}
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_CLOSED
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"device_model",
|
||||||
|
[
|
||||||
|
"Roller Shade",
|
||||||
|
"Blind Tilt",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_cover_set_attributes_position_is_none(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status, device_model
|
||||||
|
) -> None:
|
||||||
|
"""Test cover_set_attributes position is none."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType=device_model,
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
mock_get_status.side_effect = [{"direction": "up"}, {"direction": "up"}]
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_UNKNOWN
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"device_model",
|
||||||
|
[
|
||||||
|
"Roller Shade",
|
||||||
|
"Blind Tilt",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_cover_set_attributes_coordinator_is_none(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status, device_model
|
||||||
|
) -> None:
|
||||||
|
"""Test cover set_attributes coordinator is none."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType=device_model,
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
mock_get_status.return_value = None
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_UNKNOWN
|
||||||
|
|
||||||
|
|
||||||
|
async def test_curtain_features(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test curtain features."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Curtain",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_get_status.side_effect = [
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
]
|
||||||
|
entry = await configure_integration(hass)
|
||||||
|
assert entry.state is ConfigEntryState.LOADED
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_OPEN_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", CommonCommands.ON, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_CLOSE_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", CommonCommands.OFF, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_STOP_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", CurtainCommands.PAUSE, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_SET_COVER_POSITION,
|
||||||
|
{"position": 50, ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", CurtainCommands.SET_POSITION, "command", "0,ff,50"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_blind_tilt_features(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test blind_tilt features."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Blind Tilt",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_get_status.side_effect = [
|
||||||
|
{"slidePosition": 95, "direction": "up"},
|
||||||
|
{"slidePosition": 95, "direction": "up"},
|
||||||
|
]
|
||||||
|
entry = await configure_integration(hass)
|
||||||
|
assert entry.state is ConfigEntryState.LOADED
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_OPEN_COVER_TILT,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", BlindTiltCommands.FULLY_OPEN, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_CLOSE_COVER_TILT,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", BlindTiltCommands.CLOSE_UP, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_SET_COVER_TILT_POSITION,
|
||||||
|
{"tilt_position": 25, ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", BlindTiltCommands.SET_POSITION, "command", "up;25"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_blind_tilt_features_close_down(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test blind tilt features close_down."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Blind Tilt",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_get_status.side_effect = [
|
||||||
|
{"slidePosition": 25, "direction": "down"},
|
||||||
|
{"slidePosition": 25, "direction": "down"},
|
||||||
|
]
|
||||||
|
entry = await configure_integration(hass)
|
||||||
|
assert entry.state is ConfigEntryState.LOADED
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_CLOSE_COVER_TILT,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", BlindTiltCommands.CLOSE_DOWN, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_roller_shade_features(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test roller shade features."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Roller Shade",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_get_status.side_effect = [
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"slidePosition": 95,
|
||||||
|
},
|
||||||
|
]
|
||||||
|
entry = await configure_integration(hass)
|
||||||
|
assert entry.state is ConfigEntryState.LOADED
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_OPEN_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", RollerShadeCommands.SET_POSITION, "command", "0"
|
||||||
|
)
|
||||||
|
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_OPEN
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_CLOSE_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", RollerShadeCommands.SET_POSITION, "command", "100"
|
||||||
|
)
|
||||||
|
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_OPEN
|
||||||
|
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_SET_COVER_POSITION,
|
||||||
|
{"position": 50, ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", RollerShadeCommands.SET_POSITION, "command", "50"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_cover_set_attributes_coordinator_is_none_for_garage_door(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test cover set_attributes coordinator is none for garage_door."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Garage Door Opener",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
mock_get_status.return_value = None
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_UNKNOWN
|
||||||
|
|
||||||
|
|
||||||
|
async def test_garage_door_features_close(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test garage door features close."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Garage Door Opener",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_get_status.side_effect = [
|
||||||
|
{
|
||||||
|
"doorStatus": 1,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"doorStatus": 1,
|
||||||
|
},
|
||||||
|
]
|
||||||
|
entry = await configure_integration(hass)
|
||||||
|
assert entry.state is ConfigEntryState.LOADED
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_CLOSE_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", CommonCommands.OFF, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_CLOSED
|
||||||
|
|
||||||
|
|
||||||
|
async def test_garage_door_features_open(
|
||||||
|
hass: HomeAssistant, mock_list_devices, mock_get_status
|
||||||
|
) -> None:
|
||||||
|
"""Test garage_door features open cover."""
|
||||||
|
mock_list_devices.return_value = [
|
||||||
|
Device(
|
||||||
|
version="V1.0",
|
||||||
|
deviceId="cover-id-1",
|
||||||
|
deviceName="cover-1",
|
||||||
|
deviceType="Garage Door Opener",
|
||||||
|
hubDeviceId="test-hub-id",
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
mock_get_status.side_effect = [
|
||||||
|
{
|
||||||
|
"doorStatus": 0,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"doorStatus": 0,
|
||||||
|
},
|
||||||
|
]
|
||||||
|
entry = await configure_integration(hass)
|
||||||
|
assert entry.state is ConfigEntryState.LOADED
|
||||||
|
|
||||||
|
cover_id = "cover.cover_1"
|
||||||
|
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
|
||||||
|
await hass.services.async_call(
|
||||||
|
COVER_DOMAIN,
|
||||||
|
SERVICE_OPEN_COVER,
|
||||||
|
{ATTR_ENTITY_ID: cover_id},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
mock_send_command.assert_called_once_with(
|
||||||
|
"cover-id-1", CommonCommands.ON, "command", "default"
|
||||||
|
)
|
||||||
|
|
||||||
|
await configure_integration(hass)
|
||||||
|
state = hass.states.get(cover_id)
|
||||||
|
assert state.state == STATE_OPEN
|
Reference in New Issue
Block a user