jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-13 00:07:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Switcher Runner S11 support (#123578)

Browse Source 
* switcher start s11 integration

* switcher linting

* switcher starting reauth logic

* switcher fix linting

* switcher fix linting

* switcher remove get_circuit_number

* switcher adding support for validate token

* switcher fix initial auth for new devices and fix strings

* switcher fix linting

* switcher fix utils

* Revert "switcher fix utils"

This reverts commit b162a943b94fb0a581140feb21fe871df578c16a.

* switcher revert and test

* switcher fix validate logic and strings

* switcher add tests to improve coverage

* switcher adding tests

* switcher adding test

* switcher revert back things

* switcher fix based on requested changes

* switcher tests fixes

* switcher fix based on requested changes

* switcher remove single_instance_allowed code and added tests

* Update config_flow.py

* switcher fix comment

* switcher fix tests

* switcher lint

* switcehr fix based on requested changes

* switche fix lint

* switcher small rename fix

* switcher fix based on requested changes

* switcher fix based on requested changes

* switcher fix based on requested changes

* Update tests/components/switcher_kis/test_config_flow.py

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* Update tests/components/switcher_kis/test_config_flow.py

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* Update tests/components/switcher_kis/test_config_flow.py

Co-authored-by: Shay Levy <levyshay1@gmail.com>

* Update tests/components/switcher_kis/test_config_flow.py

---------

Co-authored-by: Shay Levy <levyshay1@gmail.com>
This commit is contained in:
YogevBokobza 2024-09-20 23:19:57 +03:00 committed by GitHub
parent 65fb688164
commit 3e1da876c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 449 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
homeassistant/components/switcher_kis/__init__.py
View File

@ -8,7 +8,7 @@ from aioswitcher.bridge import SwitcherBridge
from aioswitcher.device import SwitcherBase
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.const import CONF_TOKEN, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
@ -32,6 +32,8 @@ type SwitcherConfigEntry = ConfigEntry[dict[str, SwitcherDataUpdateCoordinator]]
async def async_setup_entry(hass: HomeAssistant, entry: SwitcherConfigEntry) -> bool:
"""Set up Switcher from a config entry."""
token = entry.data.get(CONF_TOKEN)
@callback
def on_device_data_callback(device: SwitcherBase) -> None:
"""Use as a callback for device data."""
@ -45,14 +47,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: SwitcherConfigEntry) ->
# New device - create device
_LOGGER.info(
"Discovered Switcher device - id: %s, key: %s, name: %s, type: %s (%s)",
"Discovered Switcher device - id: %s, key: %s, name: %s, type: %s (%s), is_token_needed: %s",
device.device_id,
device.device_key,
device.name,
device.device_type.value,
device.device_type.hex_rep,
device.token_needed,
)
if device.token_needed and not token:
entry.async_start_reauth(hass)
return
coordinator = SwitcherDataUpdateCoordinator(hass, entry, device)
coordinator.async_setup()
coordinators[device.device_id] = coordinator

114
homeassistant/components/switcher_kis/config_flow.py
View File

