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

Add reconfigure step to Onkyo config flow (#129088)

Browse Source
This commit is contained in:
Artur Pragacz 2024-10-30 14:31:43 +01:00 committed by GitHub
parent 0cd5deaa3f
commit ed6123a3e6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 201 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

105
homeassistant/components/onkyo/config_flow.py
View File

@ -6,6 +6,7 @@ from typing import Any
import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
@ -46,13 +47,11 @@ CONF_DEVICE = "device"
INPUT_SOURCES_ALL_MEANINGS = [
input_source.value_meaning for input_source in InputSource
]
STEP_MANUAL_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
STEP_CONFIGURE_SCHEMA = vol.Schema(
{
vol.Required(
OPTION_VOLUME_RESOLUTION,
default=OPTION_VOLUME_RESOLUTION_DEFAULT,
): vol.In(VOLUME_RESOLUTION_ALLOWED),
vol.Required(OPTION_INPUT_SOURCES, default=[]): SelectSelector(
vol.Required(OPTION_VOLUME_RESOLUTION): vol.In(VOLUME_RESOLUTION_ALLOWED),
vol.Required(OPTION_INPUT_SOURCES): SelectSelector(
SelectSelectorConfig(
options=INPUT_SOURCES_ALL_MEANINGS,
multiple=True,
@ -96,15 +95,28 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
errors["base"] = "cannot_connect"
else:
self._receiver_info = info
await self.async_set_unique_id(
info.identifier, raise_on_progress=False
)
self._abort_if_unique_id_configured(updates=user_input)
if self.source == SOURCE_RECONFIGURE:
self._abort_if_unique_id_mismatch()
else:
self._abort_if_unique_id_configured()
return await self.async_step_configure_receiver()
suggested_values = user_input
if suggested_values is None and self.source == SOURCE_RECONFIGURE:
suggested_values = {
CONF_HOST: self._get_reconfigure_entry().data[CONF_HOST]
}
return self.async_show_form(
step_id="manual",
data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
data_schema=self.add_suggested_values_to_schema(
STEP_MANUAL_SCHEMA, suggested_values
),
errors=errors,
)
@ -160,6 +172,12 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle the configuration of a single receiver."""
errors = {}
entry = None
entry_options = None
if self.source == SOURCE_RECONFIGURE:
entry = self._get_reconfigure_entry()
entry_options = entry.options
if user_input is not None:
source_meanings: list[str] = user_input[OPTION_INPUT_SOURCES]
if not source_meanings:
@ -168,33 +186,80 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
sources_store: dict[str, str] = {}
for source_meaning in source_meanings:
source = InputSource.from_meaning(source_meaning)
sources_store[source.value] = source_meaning
result = self.async_create_entry(
title=self._receiver_info.model_name,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: user_input[OPTION_VOLUME_RESOLUTION],
OPTION_MAX_VOLUME: OPTION_MAX_VOLUME_DEFAULT,
OPTION_INPUT_SOURCES: sources_store,
},
)
source_name = source_meaning
if entry_options is not None:
source_name = entry_options[OPTION_INPUT_SOURCES].get(
source.value, source_name
)
sources_store[source.value] = source_name
volume_resolution = user_input[OPTION_VOLUME_RESOLUTION]
if entry_options is None:
result = self.async_create_entry(
title=self._receiver_info.model_name,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: volume_resolution,
OPTION_MAX_VOLUME: OPTION_MAX_VOLUME_DEFAULT,
OPTION_INPUT_SOURCES: sources_store,
},
)
else:
assert entry is not None
result = self.async_update_reload_and_abort(
entry,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: volume_resolution,
OPTION_MAX_VOLUME: entry_options[OPTION_MAX_VOLUME],
OPTION_INPUT_SOURCES: sources_store,
},
)
_LOGGER.debug("Configured receiver, result: %s", result)
return result
_LOGGER.debug("Configuring receiver, info: %s", self._receiver_info)
suggested_values = user_input
if suggested_values is None:
if entry_options is None:
suggested_values = {
OPTION_VOLUME_RESOLUTION: OPTION_VOLUME_RESOLUTION_DEFAULT,
OPTION_INPUT_SOURCES: [],
}
else:
suggested_values = {
OPTION_VOLUME_RESOLUTION: entry_options[OPTION_VOLUME_RESOLUTION],
OPTION_INPUT_SOURCES: [
InputSource(input_source).value_meaning
for input_source in entry_options[OPTION_INPUT_SOURCES]
],
}
return self.async_show_form(
step_id="configure_receiver",
data_schema=STEP_CONFIGURE_SCHEMA,
data_schema=self.add_suggested_values_to_schema(
STEP_CONFIGURE_SCHEMA, suggested_values
),
errors=errors,
description_placeholders={
"name": f"{self._receiver_info.model_name} ({self._receiver_info.host})"
},
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle reconfiguration of the receiver."""
return await self.async_step_manual()
async def async_step_import(self, user_input: dict[str, Any]) -> ConfigFlowResult:
"""Import the yaml config."""
_LOGGER.debug("Import flow user input: %s", user_input)

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

@ -33,6 +33,8 @@
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unique_id_mismatch": "The serial number of the device does not match the previous serial number",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
},

26
tests/components/onkyo/__init__.py
View File

@ -19,9 +19,9 @@ def create_receiver_info(id: int) -> ReceiverInfo:
)
def create_empty_config_entry() -> MockConfigEntry:
"""Create an empty config entry for use in unit tests."""
config = {CONF_HOST: ""}
def create_config_entry_from_info(info: ReceiverInfo) -> MockConfigEntry:
"""Create a config entry from receiver info."""
data = {CONF_HOST: info.host}
options = {
"volume_resolution": 80,
"input_sources": {"12": "tv"},
@ -29,7 +29,25 @@ def create_empty_config_entry() -> MockConfigEntry:
}
return MockConfigEntry(
data=config,
data=data,
options=options,
title=info.model_name,
domain="onkyo",
unique_id=info.identifier,
)
def create_empty_config_entry() -> MockConfigEntry:
"""Create an empty config entry for use in unit tests."""
data = {CONF_HOST: ""}
options = {
"volume_resolution": 80,
"input_sources": {"12": "tv"},
"max_volume": 100,
}
return MockConfigEntry(
data=data,
options=options,
title="Unit test Onkyo",
domain="onkyo",

111
tests/components/onkyo/test_config_flow.py
View File

@ -8,13 +8,22 @@ import pytest
from homeassistant import config_entries
from homeassistant.components.onkyo import InputSource
from homeassistant.components.onkyo.config_flow import OnkyoConfigFlow
from homeassistant.components.onkyo.const import DOMAIN
from homeassistant.components.onkyo.const import (
DOMAIN,
OPTION_MAX_VOLUME,
OPTION_VOLUME_RESOLUTION,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType, InvalidData
from . import create_empty_config_entry, create_receiver_info, setup_integration
from . import (
create_config_entry_from_info,
create_empty_config_entry,
create_receiver_info,
setup_integration,
)
from tests.common import Mock, MockConfigEntry
@ -240,7 +249,7 @@ async def test_configure_empty_source_list(hass: HomeAssistant) -> None:
configure_result = await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"input_sources": []},
user_input={"volume_resolution": 200, "input_sources": []},
)
assert configure_result["errors"] == {
@ -273,13 +282,11 @@ async def test_configure_no_resolution(hass: HomeAssistant) -> None:
user_input={CONF_HOST: "sample-host-name"},
)
configure_result = await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"input_sources": ["TV"]},
)
assert configure_result["type"] is FlowResultType.CREATE_ENTRY
assert configure_result["options"]["volume_resolution"] == 50
with pytest.raises(InvalidData):
await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"input_sources": ["TV"]},
)
async def test_configure_resolution_set(hass: HomeAssistant) -> None:
@ -295,25 +302,24 @@ async def test_configure_resolution_set(hass: HomeAssistant) -> None:
{"next_step_id": "manual"},
)
mock_info = Mock()
mock_info.identifier = "mock_id"
receiver_info = create_receiver_info(1)
with patch(
"homeassistant.components.onkyo.config_flow.async_interview",
return_value=mock_info,
return_value=receiver_info,
):
select_result = await hass.config_entries.flow.async_configure(
form_result["flow_id"],
user_input={CONF_HOST: "sample-host-name"},
)
configure_result = await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"volume_resolution": 200, "input_sources": ["TV"]},
)
configure_result = await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"volume_resolution": 200, "input_sources": ["TV"]},
)
assert configure_result["type"] is FlowResultType.CREATE_ENTRY
assert configure_result["options"]["volume_resolution"] == 200
assert configure_result["type"] is FlowResultType.CREATE_ENTRY
assert configure_result["options"]["volume_resolution"] == 200
async def test_configure_invalid_resolution_set(hass: HomeAssistant) -> None:
@ -348,6 +354,73 @@ async def test_configure_invalid_resolution_set(hass: HomeAssistant) -> None:
)
async def test_reconfigure(hass: HomeAssistant) -> None:
"""Test the reconfigure config flow."""
receiver_info = create_receiver_info(1)
config_entry = create_config_entry_from_info(receiver_info)
await setup_integration(hass, config_entry, receiver_info)
old_host = config_entry.data[CONF_HOST]
old_max_volume = config_entry.options[OPTION_MAX_VOLUME]
result = await config_entry.start_reconfigure_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
with patch(
"homeassistant.components.onkyo.config_flow.async_interview",
return_value=receiver_info,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"host": receiver_info.host}
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "configure_receiver"
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={"volume_resolution": 200, "input_sources": ["TUNER"]},
)
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reconfigure_successful"
assert config_entry.data[CONF_HOST] == old_host
assert config_entry.options[OPTION_VOLUME_RESOLUTION] == 200
assert config_entry.options[OPTION_MAX_VOLUME] == old_max_volume
async def test_reconfigure_new_device(hass: HomeAssistant) -> None:
"""Test the reconfigure config flow with new device."""
receiver_info = create_receiver_info(1)
config_entry = create_config_entry_from_info(receiver_info)
await setup_integration(hass, config_entry, receiver_info)
old_unique_id = receiver_info.identifier
result = await config_entry.start_reconfigure_flow(hass)
receiver_info_2 = create_receiver_info(2)
with patch(
"homeassistant.components.onkyo.config_flow.async_interview",
return_value=receiver_info_2,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"host": receiver_info_2.host}
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "unique_id_mismatch"
# unique id should remain unchanged
assert config_entry.unique_id == old_unique_id
@pytest.mark.parametrize(
("user_input", "exception", "error"),
[