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

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
pull/135411/merge
Matthias Alphart 2025-02-06 23:24:09 +01:00 committed by GitHub
parent 609188bb33
commit 8a08a87a25
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 142 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tests/components/knx/README.md
View File

@ -3,17 +3,22 @@
A KNXTestKit instance can be requested from a fixture. It provides convenience methods A KNXTestKit instance can be requested from a fixture. It provides convenience methods
to test outgoing KNX telegrams and inject incoming telegrams. to test outgoing KNX telegrams and inject incoming telegrams.
To test something add a test function requesting the `hass` and `knx` fixture and To test something add a test function requesting the `hass` and `knx` fixture and
set up the KNX integration by passing a KNX config dict to `knx.setup_integration`. set up the KNX integration with `knx.setup_integration`.
You can pass a KNX YAML-config dict or a ConfigStore fixture filename to the setup method. The fixture should be placed in the `tests/components/knx/fixtures` directory.
```python ```python
async def test_something(hass, knx): async def test_some_yaml(hass: HomeAssistant, knx: KNXTestKit):
await knx.setup_integration({ await knx.setup_integration(
yaml_config={
"switch": { "switch": {
"name": "test_switch", "name": "test_switch",
"address": "1/2/3", "address": "1/2/3",
} }
} }
) )
async def test_some_config_store(hass: HomeAssistant, knx: KNXTestKit):
await knx.setup_integration(config_store_fixture="config_store_filename.json")
``` ```
## Asserting outgoing telegrams ## Asserting outgoing telegrams

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

