Teach helpers returning device and config entry for domain about children (#181923)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Erik Montnemery
2026-09-11 12:24:29 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 37675b32fc
commit b385fc9f86
14 changed files with 424 additions and 41 deletions
@@ -46,8 +46,9 @@ def async_get_entry_id_for_service_call(
) -> tuple[dr.DeviceEntry, AmazonConfigEntry]:
"""Get the entry ID related to a service call (by device ID)."""
config_entry: AmazonConfigEntry
# Callers read the device's serial number, which only a main device has
device, config_entry = service.async_get_device_and_config_entry(
call.hass, DOMAIN, call.data[ATTR_DEVICE_ID]
call.hass, DOMAIN, call.data[ATTR_DEVICE_ID], include_child_devices=False
)
return (device, config_entry)
+3 -3
View File
@@ -127,7 +127,7 @@ def _build_weekly_schedule_days(
def _resolve_config_entry(
service_call: ServiceCall,
) -> tuple[BSBLanConfigEntry, dr.DeviceEntry]:
) -> tuple[BSBLanConfigEntry, dr.AnyDeviceEntry]:
"""Resolve device_id from a service call into a loaded BSBLAN config entry."""
config_entry: BSBLanConfigEntry
device, config_entry = service.async_get_device_and_config_entry(
@@ -136,12 +136,12 @@ def _resolve_config_entry(
return config_entry, device
def _device_name(device_entry: dr.DeviceEntry) -> str:
def _device_name(device_entry: dr.AnyDeviceEntry) -> str:
"""Return the best available display name for a device."""
return device_entry.name_by_user or device_entry.name or device_entry.id
def _ensure_water_heater_device(device_entry: dr.DeviceEntry) -> None:
def _ensure_water_heater_device(device_entry: dr.AnyDeviceEntry) -> None:
"""Validate the service targets the water heater sub-device."""
for domain, identifier in device_entry.identifiers:
if domain == DOMAIN and identifier.endswith("-water-heater"):
@@ -130,24 +130,17 @@ async def async_validate_device_automation_config(
)
# Find a config entry with the same domain as the device automation
device_config_entry = None
for entry_id in device.config_entries:
if (
not (entry := hass.config_entries.async_get_entry(entry_id))
or entry.domain != validated_config[CONF_DOMAIN]
):
continue
device_config_entry = entry
break
if not device_config_entry:
_, config_entry = dr.async_get_device_and_config_entry_for_domain(
hass, device.id, domain=validated_config[CONF_DOMAIN]
)
if not config_entry:
# There's no config entry with the same domain as the device automation
raise InvalidDeviceAutomationConfig(
f"Device '{validated_config[CONF_DEVICE_ID]}' has no config entry from "
f"domain '{validated_config[CONF_DOMAIN]}'"
)
if not await hass.config_entries.async_wait_component(device_config_entry):
if not await hass.config_entries.async_wait_component(config_entry):
# The component could not be loaded, skip the dynamic validation
return validated_config
+2 -1
View File
@@ -452,8 +452,9 @@ def async_register_network_storage_services(
async def async_mount_reload(service: ServiceCall) -> None:
"""Handle service calls for Hass.io."""
# A mount is always a main device, and the check below reads its model
device, _ = async_get_device_and_config_entry(
hass, DOMAIN, service.data[ATTR_DEVICE_ID]
hass, DOMAIN, service.data[ATTR_DEVICE_ID], include_child_devices=False
)
if device.name is None or device.model != SupervisorEntityModel.MOUNT:
+3 -1
View File
@@ -71,8 +71,10 @@ class LcnServiceCall:
def get_device_connection(self, service: ServiceCall) -> DeviceConnection:
"""Get address connection object."""
entry: LcnConfigEntry
# device_connections is keyed by the ids of the main devices LCN registers
# for its modules and groups, so a child device has no connection
device, entry = async_get_device_and_config_entry(
self.hass, DOMAIN, service.data[CONF_DEVICE_ID]
self.hass, DOMAIN, service.data[CONF_DEVICE_ID], include_child_devices=False
)
return entry.runtime_data.device_connections[device.id]
+2 -2
View File
@@ -88,7 +88,7 @@ TRANSFER_SCHEMA = vol.Schema(
@callback
def _async_get_resource_id(device: dr.DeviceEntry) -> str:
def _async_get_resource_id(device: dr.AnyDeviceEntry) -> str:
"""Get the Monzo resource ID represented by a device."""
for domain, resource_id in device.identifiers:
if domain == DOMAIN:
@@ -99,7 +99,7 @@ def _async_get_resource_id(device: dr.DeviceEntry) -> str:
)
def _device_name(device: dr.DeviceEntry) -> str:
def _device_name(device: dr.AnyDeviceEntry) -> str:
"""Return the best available name for a device."""
return device.name_by_user or device.name or device.id
@@ -53,7 +53,7 @@ SERVICE_RECREATE_CONTAINER_SCHEMA = vol.Schema(
@callback
def _async_get_device_and_entry(
call: ServiceCall, device_id: str
) -> tuple[dr.DeviceEntry, PortainerConfigEntry]:
) -> tuple[dr.AnyDeviceEntry, PortainerConfigEntry]:
"""Resolve and validate the device and Portainer config entry for a device ID."""
entry: PortainerConfigEntry
device, entry = service.async_get_device_and_config_entry(
@@ -64,7 +64,7 @@ def _async_get_device_and_entry(
@callback
def _async_get_endpoint_id(
device: dr.DeviceEntry,
device: dr.AnyDeviceEntry,
config_entry: PortainerConfigEntry,
) -> int:
"""Get the endpoint ID from a device entry."""
@@ -85,7 +85,7 @@ def _async_get_endpoint_id(
@callback
def _async_get_container_and_endpoint_ids(
device: dr.DeviceEntry,
device: dr.AnyDeviceEntry,
config_entry: PortainerConfigEntry,
) -> tuple[int, str]:
"""Get the endpoint ID and container ID from a container device entry."""
@@ -73,8 +73,9 @@ def async_get_device_and_config_for_service_call(
) -> tuple[dr.DeviceEntry, TeslemetryConfigEntry]:
"""Get the device entry and config entry related to a service call."""
config_entry: TeslemetryConfigEntry
# Callers match the device's serial number, which only a main device has
device_entry, config_entry = service.async_get_device_and_config_entry(
hass, DOMAIN, call.data[CONF_DEVICE_ID]
hass, DOMAIN, call.data[CONF_DEVICE_ID], include_child_devices=False
)
return device_entry, config_entry
+2 -2
View File
@@ -286,7 +286,7 @@ def async_get_node_from_device_id(hass: HomeAssistant, device_id: str) -> ZwaveN
# Use the device config entry to validate that this is a valid zwave_js device
# and to get the client
device, config_entry = cast(
tuple[dr.DeviceEntry | None, ZwaveJSConfigEntry | None],
tuple[dr.AnyDeviceEntry | None, ZwaveJSConfigEntry | None],
dr.async_get_device_and_config_entry_for_domain(hass, device_id, domain=DOMAIN),
)
if device is None:
@@ -340,7 +340,7 @@ async def async_get_provisioning_entry_from_device_id(
# Use the device config entry to validate that this is a valid zwave_js device
# and to get the client
device, config_entry = cast(
tuple[dr.DeviceEntry | None, ZwaveJSConfigEntry | None],
tuple[dr.AnyDeviceEntry | None, ZwaveJSConfigEntry | None],
dr.async_get_device_and_config_entry_for_domain(hass, device_id, domain=DOMAIN),
)
if device is None:
+64 -10
View File
@@ -4593,26 +4593,80 @@ def async_get_device_id_by_identifier(
return device.id
@overload
def async_get_device_and_config_entry_for_domain(
hass: HomeAssistant,
device_id: str,
*,
domain: str,
include_child_devices: Literal[False],
include_main_devices: bool = True,
) -> tuple[DeviceEntry | None, ConfigEntry | None]: ...
@overload
def async_get_device_and_config_entry_for_domain(
hass: HomeAssistant,
device_id: str,
*,
domain: str,
include_child_devices: Literal[True] = True,
include_main_devices: Literal[False],
) -> tuple[ChildDeviceEntry | None, ConfigEntry | None]: ...
@overload
def async_get_device_and_config_entry_for_domain(
hass: HomeAssistant,
device_id: str,
*,
domain: str,
include_child_devices: Literal[True] = True,
include_main_devices: Literal[True] = True,
) -> tuple[AnyDeviceEntry | None, ConfigEntry | None]: ...
@callback
def async_get_device_and_config_entry_for_domain(
hass: HomeAssistant, device_id: str, *, domain: str
) -> tuple[DeviceEntry | None, ConfigEntry | None]:
hass: HomeAssistant,
device_id: str,
*,
domain: str,
include_child_devices: bool = True,
include_main_devices: bool = True,
) -> tuple[AnyDeviceEntry | None, ConfigEntry | None]:
"""Get the device and the config entry of the domain owning it.
Returns (None, None) for an unknown device id or if the device is a child
device, and (device, None) when no config entry of the domain owns the
device. A returned pair is consistent: for a pre-migration composite
device id, the device is the domain's split device, not the composite; if
several splits belong to config entries of the domain, which pair is
returned is undefined. When no split matches the domain, the restored
composite is returned as the device.
Returns (None, None) for an unknown device id, and (device, None) when no
config entry of the domain owns the device.
With include_child_devices=False a child-device id resolves to None.
With include_main_devices=False a main-device id resolves to None. A
composite-device id then resolves to None as well, because both the splits
of a composite and the restored composite itself are main devices.
A returned pair is consistent: for a pre-migration composite device id, the
device is the domain's split device, not the composite; if several splits
belong to config entries of the domain, which pair is returned is undefined.
When no split matches the domain, the restored composite is returned as the
device.
"""
registry = async_get(hass)
if (device := registry._devices.get(device_id)) is not None: # noqa: SLF001
device: AnyDeviceEntry | None = None
if include_main_devices:
device = registry._devices.get(device_id) # noqa: SLF001
if device is None and include_child_devices:
device = registry.async_get(
device_id, include_main_devices=False, include_composite_devices=False
)
if device is not None:
config_entry = hass.config_entries.async_get_entry(device.config_entry_id)
if config_entry is not None and config_entry.domain == domain:
return device, config_entry
return device, None
if not include_main_devices:
return None, None
for split in registry.async_get_devices_for_composite_device_id(device_id):
config_entry = hass.config_entries.async_get_entry(split.config_entry_id)
if config_entry is not None and config_entry.domain == domain:
+55 -5
View File
@@ -7,7 +7,7 @@ from functools import cache, partial
import inspect
import logging
from types import ModuleType
from typing import TYPE_CHECKING, Any, TypedDict, cast
from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast, overload
import voluptuous as vol
@@ -1425,17 +1425,67 @@ def _async_get_single_loaded_config_entry(
return config_entry
@overload
def async_get_device_and_config_entry(
hass: HomeAssistant,
domain: str,
device_id: str,
*,
include_child_devices: Literal[False],
include_main_devices: bool = True,
) -> tuple[device_registry.DeviceEntry, ConfigEntry]: ...
@overload
def async_get_device_and_config_entry(
hass: HomeAssistant,
domain: str,
device_id: str,
*,
include_child_devices: Literal[True] = True,
include_main_devices: Literal[False],
) -> tuple[device_registry.ChildDeviceEntry, ConfigEntry]: ...
@overload
def async_get_device_and_config_entry(
hass: HomeAssistant,
domain: str,
device_id: str,
*,
include_child_devices: Literal[True] = True,
include_main_devices: Literal[True] = True,
) -> tuple[device_registry.AnyDeviceEntry, ConfigEntry]: ...
@callback
def async_get_device_and_config_entry(
hass: HomeAssistant, domain: str, device_id: str
) -> tuple[device_registry.DeviceEntry, ConfigEntry]:
hass: HomeAssistant,
domain: str,
device_id: str,
*,
include_child_devices: bool = True,
include_main_devices: bool = True,
) -> tuple[device_registry.AnyDeviceEntry, ConfigEntry]:
"""Get and validate the device and the loaded config entry of the domain owning it.
Raises ServiceValidationError if the device is unknown, is not owned by a
config entry of the domain, or if that config entry is not loaded.
With include_child_devices=False a child-device id raises as an unknown device.
With include_main_devices=False a main-device id raises as an unknown device;
as does a composite-device id, because both the splits of a composite and the
restored composite itself are main devices.
"""
device, config_entry = device_registry.async_get_device_and_config_entry_for_domain(
hass, device_id, domain=domain
device: device_registry.AnyDeviceEntry | None
config_entry: ConfigEntry | None
# The flags are plain bools here, which matches none of the Literal overloads
device, config_entry = device_registry.async_get_device_and_config_entry_for_domain( # type: ignore[call-overload]
hass,
device_id,
domain=domain,
include_child_devices=include_child_devices,
include_main_devices=include_main_devices,
)
if device is None:
raise ServiceValidationError(
@@ -1243,6 +1243,50 @@ async def test_automation_with_dynamically_validated_trigger(
module.async_attach_trigger.assert_awaited_once()
@pytest.mark.usefixtures("fake_integration")
async def test_automation_with_child_device(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test device automation targeting a child device of the domain's config entry."""
module_cache = hass.data[loader.DATA_COMPONENTS]
module = module_cache["fake_integration.device_trigger"]
module.async_attach_trigger = AsyncMock()
module.async_validate_trigger_config = AsyncMock(wraps=lambda hass, config: config)
config_entry = MockConfigEntry(domain="fake_integration", data={})
config_entry.mock_state(hass, ConfigEntryState.LOADED)
config_entry.add_to_hass(hass)
parent_device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={("fake_integration", "parent")},
)
child_device_entry = device_registry.async_get_or_create_child(
config_entry_id=config_entry.entry_id,
identifiers={("fake_integration", "child")},
parent_device_id=parent_device_entry.id,
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"alias": "hello",
"trigger": {
"platform": "device",
"device_id": child_device_entry.id,
"domain": "fake_integration",
},
"action": {"service": "test.automation", "entity_id": "hello.world"},
}
},
)
module.async_validate_trigger_config.assert_awaited_once()
module.async_attach_trigger.assert_awaited_once()
async def test_automation_with_integration_without_device_trigger(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
+131
View File
@@ -10184,6 +10184,137 @@ async def test_async_get_device_and_config_entry_for_domain_composite(
assert device.config_entries == {entry_a.entry_id, entry_b.entry_id}
async def test_async_get_device_and_config_entry_for_domain_child_devices(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test getting the device and config entry of a domain owning a child device."""
entry = MockConfigEntry(domain="domain_a")
entry.add_to_hass(hass)
parent = device_registry.async_get_or_create(
config_entry_id=entry.entry_id, identifiers={("domain_a", "1")}
)
child = device_registry.async_get_or_create_child(
config_entry_id=entry.entry_id,
identifiers={("domain_a", "1_1")},
parent_device_id=parent.id,
)
# A child device is paired with the entry owning it
assert dr.async_get_device_and_config_entry_for_domain(
hass, child.id, domain="domain_a"
) == (child, entry)
# A domain not owning the child still gets the child
assert dr.async_get_device_and_config_entry_for_domain(
hass, child.id, domain="domain_b"
) == (child, None)
# With include_child_devices=False the child is treated as absent
assert dr.async_get_device_and_config_entry_for_domain(
hass, child.id, domain="domain_a", include_child_devices=False
) == (None, None)
# A main device is unaffected by include_child_devices
assert dr.async_get_device_and_config_entry_for_domain(
hass, parent.id, domain="domain_a"
) == (parent, entry)
assert dr.async_get_device_and_config_entry_for_domain(
hass, parent.id, domain="domain_a", include_child_devices=False
) == (parent, entry)
# An unknown device id
assert dr.async_get_device_and_config_entry_for_domain(
hass, "unknown_id", domain="domain_a"
) == (None, None)
async def test_async_get_device_and_config_entry_for_domain_no_main_devices(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test getting the device and config entry with main devices excluded."""
entry = MockConfigEntry(domain="domain_a")
entry.add_to_hass(hass)
parent = device_registry.async_get_or_create(
config_entry_id=entry.entry_id, identifiers={("domain_a", "1")}
)
child = device_registry.async_get_or_create_child(
config_entry_id=entry.entry_id,
identifiers={("domain_a", "1_1")},
parent_device_id=parent.id,
)
# A main device is not resolved
assert dr.async_get_device_and_config_entry_for_domain(
hass, parent.id, domain="domain_a", include_main_devices=False
) == (None, None)
# A child-only lookup resolves the child
assert dr.async_get_device_and_config_entry_for_domain(
hass, child.id, domain="domain_a", include_main_devices=False
) == (child, entry)
assert dr.async_get_device_and_config_entry_for_domain(
hass, child.id, domain="domain_b", include_main_devices=False
) == (child, None)
# Neither main nor child devices are resolved with both flags off
assert dr.async_get_device_and_config_entry_for_domain(
hass,
child.id,
domain="domain_a",
include_child_devices=False,
include_main_devices=False,
) == (None, None)
assert dr.async_get_device_and_config_entry_for_domain(
hass,
parent.id,
domain="domain_a",
include_child_devices=False,
include_main_devices=False,
) == (None, None)
@pytest.mark.parametrize("load_registries", [False])
async def test_async_get_device_and_config_entry_for_domain_composite_flags(
hass: HomeAssistant, hass_storage: dict[str, Any]
) -> None:
"""Test the include_* flags for a composite device id."""
entry_a = MockConfigEntry(domain="domain_a")
entry_a.add_to_hass(hass)
entry_b = MockConfigEntry(domain="domain_b")
entry_b.add_to_hass(hass)
hass_storage[dr.STORAGE_KEY] = _composite_device_storage(entry_a, entry_b)
dr.async_setup(hass)
await dr.async_load(hass)
device_registry = dr.async_get(hass)
split_a = _get_device_for_config_entry(
device_registry, entry_a.entry_id, identifiers={("domain_a", "1")}
)
# A composite device id resolves to the domain's split regardless of
# include_child_devices
assert dr.async_get_device_and_config_entry_for_domain(
hass, COMPOSITE_ID, domain="domain_a", include_child_devices=False
) == (split_a, entry_a)
# A domain owning none of the splits still gets the restored composite
device, config_entry = dr.async_get_device_and_config_entry_for_domain(
hass, COMPOSITE_ID, domain="domain_c", include_child_devices=False
)
assert config_entry is None
assert device is not None
assert device.id == COMPOSITE_ID
# A composite device and its splits are main devices: include_main_devices=False
# suppresses both the split lookup and the restored composite
assert dr.async_get_device_and_config_entry_for_domain(
hass, COMPOSITE_ID, domain="domain_a", include_main_devices=False
) == (None, None)
assert dr.async_get_device_and_config_entry_for_domain(
hass, COMPOSITE_ID, domain="domain_c", include_main_devices=False
) == (None, None)
assert dr.async_get_device_and_config_entry_for_domain(
hass,
COMPOSITE_ID,
domain="domain_a",
include_child_devices=False,
include_main_devices=False,
) == (None, None)
@pytest.mark.parametrize("load_registries", [False])
async def test_clear_config_entry_clears_composite_primary_config_entry(
hass: HomeAssistant, hass_storage: dict[str, Any]
+106
View File
@@ -3511,3 +3511,109 @@ async def test_get_service_device_and_config_entry(
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(hass, domain, device.id)
assert err.value.translation_key == "service_config_entry_not_loaded"
async def test_get_service_device_and_config_entry_child_devices(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test that we can get a child device and its config entry."""
domain = "mock_integration"
entry = MockConfigEntry(domain=domain)
entry.add_to_hass(hass)
entry.mock_state(hass, config_entries.ConfigEntryState.LOADED)
parent = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(domain, "unique_id")},
name="Mock device",
)
child = device_registry.async_get_or_create_child(
config_entry_id=entry.entry_id,
identifiers={(domain, "unique_id_child")},
parent_device_id=parent.id,
name="Mock child device",
)
# A child device is paired with the entry owning it
assert service.async_get_device_and_config_entry(hass, domain, child.id) == (
child,
entry,
)
# With include_child_devices=False the child raises as an unknown device
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(
hass, domain, child.id, include_child_devices=False
)
assert err.value.translation_key == "service_device_not_found"
assert err.value.translation_placeholders == {"device_id": child.id}
# The child exists, but is not owned by a config entry of the domain
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(hass, "another_domain", child.id)
assert err.value.translation_key == "service_device_wrong_domain"
assert err.value.translation_placeholders == {
"device_name": "Mock child device",
"domain": "another_domain",
}
# The child exists, but its config entry is not loaded
entry.mock_state(hass, config_entries.ConfigEntryState.NOT_LOADED)
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(hass, domain, child.id)
assert err.value.translation_key == "service_config_entry_not_loaded"
async def test_get_service_device_and_config_entry_no_main_devices(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test getting a device and its config entry with main devices excluded."""
domain = "mock_integration"
entry = MockConfigEntry(domain=domain)
entry.add_to_hass(hass)
entry.mock_state(hass, config_entries.ConfigEntryState.LOADED)
parent = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(domain, "unique_id")},
name="Mock device",
)
child = device_registry.async_get_or_create_child(
config_entry_id=entry.entry_id,
identifiers={(domain, "unique_id_child")},
parent_device_id=parent.id,
name="Mock child device",
)
# A main device raises as an unknown device
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(
hass, domain, parent.id, include_main_devices=False
)
assert err.value.translation_key == "service_device_not_found"
assert err.value.translation_placeholders == {"device_id": parent.id}
# A child-only lookup resolves the child
assert service.async_get_device_and_config_entry(
hass, domain, child.id, include_main_devices=False
) == (child, entry)
# A child device raises as an unknown device with both flags off
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(
hass,
domain,
child.id,
include_child_devices=False,
include_main_devices=False,
)
assert err.value.translation_key == "service_device_not_found"
# Neither does a main device with both flags off
with pytest.raises(exceptions.ServiceValidationError) as err:
service.async_get_device_and_config_entry(
hass,
domain,
parent.id,
include_child_devices=False,
include_main_devices=False,
)
assert err.value.translation_key == "service_device_not_found"