Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 12:59:34 +00:00
Code Activity

Switch LCN integration to local polling (#152601)

Browse Source
This commit is contained in:
Andre Lengwenus
2025-11-17 15:50:26 +01:00
committed by GitHub
parent 2fe20553b3
commit 10c12623bf
20 changed files with 226 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
homeassistant/components/lcn/binary_sensor.py
View File

@@ -1,6 +1,7 @@
"""Support for LCN binary sensors."""
from collections.abc import Iterable
from datetime import timedelta
from functools import partial
import pypck
@@ -19,6 +20,7 @@ from .entity import LcnEntity
from .helpers import InputType, LcnConfigEntry
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=1)
def add_lcn_entities(
@@ -69,21 +71,11 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
config[CONF_DOMAIN_DATA][CONF_SOURCE]
]
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(
self.bin_sensor_port
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(
self.bin_sensor_port
)
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_binary_sensors(
SCAN_INTERVAL.seconds
)
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""

28
homeassistant/components/lcn/climate.py
View File

@@ -1,6 +1,8 @@
"""Support for LCN climate control."""
import asyncio
from collections.abc import Iterable
from datetime import timedelta
from functools import partial
from typing import Any, cast
@@ -36,6 +38,7 @@ from .entity import LcnEntity
from .helpers import InputType, LcnConfigEntry
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=1)
def add_lcn_entities(
@@ -110,20 +113,6 @@ class LcnClimate(LcnEntity, ClimateEntity):
ClimateEntityFeature.TURN_OFF | ClimateEntityFeature.TURN_ON
)
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.variable)
await self.device_connection.activate_status_request_handler(self.setpoint)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.variable)
await self.device_connection.cancel_status_request_handler(self.setpoint)
@property
def temperature_unit(self) -> str:
"""Return the unit of measurement."""
@@ -192,6 +181,17 @@ class LcnClimate(LcnEntity, ClimateEntity):
self._target_temperature = temperature
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await asyncio.gather(
self.device_connection.request_status_variable(
self.variable, SCAN_INTERVAL.seconds
),
self.device_connection.request_status_variable(
self.setpoint, SCAN_INTERVAL.seconds
),
)
def input_received(self, input_obj: InputType) -> None:
"""Set temperature value when LCN input object is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusVar):

66
homeassistant/components/lcn/cover.py
View File

@@ -1,6 +1,8 @@
"""Support for LCN covers."""
import asyncio
from collections.abc import Iterable
from datetime import timedelta
from functools import partial
from typing import Any
@@ -27,6 +29,7 @@ from .entity import LcnEntity
from .helpers import InputType, LcnConfigEntry
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=1)
def add_lcn_entities(
@@ -73,7 +76,7 @@ async def async_setup_entry(
class LcnOutputsCover(LcnEntity, CoverEntity):
"""Representation of a LCN cover connected to output ports."""
_attr_is_closed = False
_attr_is_closed = True
_attr_is_closing = False
_attr_is_opening = False
_attr_assumed_state = True
@@ -93,28 +96,6 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
else:
self.reverse_time = None
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTUP"]
)
await self.device_connection.activate_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTDOWN"]
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTUP"]
)
await self.device_connection.cancel_status_request_handler(
pypck.lcn_defs.OutputPort["OUTPUTDOWN"]
)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
state = pypck.lcn_defs.MotorStateModifier.DOWN
@@ -147,6 +128,18 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
self._attr_is_opening = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
if not self.device_connection.is_group:
await asyncio.gather(
self.device_connection.request_status_output(
pypck.lcn_defs.OutputPort["OUTPUTUP"], SCAN_INTERVAL.seconds
),
self.device_connection.request_status_output(
pypck.lcn_defs.OutputPort["OUTPUTDOWN"], SCAN_INTERVAL.seconds
),
)
def input_received(self, input_obj: InputType) -> None:
"""Set cover states when LCN input object (command) is received."""
if (
@@ -175,7 +168,7 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
class LcnRelayCover(LcnEntity, CoverEntity):
"""Representation of a LCN cover connected to relays."""
_attr_is_closed = False
_attr_is_closed = True
_attr_is_closing = False
_attr_is_opening = False
_attr_assumed_state = True
@@ -206,20 +199,6 @@ class LcnRelayCover(LcnEntity, CoverEntity):
self._is_closing = False
self._is_opening = False
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(
self.motor, self.positioning_mode
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.motor)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
if not await self.device_connection.control_motor_relays(
@@ -274,6 +253,17 @@ class LcnRelayCover(LcnEntity, CoverEntity):
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
coros = [self.device_connection.request_status_relays(SCAN_INTERVAL.seconds)]
if self.positioning_mode == pypck.lcn_defs.MotorPositioningMode.BS4:
coros.append(
self.device_connection.request_status_motor_position(
self.motor, self.positioning_mode, SCAN_INTERVAL.seconds
)
)
await asyncio.gather(*coros)
def input_received(self, input_obj: InputType) -> None:
"""Set cover states when LCN input object (command) is received."""
if isinstance(input_obj, pypck.inputs.ModStatusRelays):

18
homeassistant/components/lcn/entity.py
View File

@@ -22,7 +22,6 @@ from .helpers import (
class LcnEntity(Entity):
"""Parent class for all entities associated with the LCN component."""
_attr_should_poll = False
_attr_has_entity_name = True
device_connection: DeviceConnectionType
@@ -57,15 +56,24 @@ class LcnEntity(Entity):
).lower(),
)
@property
def should_poll(self) -> bool:
"""Groups may not poll for a status."""
return not self.device_connection.is_group
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
self.device_connection = get_device_connection(
self.hass, self.config[CONF_ADDRESS], self.config_entry
)
if not self.device_connection.is_group:
self._unregister_for_inputs = self.device_connection.register_for_inputs(
self.input_received
)
if self.device_connection.is_group:
return
self._unregister_for_inputs = self.device_connection.register_for_inputs(
self.input_received
)
self.schedule_update_ha_state(force_refresh=True)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""

14
homeassistant/components/lcn/helpers.py
View File

@@ -251,13 +251,19 @@ async def async_update_device_config(
"""Fill missing values in device_config with infos from LCN bus."""
# fetch serial info if device is module
if not (is_group := device_config[CONF_ADDRESS][2]): # is module
await device_connection.serial_known
await device_connection.serials_known()
if device_config[CONF_HARDWARE_SERIAL] == -1:
device_config[CONF_HARDWARE_SERIAL] = device_connection.hardware_serial
device_config[CONF_HARDWARE_SERIAL] = (
device_connection.serials.hardware_serial
)
if device_config[CONF_SOFTWARE_SERIAL] == -1:
device_config[CONF_SOFTWARE_SERIAL] = device_connection.software_serial
device_config[CONF_SOFTWARE_SERIAL] = (
device_connection.serials.software_serial
)
if device_config[CONF_HARDWARE_TYPE] == -1:
device_config[CONF_HARDWARE_TYPE] = device_connection.hardware_type.value
device_config[CONF_HARDWARE_TYPE] = (
device_connection.serials.hardware_type.value
)
# fetch name if device is module
if device_config[CONF_NAME] != "":

36
homeassistant/components/lcn/light.py
View File

@@ -1,6 +1,7 @@
"""Support for LCN lights."""
from collections.abc import Iterable
from datetime import timedelta
from functools import partial
from typing import Any
@@ -33,6 +34,7 @@ from .helpers import InputType, LcnConfigEntry
BRIGHTNESS_SCALE = (1, 100)
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=1)
def add_lcn_entities(
@@ -100,18 +102,6 @@ class LcnOutputLight(LcnEntity, LightEntity):
self._attr_color_mode = ColorMode.ONOFF
self._attr_supported_color_modes = {self._attr_color_mode}
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
if ATTR_TRANSITION in kwargs:
@@ -157,6 +147,12 @@ class LcnOutputLight(LcnEntity, LightEntity):
self._attr_is_on = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_output(
self.output, SCAN_INTERVAL.seconds
)
def input_received(self, input_obj: InputType) -> None:
"""Set light state when LCN input object (command) is received."""
if (
@@ -184,18 +180,6 @@ class LcnRelayLight(LcnEntity, LightEntity):
self.output = pypck.lcn_defs.RelayPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
states = [pypck.lcn_defs.RelayStateModifier.NOCHANGE] * 8
@@ -214,6 +198,10 @@ class LcnRelayLight(LcnEntity, LightEntity):
self._attr_is_on = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_relays(SCAN_INTERVAL.seconds)
def input_received(self, input_obj: InputType) -> None:
"""Set light state when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusRelays):

