mirror of
https://github.com/home-assistant/core.git
synced 2026-08-15 01:32:55 +01:00
238 lines
8.0 KiB
Python
238 lines
8.0 KiB
Python
"""Test KNX devices."""
|
|
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
|
|
from homeassistant.components.knx.const import DOMAIN, KNX_ADDRESS
|
|
from homeassistant.components.knx.storage.config_store import (
|
|
STORAGE_KEY as KNX_CONFIG_STORAGE_KEY,
|
|
)
|
|
from homeassistant.const import SERVICE_RELOAD, Platform
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
|
from homeassistant.setup import async_setup_component
|
|
|
|
from .conftest import KNXTestKit, _patch_telegram_store
|
|
|
|
from tests.common import MockConfigEntry
|
|
from tests.typing import WebSocketGenerator
|
|
|
|
|
|
async def test_create_device(
|
|
hass: HomeAssistant,
|
|
knx: KNXTestKit,
|
|
device_registry: dr.DeviceRegistry,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test device creation."""
|
|
await knx.setup_integration()
|
|
client = await hass_ws_client(hass)
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "knx/create_device",
|
|
"name": "Test Device",
|
|
}
|
|
)
|
|
res = await client.receive_json()
|
|
assert res["success"], res
|
|
assert res["result"]["name"] == "Test Device"
|
|
assert res["result"]["manufacturer"] == "KNX"
|
|
assert res["result"]["identifiers"]
|
|
assert res["result"]["config_entries"][0] == knx.mock_config_entry.entry_id
|
|
|
|
device_identifier = res["result"]["identifiers"][0][1]
|
|
assert device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, device_identifier), knx.mock_config_entry.entry_id
|
|
)
|
|
device_id = res["result"]["id"]
|
|
assert device_registry.async_get(device_id)
|
|
|
|
|
|
async def test_remove_device(
|
|
hass: HomeAssistant,
|
|
knx: KNXTestKit,
|
|
device_registry: dr.DeviceRegistry,
|
|
entity_registry: er.EntityRegistry,
|
|
hass_ws_client: WebSocketGenerator,
|
|
hass_storage: dict[str, Any],
|
|
) -> None:
|
|
"""Test device removal."""
|
|
assert await async_setup_component(hass, "config", {})
|
|
await knx.setup_integration(config_store_fixture="config_store_light_switch.json")
|
|
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/45", response=True, ignore_order=True) # test switch
|
|
|
|
assert hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"].get("switch")
|
|
test_device = device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "knx_vdev_4c80a564f5fe5da701ed293966d6384d"),
|
|
knx.mock_config_entry.entry_id,
|
|
)
|
|
device_id = test_device.id
|
|
device_entities = entity_registry.entities.get_entries_for_device_id(device_id)
|
|
assert len(device_entities) == 1
|
|
|
|
response = await client.remove_device(device_id, knx.mock_config_entry.entry_id)
|
|
assert response["success"]
|
|
assert not device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "knx_vdev_4c80a564f5fe5da701ed293966d6384d"),
|
|
knx.mock_config_entry.entry_id,
|
|
)
|
|
assert not entity_registry.entities.get_entries_for_device_id(device_id)
|
|
assert not hass_storage[KNX_CONFIG_STORAGE_KEY]["data"]["entities"].get("switch")
|
|
|
|
|
|
async def test_remove_yaml_device_blocked(
|
|
hass: HomeAssistant,
|
|
knx: KNXTestKit,
|
|
device_registry: dr.DeviceRegistry,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""A device with YAML-configured entities can not be removed from the UI."""
|
|
assert await async_setup_component(hass, "config", {})
|
|
await knx.setup_integration(
|
|
{
|
|
Platform.SWITCH: {
|
|
"name": "test",
|
|
KNX_ADDRESS: "1/1/1",
|
|
"device": {"id": "living_room", "name": "Living room"},
|
|
}
|
|
}
|
|
)
|
|
client = await hass_ws_client(hass)
|
|
|
|
device = device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "living_room"), knx.mock_config_entry.entry_id
|
|
)
|
|
assert device is not None
|
|
|
|
response = await client.remove_device(device.id, knx.mock_config_entry.entry_id)
|
|
assert not response["success"]
|
|
assert device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "living_room"), knx.mock_config_entry.entry_id
|
|
)
|
|
|
|
|
|
async def test_remove_device_ignores_foreign_platform_entities(
|
|
hass: HomeAssistant,
|
|
knx: KNXTestKit,
|
|
device_registry: dr.DeviceRegistry,
|
|
entity_registry: er.EntityRegistry,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""An entity from another integration on the device does not block removal."""
|
|
assert await async_setup_component(hass, "config", {})
|
|
await knx.setup_integration()
|
|
client = await hass_ws_client(hass)
|
|
|
|
await client.send_json_auto_id({"type": "knx/create_device", "name": "Test Device"})
|
|
res = await client.receive_json()
|
|
assert res["success"], res
|
|
device_id = res["result"]["id"]
|
|
|
|
entity_registry.async_get_or_create(
|
|
"sensor",
|
|
"other_integration",
|
|
"other_unique_id",
|
|
device_id=device_id,
|
|
)
|
|
|
|
response = await client.remove_device(device_id, knx.mock_config_entry.entry_id)
|
|
assert response["success"]
|
|
assert not device_registry.async_get(device_id)
|
|
|
|
|
|
async def test_remove_device_ignores_other_config_entry_knx_entities(
|
|
hass: HomeAssistant,
|
|
knx: KNXTestKit,
|
|
device_registry: dr.DeviceRegistry,
|
|
entity_registry: er.EntityRegistry,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""A knx-platform entity owned by a different config entry does not block removal."""
|
|
assert await async_setup_component(hass, "config", {})
|
|
await knx.setup_integration()
|
|
client = await hass_ws_client(hass)
|
|
|
|
await client.send_json_auto_id({"type": "knx/create_device", "name": "Test Device"})
|
|
res = await client.receive_json()
|
|
assert res["success"], res
|
|
device_id = res["result"]["id"]
|
|
|
|
other_entry = MockConfigEntry(domain=DOMAIN)
|
|
other_entry.add_to_hass(hass)
|
|
entity_registry.async_get_or_create(
|
|
"sensor",
|
|
DOMAIN,
|
|
"other_config_entry_unique_id",
|
|
config_entry=other_entry,
|
|
device_id=device_id,
|
|
)
|
|
|
|
response = await client.remove_device(device_id, knx.mock_config_entry.entry_id)
|
|
assert response["success"]
|
|
assert not device_registry.async_get(device_id)
|
|
|
|
|
|
@pytest.mark.parametrize("entity_count", [1, 2])
|
|
async def test_yaml_device_name_updates_on_reload(
|
|
hass: HomeAssistant,
|
|
knx: KNXTestKit,
|
|
device_registry: dr.DeviceRegistry,
|
|
entity_count: int,
|
|
) -> None:
|
|
"""Renaming a YAML `device` and reloading updates the device registry.
|
|
|
|
The current YAML `name` wins on every (re-)setup - it is not fixed to
|
|
whatever was configured when the device was first created. This holds
|
|
regardless of how many entities reference the device.
|
|
"""
|
|
|
|
def _config(device_name: str) -> dict[str, Any]:
|
|
entities = [
|
|
{
|
|
"name": "a",
|
|
KNX_ADDRESS: "1/1/1",
|
|
"device": {"id": "as_df", "name": device_name},
|
|
}
|
|
]
|
|
if entity_count > 1:
|
|
# No `name` here: this entity must not block the rename.
|
|
entities.append(
|
|
{"name": "b", KNX_ADDRESS: "1/1/2", "device": {"id": "as_df"}}
|
|
)
|
|
return {Platform.SWITCH: entities}
|
|
|
|
await knx.setup_integration(_config("Initial name"))
|
|
|
|
device = device_registry.async_get_device_by_identifier(
|
|
(DOMAIN, "as_df"), knx.mock_config_entry.entry_id
|
|
)
|
|
assert device is not None
|
|
assert device.name == "Initial name"
|
|
device_id = device.id
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.config.async_hass_config_yaml",
|
|
AsyncMock(return_value={DOMAIN: _config("My device")}),
|
|
),
|
|
_patch_telegram_store(real_store=False),
|
|
patch(
|
|
"xknx.xknx.knx_interface_factory",
|
|
return_value=Mock(
|
|
start=AsyncMock(), stop=AsyncMock(), gateway_info=AsyncMock()
|
|
),
|
|
),
|
|
):
|
|
await hass.services.async_call(DOMAIN, SERVICE_RELOAD, blocking=True)
|
|
|
|
# Same device is updated in place, not duplicated.
|
|
device = device_registry.async_get(device_id)
|
|
assert device is not None
|
|
assert device.name == "My device"
|