jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-28 07:37:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplify and add tests

Browse Source
This commit is contained in:
jbouwh 2025-07-06 22:38:02 +00:00
parent 02d911b7c7
commit 355881e60b
5 changed files with 179 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/mqtt/entity.py
View File

@ -291,7 +291,6 @@ def async_setup_entity_entry_helper(
"entry_id": entry.entry_id,
"subentry_id": subentry_id,
"name": name,
"migration_type": migration_type,
},
translation_placeholders={"name": name},
translation_key=migration_type,

37
homeassistant/components/mqtt/repairs.py
View File

@ -15,16 +15,13 @@ from .const import DOMAIN
class MQTTDeviceEntryMigration(RepairsFlow):
"""Handler to migrate device from subentry to main entry and reload."""
"""Handler to remove subentry for migrated MQTT device."""
def __init__(
self, entry_id: str, subentry_id: str, name: str, migration_type: str
) -> None:
def __init__(self, entry_id: str, subentry_id: str, name: str) -> None:
"""Initialize the flow."""
self.entry_id = entry_id
self.subentry_id = subentry_id
self.name = name
self.migration_type = migration_type
async def async_step_init(
self, user_input: dict[str, str] | None = None
@ -36,9 +33,7 @@ class MQTTDeviceEntryMigration(RepairsFlow):
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Handle the confirm step of a fix flow."""
if user_input is not None and self.migration_type == "subentry_migration_yaml":
# Via YAML the device was already registered and bound to the entry,
# so it is safe to remove the subentry from here.
if user_input is not None:
device_registry = dr.async_get(self.hass)
subentry_device = device_registry.async_get_device(
identifiers={(DOMAIN, self.subentry_id)}
@ -49,29 +44,6 @@ class MQTTDeviceEntryMigration(RepairsFlow):
assert subentry_device is not None
self.hass.config_entries.async_remove_subentry(entry, self.subentry_id)
return self.async_create_entry(data={})
if (
user_input is not None
and self.migration_type == "subentry_migration_discovery"
):
# The device offered via discovery was already set up through the subentry,
# so we need to update the device before removing the subentry and reload.
device_registry = dr.async_get(self.hass)
subentry_device = device_registry.async_get_device(
identifiers={(DOMAIN, self.subentry_id)}
)
entry = self.hass.config_entries.async_get_entry(self.entry_id)
if TYPE_CHECKING:
assert entry is not None
assert subentry_device is not None
device_registry.async_update_device(
subentry_device.id,
remove_config_entry_id=self.entry_id,
remove_config_subentry_id=self.subentry_id,
add_config_entry_id=self.entry_id,
)
self.hass.config_entries.async_remove_subentry(entry, self.subentry_id)
self.hass.config_entries.async_schedule_reload(self.entry_id)
return self.async_create_entry(data={})
return self.async_show_form(
step_id="confirm",
@ -91,15 +63,12 @@ async def async_create_fix_flow(
entry_id = data["entry_id"]
subentry_id = data["subentry_id"]
name = data["name"]
migration_type = data["migration_type"]
if TYPE_CHECKING:
assert isinstance(entry_id, str)
assert isinstance(subentry_id, str)
assert isinstance(name, str)
assert isinstance(migration_type, str)
return MQTTDeviceEntryMigration(
entry_id=entry_id,
subentry_id=subentry_id,
name=name,
migration_type=migration_type,
)

2
homeassistant/components/mqtt/strings.json
View File

@ -10,7 +10,7 @@
"step": {
"confirm": {
"title": "[%key:component::mqtt::issues::subentry_migration_discovery::title%]",
"description": "Exported MQTT Device \"{name}\" identified via MQTT discovery. Confirm that the MQTT device is migrated to the main MQTT configuration, the replaced subentry of the MQTT device is to be removed, and MQTT is to be reloaded. Make sure that the discovery is retained at the MQTT broker, or is resent after the reload is completed, so that the MQTT will be set up correctly."
"description": "Exported MQTT Device \"{name}\" identified via MQTT discovery. Confirm that the MQTT device is migrated to the main MQTT configuration, the replaced subentry of the MQTT device is to be removed. Make sure that the discovery is retained at the MQTT broker, or is resent after the subentry is removed, so that the MQTT device will be set up correctly."
}
}
}

24
tests/components/mqtt/test_config_flow.py
View File

@ -4232,14 +4232,27 @@ async def test_subentry_reconfigure_availablity(
)
],
)
@pytest.mark.parametrize(("flow_step"), ["export_yaml", "export_discovery"])
@pytest.mark.parametrize(
("flow_step", "field_suggestions"),
[
("export_yaml", {"yaml": "identifiers:\n - {}\n"}),
(
"export_discovery",
{
"discovery_topic": "homeassistant/device/{}/config",
"discovery_payload": '"identifiers": [\n "{}"\n',
},
),
],
)
async def test_subentry_reconfigure_export_settings(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
device_registry: dr.DeviceRegistry,
flow_step: str,
field_suggestions: dict[str, str],
) -> None:
"""Test the subentry ConfigFlow reconfigure and update device properties."""
"""Test the subentry ConfigFlow reconfigure export feature."""
await mqtt_mock_entry()
config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
subentry_id: str
@ -4285,6 +4298,13 @@ async def test_subentry_reconfigure_export_settings(
"url": "https://www.home-assistant.io/integrations/mqtt/"
}
# Assert the export is correct
for field in result["data_schema"].schema:
assert (
field_suggestions[field].format(subentry_id)
in field.description["suggested_value"]
)
# Back to summary menu
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],

153
tests/components/mqtt/test_repairs.py Normal file
View File

@ -0,0 +1,153 @@
"""Test repairs for MQTT."""
from collections.abc import Coroutine
from copy import deepcopy
from typing import Any
from unittest.mock import patch
import pytest
from homeassistant.components import mqtt
from homeassistant.config_entries import ConfigSubentry, ConfigSubentryData
from homeassistant.const import SERVICE_RELOAD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import device_registry as dr, issue_registry as ir
from homeassistant.util.yaml import parse_yaml
from .common import MOCK_NOTIFY_SUBENTRY_DATA_MULTI, async_fire_mqtt_message
from tests.common import MockConfigEntry, async_capture_events
from tests.components.repairs import (
async_process_repairs_platforms,
process_repair_fix_flow,
start_repair_fix_flow,
)
from tests.conftest import ClientSessionGenerator
from tests.typing import MqttMockHAClientGenerator
async def help_setup_yaml(hass: HomeAssistant, config: dict[str, str]) -> None:
"""Help to set up an exported MQTT device via YAML."""
with patch(
"homeassistant.config.load_yaml_config_file",
return_value=parse_yaml(config["yaml"]),
):
await hass.services.async_call(
mqtt.DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
async def help_setup_discovery(hass: HomeAssistant, config: dict[str, str]) -> None:
"""Help to set up an exported MQTT device via YAML."""
async_fire_mqtt_message(
hass, config["discovery_topic"], config["discovery_payload"]
)
await hass.async_block_till_done(wait_background_tasks=True)
@pytest.mark.parametrize(
"mqtt_config_subentries_data",
[
(
ConfigSubentryData(
data=MOCK_NOTIFY_SUBENTRY_DATA_MULTI,
subentry_type="device",
title="Mock subentry",
),
)
],
)
@pytest.mark.parametrize(
("flow_step", "setup_helper", "translation_key"),
[
("export_yaml", help_setup_yaml, "subentry_migration_yaml"),
("export_discovery", help_setup_discovery, "subentry_migration_discovery"),
],
)
async def test_subentry_reconfigure_export_settings(
hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator,
device_registry: dr.DeviceRegistry,
hass_client: ClientSessionGenerator,
flow_step: str,
setup_helper: Coroutine[Any, Any, None],
translation_key: str,
) -> None:
"""Test the subentry ConfigFlow YAML export with migration to YAML."""
await mqtt_mock_entry()
config_entry: MockConfigEntry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
subentry_id: str
subentry: ConfigSubentry
subentry_id, subentry = next(iter(config_entry.subentries.items()))
result = await config_entry.start_subentry_reconfigure_flow(hass, subentry_id)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "summary_menu"
# assert we have a device for the subentry
device = device_registry.async_get_device(identifiers={(mqtt.DOMAIN, subentry_id)})
assert device is not None
# assert we entity for all subentry components
components = deepcopy(dict(subentry.data))["components"]
assert len(components) == 2
# assert menu options, we have the option to export
assert result["menu_options"] == [
"entity",
"update_entity",
"delete_entity",
"device",
"availability",
"export",
]
# Open export menu
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{"next_step_id": "export"},
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "export"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{"next_step_id": flow_step},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == flow_step
assert result["description_placeholders"] == {
"url": "https://www.home-assistant.io/integrations/mqtt/"
}
# Copy the suggested values for an export
suggested_values_from_schema = {
field: field.description["suggested_value"]
for field in result["data_schema"].schema
}
# Try to set up the exported config
events = async_capture_events(hass, ir.EVENT_REPAIRS_ISSUE_REGISTRY_UPDATED)
await setup_helper(hass, suggested_values_from_schema)
# Assert a repair flow was created
assert len(events) == 1
issue_id = events[0].data["issue_id"]
issue_registry = ir.async_get(hass)
repair_issue = issue_registry.async_get_issue(mqtt.DOMAIN, issue_id)
assert repair_issue.translation_key == translation_key
await async_process_repairs_platforms(hass)
client = await hass_client()
data = await start_repair_fix_flow(client, mqtt.DOMAIN, issue_id)
flow_id = data["flow_id"]
assert data["description_placeholders"] == {"name": "Milk notifier"}
assert data["step_id"] == "confirm"
data = await process_repair_fix_flow(client, flow_id)
assert data["type"] == "create_entry"