4
homeassistant/components/lcn/manifest.json
View File

@@ -6,8 +6,8 @@
"config_flow": true,
"dependencies": ["http", "websocket_api"],
"documentation": "https://www.home-assistant.io/integrations/lcn",
"iot_class": "local_push",
"iot_class": "local_polling",
"loggers": ["pypck"],
"quality_scale": "bronze",
"requirements": ["pypck==0.8.12", "lcn-frontend==0.2.7"]
"requirements": ["pypck==0.9.2", "lcn-frontend==0.2.7"]
}

35
homeassistant/components/lcn/sensor.py
View File

@@ -1,6 +1,7 @@
"""Support for LCN sensors."""
from collections.abc import Iterable
from datetime import timedelta
from functools import partial
from itertools import chain
@@ -40,6 +41,8 @@ from .entity import LcnEntity
from .helpers import InputType, LcnConfigEntry
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=1)
DEVICE_CLASS_MAPPING = {
pypck.lcn_defs.VarUnit.CELSIUS: SensorDeviceClass.TEMPERATURE,
@@ -128,17 +131,11 @@ class LcnVariableSensor(LcnEntity, SensorEntity):
)
self._attr_device_class = DEVICE_CLASS_MAPPING.get(self.unit)
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.variable)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.variable)
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_variable(
self.variable, SCAN_INTERVAL.seconds
)
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""
@@ -170,17 +167,11 @@ class LcnLedLogicSensor(LcnEntity, SensorEntity):
config[CONF_DOMAIN_DATA][CONF_SOURCE]
]
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.source)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.source)
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_led_and_logic_ops(
SCAN_INTERVAL.seconds
)
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""

