starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 17:31:15 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add switch platform to flow-it (#181386)

Browse Source
This commit is contained in:
Alberto Geniola
2026-09-06 11:39:57 +02:00
committed by GitHub
parent e9a65206d5
commit c023f8e7d5
5 changed files with 316 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/flow_it/__init__.py
+1
View File
@@ -15,6 +15,7 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [
Platform.FAN,
Platform.SWITCH,
]
homeassistant/components/flow_it/strings.json
+10
View File
@@ -36,6 +36,16 @@
}
}
},
"entity": {
"switch": {
"flow_in": {
"name": "Air intake"
},
"flow_out": {
"name": "Air exhaust"
}
}
},
"exceptions": {
"auth_failed": {
"message": "Authentication failed when communicating with Flow-it VMC"
homeassistant/components/flow_it/switch.py
+102
View File
@@ -0,0 +1,102 @@
"""Switch platform for Flow-it."""
from typing import Any, override
from flow_it_api.client import FlowItVMCMachine
from flow_it_api.exceptions import FlowItAuthError, FlowItCommandError, FlowItError
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import FlowItConfigEntry, FlowItCoordinator
from .entity import FlowItVmcEntity
SWITCHES: tuple[SwitchEntityDescription, ...] = (
SwitchEntityDescription(
key="flow_in",
translation_key="flow_in",
entity_category=EntityCategory.CONFIG,
),
SwitchEntityDescription(
key="flow_out",
translation_key="flow_out",
entity_category=EntityCategory.CONFIG,
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: FlowItConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Flow-it switches."""
data = config_entry.runtime_data
async_add_entities(
FlowItVmcFlowSwitch(data.coordinator, data.vmc, description)
for description in SWITCHES
)
class FlowItVmcFlowSwitch(FlowItVmcEntity, SwitchEntity):
"""Flow-it flow switch entity."""
entity_description: SwitchEntityDescription
def __init__(
self,
coordinator: FlowItCoordinator,
vmc: FlowItVMCMachine,
description: SwitchEntityDescription,
) -> None:
"""Initialize the switch."""
super().__init__(coordinator, vmc, description)
@override
@property
def is_on(self) -> bool | None:
"""Return true if the switch is on."""
mode = self.coordinator.data.state.data.mode
if self.entity_description.key == "flow_in":
return bool(mode.flowIn) # codespell:ignore flowin
return bool(mode.flowOut)
@override
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._async_set_flow(True)
@override
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self._async_set_flow(False)
async def _async_set_flow(self, state: bool) -> None:
"""Set the flow state."""
mode = self.coordinator.data.state.data.mode
speed = mode.speed
flow_in = (
state
if self.entity_description.key == "flow_in"
else mode.flowIn # codespell:ignore flowin
)
flow_out = state if self.entity_description.key == "flow_out" else mode.flowOut
try:
await self.vmc.send_command(speed, flow_in=flow_in, flow_out=flow_out)
except FlowItAuthError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="auth_failed",
) from err
except (FlowItCommandError, FlowItError) as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_failed",
) from err
await self.coordinator.async_refresh()
tests/components/flow_it/snapshots/test_switch.ambr
+101
View File
@@ -0,0 +1,101 @@
# serializer version: 1
# name: test_switch_setup[switch.001122334455_air_exhaust-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.001122334455_air_exhaust',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Air exhaust',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Air exhaust',
'platform': 'flow_it',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'flow_out',
'unique_id': '001122334455_flow_out',
'unit_of_measurement': None,
})
# ---
# name: test_switch_setup[switch.001122334455_air_exhaust-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: '001122334455 Air exhaust',
}),
'context': <ANY>,
'entity_id': 'switch.001122334455_air_exhaust',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_switch_setup[switch.001122334455_air_intake-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.001122334455_air_intake',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Air intake',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Air intake',
'platform': 'flow_it',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'flow_in',
'unique_id': '001122334455_flow_in',
'unit_of_measurement': None,
})
# ---
# name: test_switch_setup[switch.001122334455_air_intake-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: '001122334455 Air intake',
}),
'context': <ANY>,
'entity_id': 'switch.001122334455_air_intake',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/flow_it/test_switch.py
+102
View File
@@ -0,0 +1,102 @@
"""Test Flow-it switch platform."""
from unittest.mock import AsyncMock, patch
from flow_it_api.exceptions import FlowItAuthError, FlowItCommandError, FlowItError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
AIR_INTAKE_ENTITY_ID = "switch.001122334455_air_intake"
AIR_EXHAUST_ENTITY_ID = "switch.001122334455_air_exhaust"
async def test_switch_setup(
hass: HomeAssistant,
mock_flow_it: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test switch platform setup and entity registry."""
with patch("homeassistant.components.flow_it.PLATFORMS", [Platform.SWITCH]):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("entity_id", "service", "expected_flow_in", "expected_flow_out"),
[
(AIR_INTAKE_ENTITY_ID, SERVICE_TURN_ON, True, True),
(AIR_INTAKE_ENTITY_ID, SERVICE_TURN_OFF, False, True),
(AIR_EXHAUST_ENTITY_ID, SERVICE_TURN_ON, True, True),
(AIR_EXHAUST_ENTITY_ID, SERVICE_TURN_OFF, True, False),
],
)
async def test_switch_turn_on_off(
hass: HomeAssistant,
mock_flow_it: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_id: str,
service: str,
expected_flow_in: bool,
expected_flow_out: bool,
) -> None:
"""Test turning on and off the switches."""
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_flow_it.return_value.send_command.assert_awaited_once_with(
"2",
flow_in=expected_flow_in,
flow_out=expected_flow_out,
)
@pytest.mark.parametrize(
("exception", "expected_exception"),
[
(FlowItAuthError("Auth error"), ConfigEntryAuthFailed),
(FlowItCommandError("Command error"), HomeAssistantError),
(FlowItError("Generic error"), HomeAssistantError),
],
)
async def test_switch_exceptions(
hass: HomeAssistant,
mock_flow_it: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
expected_exception: type[Exception],
) -> None:
"""Test exception handling during switch commands."""
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_flow_it.return_value.send_command.side_effect = exception
with pytest.raises(expected_exception):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: AIR_INTAKE_ENTITY_ID},
blocking=True,
)