jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity
core/tests/components/knx/test_init.py
Matthias Alphart 8a08a87a25
KNX tests: Support loading different config store fixtures (#136961) 
* Support loading different config store fixtures

* Add config store test for binary sensor

* Update README.md

* remove unused fixture

* AddAdd config store test for switch
2025-02-06 23:24:09 +01:00

369 lines
13 KiB
Python
Raw Permalink Blame History

"""Test KNX init."""
from datetime import timedelta
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from xknx.io import (
DEFAULT_MCAST_GRP,
DEFAULT_MCAST_PORT,
ConnectionConfig,
ConnectionType,
SecureConfig,
)
from homeassistant.components.knx.config_flow import (
DEFAULT_ENTRY_DATA,
DEFAULT_ROUTING_IA,
)
from homeassistant.components.knx.const import (
CONF_KNX_AUTOMATIC,
CONF_KNX_CONNECTION_TYPE,
CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_INDIVIDUAL_ADDRESS,
CONF_KNX_KNXKEY_FILENAME,
CONF_KNX_KNXKEY_PASSWORD,
CONF_KNX_LOCAL_IP,
CONF_KNX_MCAST_GRP,
CONF_KNX_MCAST_PORT,
CONF_KNX_RATE_LIMIT,
CONF_KNX_ROUTE_BACK,
CONF_KNX_ROUTING,
CONF_KNX_ROUTING_BACKBONE_KEY,
CONF_KNX_ROUTING_SECURE,
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE,
CONF_KNX_SECURE_DEVICE_AUTHENTICATION,
CONF_KNX_SECURE_USER_ID,
CONF_KNX_SECURE_USER_PASSWORD,
CONF_KNX_STATE_UPDATER,
CONF_KNX_TUNNELING,
CONF_KNX_TUNNELING_TCP,
CONF_KNX_TUNNELING_TCP_SECURE,
DOMAIN as KNX_DOMAIN,
KNXConfigEntryData,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.core import HomeAssistant
from . import KnxEntityGenerator
from .conftest import KNXTestKit
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.parametrize(
("config_entry_data", "connection_config"),
[
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
},
ConnectionConfig(threaded=True),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_LOCAL_IP: "192.168.1.1",
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
},
ConnectionConfig(
connection_type=ConnectionType.ROUTING,
individual_address=DEFAULT_ROUTING_IA,
multicast_group=DEFAULT_MCAST_GRP,
multicast_port=DEFAULT_MCAST_PORT,
local_ip="192.168.1.1",
threaded=True,
),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
},
ConnectionConfig(
connection_type=ConnectionType.TUNNELING,
route_back=False,
gateway_ip="192.168.0.2",
gateway_port=3675,
local_ip="192.168.1.112",
auto_reconnect=True,
threaded=True,
),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
CONF_KNX_KNXKEY_FILENAME: "knx/keyring.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password",
},
ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP,
gateway_ip="192.168.0.2",
gateway_port=3675,
auto_reconnect=True,
secure_config=SecureConfig(
knxkeys_file_path="keyring.knxkeys", knxkeys_password="password"
),
threaded=True,
),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password",
},
ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP_SECURE,
gateway_ip="192.168.0.2",
gateway_port=3675,
secure_config=SecureConfig(
knxkeys_file_path="testcase.knxkeys", knxkeys_password="password"
),
auto_reconnect=True,
threaded=True,
),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
CONF_KNX_SECURE_USER_ID: 2,
CONF_KNX_SECURE_USER_PASSWORD: "password",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth",
},
ConnectionConfig(
connection_type=ConnectionType.TUNNELING_TCP_SECURE,
gateway_ip="192.168.0.2",
gateway_port=3675,
secure_config=SecureConfig(
device_authentication_password="device_auth",
user_password="password",
user_id=2,
),
auto_reconnect=True,
threaded=True,
),
),
(
{
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE,
CONF_KNX_LOCAL_IP: "192.168.1.1",
CONF_KNX_RATE_LIMIT: CONF_KNX_DEFAULT_RATE_LIMIT,
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_MCAST_PORT: DEFAULT_MCAST_PORT,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_INDIVIDUAL_ADDRESS: DEFAULT_ROUTING_IA,
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44",
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000,
},
ConnectionConfig(
connection_type=ConnectionType.ROUTING_SECURE,
individual_address=DEFAULT_ROUTING_IA,
multicast_group=DEFAULT_MCAST_GRP,
multicast_port=DEFAULT_MCAST_PORT,
secure_config=SecureConfig(
backbone_key="bbaacc44bbaacc44bbaacc44bbaacc44",
latency_ms=2000,
),
local_ip="192.168.1.1",
threaded=True,
),
),
],
)
async def test_init_connection_handling(
hass: HomeAssistant,
knx: KNXTestKit,
config_entry_data: KNXConfigEntryData,
connection_config: ConnectionConfig,
) -> None:
"""Test correctly generating connection config."""
config_entry = MockConfigEntry(
title="KNX",
domain=KNX_DOMAIN,
data=config_entry_data,
)
knx.mock_config_entry = config_entry
await knx.setup_integration()
assert hass.data.get(KNX_DOMAIN) is not None
original_connection_config = (
hass.data[KNX_DOMAIN].connection_config().__dict__.copy()
)
del original_connection_config["secure_config"]
connection_config_dict = connection_config.__dict__.copy()
del connection_config_dict["secure_config"]
assert original_connection_config == connection_config_dict
if connection_config.secure_config is not None:
assert (
hass.data[KNX_DOMAIN].connection_config().secure_config.knxkeys_password
== connection_config.secure_config.knxkeys_password
)
assert (
hass.data[KNX_DOMAIN].connection_config().secure_config.user_password
== connection_config.secure_config.user_password
)
assert (
hass.data[KNX_DOMAIN].connection_config().secure_config.user_id
== connection_config.secure_config.user_id
)
assert (
hass.data[KNX_DOMAIN]
.connection_config()
.secure_config.device_authentication_password
== connection_config.secure_config.device_authentication_password
)
if connection_config.secure_config.knxkeys_file_path is not None:
assert (
connection_config.secure_config.knxkeys_file_path
in hass.data[KNX_DOMAIN]
.connection_config()
.secure_config.knxkeys_file_path
)
async def _init_switch_and_wait_for_first_state_updater_run(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
freezer: FrozenDateTimeFactory,
config_entry_data: KNXConfigEntryData,
) -> None:
"""Return a config entry with default data."""
config_entry = MockConfigEntry(
title="KNX", domain=KNX_DOMAIN, data=config_entry_data
)
knx.mock_config_entry = config_entry
await knx.setup_integration()
await create_ui_entity(
platform=Platform.SWITCH,
knx_data={
"ga_switch": {"write": "1/1/1", "state": "2/2/2"},
"respond_to_read": True,
"sync_state": True, # True uses xknx default state updater
"invert": False,
},
)
# created entity sends read-request to KNX bus on connection
await knx.assert_read("2/2/2")
await knx.receive_response("2/2/2", True)
freezer.tick(timedelta(minutes=59))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
await knx.assert_no_telegram()
freezer.tick(timedelta(minutes=1)) # 60 minutes passed
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
async def test_default_state_updater_enabled(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test default state updater is applied to xknx device instances."""
config_entry = DEFAULT_ENTRY_DATA | KNXConfigEntryData(
connection_type=CONF_KNX_AUTOMATIC, # missing in default data
state_updater=True,
)
await _init_switch_and_wait_for_first_state_updater_run(
hass, knx, create_ui_entity, freezer, config_entry
)
await knx.assert_read("2/2/2")
await knx.receive_response("2/2/2", True)
async def test_default_state_updater_disabled(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test default state updater is applied to xknx device instances."""
config_entry = DEFAULT_ENTRY_DATA | KNXConfigEntryData(
connection_type=CONF_KNX_AUTOMATIC, # missing in default data
state_updater=False,
)
await _init_switch_and_wait_for_first_state_updater_run(
hass, knx, create_ui_entity, freezer, config_entry
)
await knx.assert_no_telegram()
async def test_async_remove_entry(
hass: HomeAssistant,
knx: KNXTestKit,
) -> None:
"""Test async_setup_entry (for coverage)."""
config_entry = MockConfigEntry(
title="KNX",
domain=KNX_DOMAIN,
data={
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
},
)
knx.mock_config_entry = config_entry
await knx.setup_integration()
with (
patch("pathlib.Path.unlink") as unlink_mock,
patch("pathlib.Path.rmdir") as rmdir_mock,
):
assert await hass.config_entries.async_remove(config_entry.entry_id)
assert unlink_mock.call_count == 4
rmdir_mock.assert_called_once()
assert hass.config_entries.async_entries() == []
assert config_entry.state is ConfigEntryState.NOT_LOADED
Reference in New Issue View Git Blame Copy Permalink