Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add KNX interface device with diagnostic entities (#89213)

pull/89951/head^2
Matthias Alphart 2023-03-19 02:13:52 -11:00 committed by GitHub
parent 0441a64c69
commit 557b9c7d51
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 347 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
homeassistant/components/knx/__init__.py
View File

@ -69,6 +69,7 @@ from .const import (
KNX_ADDRESS,
SUPPORTED_PLATFORMS,
)
from .device import KNXInterfaceDevice
from .expose import KNXExposeSensor, KNXExposeTime, create_knx_exposure
from .schema import (
BinarySensorSchema,
@ -254,13 +255,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
knx_module.exposures.append(
create_knx_exposure(hass, knx_module.xknx, expose_config)
)
# always forward sensor for system entities (telegram counter, etc.)
await hass.config_entries.async_forward_entry_setup(entry, Platform.SENSOR)
await hass.config_entries.async_forward_entry_setups(
entry,
[
platform
for platform in SUPPORTED_PLATFORMS
if platform in config and platform is not Platform.NOTIFY
if platform in config and platform not in (Platform.SENSOR, Platform.NOTIFY)
],
)
@ -366,10 +368,17 @@ class KNXModule:
self.service_exposures: dict[str, KNXExposeSensor | KNXExposeTime] = {}
self.entry = entry
self.init_xknx()
self.xknx = XKNX(
connection_config=self.connection_config(),
rate_limit=self.entry.data[CONF_KNX_RATE_LIMIT],
state_updater=self.entry.data[CONF_KNX_STATE_UPDATER],
)
self.xknx.connection_manager.register_connection_state_changed_cb(
self.connection_state_changed_cb
)
self.interface_device = KNXInterfaceDevice(
hass=hass, entry=entry, xknx=self.xknx
)
self._address_filter_transcoder: dict[AddressFilter, type[DPTBase]] = {}
self._group_address_transcoder: dict[DeviceGroupAddress, type[DPTBase]] = {}
@ -382,14 +391,6 @@ class KNXModule:
)
self.entry.async_on_unload(self.entry.add_update_listener(async_update_entry))
def init_xknx(self) -> None:
"""Initialize XKNX object."""
self.xknx = XKNX(
connection_config=self.connection_config(),
rate_limit=self.entry.data[CONF_KNX_RATE_LIMIT],
state_updater=self.entry.data[CONF_KNX_STATE_UPDATER],
)
async def start(self) -> None:
"""Start XKNX object. Connect to tunneling or Routing device."""
await self.xknx.start()

51
homeassistant/components/knx/device.py Normal file
View File

@ -0,0 +1,51 @@
"""Handle KNX Devices."""
from __future__ import annotations
from xknx import XKNX
from xknx.core import XknxConnectionState
from xknx.io.gateway_scanner import GatewayDescriptor
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from .const import DOMAIN
class KNXInterfaceDevice:
"""Class for KNX Interface Device handling."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry, xknx: XKNX) -> None:
"""Initialize interface device class."""
self.device_registry = dr.async_get(hass)
self.gateway_descriptor: GatewayDescriptor | None = None
self.xknx = xknx
_device_id = (DOMAIN, f"_{entry.entry_id}_interface")
self.device = self.device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
default_name="KNX Interface",
identifiers={_device_id},
)
self.device_info = DeviceInfo(identifiers={_device_id})
self.xknx.connection_manager.register_connection_state_changed_cb(
self.connection_state_changed_cb
)
async def update(self) -> None:
"""Update interface properties on new connection."""
self.gateway_descriptor = await self.xknx.knxip_interface.gateway_info()
self.device_registry.async_update_device(
device_id=self.device.id,
model=str(self.gateway_descriptor.name)
if self.gateway_descriptor
else None,
)
async def connection_state_changed_cb(self, state: XknxConnectionState) -> None:
"""Call invoked after a KNX connection state change was received."""
if state is XknxConnectionState.CONNECTED:
await self.update()

151
homeassistant/components/knx/sensor.py
View File

@ -1,9 +1,13 @@
"""Support for KNX/IP sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any
from xknx import XKNX
from xknx.core.connection_state import XknxConnectionState, XknxConnectionType
from xknx.devices import Sensor as XknxSensor
from homeassistant import config_entries
@ -11,12 +15,15 @@ from homeassistant.components.sensor import (
CONF_STATE_CLASS,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_ENTITY_CATEGORY,
CONF_NAME,
CONF_TYPE,
EntityCategory,
Platform,
)
from homeassistant.core import HomeAssistant
@ -24,10 +31,95 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, StateType
from homeassistant.util.enum import try_parse_enum
from . import KNXModule
from .const import ATTR_SOURCE, DATA_KNX_CONFIG, DOMAIN
from .knx_entity import KnxEntity
from .schema import SensorSchema
SCAN_INTERVAL = timedelta(seconds=10)
@dataclass
class KNXSystemEntityDescription(SensorEntityDescription):
"""Class describing KNX system sensor entities."""
always_available: bool = True
entity_category: EntityCategory = EntityCategory.DIAGNOSTIC
has_entity_name: bool = True
should_poll: bool = True
value_fn: Callable[[KNXModule], StateType | datetime] = lambda knx: None
SYSTEM_ENTITY_DESCRIPTIONS = (
KNXSystemEntityDescription(
key="individual_address",
name="Individual Address",
always_available=False,
icon="mdi:router-network",
should_poll=False,
value_fn=lambda knx: str(knx.xknx.current_address),
),
KNXSystemEntityDescription(
key="connected_since",
name="Connected since",
always_available=False,
device_class=SensorDeviceClass.TIMESTAMP,
should_poll=False,
value_fn=lambda knx: knx.xknx.connection_manager.connected_since,
),
KNXSystemEntityDescription(
key="connection_type",
name="Connection type",
always_available=False,
device_class=SensorDeviceClass.ENUM,
options=[opt.value for opt in XknxConnectionType],
should_poll=False,
value_fn=lambda knx: knx.xknx.connection_manager.connection_type.value, # type: ignore[no-any-return]
),
KNXSystemEntityDescription(
key="telegrams_incoming",
name="Telegrams incoming",
icon="mdi:upload-network",
entity_registry_enabled_default=False,
force_update=True,
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda knx: knx.xknx.connection_manager.cemi_count_incoming,
),
KNXSystemEntityDescription(
key="telegrams_incoming_error",
name="Telegrams incoming Error",
icon="mdi:help-network",
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda knx: knx.xknx.connection_manager.cemi_count_incoming_error,
),
KNXSystemEntityDescription(
key="telegrams_outgoing",
name="Telegrams outgoing",
icon="mdi:download-network",
entity_registry_enabled_default=False,
force_update=True,
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda knx: knx.xknx.connection_manager.cemi_count_outgoing,
),
KNXSystemEntityDescription(
key="telegrams_outgoing_error",
name="Telegrams outgoing Error",
icon="mdi:close-network",
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda knx: knx.xknx.connection_manager.cemi_count_outgoing_error,
),
KNXSystemEntityDescription(
key="telegram_count",
name="Telegrams",
icon="mdi:plus-network",
force_update=True,
state_class=SensorStateClass.TOTAL_INCREASING,
value_fn=lambda knx: knx.xknx.connection_manager.cemi_count_outgoing
+ knx.xknx.connection_manager.cemi_count_incoming
+ knx.xknx.connection_manager.cemi_count_incoming_error,
),
)
async def async_setup_entry(
hass: HomeAssistant,
@ -35,10 +127,18 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensor(s) for KNX platform."""
xknx: XKNX = hass.data[DOMAIN].xknx
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][Platform.SENSOR]
knx_module: KNXModule = hass.data[DOMAIN]
async_add_entities(KNXSensor(xknx, entity_config) for entity_config in config)
async_add_entities(
KNXSystemSensor(knx_module, description)
for description in SYSTEM_ENTITY_DESCRIPTIONS
)
config: list[ConfigType] = hass.data[DATA_KNX_CONFIG].get(Platform.SENSOR)
if config:
async_add_entities(
KNXSensor(knx_module.xknx, entity_config) for entity_config in config
)
def _create_sensor(xknx: XKNX, config: ConfigType) -> XknxSensor:
@ -87,3 +187,48 @@ class KNXSensor(KnxEntity, SensorEntity):
if self._device.last_telegram is not None:
attr[ATTR_SOURCE] = str(self._device.last_telegram.source_address)
return attr
class KNXSystemSensor(SensorEntity):
"""Representation of a KNX system sensor."""
def __init__(
self,
knx: KNXModule,
description: KNXSystemEntityDescription,
) -> None:
"""Initialize of a KNX system sensor."""
self.entity_description: KNXSystemEntityDescription = description
self.knx = knx
self._attr_device_info = knx.interface_device.device_info
self._attr_should_poll = description.should_poll
self._attr_unique_id = f"_{knx.entry.entry_id}_{description.key}"
@property
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.knx)
@property
def available(self) -> bool:
"""Return True if entity is available."""
if self.entity_description.always_available:
return True
return self.knx.xknx.connection_manager.state is XknxConnectionState.CONNECTED
async def after_update_callback(self, _: XknxConnectionState) -> None:
"""Call after device was updated."""
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Store register state change callback."""
self.knx.xknx.connection_manager.register_connection_state_changed_cb(
self.after_update_callback
)
async def async_will_remove_from_hass(self) -> None:
"""Disconnect device object when removed."""
self.knx.xknx.connection_manager.unregister_connection_state_changed_cb(
self.after_update_callback
)

5
tests/components/knx/conftest.py
View File

@ -6,7 +6,7 @@ from unittest.mock import DEFAULT, AsyncMock, Mock, patch
import pytest
from xknx import XKNX
from xknx.core import XknxConnectionState
from xknx.core import XknxConnectionState, XknxConnectionType
from xknx.dpt import DPTArray, DPTBinary
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
from xknx.telegram import Telegram, TelegramDirection
@ -67,7 +67,8 @@ class KNXTestKit:
# set XknxConnectionState.CONNECTED to avoid `unavailable` entities at startup
# and start StateUpdater. This would be awaited on normal startup too.
await self.xknx.connection_manager.connection_state_changed(
XknxConnectionState.CONNECTED
state=XknxConnectionState.CONNECTED,
connection_type=XknxConnectionType.TUNNEL_TCP,
)
def knx_ip_interface_mock():

22
tests/components/knx/test_binary_sensor.py
View File

@ -38,7 +38,6 @@ async def test_binary_sensor_entity_category(
]
}
)
assert len(hass.states.async_all()) == 1
await knx.assert_read("1/1/1")
await knx.receive_response("1/1/1", True)
@ -65,7 +64,6 @@ async def test_binary_sensor(hass: HomeAssistant, knx: KNXTestKit) -> None:
]
}
)
assert len(hass.states.async_all()) == 2
# StateUpdater initialize state
await knx.assert_read("1/1/1")
@ -103,8 +101,6 @@ async def test_binary_sensor_ignore_internal_state(
hass: HomeAssistant, knx: KNXTestKit
) -> None:
"""Test KNX binary_sensor with ignore_internal_state."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
BinarySensorSchema.PLATFORM: [
@ -122,39 +118,36 @@ async def test_binary_sensor_ignore_internal_state(
]
}
)
assert len(hass.states.async_all()) == 2
# binary_sensor defaults to STATE_OFF - state change form None
assert len(events) == 2
events = async_capture_events(hass, "state_changed")
# receive initial ON telegram
await knx.receive_write("1/1/1", True)
await knx.receive_write("2/2/2", True)
await hass.async_block_till_done()
assert len(events) == 4
assert len(events) == 2
# receive second ON telegram - ignore_internal_state shall force state_changed event
await knx.receive_write("1/1/1", True)
await knx.receive_write("2/2/2", True)
await hass.async_block_till_done()
assert len(events) == 5
assert len(events) == 3
# receive first OFF telegram
await knx.receive_write("1/1/1", False)
await knx.receive_write("2/2/2", False)
await hass.async_block_till_done()
assert len(events) == 7
assert len(events) == 5
# receive second OFF telegram - ignore_internal_state shall force state_changed event
await knx.receive_write("1/1/1", False)
await knx.receive_write("2/2/2", False)
await hass.async_block_till_done()
assert len(events) == 8
assert len(events) == 6
async def test_binary_sensor_counter(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test KNX binary_sensor with context timeout."""
async_fire_time_changed(hass, dt.utcnow())
events = async_capture_events(hass, "state_changed")
context_timeout = 1
await knx.setup_integration(
@ -169,9 +162,7 @@ async def test_binary_sensor_counter(hass: HomeAssistant, knx: KNXTestKit) -> No
]
}
)
assert len(hass.states.async_all()) == 1
assert len(events) == 1
events.pop()
events = async_capture_events(hass, "state_changed")
# receive initial ON telegram
await knx.receive_write("2/2/2", True)
@ -236,7 +227,6 @@ async def test_binary_sensor_reset(hass: HomeAssistant, knx: KNXTestKit) -> None
]
}
)
assert len(hass.states.async_all()) == 1
# receive ON telegram
await knx.receive_write("2/2/2", True)