3
homeassistant/components/lcn/services.py
View File

@@ -380,9 +380,6 @@ class LockKeys(LcnServiceCall):
else:
await device_connection.lock_keys(table_id, states)
handler = device_connection.status_requests_handler
await handler.request_status_locked_keys_timeout()
class DynText(LcnServiceCall):
"""Send dynamic text to LCN-GTxD displays."""

74
homeassistant/components/lcn/switch.py
View File

@@ -1,6 +1,7 @@
"""Support for LCN switches."""
from collections.abc import Iterable
from datetime import timedelta
from functools import partial
from typing import Any
@@ -17,6 +18,7 @@ from .entity import LcnEntity
from .helpers import InputType, LcnConfigEntry
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(minutes=1)
def add_lcn_switch_entities(
@@ -77,18 +79,6 @@ class LcnOutputSwitch(LcnEntity, SwitchEntity):
self.output = pypck.lcn_defs.OutputPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
if not await self.device_connection.dim_output(self.output.value, 100, 0):
@@ -103,6 +93,12 @@ class LcnOutputSwitch(LcnEntity, SwitchEntity):
self._attr_is_on = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_output(
self.output, SCAN_INTERVAL.seconds
)
def input_received(self, input_obj: InputType) -> None:
"""Set switch state when LCN input object (command) is received."""
if (
@@ -126,18 +122,6 @@ class LcnRelaySwitch(LcnEntity, SwitchEntity):
self.output = pypck.lcn_defs.RelayPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
states = [pypck.lcn_defs.RelayStateModifier.NOCHANGE] * 8
@@ -156,6 +140,10 @@ class LcnRelaySwitch(LcnEntity, SwitchEntity):
self._attr_is_on = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_relays(SCAN_INTERVAL.seconds)
def input_received(self, input_obj: InputType) -> None:
"""Set switch state when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusRelays):
@@ -179,22 +167,6 @@ class LcnRegulatorLockSwitch(LcnEntity, SwitchEntity):
]
self.reg_id = pypck.lcn_defs.Var.to_set_point_id(self.setpoint_variable)
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(
self.setpoint_variable
)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(
self.setpoint_variable
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
if not await self.device_connection.lock_regulator(self.reg_id, True):
@@ -209,6 +181,12 @@ class LcnRegulatorLockSwitch(LcnEntity, SwitchEntity):
self._attr_is_on = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_variable(
self.setpoint_variable, SCAN_INTERVAL.seconds
)
def input_received(self, input_obj: InputType) -> None:
"""Set switch state when LCN input object (command) is received."""
if (
@@ -234,18 +212,6 @@ class LcnKeyLockSwitch(LcnEntity, SwitchEntity):
self.table_id = ord(self.key.name[0]) - 65
self.key_id = int(self.key.name[1]) - 1
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.key)
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.key)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
states = [pypck.lcn_defs.KeyLockStateModifier.NOCHANGE] * 8
@@ -268,6 +234,10 @@ class LcnKeyLockSwitch(LcnEntity, SwitchEntity):
self._attr_is_on = False
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the state of the entity."""
await self.device_connection.request_status_locked_keys(SCAN_INTERVAL.seconds)
def input_received(self, input_obj: InputType) -> None:
"""Set switch state when LCN input object (command) is received."""
if (

2
homeassistant/generated/integrations.json
View File

@@ -3418,7 +3418,7 @@
"name": "LCN",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
"iot_class": "local_polling"
},
"ld2410_ble": {
"name": "LD2410 BLE",

2
requirements_all.txt generated
View File

@@ -2269,7 +2269,7 @@ pypaperless==4.1.1
pypca==0.0.7
# homeassistant.components.lcn
pypck==0.8.12
pypck==0.9.2
# homeassistant.components.pglab
pypglab==0.0.5

2
requirements_test_all.txt generated
View File

@@ -1892,7 +1892,7 @@ pypalazzetti==0.1.20
pypaperless==4.1.1
# homeassistant.components.lcn
pypck==0.8.12
pypck==0.9.2
# homeassistant.components.pglab
pypglab==0.0.5

32
tests/components/lcn/conftest.py
View File

@@ -5,8 +5,8 @@ from typing import Any
from unittest.mock import AsyncMock, Mock, patch
import pypck
import pypck.module
from pypck.module import GroupConnection, ModuleConnection
from pypck import lcn_defs
from pypck.module import GroupConnection, ModuleConnection, Serials
import pytest
from homeassistant.components.lcn import PchkConnectionManager
@@ -25,16 +25,28 @@ LATEST_CONFIG_ENTRY_VERSION = (LcnFlowHandler.VERSION, LcnFlowHandler.MINOR_VERS
class MockModuleConnection(ModuleConnection):
"""Fake a LCN module connection."""
status_request_handler = AsyncMock()
activate_status_request_handler = AsyncMock()
cancel_status_request_handler = AsyncMock()
request_name = AsyncMock(return_value="TestModule")
request_serials = AsyncMock(
return_value=Serials(
hardware_serial=0x1A20A1234,
manu=0x01,
software_serial=0x190B11,
hardware_type=lcn_defs.HardwareType.UPP,
)
)
send_command = AsyncMock(return_value=True)
request_status_output = AsyncMock()
request_status_relays = AsyncMock()
request_status_motor_position = AsyncMock()
request_status_binary_sensors = AsyncMock()
request_status_variable = AsyncMock()
request_status_led_and_logic_ops = AsyncMock()
request_status_locked_keys = AsyncMock()
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Construct ModuleConnection instance."""
super().__init__(*args, **kwargs)
self.serials_request_handler.serial_known.set()
self._serials_known.set()
class MockGroupConnection(GroupConnection):
@@ -55,14 +67,10 @@ class MockPchkConnectionManager(PchkConnectionManager):
async def async_close(self) -> None:
"""Mock closing a connection to PCHK."""
def get_address_conn(self, addr, request_serials=False):
"""Get LCN address connection."""
return super().get_address_conn(addr, request_serials)
@patch.object(pypck.connection, "ModuleConnection", MockModuleConnection)
def get_module_conn(self, addr, request_serials=False):
def get_module_conn(self, addr):
"""Get LCN module connection."""
return super().get_module_conn(addr, request_serials)
return super().get_module_conn(addr)
@patch.object(pypck.connection, "GroupConnection", MockGroupConnection)
def get_group_conn(self, addr):

8
tests/components/lcn/snapshots/test_cover.ambr
View File

@@ -46,7 +46,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'open',
'state': 'closed',
})
# ---
# name: test_setup_lcn_cover[cover.testmodule_cover_relays-entry]
@@ -96,7 +96,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'open',
'state': 'closed',
})
# ---
# name: test_setup_lcn_cover[cover.testmodule_cover_relays_bs4-entry]
@@ -146,7 +146,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'open',
'state': 'closed',
})
# ---
# name: test_setup_lcn_cover[cover.testmodule_cover_relays_module-entry]
@@ -196,6 +196,6 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'open',
'state': 'closed',
})
# ---

18
tests/components/lcn/snapshots/test_init.ambr
View File

@@ -9,10 +9,10 @@
7,
False,
),
'hardware_serial': -1,
'hardware_type': -1,
'hardware_serial': 7013536308,
'hardware_type': 11,
'name': 'TestModule',
'software_serial': -1,
'software_serial': 1641233,
}),
]),
'dim_mode': 'STEPS200',
@@ -50,10 +50,10 @@
7,
False,
),
'hardware_serial': -1,
'hardware_type': -1,
'hardware_serial': 7013536308,
'hardware_type': 11,
'name': 'TestModule',
'software_serial': -1,
'software_serial': 1641233,
}),
]),
'dim_mode': 'STEPS200',
@@ -143,10 +143,10 @@
7,
False,
),
'hardware_serial': -1,
'hardware_type': -1,
'hardware_serial': 7013536308,
'hardware_type': 11,
'name': 'TestModule',
'software_serial': -1,
'software_serial': 1641233,
}),
]),
'dim_mode': 'STEPS200',

11
tests/components/lcn/test_climate.py
View File

@@ -52,8 +52,15 @@ async def test_set_hvac_mode_heat(hass: HomeAssistant, entry: MockConfigEntry) -
await init_integration(hass, entry)
with patch.object(MockModuleConnection, "lock_regulator") as lock_regulator:
state = hass.states.get("climate.testmodule_climate1")
state.state = HVACMode.OFF
await hass.services.async_call(
DOMAIN_CLIMATE,
SERVICE_SET_HVAC_MODE,
{
ATTR_ENTITY_ID: "climate.testmodule_climate1",
ATTR_HVAC_MODE: HVACMode.OFF,
},
blocking=True,
)
# command failed
lock_regulator.return_value = False

41
tests/components/lcn/test_cover.py
View File

@@ -63,7 +63,8 @@ async def test_outputs_open(hass: HomeAssistant, entry: MockConfigEntry) -> None
MockModuleConnection, "control_motor_outputs"
) as control_motor_outputs:
state = hass.states.get(COVER_OUTPUTS)
state.state = CoverState.CLOSED
assert state is not None
assert state.state == CoverState.CLOSED
# command failed
control_motor_outputs.return_value = False
@@ -110,8 +111,12 @@ async def test_outputs_close(hass: HomeAssistant, entry: MockConfigEntry) -> Non
with patch.object(
MockModuleConnection, "control_motor_outputs"
) as control_motor_outputs:
state = hass.states.get(COVER_OUTPUTS)
state.state = CoverState.OPEN
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
# command failed
control_motor_outputs.return_value = False
@@ -158,8 +163,12 @@ async def test_outputs_stop(hass: HomeAssistant, entry: MockConfigEntry) -> None
with patch.object(
MockModuleConnection, "control_motor_outputs"
) as control_motor_outputs:
state = hass.states.get(COVER_OUTPUTS)
state.state = CoverState.CLOSING
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: COVER_OUTPUTS},
blocking=True,
)
# command failed
control_motor_outputs.return_value = False
@@ -203,7 +212,8 @@ async def test_relays_open(hass: HomeAssistant, entry: MockConfigEntry) -> None:
MockModuleConnection, "control_motor_relays"
) as control_motor_relays:
state = hass.states.get(COVER_RELAYS)
state.state = CoverState.CLOSED
assert state is not None
assert state.state == CoverState.CLOSED
# command failed
control_motor_relays.return_value = False
@@ -250,8 +260,12 @@ async def test_relays_close(hass: HomeAssistant, entry: MockConfigEntry) -> None
with patch.object(
MockModuleConnection, "control_motor_relays"
) as control_motor_relays:
state = hass.states.get(COVER_RELAYS)
state.state = CoverState.OPEN
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
# command failed
control_motor_relays.return_value = False
@@ -298,8 +312,12 @@ async def test_relays_stop(hass: HomeAssistant, entry: MockConfigEntry) -> None:
with patch.object(
MockModuleConnection, "control_motor_relays"
) as control_motor_relays:
state = hass.states.get(COVER_RELAYS)
state.state = CoverState.CLOSING
await hass.services.async_call(
DOMAIN_COVER,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: COVER_RELAYS},
blocking=True,
)
# command failed
control_motor_relays.return_value = False
@@ -360,7 +378,8 @@ async def test_relays_set_position(
MockModuleConnection, "control_motor_relays_position"
) as control_motor_relays_position:
state = hass.states.get(entity_id)
state.state = CoverState.CLOSED
assert state is not None
assert state.state == CoverState.CLOSED
# command failed
control_motor_relays_position.return_value = False

8
tests/components/lcn/test_light.py
View File

@@ -209,8 +209,12 @@ async def test_relay_turn_off(hass: HomeAssistant, entry: MockConfigEntry) -> No
states = [RelayStateModifier.NOCHANGE] * 8
states[0] = RelayStateModifier.OFF
state = hass.states.get(LIGHT_RELAY1)
state.state = STATE_ON
await hass.services.async_call(
DOMAIN_LIGHT,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: LIGHT_RELAY1},
blocking=True,
)
# command failed
control_relays.return_value = False

32
tests/components/lcn/test_switch.py
View File

@@ -93,8 +93,12 @@ async def test_output_turn_off(hass: HomeAssistant, entry: MockConfigEntry) -> N
await init_integration(hass, entry)
with patch.object(MockModuleConnection, "dim_output") as dim_output:
state = hass.states.get(SWITCH_OUTPUT1)
state.state = STATE_ON
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: SWITCH_OUTPUT1},
blocking=True,
)
# command failed
dim_output.return_value = False
@@ -176,8 +180,12 @@ async def test_relay_turn_off(hass: HomeAssistant, entry: MockConfigEntry) -> No
states = [RelayStateModifier.NOCHANGE] * 8
states[0] = RelayStateModifier.OFF
state = hass.states.get(SWITCH_RELAY1)
state.state = STATE_ON
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: SWITCH_RELAY1},
blocking=True,
)
# command failed
control_relays.return_value = False
@@ -257,8 +265,12 @@ async def test_regulatorlock_turn_off(
await init_integration(hass, entry)
with patch.object(MockModuleConnection, "lock_regulator") as lock_regulator:
state = hass.states.get(SWITCH_REGULATOR1)
state.state = STATE_ON
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: SWITCH_REGULATOR1},
blocking=True,
)
# command failed
lock_regulator.return_value = False
@@ -340,8 +352,12 @@ async def test_keylock_turn_off(hass: HomeAssistant, entry: MockConfigEntry) ->
states = [KeyLockStateModifier.NOCHANGE] * 8
states[0] = KeyLockStateModifier.OFF
state = hass.states.get(SWITCH_KEYLOCKK1)
state.state = STATE_ON
await hass.services.async_call(
DOMAIN_SWITCH,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: SWITCH_KEYLOCKK1},
blocking=True,
)
# command failed
lock_keys.return_value = False