mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 09:17:53 +00:00
Full test coverage for KNX integration (#69697)
* Full test coverage for KNX integration * Allow for 0 values, since we all live at those kind of areas in the world * Remove unneeded method * Add missing test for climate mode * Cleanup test and remove fixture that was used only once
This commit is contained in:
Marvin Wichmann
GitHub
parent
d060d57a87
commit
b5b514b62f
@ -593,10 +593,6 @@ omit =
|
||||
homeassistant/components/keyboard_remote/*
|
||||
homeassistant/components/kira/*
|
||||
homeassistant/components/kiwi/lock.py
|
||||
homeassistant/components/knx/__init__.py
|
||||
homeassistant/components/knx/climate.py
|
||||
homeassistant/components/knx/cover.py
|
||||
homeassistant/components/knx/notify.py
|
||||
homeassistant/components/kodi/__init__.py
|
||||
homeassistant/components/kodi/browse_media.py
|
||||
homeassistant/components/kodi/const.py
|
||||
|
||||
@ -151,12 +151,6 @@ class KNXClimate(KnxEntity, ClimateEntity):
|
||||
)
|
||||
self.default_hvac_mode: str = config[ClimateSchema.CONF_DEFAULT_CONTROLLER_MODE]
|
||||
|
||||
async def async_update(self) -> None:
|
||||
"""Request a state update from KNX bus."""
|
||||
await self._device.sync()
|
||||
if self._device.mode is not None:
|
||||
await self._device.mode.sync()
|
||||
|
||||
@property
|
||||
def current_temperature(self) -> float | None:
|
||||
"""Return the current temperature."""
|
||||
@ -181,10 +175,10 @@ class KNXClimate(KnxEntity, ClimateEntity):
|
||||
|
||||
async def async_set_temperature(self, **kwargs: Any) -> None:
|
||||
"""Set new target temperature."""
|
||||
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
|
||||
return
|
||||
await self._device.set_target_temperature(temperature)
|
||||
self.async_write_ha_state()
|
||||
temperature = kwargs.get(ATTR_TEMPERATURE)
|
||||
if temperature is not None:
|
||||
await self._device.set_target_temperature(temperature)
|
||||
self.async_write_ha_state()
|
||||
|
||||
@property
|
||||
def hvac_mode(self) -> str:
|
||||
|
||||
@ -160,10 +160,10 @@ class KNXCover(KnxEntity, CoverEntity):
|
||||
@property
|
||||
def current_cover_tilt_position(self) -> int | None:
|
||||
"""Return current tilt position of cover."""
|
||||
if not self._device.supports_angle:
|
||||
return None
|
||||
ang = self._device.current_angle()
|
||||
return 100 - ang if ang is not None else None
|
||||
if self._device.supports_angle:
|
||||
ang = self._device.current_angle()
|
||||
return 100 - ang if ang is not None else None
|
||||
return None
|
||||
|
||||
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
|
||||
"""Move the cover tilt to a specific position."""
|
||||
|
||||
245
tests/components/knx/test_climate.py
Normal file
245
tests/components/knx/test_climate.py
Normal file
@ -0,0 +1,245 @@
|
||||
"""Test KNX climate."""
|
||||
from homeassistant.components.climate import HVAC_MODE_HEAT, HVAC_MODE_OFF
|
||||
from homeassistant.components.climate.const import PRESET_ECO, PRESET_SLEEP
|
||||
from homeassistant.components.knx.schema import ClimateSchema
|
||||
from homeassistant.const import CONF_NAME, STATE_IDLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .conftest import KNXTestKit
|
||||
|
||||
from tests.common import async_capture_events
|
||||
|
||||
|
||||
async def test_climate_basic_temperature_set(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX climate basic."""
|
||||
events = async_capture_events(hass, "state_changed")
|
||||
await knx.setup_integration(
|
||||
{
|
||||
ClimateSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
ClimateSchema.CONF_TEMPERATURE_ADDRESS: "1/2/3",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_ADDRESS: "1/2/4",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_STATE_ADDRESS: "1/2/5",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# read temperature
|
||||
await knx.assert_read("1/2/3")
|
||||
# read target temperature
|
||||
await knx.assert_read("1/2/5")
|
||||
|
||||
# set new temperature
|
||||
await hass.services.async_call(
|
||||
"climate",
|
||||
"set_temperature",
|
||||
{"entity_id": "climate.test", "temperature": 20},
|
||||
blocking=True,
|
||||
)
|
||||
await knx.assert_write("1/2/4", (7, 208))
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
|
||||
async def test_climate_hvac_mode(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX climate hvac mode."""
|
||||
events = async_capture_events(hass, "state_changed")
|
||||
await knx.setup_integration(
|
||||
{
|
||||
ClimateSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
ClimateSchema.CONF_TEMPERATURE_ADDRESS: "1/2/3",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_ADDRESS: "1/2/4",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_STATE_ADDRESS: "1/2/5",
|
||||
ClimateSchema.CONF_CONTROLLER_MODE_ADDRESS: "1/2/6",
|
||||
ClimateSchema.CONF_CONTROLLER_MODE_STATE_ADDRESS: "1/2/7",
|
||||
ClimateSchema.CONF_ON_OFF_ADDRESS: "1/2/8",
|
||||
ClimateSchema.CONF_OPERATION_MODES: ["Auto"],
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
await hass.async_block_till_done()
|
||||
# read states state updater
|
||||
await knx.assert_read("1/2/7")
|
||||
await knx.assert_read("1/2/3")
|
||||
# StateUpdater initialize state
|
||||
await knx.receive_response("1/2/7", True)
|
||||
await knx.receive_response("1/2/3", (0x21,))
|
||||
# StateUpdater semaphore allows 2 concurrent requests
|
||||
# read target temperature state
|
||||
await knx.assert_read("1/2/5")
|
||||
|
||||
# turn hvac off
|
||||
await hass.services.async_call(
|
||||
"climate",
|
||||
"set_hvac_mode",
|
||||
{"entity_id": "climate.test", "hvac_mode": HVAC_MODE_OFF},
|
||||
blocking=True,
|
||||
)
|
||||
await knx.assert_write("1/2/8", False)
|
||||
|
||||
# turn hvac on
|
||||
await hass.services.async_call(
|
||||
"climate",
|
||||
"set_hvac_mode",
|
||||
{"entity_id": "climate.test", "hvac_mode": HVAC_MODE_HEAT},
|
||||
blocking=True,
|
||||
)
|
||||
await knx.assert_write("1/2/8", True)
|
||||
await knx.assert_write("1/2/6", (0x01,))
|
||||
|
||||
|
||||
async def test_climate_preset_mode(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX climate preset mode."""
|
||||
events = async_capture_events(hass, "state_changed")
|
||||
await knx.setup_integration(
|
||||
{
|
||||
ClimateSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
ClimateSchema.CONF_TEMPERATURE_ADDRESS: "1/2/3",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_ADDRESS: "1/2/4",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_STATE_ADDRESS: "1/2/5",
|
||||
ClimateSchema.CONF_OPERATION_MODE_ADDRESS: "1/2/6",
|
||||
ClimateSchema.CONF_OPERATION_MODE_STATE_ADDRESS: "1/2/7",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
await hass.async_block_till_done()
|
||||
# read states state updater
|
||||
await knx.assert_read("1/2/7")
|
||||
await knx.assert_read("1/2/3")
|
||||
# StateUpdater initialize state
|
||||
await knx.receive_response("1/2/7", True)
|
||||
await knx.receive_response("1/2/3", (0x01,))
|
||||
# StateUpdater semaphore allows 2 concurrent requests
|
||||
# read target temperature state
|
||||
await knx.assert_read("1/2/5")
|
||||
|
||||
# set preset mode
|
||||
await hass.services.async_call(
|
||||
"climate",
|
||||
"set_preset_mode",
|
||||
{"entity_id": "climate.test", "preset_mode": PRESET_ECO},
|
||||
blocking=True,
|
||||
)
|
||||
await knx.assert_write("1/2/6", (0x04,))
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# set preset mode
|
||||
await hass.services.async_call(
|
||||
"climate",
|
||||
"set_preset_mode",
|
||||
{"entity_id": "climate.test", "preset_mode": PRESET_SLEEP},
|
||||
blocking=True,
|
||||
)
|
||||
await knx.assert_write("1/2/6", (0x03,))
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
assert len(knx.xknx.devices) == 2
|
||||
assert len(knx.xknx.devices[0].device_updated_cbs) == 2
|
||||
assert len(knx.xknx.devices[1].device_updated_cbs) == 2
|
||||
# test removing also removes hooks
|
||||
er = entity_registry.async_get(hass)
|
||||
er.async_remove("climate.test")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(knx.xknx.devices) == 2
|
||||
assert len(knx.xknx.devices[0].device_updated_cbs) == 1
|
||||
assert len(knx.xknx.devices[1].device_updated_cbs) == 1
|
||||
|
||||
|
||||
async def test_update_entity(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test update climate entity for KNX."""
|
||||
events = async_capture_events(hass, "state_changed")
|
||||
await knx.setup_integration(
|
||||
{
|
||||
ClimateSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
ClimateSchema.CONF_TEMPERATURE_ADDRESS: "1/2/3",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_ADDRESS: "1/2/4",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_STATE_ADDRESS: "1/2/5",
|
||||
ClimateSchema.CONF_OPERATION_MODE_ADDRESS: "1/2/6",
|
||||
ClimateSchema.CONF_OPERATION_MODE_STATE_ADDRESS: "1/2/7",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert await async_setup_component(hass, "homeassistant", {})
|
||||
await hass.async_block_till_done()
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
await hass.async_block_till_done()
|
||||
# read states state updater
|
||||
await knx.assert_read("1/2/7")
|
||||
await knx.assert_read("1/2/3")
|
||||
# StateUpdater initialize state
|
||||
await knx.receive_response("1/2/7", True)
|
||||
await knx.receive_response("1/2/3", (0x01,))
|
||||
# StateUpdater semaphore allows 2 concurrent requests
|
||||
await knx.assert_read("1/2/5")
|
||||
|
||||
# verify update entity retriggers group value reads to the bus
|
||||
await hass.services.async_call(
|
||||
"homeassistant",
|
||||
"update_entity",
|
||||
target={"entity_id": "climate.test"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
await knx.assert_read("1/2/3")
|
||||
await knx.assert_read("1/2/5")
|
||||
await knx.assert_read("1/2/7")
|
||||
|
||||
|
||||
async def test_command_value_idle_mode(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX climate command_value."""
|
||||
events = async_capture_events(hass, "state_changed")
|
||||
await knx.setup_integration(
|
||||
{
|
||||
ClimateSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
ClimateSchema.CONF_TEMPERATURE_ADDRESS: "1/2/3",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_ADDRESS: "1/2/4",
|
||||
ClimateSchema.CONF_TARGET_TEMPERATURE_STATE_ADDRESS: "1/2/5",
|
||||
ClimateSchema.CONF_COMMAND_VALUE_STATE_ADDRESS: "1/2/6",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
await hass.async_block_till_done()
|
||||
# read states state updater
|
||||
await knx.assert_read("1/2/3")
|
||||
await knx.assert_read("1/2/5")
|
||||
# StateUpdater initialize state
|
||||
await knx.receive_response("1/2/6", (0x32,))
|
||||
await knx.receive_response("1/2/3", (0x0C, 0x1A))
|
||||
|
||||
assert len(events) == 2
|
||||
events.pop()
|
||||
|
||||
knx.assert_state("climate.test", HVAC_MODE_HEAT, command_value=20)
|
||||
|
||||
await knx.receive_write("1/2/6", (0x00,))
|
||||
|
||||
knx.assert_state(
|
||||
"climate.test", HVAC_MODE_HEAT, command_value=0, hvac_action=STATE_IDLE
|
||||
)
|
||||
112
tests/components/knx/test_cover.py
Normal file
112
tests/components/knx/test_cover.py
Normal file
@ -0,0 +1,112 @@
|
||||
"""Test KNX cover."""
|
||||
from homeassistant.components.knx.schema import CoverSchema
|
||||
from homeassistant.const import CONF_NAME, STATE_CLOSING
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import KNXTestKit
|
||||
|
||||
from tests.common import async_capture_events
|
||||
|
||||
|
||||
async def test_cover_basic(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX cover basic."""
|
||||
events = async_capture_events(hass, "state_changed")
|
||||
await knx.setup_integration(
|
||||
{
|
||||
CoverSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
CoverSchema.CONF_MOVE_LONG_ADDRESS: "1/0/0",
|
||||
CoverSchema.CONF_MOVE_SHORT_ADDRESS: "1/0/1",
|
||||
CoverSchema.CONF_POSITION_STATE_ADDRESS: "1/0/2",
|
||||
CoverSchema.CONF_POSITION_ADDRESS: "1/0/3",
|
||||
CoverSchema.CONF_ANGLE_STATE_ADDRESS: "1/0/4",
|
||||
CoverSchema.CONF_ANGLE_ADDRESS: "1/0/5",
|
||||
}
|
||||
}
|
||||
)
|
||||
assert len(hass.states.async_all()) == 1
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# read position state address and angle state address
|
||||
await knx.assert_read("1/0/2")
|
||||
await knx.assert_read("1/0/4")
|
||||
|
||||
# open cover
|
||||
await hass.services.async_call(
|
||||
"cover", "open_cover", target={"entity_id": "cover.test"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/0/0", False)
|
||||
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# close cover
|
||||
await hass.services.async_call(
|
||||
"cover", "close_cover", target={"entity_id": "cover.test"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/0/0", True)
|
||||
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# stop cover
|
||||
await hass.services.async_call(
|
||||
"cover", "stop_cover", target={"entity_id": "cover.test"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/0/1", True)
|
||||
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# set cover position
|
||||
await hass.services.async_call(
|
||||
"cover",
|
||||
"set_cover_position",
|
||||
{"position": 25},
|
||||
target={"entity_id": "cover.test"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# in KNX this will result in a payload of 191, percent values are encoded from 0 to 255
|
||||
# We need to transpile the position by using 100 - position due to the way KNX actuators work
|
||||
await knx.assert_write("1/0/3", (0xBF,))
|
||||
|
||||
knx.assert_state(
|
||||
"cover.test",
|
||||
STATE_CLOSING,
|
||||
)
|
||||
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# set cover tilt position
|
||||
await hass.services.async_call(
|
||||
"cover",
|
||||
"set_cover_tilt_position",
|
||||
{"tilt_position": 25},
|
||||
target={"entity_id": "cover.test"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# in KNX this will result in a payload of 191, percent values are encoded from 0 to 255
|
||||
# We need to transpile the position by using 100 - position due to the way KNX actuators work
|
||||
await knx.assert_write("1/0/5", (0xBF,))
|
||||
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# close cover tilt
|
||||
await hass.services.async_call(
|
||||
"cover", "close_cover_tilt", target={"entity_id": "cover.test"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/0/1", True)
|
||||
|
||||
assert len(events) == 1
|
||||
events.pop()
|
||||
|
||||
# open cover tilt
|
||||
await hass.services.async_call(
|
||||
"cover", "open_cover_tilt", target={"entity_id": "cover.test"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/0/1", False)
|
||||
137
tests/components/knx/test_notify.py
Normal file
137
tests/components/knx/test_notify.py
Normal file
@ -0,0 +1,137 @@
|
||||
"""Test KNX notify."""
|
||||
|
||||
from homeassistant.components.knx.const import KNX_ADDRESS
|
||||
from homeassistant.components.knx.schema import NotifySchema
|
||||
from homeassistant.const import CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import KNXTestKit
|
||||
|
||||
|
||||
async def test_notify_simple(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX notify can send to one device."""
|
||||
await knx.setup_integration(
|
||||
{
|
||||
NotifySchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
KNX_ADDRESS: "1/0/0",
|
||||
}
|
||||
}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await hass.services.async_call(
|
||||
"notify", "notify", {"target": "test", "message": "I love KNX"}, blocking=True
|
||||
)
|
||||
|
||||
await knx.assert_write(
|
||||
"1/0/0",
|
||||
(
|
||||
0x49,
|
||||
0x20,
|
||||
0x6C,
|
||||
0x6F,
|
||||
0x76,
|
||||
0x65,
|
||||
0x20,
|
||||
0x4B,
|
||||
0x4E,
|
||||
0x58,
|
||||
0x0,
|
||||
0x0,
|
||||
0x0,
|
||||
0x0,
|
||||
),
|
||||
)
|
||||
|
||||
await hass.services.async_call(
|
||||
"notify",
|
||||
"notify",
|
||||
{
|
||||
"target": "test",
|
||||
"message": "I love KNX, but this text is too long for KNX, poor KNX",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
await knx.assert_write(
|
||||
"1/0/0",
|
||||
(
|
||||
0x49,
|
||||
0x20,
|
||||
0x6C,
|
||||
0x6F,
|
||||
0x76,
|
||||
0x65,
|
||||
0x20,
|
||||
0x4B,
|
||||
0x4E,
|
||||
0x58,
|
||||
0x2C,
|
||||
0x20,
|
||||
0x62,
|
||||
0x75,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def test_notify_multiple_sends_to_all(hass: HomeAssistant, knx: KNXTestKit):
|
||||
"""Test KNX notify can send to all devices."""
|
||||
await knx.setup_integration(
|
||||
{
|
||||
NotifySchema.PLATFORM: [
|
||||
{
|
||||
CONF_NAME: "test",
|
||||
KNX_ADDRESS: "1/0/0",
|
||||
},
|
||||
{
|
||||
CONF_NAME: "test2",
|
||||
KNX_ADDRESS: "1/0/1",
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await hass.services.async_call(
|
||||
"notify", "notify", {"message": "I love KNX"}, blocking=True
|
||||
)
|
||||
|
||||
await knx.assert_write(
|
||||
"1/0/0",
|
||||
(
|
||||
0x49,
|
||||
0x20,
|
||||
0x6C,
|
||||
0x6F,
|
||||
0x76,
|
||||
0x65,
|
||||
0x20,
|
||||
0x4B,
|
||||
0x4E,
|
||||
0x58,
|
||||
0x0,
|
||||
0x0,
|
||||
0x0,
|
||||
0x0,
|
||||
),
|
||||
)
|
||||
await knx.assert_write(
|
||||
"1/0/1",
|
||||
(
|
||||
0x49,
|
||||
0x20,
|
||||
0x6C,
|
||||
0x6F,
|
||||
0x76,
|
||||
0x65,
|
||||
0x20,
|
||||
0x4B,
|
||||
0x4E,
|
||||
0x58,
|
||||
0x0,
|
||||
0x0,
|
||||
0x0,
|
||||
0x0,
|
||||
),
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user