5
tests/components/knx/test_button.py
View File

@ -18,7 +18,6 @@ from tests.common import async_capture_events, async_fire_time_changed
async def test_button_simple(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test KNX button with default payload."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
ButtonSchema.PLATFORM: {
@ -27,9 +26,7 @@ async def test_button_simple(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
assert len(events) == 1
events.pop()
events = async_capture_events(hass, "state_changed")
# press button
await hass.services.async_call(

20
tests/components/knx/test_climate.py
View File

@ -20,7 +20,6 @@ async def test_climate_basic_temperature_set(
hass: HomeAssistant, knx: KNXTestKit
) -> None:
"""Test KNX climate basic."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
ClimateSchema.PLATFORM: {
@ -31,9 +30,7 @@ async def test_climate_basic_temperature_set(
}
}
)
assert len(hass.states.async_all()) == 1
assert len(events) == 1
events.pop()
events = async_capture_events(hass, "state_changed")
# read temperature
await knx.assert_read("1/2/3")
@ -57,7 +54,6 @@ async def test_climate_basic_temperature_set(
async def test_climate_hvac_mode(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test KNX climate hvac mode."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
ClimateSchema.PLATFORM: {
@ -72,9 +68,7 @@ async def test_climate_hvac_mode(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
assert len(events) == 1
events.pop()
async_capture_events(hass, "state_changed")
await hass.async_block_till_done()
# read states state updater
@ -112,7 +106,6 @@ async def test_climate_preset_mode(
hass: HomeAssistant, knx: KNXTestKit, entity_registry: er.EntityRegistry
) -> None:
"""Test KNX climate preset mode."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
ClimateSchema.PLATFORM: {
@ -125,9 +118,7 @@ async def test_climate_preset_mode(
}
}
)
assert len(hass.states.async_all()) == 1
assert len(events) == 1
events.pop()
events = async_capture_events(hass, "state_changed")
await hass.async_block_till_done()
# read states state updater
@ -177,7 +168,6 @@ async def test_climate_preset_mode(
async def test_update_entity(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test update climate entity for KNX."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
ClimateSchema.PLATFORM: {
@ -192,9 +182,7 @@ async def test_update_entity(hass: HomeAssistant, knx: KNXTestKit) -> None:
)
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()
async_capture_events(hass, "state_changed")
await hass.async_block_till_done()
# read states state updater

5
tests/components/knx/test_cover.py
View File

@ -11,7 +11,6 @@ from tests.common import async_capture_events
async def test_cover_basic(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test KNX cover basic."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
CoverSchema.PLATFORM: {
@ -25,9 +24,7 @@ async def test_cover_basic(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
assert len(events) == 1
events.pop()
events = async_capture_events(hass, "state_changed")
# read position state address and angle state address
await knx.assert_read("1/0/2")

7
tests/components/knx/test_expose.py
View File

@ -28,7 +28,6 @@ async def test_binary_expose(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
},
)
assert not hass.states.async_all()
# Change state to on
hass.states.async_set(entity_id, "on", {})
@ -57,7 +56,6 @@ async def test_expose_attribute(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
},
)
assert not hass.states.async_all()
# Before init no response shall be sent
await knx.receive_read("1/1/8")
@ -105,7 +103,6 @@ async def test_expose_attribute_with_default(
}
},
)
assert not hass.states.async_all()
# Before init default value shall be sent as response
await knx.receive_read("1/1/8")
@ -152,7 +149,6 @@ async def test_expose_string(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
},
)
assert not hass.states.async_all()
# Before init default value shall be sent as response
await knx.receive_read("1/1/8")
@ -185,7 +181,6 @@ async def test_expose_cooldown(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
},
)
assert not hass.states.async_all()
# Change state to 1
hass.states.async_set(entity_id, "1", {})
await knx.assert_write("1/1/8", (1,))
@ -220,7 +215,6 @@ async def test_expose_conversion_exception(
}
},
)
assert not hass.states.async_all()
# Before init default value shall be sent as response
await knx.receive_read("1/1/8")
@ -253,7 +247,6 @@ async def test_expose_with_date(
}
}
)
assert not hass.states.async_all()
await knx.assert_write("1/1/8", (0x7A, 0x1, 0x7, 0xE9, 0xD, 0xE, 0x20, 0x80))

