From b385fc9f8608407aca387db12920a69d95c4fcd7 Mon Sep 17 00:00:00 2001 From: Erik Montnemery Date: Fri, 11 Sep 2026 12:24:29 +0200 Subject: [PATCH] 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> --- .../components/alexa_devices/services.py | 3 +- homeassistant/components/bsblan/services.py | 6 +- .../components/device_automation/helpers.py | 17 +-- homeassistant/components/hassio/services.py | 3 +- homeassistant/components/lcn/services.py | 4 +- homeassistant/components/monzo/services.py | 4 +- .../components/portainer/services.py | 6 +- .../components/teslemetry/services.py | 3 +- homeassistant/components/zwave_js/helpers.py | 4 +- homeassistant/helpers/device_registry.py | 74 ++++++++-- homeassistant/helpers/service.py | 60 +++++++- .../components/device_automation/test_init.py | 44 ++++++ tests/helpers/test_device_registry.py | 131 ++++++++++++++++++ tests/helpers/test_service.py | 106 ++++++++++++++ 14 files changed, 424 insertions(+), 41 deletions(-) diff --git a/homeassistant/components/alexa_devices/services.py b/homeassistant/components/alexa_devices/services.py index 2abf2563772d..a755e61fe736 100644 --- a/homeassistant/components/alexa_devices/services.py +++ b/homeassistant/components/alexa_devices/services.py @@ -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) diff --git a/homeassistant/components/bsblan/services.py b/homeassistant/components/bsblan/services.py index 622c615ce2c8..2685cad5d607 100644 --- a/homeassistant/components/bsblan/services.py +++ b/homeassistant/components/bsblan/services.py @@ -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"): diff --git a/homeassistant/components/device_automation/helpers.py b/homeassistant/components/device_automation/helpers.py index 6b91465a7e42..b34b50df0f6d 100644 --- a/homeassistant/components/device_automation/helpers.py +++ b/homeassistant/components/device_automation/helpers.py @@ -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 diff --git a/homeassistant/components/hassio/services.py b/homeassistant/components/hassio/services.py index c029052be2a6..079e63454caf 100644 --- a/homeassistant/components/hassio/services.py +++ b/homeassistant/components/hassio/services.py @@ -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: diff --git a/homeassistant/components/lcn/services.py b/homeassistant/components/lcn/services.py index ced2ed7023fc..4db077c4a654 100644 --- a/homeassistant/components/lcn/services.py +++ b/homeassistant/components/lcn/services.py @@ -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] diff --git a/homeassistant/components/monzo/services.py b/homeassistant/components/monzo/services.py index 08040f2e51ae..0c5ee510c78a 100644 --- a/homeassistant/components/monzo/services.py +++ b/homeassistant/components/monzo/services.py @@ -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 diff --git a/homeassistant/components/portainer/services.py b/homeassistant/components/portainer/services.py index c20316458282..f191998e7cec 100644 --- a/homeassistant/components/portainer/services.py +++ b/homeassistant/components/portainer/services.py @@ -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.""" diff --git a/homeassistant/components/teslemetry/services.py b/homeassistant/components/teslemetry/services.py index 022de84d55dd..58a5fa12f2e6 100644 --- a/homeassistant/components/teslemetry/services.py +++ b/homeassistant/components/teslemetry/services.py @@ -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 diff --git a/homeassistant/components/zwave_js/helpers.py b/homeassistant/components/zwave_js/helpers.py index 93b46d0940e4..d6d19a195904 100644 --- a/homeassistant/components/zwave_js/helpers.py +++ b/homeassistant/components/zwave_js/helpers.py @@ -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: diff --git a/homeassistant/helpers/device_registry.py b/homeassistant/helpers/device_registry.py index 0144fbe94c64..9ad99a6e378d 100644 --- a/homeassistant/helpers/device_registry.py +++ b/homeassistant/helpers/device_registry.py @@ -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: diff --git a/homeassistant/helpers/service.py b/homeassistant/helpers/service.py index 9e4ea2a1b187..d4e9534c6337 100644 --- a/homeassistant/helpers/service.py +++ b/homeassistant/helpers/service.py @@ -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( diff --git a/tests/components/device_automation/test_init.py b/tests/components/device_automation/test_init.py index 4cc8658f3e0b..ba5d1b9fcee7 100644 --- a/tests/components/device_automation/test_init.py +++ b/tests/components/device_automation/test_init.py @@ -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: diff --git a/tests/helpers/test_device_registry.py b/tests/helpers/test_device_registry.py index d908c75b9d1c..9d654bd7440f 100644 --- a/tests/helpers/test_device_registry.py +++ b/tests/helpers/test_device_registry.py @@ -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] diff --git a/tests/helpers/test_service.py b/tests/helpers/test_service.py index 4250e65e65c4..f9c4ecdc4572 100644 --- a/tests/helpers/test_service.py +++ b/tests/helpers/test_service.py @@ -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"