@ -2,9 +2,117 @@
from __future__ import annotations
from homeassistant.helpers import config_entry_flow
from collections.abc import Mapping
import logging
from typing import Any, Final
from aioswitcher.bridge import SwitcherBase
from aioswitcher.device.tools import validate_token
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_TOKEN, CONF_USERNAME
from .const import DOMAIN
from .utils import async_has_devices
from .utils import async_discover_devices
config_entry_flow.register_discovery_flow(DOMAIN, "Switcher", async_has_devices)
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA: Final = vol.Schema(
{
vol.Required(CONF_USERNAME, default=""): str,
vol.Required(CONF_TOKEN, default=""): str,
}
)
class SwitcherFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle Switcher config flow."""
VERSION = 1
entry: ConfigEntry | None = None
username: str | None = None
token: str | None = None
discovered_devices: dict[str, SwitcherBase] = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the start of the config flow."""
self.discovered_devices = await async_discover_devices()
return self.async_show_form(step_id="confirm")
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle user-confirmation of the config flow."""
if len(self.discovered_devices) == 0:
return self.async_abort(reason="no_devices_found")
for device_id, device in self.discovered_devices.items():
if device.token_needed:
_LOGGER.debug("Device with ID %s requires a token", device_id)
return await self.async_step_credentials()
return await self._create_entry()
async def async_step_credentials(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the credentials step."""
errors: dict[str, str] = {}
if user_input is not None:
self.username = user_input.get(CONF_USERNAME)
self.token = user_input.get(CONF_TOKEN)
token_is_valid = await validate_token(
user_input[CONF_USERNAME], user_input[CONF_TOKEN]
)
if token_is_valid:
return await self._create_entry()
errors["base"] = "invalid_auth"
return self.async_show_form(
step_id="credentials", data_schema=CONFIG_SCHEMA, errors=errors
)
async def async_step_reauth(
self, user_input: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle configuration by re-auth."""
self.entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that informs the user that reauth is required."""
errors: dict[str, str] = {}
assert self.entry is not None
if user_input is not None:
token_is_valid = await validate_token(
user_input[CONF_USERNAME], user_input[CONF_TOKEN]
)
if token_is_valid:
return self.async_update_reload_and_abort(
self.entry, data={**self.entry.data, **user_input}
)
errors["base"] = "invalid_auth"
return self.async_show_form(
step_id="reauth_confirm",
data_schema=CONFIG_SCHEMA,
errors=errors,
)
async def _create_entry(self) -> ConfigFlowResult:
return self.async_create_entry(
title="Switcher",
data={
CONF_USERNAME: self.username,
CONF_TOKEN: self.token,
},
)

7
homeassistant/components/switcher_kis/coordinator.py
View File

@ -8,6 +8,7 @@ import logging
from aioswitcher.device import SwitcherBase
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TOKEN
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, update_coordinator
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -23,7 +24,10 @@ class SwitcherDataUpdateCoordinator(
"""Switcher device data update coordinator."""
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, device: SwitcherBase
self,
hass: HomeAssistant,
entry: ConfigEntry,
device: SwitcherBase,
) -> None:
"""Initialize the Switcher device coordinator."""
super().__init__(
@ -34,6 +38,7 @@ class SwitcherDataUpdateCoordinator(
)
self.entry = entry
self.data = device
self.token = entry.data.get(CONF_TOKEN)
async def _async_update_data(self) -> SwitcherBase:
"""Mark device offline if no data."""

25
homeassistant/components/switcher_kis/cover.py
View File

@ -42,8 +42,11 @@ async def async_setup_entry(
@callback
def async_add_cover(coordinator: SwitcherDataUpdateCoordinator) -> None:
"""Add cover from Switcher device."""
if coordinator.data.device_type.category == DeviceCategory.SHUTTER:
async_add_entities([SwitcherCoverEntity(coordinator)])
if coordinator.data.device_type.category in (
DeviceCategory.SHUTTER,
DeviceCategory.SINGLE_SHUTTER_DUAL_LIGHT,
):
async_add_entities([SwitcherCoverEntity(coordinator, 0)])
config_entry.async_on_unload(
async_dispatcher_connect(hass, SIGNAL_DEVICE_ADD, async_add_cover)
@ -65,9 +68,14 @@ class SwitcherCoverEntity(
| CoverEntityFeature.STOP
)
def __init__(self, coordinator: SwitcherDataUpdateCoordinator) -> None:
def __init__(
self,
coordinator: SwitcherDataUpdateCoordinator,
cover_id: int | None = None,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._cover_id = cover_id
self._attr_unique_id = f"{coordinator.device_id}-{coordinator.mac_address}"
self._attr_device_info = DeviceInfo(
@ -102,6 +110,7 @@ class SwitcherCoverEntity(
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
self.coordinator.token,
) as swapi:
response = await getattr(swapi, api)(*args)
except (TimeoutError, OSError, RuntimeError) as err:
@ -117,16 +126,18 @@ class SwitcherCoverEntity(
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close cover."""
await self._async_call_api(API_SET_POSITON, 0)
await self._async_call_api(API_SET_POSITON, 0, self._cover_id)
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open cover."""
await self._async_call_api(API_SET_POSITON, 100)
await self._async_call_api(API_SET_POSITON, 100, self._cover_id)
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
await self._async_call_api(API_SET_POSITON, kwargs[ATTR_POSITION])
await self._async_call_api(
API_SET_POSITON, kwargs[ATTR_POSITION], self._cover_id
)
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
await self._async_call_api(API_STOP)
await self._async_call_api(API_STOP, self._cover_id)

20
homeassistant/components/switcher_kis/strings.json
View File

@ -3,11 +3,29 @@
"step": {
"confirm": {
"description": "[%key:common::config_flow::description::confirm_setup%]"
},
"credentials": {
"description": "Found a Switcher device that requires a token\nEnter your username and token\nFor more information see https://www.home-assistant.io/integrations/switcher_kis/#prerequisites",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"token": "[%key:common::config_flow::data::access_token%]"
}
},
"reauth_confirm": {
"description": "Found a Switcher device that requires a token\nEnter your username and token\nFor more information see https://www.home-assistant.io/integrations/switcher_kis/#prerequisites",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"token": "[%key:common::config_flow::data::access_token%]"
}
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"abort": {
"single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"entity": {

4
homeassistant/components/switcher_kis/utils.py
View File

@ -16,7 +16,7 @@ from .const import DISCOVERY_TIME_SEC
_LOGGER = logging.getLogger(__name__)
async def async_has_devices(hass: HomeAssistant) -> bool:
async def async_discover_devices() -> dict[str, SwitcherBase]:
"""Discover Switcher devices."""
_LOGGER.debug("Starting discovery")
discovered_devices = {}
@ -35,7 +35,7 @@ async def async_has_devices(hass: HomeAssistant) -> bool:
await bridge.stop()
_LOGGER.debug("Finished discovery, discovered devices: %s", len(discovered_devices))
return len(discovered_devices) > 0
return discovered_devices
@singleton.singleton("switcher_breeze_remote_manager")

13
tests/components/switcher_kis/__init__.py
View File

@ -1,14 +1,23 @@
"""Test cases and object for the Switcher integration tests."""
from homeassistant.components.switcher_kis.const import DOMAIN
from homeassistant.const import CONF_TOKEN, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def init_integration(hass: HomeAssistant) -> MockConfigEntry:
async def init_integration(
hass: HomeAssistant, username: str | None = None, token: str | None = None
) -> MockConfigEntry:
"""Set up the Switcher integration in Home Assistant."""
entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
data = {}
if username is not None:
data[CONF_USERNAME] = username
if token is not None:
data[CONF_TOKEN] = token
entry = MockConfigEntry(domain=DOMAIN, data=data, unique_id=DOMAIN)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)

24
tests/components/switcher_kis/consts.py
View File

@ -6,6 +6,7 @@ from aioswitcher.device import (
ShutterDirection,
SwitcherPowerPlug,
SwitcherShutter,
SwitcherSingleShutterDualLight,
SwitcherThermostat,
SwitcherWaterHeater,
ThermostatFanLevel,
@ -19,14 +20,17 @@ DUMMY_DEVICE_ID1 = "a123bc"
DUMMY_DEVICE_ID2 = "cafe12"
DUMMY_DEVICE_ID3 = "bada77"
DUMMY_DEVICE_ID4 = "bbd164"
DUMMY_DEVICE_ID5 = "bcdb64"
DUMMY_DEVICE_KEY1 = "18"
DUMMY_DEVICE_KEY2 = "01"
DUMMY_DEVICE_KEY3 = "12"
DUMMY_DEVICE_KEY4 = "07"
DUMMY_DEVICE_KEY5 = "15"
DUMMY_DEVICE_NAME1 = "Plug 23BC"
DUMMY_DEVICE_NAME2 = "Heater FE12"
DUMMY_DEVICE_NAME3 = "Breeze AB39"
DUMMY_DEVICE_NAME4 = "Runner DD77"
DUMMY_DEVICE_NAME5 = "RunnerS11 6CF5"
DUMMY_DEVICE_PASSWORD = "12345678"
DUMMY_ELECTRIC_CURRENT1 = 0.5
DUMMY_ELECTRIC_CURRENT2 = 12.8
@ -34,14 +38,17 @@ DUMMY_IP_ADDRESS1 = "192.168.100.157"
DUMMY_IP_ADDRESS2 = "192.168.100.158"
DUMMY_IP_ADDRESS3 = "192.168.100.159"
DUMMY_IP_ADDRESS4 = "192.168.100.160"
DUMMY_IP_ADDRESS5 = "192.168.100.161"
DUMMY_MAC_ADDRESS1 = "A1:B2:C3:45:67:D8"
DUMMY_MAC_ADDRESS2 = "A1:B2:C3:45:67:D9"
DUMMY_MAC_ADDRESS3 = "A1:B2:C3:45:67:DA"
DUMMY_MAC_ADDRESS4 = "A1:B2:C3:45:67:DB"
DUMMY_MAC_ADDRESS5 = "A1:B2:C3:45:67:DC"
DUMMY_TOKEN_NEEDED1 = False
DUMMY_TOKEN_NEEDED2 = False
DUMMY_TOKEN_NEEDED3 = False
DUMMY_TOKEN_NEEDED4 = False
DUMMY_TOKEN_NEEDED5 = True
DUMMY_PHONE_ID = "1234"
DUMMY_POWER_CONSUMPTION1 = 100
DUMMY_POWER_CONSUMPTION2 = 2780
@ -55,6 +62,9 @@ DUMMY_SWING = ThermostatSwing.OFF
DUMMY_REMOTE_ID = "ELEC7001"
DUMMY_POSITION = 54
DUMMY_DIRECTION = ShutterDirection.SHUTTER_STOP
DUMMY_USERNAME = "email"
DUMMY_TOKEN = "zvVvd7JxtN7CgvkD1Psujw=="
DUMMY_LIGHTS = [DeviceState.ON, DeviceState.ON]
DUMMY_PLUG_DEVICE = SwitcherPowerPlug(
DeviceType.POWER_PLUG,
@ -97,6 +107,20 @@ DUMMY_SHUTTER_DEVICE = SwitcherShutter(
DUMMY_DIRECTION,
)
DUMMY_SINGLE_SHUTTER_DUAL_LIGHT_DEVICE = SwitcherSingleShutterDualLight(
DeviceType.RUNNER_S11,
DeviceState.ON,
DUMMY_DEVICE_ID5,
DUMMY_DEVICE_KEY5,
DUMMY_IP_ADDRESS5,
DUMMY_MAC_ADDRESS5,
DUMMY_DEVICE_NAME5,
DUMMY_TOKEN_NEEDED5,
DUMMY_POSITION,
DUMMY_DIRECTION,
DUMMY_LIGHTS,
)
DUMMY_THERMOSTAT_DEVICE = SwitcherThermostat(
DeviceType.BREEZE,
DeviceState.ON,

169
tests/components/switcher_kis/test_config_flow.py
View File

@ -6,10 +6,17 @@ import pytest
from homeassistant import config_entries
from homeassistant.components.switcher_kis.const import DOMAIN
from homeassistant.const import CONF_TOKEN, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .consts import DUMMY_PLUG_DEVICE, DUMMY_WATER_HEATER_DEVICE
from .consts import (
DUMMY_PLUG_DEVICE,
DUMMY_SINGLE_SHUTTER_DUAL_LIGHT_DEVICE,
DUMMY_TOKEN,
DUMMY_USERNAME,
DUMMY_WATER_HEATER_DEVICE,
)
from tests.common import MockConfigEntry
@ -43,13 +50,96 @@ async def test_user_setup(
assert mock_bridge.is_running is False
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "Switcher"
assert result2["result"].data == {}
assert result2["result"].data == {CONF_USERNAME: None, CONF_TOKEN: None}
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
"mock_bridge",
[
[
DUMMY_SINGLE_SHUTTER_DUAL_LIGHT_DEVICE,
]
],
indirect=True,
)
async def test_user_setup_found_token_device_valid_token(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_bridge
) -> None:
"""Test we can finish a config flow with token device found."""
with patch("homeassistant.components.switcher_kis.utils.DISCOVERY_TIME_SEC", 0):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm"
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert mock_bridge.is_running is False
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "credentials"
with patch(
"homeassistant.components.switcher_kis.config_flow.validate_token",
return_value=True,
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_USERNAME: DUMMY_USERNAME, CONF_TOKEN: DUMMY_TOKEN},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Switcher"
assert result3["result"].data == {
CONF_USERNAME: DUMMY_USERNAME,
CONF_TOKEN: DUMMY_TOKEN,
}
@pytest.mark.parametrize(
"mock_bridge",
[
[
DUMMY_SINGLE_SHUTTER_DUAL_LIGHT_DEVICE,
]
],
indirect=True,
)
async def test_user_setup_found_token_device_invalid_token(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_bridge
) -> None:
"""Test we can finish a config flow with token device found."""
with patch("homeassistant.components.switcher_kis.utils.DISCOVERY_TIME_SEC", 0):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm"
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "credentials"
with patch(
"homeassistant.components.switcher_kis.config_flow.validate_token",
return_value=False,
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{CONF_USERNAME: DUMMY_USERNAME, CONF_TOKEN: DUMMY_TOKEN},
)
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {"base": "invalid_auth"}
async def test_user_setup_abort_no_devices_found(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_bridge
) -> None:
@ -84,3 +174,78 @@ async def test_single_instance(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "single_instance_allowed"
@pytest.mark.parametrize(
("user_input"),
[
({CONF_USERNAME: DUMMY_USERNAME, CONF_TOKEN: DUMMY_TOKEN}),
],
)
async def test_reauth_successful(
hass: HomeAssistant,
user_input: dict[str, str],
) -> None:
"""Test starting a reauthentication flow."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_USERNAME: DUMMY_USERNAME, CONF_TOKEN: DUMMY_TOKEN},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=entry.data,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with patch(
"homeassistant.components.switcher_kis.config_flow.validate_token",
return_value=True,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=user_input,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
async def test_reauth_invalid_auth(hass: HomeAssistant) -> None:
"""Test reauthentication flow with invalid credentials."""
entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_USERNAME: DUMMY_USERNAME, CONF_TOKEN: DUMMY_TOKEN},
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": entry.entry_id,
},
data=entry.data,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with patch(
"homeassistant.components.switcher_kis.config_flow.validate_token",
return_value=False,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_USERNAME: "invalid_user", CONF_TOKEN: "invalid_token"},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}

122
tests/components/switcher_kis/test_cover.py
View File

@ -25,21 +25,39 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.util import slugify
from . import init_integration
from .consts import DUMMY_SHUTTER_DEVICE as DEVICE
from .consts import (
DUMMY_SHUTTER_DEVICE as DEVICE,
DUMMY_SINGLE_SHUTTER_DUAL_LIGHT_DEVICE as DEVICE2,
DUMMY_TOKEN as TOKEN,
DUMMY_USERNAME as USERNAME,
)
ENTITY_ID = f"{COVER_DOMAIN}.{slugify(DEVICE.name)}"
ENTITY_ID2 = f"{COVER_DOMAIN}.{slugify(DEVICE2.name)}"
@pytest.mark.parametrize("mock_bridge", [[DEVICE]], indirect=True)
@pytest.mark.parametrize(
("device", "entity_id"),
[
(DEVICE, ENTITY_ID),
(DEVICE2, ENTITY_ID2),
],
)
@pytest.mark.parametrize("mock_bridge", [[DEVICE, DEVICE2]], indirect=True)
async def test_cover(
hass: HomeAssistant, mock_bridge, mock_api, monkeypatch: pytest.MonkeyPatch
hass: HomeAssistant,
mock_bridge,
mock_api,
monkeypatch: pytest.MonkeyPatch,
device,
entity_id,
) -> None:
"""Test cover services."""
await init_integration(hass)
await init_integration(hass, USERNAME, TOKEN)
assert mock_bridge
# Test initial state - open
state = hass.states.get(ENTITY_ID)
state = hass.states.get(entity_id)
assert state.state == STATE_OPEN
# Test set position
@ -49,17 +67,17 @@ async def test_cover(
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_POSITION: 77},
{ATTR_ENTITY_ID: entity_id, ATTR_POSITION: 77},
blocking=True,
)
monkeypatch.setattr(DEVICE, "position", 77)
mock_bridge.mock_callbacks([DEVICE])
monkeypatch.setattr(device, "position", 77)
mock_bridge.mock_callbacks([device])
await hass.async_block_till_done()
assert mock_api.call_count == 2
mock_control_device.assert_called_once_with(77)
state = hass.states.get(ENTITY_ID)
mock_control_device.assert_called_once_with(77, 0)
state = hass.states.get(entity_id)
assert state.state == STATE_OPEN
assert state.attributes[ATTR_CURRENT_POSITION] == 77
@ -70,17 +88,17 @@ async def test_cover(
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
monkeypatch.setattr(DEVICE, "direction", ShutterDirection.SHUTTER_UP)
mock_bridge.mock_callbacks([DEVICE])
monkeypatch.setattr(device, "direction", ShutterDirection.SHUTTER_UP)
mock_bridge.mock_callbacks([device])
await hass.async_block_till_done()
assert mock_api.call_count == 4
mock_control_device.assert_called_once_with(100)
state = hass.states.get(ENTITY_ID)
mock_control_device.assert_called_once_with(100, 0)
state = hass.states.get(entity_id)
assert state.state == STATE_OPENING
# Test close
@ -90,17 +108,17 @@ async def test_cover(
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
monkeypatch.setattr(DEVICE, "direction", ShutterDirection.SHUTTER_DOWN)
mock_bridge.mock_callbacks([DEVICE])
monkeypatch.setattr(device, "direction", ShutterDirection.SHUTTER_DOWN)
mock_bridge.mock_callbacks([device])
await hass.async_block_till_done()
assert mock_api.call_count == 6
mock_control_device.assert_called_once_with(0)
state = hass.states.get(ENTITY_ID)
mock_control_device.assert_called_once_with(0, 0)
state = hass.states.get(entity_id)
assert state.state == STATE_CLOSING
# Test stop
@ -110,37 +128,50 @@ async def test_cover(
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: ENTITY_ID},
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
monkeypatch.setattr(DEVICE, "direction", ShutterDirection.SHUTTER_STOP)
mock_bridge.mock_callbacks([DEVICE])
monkeypatch.setattr(device, "direction", ShutterDirection.SHUTTER_STOP)
mock_bridge.mock_callbacks([device])
await hass.async_block_till_done()
assert mock_api.call_count == 8
mock_control_device.assert_called_once()
state = hass.states.get(ENTITY_ID)
mock_control_device.assert_called_once_with(0)
state = hass.states.get(entity_id)
assert state.state == STATE_OPEN
# Test closed on position == 0
monkeypatch.setattr(DEVICE, "position", 0)
mock_bridge.mock_callbacks([DEVICE])
monkeypatch.setattr(device, "position", 0)
mock_bridge.mock_callbacks([device])
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
state = hass.states.get(entity_id)
assert state.state == STATE_CLOSED
assert state.attributes[ATTR_CURRENT_POSITION] == 0
@pytest.mark.parametrize("mock_bridge", [[DEVICE]], indirect=True)
async def test_cover_control_fail(hass: HomeAssistant, mock_bridge, mock_api) -> None:
@pytest.mark.parametrize(
("device", "entity_id"),
[
(DEVICE, ENTITY_ID),
(DEVICE2, ENTITY_ID2),
],
)
@pytest.mark.parametrize("mock_bridge", [[DEVICE, DEVICE2]], indirect=True)
async def test_cover_control_fail(
hass: HomeAssistant,
mock_bridge,
mock_api,
device,
entity_id,
) -> None:
"""Test cover control fail."""
await init_integration(hass)
await init_integration(hass, USERNAME, TOKEN)
assert mock_bridge
# Test initial state - open
state = hass.states.get(ENTITY_ID)
state = hass.states.get(entity_id)
assert state.state == STATE_OPEN
# Test exception during set position
@ -152,20 +183,20 @@ async def test_cover_control_fail(hass: HomeAssistant, mock_bridge, mock_api) ->
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_POSITION: 44},
{ATTR_ENTITY_ID: entity_id, ATTR_POSITION: 44},
blocking=True,
)
assert mock_api.call_count == 2
mock_control_device.assert_called_once_with(44)
state = hass.states.get(ENTITY_ID)
mock_control_device.assert_called_once_with(44, 0)
state = hass.states.get(entity_id)
assert state.state == STATE_UNAVAILABLE
# Make device available again
mock_bridge.mock_callbacks([DEVICE])
mock_bridge.mock_callbacks([device])
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
state = hass.states.get(entity_id)
assert state.state == STATE_OPEN
# Test error response during set position
@ -177,11 +208,22 @@ async def test_cover_control_fail(hass: HomeAssistant, mock_bridge, mock_api) ->
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_POSITION: 27},
{ATTR_ENTITY_ID: entity_id, ATTR_POSITION: 27},
blocking=True,
)
assert mock_api.call_count == 4
mock_control_device.assert_called_once_with(27)
state = hass.states.get(ENTITY_ID)
mock_control_device.assert_called_once_with(27, 0)
state = hass.states.get(entity_id)
assert state.state == STATE_UNAVAILABLE
@pytest.mark.parametrize("mock_bridge", [[DEVICE2]], indirect=True)
async def test_cover2_no_token(
hass: HomeAssistant, mock_bridge, mock_api, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test single cover dual light without token services."""
await init_integration(hass)
assert mock_bridge
assert mock_api.call_count == 0