3
tests/components/knx/test_fan.py
View File

@ -17,7 +17,6 @@ async def test_fan_percent(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
# turn on fan with default speed (50%)
await hass.services.async_call(
@ -63,7 +62,6 @@ async def test_fan_step(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
# turn on fan with default speed (50% - step 2)
await hass.services.async_call(
@ -116,7 +114,6 @@ async def test_fan_oscillation(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
# turn on oscillation
await hass.services.async_call(

112
tests/components/knx/test_interface_device.py Normal file
View File

@ -0,0 +1,112 @@
"""Test KNX scene."""
from unittest.mock import patch
from xknx.core import XknxConnectionState, XknxConnectionType
from xknx.telegram import IndividualAddress
from homeassistant.components.knx.sensor import SCAN_INTERVAL
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt
from .conftest import KNXTestKit
from tests.common import async_capture_events, async_fire_time_changed
async def test_diagnostic_entities(
hass: HomeAssistant, knx: KNXTestKit, entity_registry: er.EntityRegistry
) -> None:
"""Test diagnostic entities."""
await knx.setup_integration({})
for entity_id in [
"sensor.knx_interface_individual_address",
"sensor.knx_interface_connected_since",
"sensor.knx_interface_connection_type",
"sensor.knx_interface_telegrams_incoming",
"sensor.knx_interface_telegrams_incoming_error",
"sensor.knx_interface_telegrams_outgoing",
"sensor.knx_interface_telegrams_outgoing_error",
"sensor.knx_interface_telegrams",
]:
entity = entity_registry.async_get(entity_id)
assert entity.entity_category is EntityCategory.DIAGNOSTIC
for entity_id in [
"sensor.knx_interface_telegrams_incoming",
"sensor.knx_interface_telegrams_outgoing",
]:
entity = entity_registry.async_get(entity_id)
assert entity.disabled is True
knx.xknx.connection_manager.cemi_count_incoming = 20
knx.xknx.connection_manager.cemi_count_incoming_error = 1
knx.xknx.connection_manager.cemi_count_outgoing = 10
knx.xknx.connection_manager.cemi_count_outgoing_error = 2
events = async_capture_events(hass, "state_changed")
async_fire_time_changed(hass, dt.utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done()
assert len(events) == 3 # 5 polled sensors - 2 disabled
events.clear()
for entity_id, test_state in [
("sensor.knx_interface_individual_address", "0.0.0"),
("sensor.knx_interface_connection_type", "Tunnel TCP"),
# skipping connected_since timestamp
("sensor.knx_interface_telegrams_incoming_error", "1"),
("sensor.knx_interface_telegrams_outgoing_error", "2"),
("sensor.knx_interface_telegrams", "31"),
]:
assert hass.states.get(entity_id).state == test_state
await knx.xknx.connection_manager.connection_state_changed(
state=XknxConnectionState.DISCONNECTED
)
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(events) == 4 # 3 not always_available + 3 force_update - 2 disabled
events.clear()
knx.xknx.current_address = IndividualAddress("1.1.1")
await knx.xknx.connection_manager.connection_state_changed(
state=XknxConnectionState.CONNECTED,
connection_type=XknxConnectionType.TUNNEL_UDP,
)
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(events) == 6 # all diagnostic sensors - counters are reset on connect
for entity_id, test_state in [
("sensor.knx_interface_individual_address", "1.1.1"),
("sensor.knx_interface_connection_type", "Tunnel UDP"),
# skipping connected_since timestamp
("sensor.knx_interface_telegrams_incoming_error", "0"),
("sensor.knx_interface_telegrams_outgoing_error", "0"),
("sensor.knx_interface_telegrams", "0"),
]:
assert hass.states.get(entity_id).state == test_state
async def test_removed_entity(
hass: HomeAssistant, knx: KNXTestKit, entity_registry: er.EntityRegistry
) -> None:
"""Test unregister callback when entity is removed."""
await knx.setup_integration({})
with patch.object(
knx.xknx.connection_manager, "unregister_connection_state_changed_cb"
) as unregister_mock:
entity_registry.async_update_entity(
"sensor.knx_interface_connected_since",
disabled_by=er.RegistryEntryDisabler.USER,
)
await hass.async_block_till_done()
unregister_mock.assert_called_once()

1
tests/components/knx/test_light.py
View File

@ -36,7 +36,6 @@ async def test_light_simple(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
knx.assert_state("light.test", STATE_OFF)
# turn on light

8
tests/components/knx/test_scene.py
View File

@ -9,7 +9,9 @@ from homeassistant.helpers import entity_registry as er
from .conftest import KNXTestKit
async def test_activate_knx_scene(hass: HomeAssistant, knx: KNXTestKit) -> None:
async def test_activate_knx_scene(
hass: HomeAssistant, knx: KNXTestKit, entity_registry: er.EntityRegistry
) -> None:
"""Test KNX scene."""
await knx.setup_integration(
{
@ -23,10 +25,8 @@ async def test_activate_knx_scene(hass: HomeAssistant, knx: KNXTestKit) -> None:
]
}
)
assert len(hass.states.async_all()) == 1
registry = er.async_get(hass)
entity = registry.async_get("scene.test")
entity = entity_registry.async_get("scene.test")
assert entity.entity_category is EntityCategory.DIAGNOSTIC
assert entity.unique_id == "1/1/1_24"

2
tests/components/knx/test_select.py
View File

@ -37,7 +37,6 @@ async def test_select_dpt_2_simple(hass: HomeAssistant, knx: KNXTestKit) -> None
}
}
)
assert len(hass.states.async_all()) == 1
state = hass.states.get("select.test")
assert state.state is STATE_UNKNOWN
@ -152,7 +151,6 @@ async def test_select_dpt_20_103_all_options(
}
}
)
assert len(hass.states.async_all()) == 1
state = hass.states.get("select.test")
assert state.state is STATE_UNKNOWN

14
tests/components/knx/test_sensor.py
View File

@ -21,7 +21,6 @@ async def test_sensor(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
state = hass.states.get("sensor.test")
assert state.state is STATE_UNKNOWN
@ -44,7 +43,6 @@ async def test_sensor(hass: HomeAssistant, knx: KNXTestKit) -> None:
async def test_always_callback(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test KNX sensor with always_callback."""
events = async_capture_events(hass, "state_changed")
await knx.setup_integration(
{
SensorSchema.PLATFORM: [
@ -64,32 +62,30 @@ async def test_always_callback(hass: HomeAssistant, knx: KNXTestKit) -> None:
]
}
)
assert len(hass.states.async_all()) == 2
# state changes form None to "unknown"
assert len(events) == 2
events = async_capture_events(hass, "state_changed")
# receive initial telegram
await knx.receive_write("1/1/1", (0x42,))
await knx.receive_write("2/2/2", (0x42,))
await hass.async_block_till_done()
assert len(events) == 4
assert len(events) == 2
# receive second telegram with identical payload
# always_callback shall force state_changed event
await knx.receive_write("1/1/1", (0x42,))
await knx.receive_write("2/2/2", (0x42,))
await hass.async_block_till_done()
assert len(events) == 5
assert len(events) == 3
# receive telegram with different payload
await knx.receive_write("1/1/1", (0xFA,))
await knx.receive_write("2/2/2", (0xFA,))
await hass.async_block_till_done()
assert len(events) == 7
assert len(events) == 5
# receive telegram with second payload again
# always_callback shall force state_changed event
await knx.receive_write("1/1/1", (0xFA,))
await knx.receive_write("2/2/2", (0xFA,))
await hass.async_block_till_done()
assert len(events) == 8
assert len(events) == 6

2
tests/components/knx/test_switch.py
View File

@ -23,7 +23,6 @@ async def test_switch_simple(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
# turn on switch
await hass.services.async_call(
@ -66,7 +65,6 @@ async def test_switch_state(hass: HomeAssistant, knx: KNXTestKit) -> None:
},
}
)
assert len(hass.states.async_all()) == 1
# StateUpdater initialize state
await knx.assert_read(_STATE_ADDRESS)

1
tests/components/knx/test_weather.py
View File

@ -35,7 +35,6 @@ async def test_weather(hass: HomeAssistant, knx: KNXTestKit) -> None:
}
}
)
assert len(hass.states.async_all()) == 1
state = hass.states.get("weather.test")
assert state.state is ATTR_CONDITION_EXCEPTIONAL