@ -44,7 +44,6 @@ from tests.common import MockConfigEntry, load_json_object_fixture
from tests.typing import WebSocketGenerator from tests.typing import WebSocketGenerator
FIXTURE_PROJECT_DATA = load_json_object_fixture("project.json", KNX_DOMAIN) FIXTURE_PROJECT_DATA = load_json_object_fixture("project.json", KNX_DOMAIN)
FIXTURE_CONFIG_STORAGE_DATA = load_json_object_fixture("config_store.json", KNX_DOMAIN)
class KNXTestKit: class KNXTestKit:
@ -52,10 +51,16 @@ class KNXTestKit:
INDIVIDUAL_ADDRESS = "1.2.3" INDIVIDUAL_ADDRESS = "1.2.3"
def __init__(self, hass: HomeAssistant, mock_config_entry: MockConfigEntry) -> None: def __init__(
self,
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
hass_storage: dict[str, Any],
) -> None:
"""Init KNX test helper class.""" """Init KNX test helper class."""
self.hass: HomeAssistant = hass self.hass: HomeAssistant = hass
self.mock_config_entry: MockConfigEntry = mock_config_entry self.mock_config_entry: MockConfigEntry = mock_config_entry
self.hass_storage: dict[str, Any] = hass_storage
self.xknx: XKNX self.xknx: XKNX
# outgoing telegrams will be put in the List instead of sent to the interface # outgoing telegrams will be put in the List instead of sent to the interface
# telegrams to an InternalGroupAddress won't be queued here # telegrams to an InternalGroupAddress won't be queued here
@ -69,7 +74,10 @@ class KNXTestKit:
assert test_state.attributes.get(attribute) == value assert test_state.attributes.get(attribute) == value
async def setup_integration( async def setup_integration(
self, config: ConfigType, add_entry_to_hass: bool = True self,
yaml_config: ConfigType | None = None,
config_store_fixture: str | None = None,
add_entry_to_hass: bool = True,
) -> None: ) -> None:
"""Create the KNX integration.""" """Create the KNX integration."""
@ -101,15 +109,21 @@ class KNXTestKit:
self.xknx = args[0] self.xknx = args[0]
return DEFAULT return DEFAULT
if config_store_fixture:
self.hass_storage[KNX_CONFIG_STORAGE_KEY] = load_json_object_fixture(
config_store_fixture, KNX_DOMAIN
)
if add_entry_to_hass: if add_entry_to_hass:
self.mock_config_entry.add_to_hass(self.hass) self.mock_config_entry.add_to_hass(self.hass)
knx_config = {KNX_DOMAIN: yaml_config or {}}
with patch( with patch(
"xknx.xknx.knx_interface_factory", "xknx.xknx.knx_interface_factory",
return_value=knx_ip_interface_mock(), return_value=knx_ip_interface_mock(),
side_effect=fish_xknx, side_effect=fish_xknx,
): ):
await async_setup_component(self.hass, KNX_DOMAIN, {KNX_DOMAIN: config}) await async_setup_component(self.hass, KNX_DOMAIN, knx_config)
await self.hass.async_block_till_done() await self.hass.async_block_till_done()
######################## ########################
@ -306,9 +320,13 @@ def mock_config_entry() -> MockConfigEntry:
@pytest.fixture @pytest.fixture
async def knx(hass: HomeAssistant, mock_config_entry: MockConfigEntry): async def knx(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
hass_storage: dict[str, Any],
):
"""Create a KNX TestKit instance.""" """Create a KNX TestKit instance."""
knx_test_kit = KNXTestKit(hass, mock_config_entry) knx_test_kit = KNXTestKit(hass, mock_config_entry, hass_storage)
yield knx_test_kit yield knx_test_kit
await knx_test_kit.assert_no_telegram() await knx_test_kit.assert_no_telegram()
@ -322,12 +340,6 @@ def load_knxproj(hass_storage: dict[str, Any]) -> None:
} }
@pytest.fixture
def load_config_store(hass_storage: dict[str, Any]) -> None:
"""Mock KNX config store data."""
hass_storage[KNX_CONFIG_STORAGE_KEY] = FIXTURE_CONFIG_STORAGE_DATA
@pytest.fixture @pytest.fixture
async def create_ui_entity( async def create_ui_entity(
hass: HomeAssistant, hass: HomeAssistant,

27
tests/components/knx/fixtures/config_store_binarysensor.json Normal file
View File

@ -0,0 +1,27 @@
{
"version": 1,
"minor_version": 1,
"key": "knx/config_store.json",
"data": {
"entities": {
"light": {},
"binary_sensor": {
"knx_es_01JJP1XDQRXB0W6YYGXW6Y1X10": {
"entity": {
"name": "test",
"device_info": null,
"entity_category": null
},
"knx": {
"ga_sensor": {
"state": "3/2/21",
"passive": []
},
"respond_to_read": false,
"sync_state": true
}
}
}
}
}
}

0
tests/components/knx/fixtures/config_store.json → tests/components/knx/fixtures/config_store_light_switch.json
View File

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

@ -329,7 +329,7 @@ async def test_binary_sensor_ui_create(
knx_data: dict[str, Any], knx_data: dict[str, Any],
) -> None: ) -> None:
"""Test creating a binary sensor.""" """Test creating a binary sensor."""
await knx.setup_integration({}) await knx.setup_integration()
await create_ui_entity( await create_ui_entity(
platform=Platform.BINARY_SENSOR, platform=Platform.BINARY_SENSOR,
entity_data={"name": "test"}, entity_data={"name": "test"},
@ -340,3 +340,10 @@ async def test_binary_sensor_ui_create(
await knx.receive_response("2/2/2", not knx_data.get("invert")) await knx.receive_response("2/2/2", not knx_data.get("invert"))
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state is STATE_ON assert state.state is STATE_ON
async def test_binary_sensor_ui_load(knx: KNXTestKit) -> None:
"""Test loading a binary sensor from storage."""
await knx.setup_integration(config_store_fixture="config_store_binarysensor.json")
await knx.assert_read("3/2/21", response=True, ignore_order=True)
knx.assert_state("binary_sensor.test", STATE_ON)

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

@ -1278,7 +1278,7 @@ async def test_options_flow_connection_type(
# usage of the already running XKNX instance for gateway scanner # usage of the already running XKNX instance for gateway scanner
gateway = _gateway_descriptor("192.168.0.1", 3675) gateway = _gateway_descriptor("192.168.0.1", 3675)
await knx.setup_integration({}) await knx.setup_integration()
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch( with patch(

18
tests/components/knx/test_config_store.py
View File

@ -25,7 +25,7 @@ async def test_create_entity(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test entity creation.""" """Test entity creation."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
test_name = "Test no device" test_name = "Test no device"
@ -69,7 +69,7 @@ async def test_create_entity_error(
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
) -> None: ) -> None:
"""Test unsuccessful entity creation.""" """Test unsuccessful entity creation."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
# create entity with invalid platform # create entity with invalid platform
@ -116,7 +116,7 @@ async def test_update_entity(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test entity update.""" """Test entity update."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
test_entity = await create_ui_entity( test_entity = await create_ui_entity(
@ -163,7 +163,7 @@ async def test_update_entity_error(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test entity update.""" """Test entity update."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
test_entity = await create_ui_entity( test_entity = await create_ui_entity(
@ -238,7 +238,7 @@ async def test_delete_entity(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test entity deletion.""" """Test entity deletion."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
test_entity = await create_ui_entity( test_entity = await create_ui_entity(
@ -270,7 +270,7 @@ async def test_delete_entity_error(
hass_storage: dict[str, Any], hass_storage: dict[str, Any],
) -> None: ) -> None:
"""Test unsuccessful entity deletion.""" """Test unsuccessful entity deletion."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
# delete unknown entity # delete unknown entity
@ -307,7 +307,7 @@ async def test_get_entity_config(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test entity config retrieval.""" """Test entity config retrieval."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
test_entity = await create_ui_entity( test_entity = await create_ui_entity(
@ -355,7 +355,7 @@ async def test_get_entity_config_error(
error_message_start: str, error_message_start: str,
) -> None: ) -> None:
"""Test entity config retrieval errors.""" """Test entity config retrieval errors."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json_auto_id( await client.send_json_auto_id(
@ -376,7 +376,7 @@ async def test_validate_entity(
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
) -> None: ) -> None:
"""Test entity validation.""" """Test entity validation."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json_auto_id( await client.send_json_auto_id(

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

@ -22,7 +22,7 @@ async def test_create_device(
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
) -> None: ) -> None:
"""Test device creation.""" """Test device creation."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json_auto_id( await client.send_json_auto_id(
@ -50,12 +50,11 @@ async def test_remove_device(
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
load_config_store: None,
hass_storage: dict[str, Any], hass_storage: dict[str, Any],
) -> None: ) -> None:
"""Test device removal.""" """Test device removal."""
assert await async_setup_component(hass, "config", {}) assert await async_setup_component(hass, "config", {})
await knx.setup_integration({}) await knx.setup_integration(config_store_fixture="config_store_light_switch.json")
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await knx.assert_read("1/0/21", response=True, ignore_order=True) # test light await knx.assert_read("1/0/21", response=True, ignore_order=True) # test light

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

@ -28,7 +28,7 @@ async def test_if_fires_on_telegram(
knx: KNXTestKit, knx: KNXTestKit,
) -> None: ) -> None:
"""Test telegram device triggers firing.""" """Test telegram device triggers firing."""
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )
@ -124,7 +124,7 @@ async def test_default_if_fires_on_telegram(
# by default (without a user changing any) extra_fields are not added to the trigger and # by default (without a user changing any) extra_fields are not added to the trigger and
# pre 2024.2 device triggers did only support "destination" field so they didn't have # pre 2024.2 device triggers did only support "destination" field so they didn't have
# "group_value_write", "group_value_response", "group_value_read", "incoming", "outgoing" # "group_value_write", "group_value_response", "group_value_read", "incoming", "outgoing"
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )
@ -206,7 +206,7 @@ async def test_remove_device_trigger(
) -> None: ) -> None:
"""Test for removed callback when device trigger not used.""" """Test for removed callback when device trigger not used."""
automation_name = "telegram_trigger_automation" automation_name = "telegram_trigger_automation"
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )
@ -256,7 +256,7 @@ async def test_get_triggers(
knx: KNXTestKit, knx: KNXTestKit,
) -> None: ) -> None:
"""Test we get the expected device triggers from knx.""" """Test we get the expected device triggers from knx."""
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )
@ -279,7 +279,7 @@ async def test_get_trigger_capabilities(
knx: KNXTestKit, knx: KNXTestKit,
) -> None: ) -> None:
"""Test we get the expected capabilities telegram device trigger.""" """Test we get the expected capabilities telegram device trigger."""
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )
@ -361,7 +361,7 @@ async def test_invalid_device_trigger(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test invalid telegram device trigger configuration.""" """Test invalid telegram device trigger configuration."""
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )
@ -404,7 +404,7 @@ async def test_invalid_trigger_configuration(
knx: KNXTestKit, knx: KNXTestKit,
) -> None: ) -> None:
"""Test invalid telegram device trigger configuration at attach_trigger.""" """Test invalid telegram device trigger configuration at attach_trigger."""
await knx.setup_integration({}) await knx.setup_integration()
device_entry = device_registry.async_get_device( device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")}
) )

13
tests/components/knx/test_diagnostic.py
View File

@ -1,5 +1,7 @@
"""Tests for the diagnostics data provided by the KNX integration.""" """Tests for the diagnostics data provided by the KNX integration."""
from typing import Any
import pytest import pytest
from syrupy import SnapshotAssertion from syrupy import SnapshotAssertion
from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT
@ -40,7 +42,7 @@ async def test_diagnostics(
snapshot: SnapshotAssertion, snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test diagnostics.""" """Test diagnostics."""
await knx.setup_integration({}) await knx.setup_integration()
# Overwrite the version for this test since we don't want to change this with every library bump # Overwrite the version for this test since we don't want to change this with every library bump
knx.xknx.version = "0.0.0" knx.xknx.version = "0.0.0"
@ -60,7 +62,7 @@ async def test_diagnostic_config_error(
snapshot: SnapshotAssertion, snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test diagnostics.""" """Test diagnostics."""
await knx.setup_integration({}) await knx.setup_integration()
# Overwrite the version for this test since we don't want to change this with every library bump # Overwrite the version for this test since we don't want to change this with every library bump
knx.xknx.version = "0.0.0" knx.xknx.version = "0.0.0"
@ -76,6 +78,7 @@ async def test_diagnostic_config_error(
async def test_diagnostic_redact( async def test_diagnostic_redact(
hass: HomeAssistant, hass: HomeAssistant,
hass_client: ClientSessionGenerator, hass_client: ClientSessionGenerator,
hass_storage: dict[str, Any],
snapshot: SnapshotAssertion, snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test diagnostics redacting data.""" """Test diagnostics redacting data."""
@ -95,8 +98,8 @@ async def test_diagnostic_redact(
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44", CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44",
}, },
) )
knx: KNXTestKit = KNXTestKit(hass, mock_config_entry) knx: KNXTestKit = KNXTestKit(hass, mock_config_entry, hass_storage)
await knx.setup_integration({}) await knx.setup_integration()
# Overwrite the version for this test since we don't want to change this with every library bump # Overwrite the version for this test since we don't want to change this with every library bump
knx.xknx.version = "0.0.0" knx.xknx.version = "0.0.0"
@ -117,7 +120,7 @@ async def test_diagnostics_project(
snapshot: SnapshotAssertion, snapshot: SnapshotAssertion,
) -> None: ) -> None:
"""Test diagnostics.""" """Test diagnostics."""
await knx.setup_integration({}) await knx.setup_integration()
knx.xknx.version = "0.0.0" knx.xknx.version = "0.0.0"
# snapshot will contain project specific fields in `project_info` # snapshot will contain project specific fields in `project_info`
assert ( assert (

6
tests/components/knx/test_init.py
View File

@ -226,7 +226,7 @@ async def test_init_connection_handling(
data=config_entry_data, data=config_entry_data,
) )
knx.mock_config_entry = config_entry knx.mock_config_entry = config_entry
await knx.setup_integration({}) await knx.setup_integration()
assert hass.data.get(KNX_DOMAIN) is not None assert hass.data.get(KNX_DOMAIN) is not None
@ -280,7 +280,7 @@ async def _init_switch_and_wait_for_first_state_updater_run(
title="KNX", domain=KNX_DOMAIN, data=config_entry_data title="KNX", domain=KNX_DOMAIN, data=config_entry_data
) )
knx.mock_config_entry = config_entry knx.mock_config_entry = config_entry
await knx.setup_integration({}) await knx.setup_integration()
await create_ui_entity( await create_ui_entity(
platform=Platform.SWITCH, platform=Platform.SWITCH,
knx_data={ knx_data={
@ -354,7 +354,7 @@ async def test_async_remove_entry(
}, },
) )
knx.mock_config_entry = config_entry knx.mock_config_entry = config_entry
await knx.setup_integration({}) await knx.setup_integration()
with ( with (
patch("pathlib.Path.unlink") as unlink_mock, patch("pathlib.Path.unlink") as unlink_mock,

6
tests/components/knx/test_interface_device.py
View File

@ -25,7 +25,7 @@ async def test_diagnostic_entities(
freezer: FrozenDateTimeFactory, freezer: FrozenDateTimeFactory,
) -> None: ) -> None:
"""Test diagnostic entities.""" """Test diagnostic entities."""
await knx.setup_integration({}) await knx.setup_integration()
for entity_id in ( for entity_id in (
"sensor.knx_interface_individual_address", "sensor.knx_interface_individual_address",
@ -103,7 +103,7 @@ async def test_removed_entity(
with patch( with patch(
"xknx.core.connection_manager.ConnectionManager.unregister_connection_state_changed_cb" "xknx.core.connection_manager.ConnectionManager.unregister_connection_state_changed_cb"
) as unregister_mock: ) as unregister_mock:
await knx.setup_integration({}) await knx.setup_integration()
entity_registry.async_update_entity( entity_registry.async_update_entity(
"sensor.knx_interface_connection_established", "sensor.knx_interface_connection_established",
@ -120,7 +120,7 @@ async def test_remove_interface_device(
) -> None: ) -> None:
"""Test device removal.""" """Test device removal."""
assert await async_setup_component(hass, "config", {}) assert await async_setup_component(hass, "config", {})
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
knx_devices = device_registry.devices.get_devices_for_config_entry_id( knx_devices = device_registry.devices.get_devices_for_config_entry_id(
knx.mock_config_entry.entry_id knx.mock_config_entry.entry_id

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

@ -1176,7 +1176,7 @@ async def test_light_ui_create(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test creating a light.""" """Test creating a light."""
await knx.setup_integration({}) await knx.setup_integration()
await create_ui_entity( await create_ui_entity(
platform=Platform.LIGHT, platform=Platform.LIGHT,
entity_data={"name": "test"}, entity_data={"name": "test"},
@ -1213,7 +1213,7 @@ async def test_light_ui_color_temp(
raw_ct: tuple[int, ...], raw_ct: tuple[int, ...],
) -> None: ) -> None:
"""Test creating a color-temp light.""" """Test creating a color-temp light."""
await knx.setup_integration({}) await knx.setup_integration()
await create_ui_entity( await create_ui_entity(
platform=Platform.LIGHT, platform=Platform.LIGHT,
entity_data={"name": "test"}, entity_data={"name": "test"},
@ -1250,7 +1250,7 @@ async def test_light_ui_multi_mode(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test creating a light with multiple color modes.""" """Test creating a light with multiple color modes."""
await knx.setup_integration({}) await knx.setup_integration()
await create_ui_entity( await create_ui_entity(
platform=Platform.LIGHT, platform=Platform.LIGHT,
entity_data={"name": "test"}, entity_data={"name": "test"},
@ -1335,13 +1335,11 @@ async def test_light_ui_multi_mode(
async def test_light_ui_load( async def test_light_ui_load(
hass: HomeAssistant,
knx: KNXTestKit, knx: KNXTestKit,
load_config_store: None,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
) -> None: ) -> None:
"""Test loading a light from storage.""" """Test loading a light from storage."""
await knx.setup_integration({}) await knx.setup_integration(config_store_fixture="config_store_light_switch.json")
await knx.assert_read("1/0/21", response=True, ignore_order=True) await knx.assert_read("1/0/21", response=True, ignore_order=True)
# unrelated switch in config store # unrelated switch in config store

12
tests/components/knx/test_services.py
View File

@ -111,7 +111,7 @@ async def test_send(
expected_apci, expected_apci,
) -> None: ) -> None:
"""Test `knx.send` service.""" """Test `knx.send` service."""
await knx.setup_integration({}) await knx.setup_integration()
await hass.services.async_call( await hass.services.async_call(
"knx", "knx",
@ -127,7 +127,7 @@ async def test_send(
async def test_read(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_read(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test `knx.read` service.""" """Test `knx.read` service."""
await knx.setup_integration({}) await knx.setup_integration()
# send read telegram # send read telegram
await hass.services.async_call("knx", "read", {"address": "1/1/1"}, blocking=True) await hass.services.async_call("knx", "read", {"address": "1/1/1"}, blocking=True)
@ -150,7 +150,7 @@ async def test_event_register(hass: HomeAssistant, knx: KNXTestKit) -> None:
events = async_capture_events(hass, "knx_event") events = async_capture_events(hass, "knx_event")
test_address = "1/2/3" test_address = "1/2/3"
await knx.setup_integration({}) await knx.setup_integration()
# no event registered # no event registered
await knx.receive_write(test_address, True) await knx.receive_write(test_address, True)
@ -200,7 +200,7 @@ async def test_exposure_register(hass: HomeAssistant, knx: KNXTestKit) -> None:
test_entity = "fake.entity" test_entity = "fake.entity"
test_attribute = "fake_attribute" test_attribute = "fake_attribute"
await knx.setup_integration({}) await knx.setup_integration()
# no exposure registered # no exposure registered
hass.states.async_set(test_entity, STATE_ON, {}) hass.states.async_set(test_entity, STATE_ON, {})
@ -265,7 +265,7 @@ async def test_reload_service(
knx: KNXTestKit, knx: KNXTestKit,
) -> None: ) -> None:
"""Test reload service.""" """Test reload service."""
await knx.setup_integration({}) await knx.setup_integration()
with ( with (
patch( patch(
@ -285,7 +285,7 @@ async def test_reload_service(
async def test_service_setup_failed(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_service_setup_failed(hass: HomeAssistant, knx: KNXTestKit) -> None:
"""Test service setup failed.""" """Test service setup failed."""
await knx.setup_integration({}) await knx.setup_integration()
await hass.config_entries.async_unload(knx.mock_config_entry.entry_id) await hass.config_entries.async_unload(knx.mock_config_entry.entry_id)
with pytest.raises(HomeAssistantError) as exc_info: with pytest.raises(HomeAssistantError) as exc_info:

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

@ -155,7 +155,7 @@ async def test_switch_ui_create(
create_ui_entity: KnxEntityGenerator, create_ui_entity: KnxEntityGenerator,
) -> None: ) -> None:
"""Test creating a switch.""" """Test creating a switch."""
await knx.setup_integration({}) await knx.setup_integration()
await create_ui_entity( await create_ui_entity(
platform=Platform.SWITCH, platform=Platform.SWITCH,
entity_data={"name": "test"}, entity_data={"name": "test"},
@ -171,3 +171,16 @@ async def test_switch_ui_create(
await knx.receive_response("2/2/2", True) await knx.receive_response("2/2/2", True)
state = hass.states.get("switch.test") state = hass.states.get("switch.test")
assert state.state is STATE_ON assert state.state is STATE_ON
async def test_switch_ui_load(knx: KNXTestKit) -> None:
"""Test loading a switch from storage."""
await knx.setup_integration(config_store_fixture="config_store_light_switch.json")
await knx.assert_read("1/0/45", response=True, ignore_order=True)
# unrelated light in config store
await knx.assert_read("1/0/21", response=True, ignore_order=True)
knx.assert_state(
"switch.none_test", # has_entity_name with unregistered device -> none_test
STATE_ON,
)

6
tests/components/knx/test_telegrams.py
View File

@ -70,7 +70,7 @@ async def test_store_telegam_history(
hass_storage: dict[str, Any], hass_storage: dict[str, Any],
) -> None: ) -> None:
"""Test storing telegram history.""" """Test storing telegram history."""
await knx.setup_integration({}) await knx.setup_integration()
await knx.receive_write("1/3/4", True) await knx.receive_write("1/3/4", True)
await hass.services.async_call( await hass.services.async_call(
@ -94,7 +94,7 @@ async def test_load_telegam_history(
) -> None: ) -> None:
"""Test telegram history restoration.""" """Test telegram history restoration."""
hass_storage["knx/telegrams_history.json"] = {"version": 1, "data": MOCK_TELEGRAMS} hass_storage["knx/telegrams_history.json"] = {"version": 1, "data": MOCK_TELEGRAMS}
await knx.setup_integration({}) await knx.setup_integration()
loaded_telegrams = hass.data[KNX_MODULE_KEY].telegrams.recent_telegrams loaded_telegrams = hass.data[KNX_MODULE_KEY].telegrams.recent_telegrams
assert assert_telegram_history(loaded_telegrams) assert assert_telegram_history(loaded_telegrams)
# TelegramDict "payload" is a tuple, this shall be restored when loading from JSON # TelegramDict "payload" is a tuple, this shall be restored when loading from JSON
@ -113,7 +113,7 @@ async def test_remove_telegam_history(
knx.mock_config_entry, knx.mock_config_entry,
data=knx.mock_config_entry.data | {CONF_KNX_TELEGRAM_LOG_SIZE: 0}, data=knx.mock_config_entry.data | {CONF_KNX_TELEGRAM_LOG_SIZE: 0},
) )
await knx.setup_integration({}, add_entry_to_hass=False) await knx.setup_integration(add_entry_to_hass=False)
# Store.async_remove() is mocked by hass_storage - check that data was removed. # Store.async_remove() is mocked by hass_storage - check that data was removed.
assert "knx/telegrams_history.json" not in hass_storage assert "knx/telegrams_history.json" not in hass_storage
assert not hass.data[KNX_MODULE_KEY].telegrams.recent_telegrams assert not hass.data[KNX_MODULE_KEY].telegrams.recent_telegrams

10
tests/components/knx/test_trigger.py
View File

@ -18,7 +18,7 @@ async def test_telegram_trigger(
knx: KNXTestKit, knx: KNXTestKit,
) -> None: ) -> None:
"""Test telegram triggers firing.""" """Test telegram triggers firing."""
await knx.setup_integration({}) await knx.setup_integration()
# "id" field added to action to test if `trigger_data` passed correctly in `async_attach_trigger` # "id" field added to action to test if `trigger_data` passed correctly in `async_attach_trigger`
assert await async_setup_component( assert await async_setup_component(
@ -105,7 +105,7 @@ async def test_telegram_trigger_dpt_option(
expected_unit: str | None, expected_unit: str | None,
) -> None: ) -> None:
"""Test telegram trigger type option.""" """Test telegram trigger type option."""
await knx.setup_integration({}) await knx.setup_integration()
assert await async_setup_component( assert await async_setup_component(
hass, hass,
automation.DOMAIN, automation.DOMAIN,
@ -190,7 +190,7 @@ async def test_telegram_trigger_options(
direction_options: dict[str, bool], direction_options: dict[str, bool],
) -> None: ) -> None:
"""Test telegram trigger options.""" """Test telegram trigger options."""
await knx.setup_integration({}) await knx.setup_integration()
assert await async_setup_component( assert await async_setup_component(
hass, hass,
automation.DOMAIN, automation.DOMAIN,
@ -266,7 +266,7 @@ async def test_remove_telegram_trigger(
) -> None: ) -> None:
"""Test for removed callback when telegram trigger not used.""" """Test for removed callback when telegram trigger not used."""
automation_name = "telegram_trigger_automation" automation_name = "telegram_trigger_automation"
await knx.setup_integration({}) await knx.setup_integration()
assert await async_setup_component( assert await async_setup_component(
hass, hass,
@ -311,7 +311,7 @@ async def test_invalid_trigger(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test invalid telegram trigger configuration.""" """Test invalid telegram trigger configuration."""
await knx.setup_integration({}) await knx.setup_integration()
caplog.clear() caplog.clear()
with caplog.at_level(logging.ERROR): with caplog.at_level(logging.ERROR):
assert await async_setup_component( assert await async_setup_component(

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

@ -20,7 +20,7 @@ async def test_knx_info_command(
hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
) -> None: ) -> None:
"""Test knx/info command.""" """Test knx/info command."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json({"id": 6, "type": "knx/info"}) await client.send_json({"id": 6, "type": "knx/info"})
@ -39,7 +39,7 @@ async def test_knx_info_command_with_project(
load_knxproj: None, load_knxproj: None,
) -> None: ) -> None:
"""Test knx/info command with loaded project.""" """Test knx/info command with loaded project."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json({"id": 6, "type": "knx/info"}) await client.send_json({"id": 6, "type": "knx/info"})
@ -65,7 +65,7 @@ async def test_knx_project_file_process(
_password = "pw-test" _password = "pw-test"
_parse_result = FIXTURE_PROJECT_DATA _parse_result = FIXTURE_PROJECT_DATA
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
assert not hass.data[KNX_MODULE_KEY].project.loaded assert not hass.data[KNX_MODULE_KEY].project.loaded
@ -100,7 +100,7 @@ async def test_knx_project_file_process_error(
hass_ws_client: WebSocketGenerator, hass_ws_client: WebSocketGenerator,
) -> None: ) -> None:
"""Test knx/project_file_process exception handling.""" """Test knx/project_file_process exception handling."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
assert not hass.data[KNX_MODULE_KEY].project.loaded assert not hass.data[KNX_MODULE_KEY].project.loaded
@ -134,7 +134,7 @@ async def test_knx_project_file_remove(
hass_storage: dict[str, Any], hass_storage: dict[str, Any],
) -> None: ) -> None:
"""Test knx/project_file_remove command.""" """Test knx/project_file_remove command."""
await knx.setup_integration({}) await knx.setup_integration()
assert hass_storage[KNX_PROJECT_STORAGE_KEY] assert hass_storage[KNX_PROJECT_STORAGE_KEY]
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
assert hass.data[KNX_MODULE_KEY].project.loaded assert hass.data[KNX_MODULE_KEY].project.loaded
@ -154,7 +154,7 @@ async def test_knx_get_project(
load_knxproj: None, load_knxproj: None,
) -> None: ) -> None:
"""Test retrieval of kxnproject from store.""" """Test retrieval of kxnproject from store."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
assert hass.data[KNX_MODULE_KEY].project.loaded assert hass.data[KNX_MODULE_KEY].project.loaded
@ -169,7 +169,7 @@ async def test_knx_group_monitor_info_command(
hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
) -> None: ) -> None:
"""Test knx/group_monitor_info command.""" """Test knx/group_monitor_info command."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json({"id": 6, "type": "knx/group_monitor_info"}) await client.send_json({"id": 6, "type": "knx/group_monitor_info"})
@ -184,7 +184,7 @@ async def test_knx_group_telegrams_command(
hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
) -> None: ) -> None:
"""Test knx/group_telegrams command.""" """Test knx/group_telegrams command."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "knx/group_telegrams"}) await client.send_json_auto_id({"type": "knx/group_telegrams"})
@ -338,7 +338,7 @@ async def test_knx_subscribe_telegrams_command_project(
load_knxproj: None, load_knxproj: None,
) -> None: ) -> None:
"""Test knx/subscribe_telegrams command with project data.""" """Test knx/subscribe_telegrams command with project data."""
await knx.setup_integration({}) await knx.setup_integration()
client = await hass_ws_client(hass) client = await hass_ws_client(hass)
await client.send_json({"id": 6, "type": "knx/subscribe_telegrams"}) await client.send_json({"id": 6, "type": "knx/subscribe_telegrams"})
res = await client.receive_json() res = await client.receive_json()
@ -405,7 +405,7 @@ async def test_websocket_when_config_entry_unloaded(
endpoint: str, endpoint: str,
) -> None: ) -> None:
"""Test websocket connection when config entry is unloaded.""" """Test websocket connection when config entry is unloaded."""
await knx.setup_integration({}) await knx.setup_integration()
await hass.config_entries.async_unload(knx.mock_config_entry.entry_id) await hass.config_entries.async_unload(knx.mock_config_entry.entry_id)
client = await hass_ws_client(hass) client = await hass_ws_client(hass)