Improve config flow tests in Onkyo (#149199)

This commit is contained in:
Artur Pragacz 2025-07-23 15:06:25 +02:00 committed by GitHub
parent 7c83fd0bf9
commit 9a9f65dc36
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 271 additions and 316 deletions

View File

@ -8,10 +8,7 @@ rules:
brands: done brands: done
common-modules: done common-modules: done
config-flow: done config-flow: done
config-flow-test-coverage: config-flow-test-coverage: done
status: todo
comment: |
Coverage is 100%, but the tests need to be improved.
dependency-transparency: done dependency-transparency: done
docs-actions: done docs-actions: done
docs-high-level-description: done docs-high-level-description: done
@ -39,9 +36,9 @@ rules:
parallel-updates: todo parallel-updates: todo
reauthentication-flow: reauthentication-flow:
status: exempt status: exempt
comment: | comment: This integration does not require authentication.
This integration does not require authentication. test-coverage: done
test-coverage: todo
# Gold # Gold
devices: todo devices: todo
diagnostics: todo diagnostics: todo

View File

@ -8,9 +8,8 @@ from aioonkyo import Code, Instruction, Kind, Receiver, Status, Zone, status
import pytest import pytest
from homeassistant.components.onkyo.const import DOMAIN from homeassistant.components.onkyo.const import DOMAIN
from homeassistant.const import CONF_HOST
from . import RECEIVER_INFO, mock_discovery from . import RECEIVER_INFO, RECEIVER_INFO_2, mock_discovery
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@ -24,7 +23,7 @@ def mock_default_discovery() -> Generator[None]:
DEVICE_INTERVIEW_TIMEOUT=1, DEVICE_INTERVIEW_TIMEOUT=1,
DEVICE_DISCOVERY_TIMEOUT=1, DEVICE_DISCOVERY_TIMEOUT=1,
), ),
mock_discovery([RECEIVER_INFO]), mock_discovery([RECEIVER_INFO, RECEIVER_INFO_2]),
): ):
yield yield
@ -164,7 +163,7 @@ def mock_receiver(
@pytest.fixture @pytest.fixture
def mock_config_entry() -> MockConfigEntry: def mock_config_entry() -> MockConfigEntry:
"""Mock a config entry.""" """Mock a config entry."""
data = {CONF_HOST: RECEIVER_INFO.host} data = {"host": RECEIVER_INFO.host}
options = { options = {
"volume_resolution": 80, "volume_resolution": 80,
"max_volume": 100, "max_volume": 100,

View File

@ -1,5 +1,7 @@
"""Test Onkyo config flow.""" """Test Onkyo config flow."""
from contextlib import AbstractContextManager, nullcontext
from aioonkyo import ReceiverInfo from aioonkyo import ReceiverInfo
import pytest import pytest
@ -15,7 +17,7 @@ from homeassistant.components.onkyo.const import (
from homeassistant.config_entries import SOURCE_USER from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType, InvalidData from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.ssdp import ( from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_FRIENDLY_NAME,
SsdpServiceInfo, SsdpServiceInfo,
@ -26,186 +28,87 @@ from . import RECEIVER_INFO, RECEIVER_INFO_2, mock_discovery, setup_integration
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
def _entry_title(receiver_info: ReceiverInfo) -> str: def _receiver_display_name(receiver_info: ReceiverInfo) -> str:
return f"{receiver_info.model_name} ({receiver_info.host})" return f"{receiver_info.model_name} ({receiver_info.host})"
async def test_user_initial_menu(hass: HomeAssistant) -> None:
"""Test initial menu."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert init_result["type"] is FlowResultType.MENU
# Check if the values are there, but ignore order
assert not set(init_result["menu_options"]) ^ {"manual", "eiscp_discovery"}
async def test_manual_valid_host(hass: HomeAssistant) -> None:
"""Test valid host entered."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
form_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
{"next_step_id": "manual"},
)
select_result = await hass.config_entries.flow.async_configure(
form_result["flow_id"],
user_input={CONF_HOST: RECEIVER_INFO.host},
)
assert select_result["step_id"] == "configure_receiver"
assert select_result["description_placeholders"]["name"] == _entry_title(
RECEIVER_INFO
)
async def test_manual_invalid_host(hass: HomeAssistant) -> None:
"""Test invalid host entered."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
form_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
{"next_step_id": "manual"},
)
with mock_discovery([]):
host_result = await hass.config_entries.flow.async_configure(
form_result["flow_id"],
user_input={CONF_HOST: "sample-host-name"},
)
assert host_result["step_id"] == "manual"
assert host_result["errors"]["base"] == "cannot_connect"
async def test_manual_valid_host_unexpected_error(hass: HomeAssistant) -> None:
"""Test valid host entered."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
form_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
{"next_step_id": "manual"},
)
with mock_discovery(None):
host_result = await hass.config_entries.flow.async_configure(
form_result["flow_id"],
user_input={CONF_HOST: "sample-host-name"},
)
assert host_result["step_id"] == "manual"
assert host_result["errors"]["base"] == "unknown"
async def test_eiscp_discovery_no_devices_found(hass: HomeAssistant) -> None:
"""Test eiscp discovery with no devices found."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
with mock_discovery([]):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"next_step_id": "eiscp_discovery"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_eiscp_discovery_unexpected_exception(hass: HomeAssistant) -> None:
"""Test eiscp discovery with an unexpected exception."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
with mock_discovery(None):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"next_step_id": "eiscp_discovery"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown"
@pytest.mark.usefixtures("mock_setup_entry") @pytest.mark.usefixtures("mock_setup_entry")
async def test_eiscp_discovery( async def test_manual(hass: HomeAssistant) -> None:
hass: HomeAssistant, """Test successful manual."""
mock_config_entry: MockConfigEntry,
) -> None:
"""Test eiscp discovery."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN, context={"source": SOURCE_USER}
context={"source": SOURCE_USER},
) )
with mock_discovery([RECEIVER_INFO, RECEIVER_INFO_2]): assert result["type"] is FlowResultType.MENU
result = await hass.config_entries.flow.async_configure( assert result["step_id"] == "user"
result["flow_id"],
{"next_step_id": "eiscp_discovery"},
)
assert result["type"] is FlowResultType.FORM
assert result["data_schema"] is not None
schema = result["data_schema"].schema
container = schema["device"].container
assert container == {RECEIVER_INFO_2.identifier: _entry_title(RECEIVER_INFO_2)}
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"], {"next_step_id": "manual"}
user_input={"device": RECEIVER_INFO_2.identifier},
) )
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO_2.host}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_receiver" assert result["step_id"] == "configure_receiver"
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={
"volume_resolution": 200, OPTION_VOLUME_RESOLUTION: 200,
"input_sources": ["TV"], OPTION_INPUT_SOURCES: ["TV"],
"listening_modes": ["THX"], OPTION_LISTENING_MODES: ["THX"],
}, },
) )
assert result["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"]["host"] == RECEIVER_INFO_2.host assert result["data"][CONF_HOST] == RECEIVER_INFO_2.host
assert result["result"].unique_id == RECEIVER_INFO_2.identifier assert result["result"].unique_id == RECEIVER_INFO_2.identifier
assert result["title"] == RECEIVER_INFO_2.model_name
@pytest.mark.parametrize(
("mock_discovery", "error_reason"),
[
(mock_discovery(None), "unknown"),
(mock_discovery([]), "cannot_connect"),
(mock_discovery([RECEIVER_INFO]), "cannot_connect"),
],
)
@pytest.mark.usefixtures("mock_setup_entry") @pytest.mark.usefixtures("mock_setup_entry")
async def test_ssdp_discovery_success(hass: HomeAssistant) -> None: async def test_manual_recoverable_error(
"""Test SSDP discovery with valid host.""" hass: HomeAssistant, mock_discovery: AbstractContextManager, error_reason: str
discovery_info = SsdpServiceInfo( ) -> None:
ssdp_location="http://192.168.0.101:8080", """Test manual with a recoverable error."""
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"}, result = await hass.config_entries.flow.async_init(
ssdp_usn="uuid:mock_usn", DOMAIN, context={"source": SOURCE_USER}
ssdp_udn="uuid:00000000-0000-0000-0000-000000000000",
ssdp_st="mock_st",
) )
result = await hass.config_entries.flow.async_init( assert result["type"] is FlowResultType.MENU
DOMAIN, assert result["step_id"] == "user"
context={"source": config_entries.SOURCE_SSDP},
data=discovery_info, result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "manual"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
with mock_discovery:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO_2.host}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
assert result["errors"] == {"base": error_reason}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO_2.host}
) )
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
@ -214,160 +117,234 @@ async def test_ssdp_discovery_success(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={
"volume_resolution": 200, OPTION_VOLUME_RESOLUTION: 200,
"input_sources": ["TV"], OPTION_INPUT_SOURCES: ["TV"],
"listening_modes": ["THX"], OPTION_LISTENING_MODES: ["THX"],
}, },
) )
assert result["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"]["host"] == RECEIVER_INFO.host assert result["data"][CONF_HOST] == RECEIVER_INFO_2.host
assert result["result"].unique_id == RECEIVER_INFO.identifier assert result["result"].unique_id == RECEIVER_INFO_2.identifier
assert result["title"] == RECEIVER_INFO_2.model_name
async def test_ssdp_discovery_already_configured( @pytest.mark.usefixtures("mock_setup_entry")
async def test_manual_error(
hass: HomeAssistant, mock_config_entry: MockConfigEntry hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None: ) -> None:
"""Test SSDP discovery with already configured device.""" """Test manual with an error."""
mock_config_entry.add_to_hass(hass) await setup_integration(hass, mock_config_entry)
discovery_info = SsdpServiceInfo(
ssdp_location="http://192.168.0.101:8080",
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"},
ssdp_usn="uuid:mock_usn",
ssdp_udn="uuid:00000000-0000-0000-0000-000000000000",
ssdp_st="mock_st",
)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN, context={"source": SOURCE_USER}
context={"source": config_entries.SOURCE_SSDP}, )
data=discovery_info,
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "manual"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO.host}
) )
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
async def test_ssdp_discovery_host_info_error(hass: HomeAssistant) -> None: @pytest.mark.usefixtures("mock_setup_entry")
"""Test SSDP discovery with host info error.""" async def test_eiscp_discovery(
discovery_info = SsdpServiceInfo( hass: HomeAssistant, mock_config_entry: MockConfigEntry
ssdp_location="http://192.168.1.100:8080", ) -> None:
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"}, """Test successful eiscp discovery."""
ssdp_usn="uuid:mock_usn", await setup_integration(hass, mock_config_entry)
ssdp_st="mock_st",
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
) )
with mock_discovery(None): assert result["type"] is FlowResultType.MENU
result = await hass.config_entries.flow.async_init( assert result["step_id"] == "user"
DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, result = await hass.config_entries.flow.async_configure(
data=discovery_info, result["flow_id"], {"next_step_id": "eiscp_discovery"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "eiscp_discovery"
devices = result["data_schema"].schema["device"].container
assert devices == {
RECEIVER_INFO_2.identifier: _receiver_display_name(RECEIVER_INFO_2)
}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"device": RECEIVER_INFO_2.identifier}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_receiver"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
OPTION_VOLUME_RESOLUTION: 200,
OPTION_INPUT_SOURCES: ["TV"],
OPTION_LISTENING_MODES: ["THX"],
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_HOST] == RECEIVER_INFO_2.host
assert result["result"].unique_id == RECEIVER_INFO_2.identifier
assert result["title"] == RECEIVER_INFO_2.model_name
@pytest.mark.parametrize(
("mock_discovery", "error_reason"),
[
(mock_discovery(None), "unknown"),
(mock_discovery([]), "no_devices_found"),
(mock_discovery([RECEIVER_INFO]), "no_devices_found"),
],
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_eiscp_discovery_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_discovery: AbstractContextManager,
error_reason: str,
) -> None:
"""Test eiscp discovery with an error."""
await setup_integration(hass, mock_config_entry)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "user"
with mock_discovery:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"next_step_id": "eiscp_discovery"}
) )
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown" assert result["reason"] == error_reason
async def test_ssdp_discovery_host_none_info(hass: HomeAssistant) -> None: @pytest.mark.usefixtures("mock_setup_entry")
"""Test SSDP discovery with host info error.""" async def test_ssdp_discovery(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test successful SSDP discovery."""
await setup_integration(hass, mock_config_entry)
discovery_info = SsdpServiceInfo( discovery_info = SsdpServiceInfo(
ssdp_location="http://192.168.1.100:8080", ssdp_location=f"http://{RECEIVER_INFO_2.host}:8080",
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"},
ssdp_usn="uuid:mock_usn",
ssdp_st="mock_st",
)
with mock_discovery([]):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_ssdp_discovery_no_location(hass: HomeAssistant) -> None:
"""Test SSDP discovery with no location."""
discovery_info = SsdpServiceInfo(
ssdp_location=None,
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"}, upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"},
ssdp_usn="uuid:mock_usn", ssdp_usn="uuid:mock_usn",
ssdp_udn="uuid:00000000-0000-0000-0000-000000000000",
ssdp_st="mock_st", ssdp_st="mock_st",
) )
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=discovery_info
context={"source": config_entries.SOURCE_SSDP},
data=discovery_info,
) )
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.FORM
assert result["reason"] == "unknown" assert result["step_id"] == "configure_receiver"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
OPTION_VOLUME_RESOLUTION: 200,
OPTION_INPUT_SOURCES: ["TV"],
OPTION_LISTENING_MODES: ["THX"],
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_HOST] == RECEIVER_INFO_2.host
assert result["result"].unique_id == RECEIVER_INFO_2.identifier
assert result["title"] == RECEIVER_INFO_2.model_name
async def test_ssdp_discovery_no_host(hass: HomeAssistant) -> None: @pytest.mark.parametrize(
"""Test SSDP discovery with no host.""" ("ssdp_location", "mock_discovery", "error_reason"),
[
(None, nullcontext(), "unknown"),
("http://", nullcontext(), "unknown"),
(f"http://{RECEIVER_INFO_2.host}:8080", mock_discovery(None), "unknown"),
(f"http://{RECEIVER_INFO_2.host}:8080", mock_discovery([]), "cannot_connect"),
(
f"http://{RECEIVER_INFO_2.host}:8080",
mock_discovery([RECEIVER_INFO]),
"cannot_connect",
),
(f"http://{RECEIVER_INFO.host}:8080", nullcontext(), "already_configured"),
],
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_ssdp_discovery_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
ssdp_location: str | None,
mock_discovery: AbstractContextManager,
error_reason: str,
) -> None:
"""Test SSDP discovery with an error."""
await setup_integration(hass, mock_config_entry)
discovery_info = SsdpServiceInfo( discovery_info = SsdpServiceInfo(
ssdp_location="http://", ssdp_location=ssdp_location,
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"}, upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"},
ssdp_usn="uuid:mock_usn", ssdp_usn="uuid:mock_usn",
ssdp_udn="uuid:00000000-0000-0000-0000-000000000000",
ssdp_st="mock_st", ssdp_st="mock_st",
) )
result = await hass.config_entries.flow.async_init( with mock_discovery:
DOMAIN, result = await hass.config_entries.flow.async_init(
context={"source": config_entries.SOURCE_SSDP}, DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=discovery_info
data=discovery_info, )
)
assert result["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unknown" assert result["reason"] == error_reason
async def test_configure_no_resolution(hass: HomeAssistant) -> None:
"""Test receiver configure with no resolution set."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
form_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
{"next_step_id": "manual"},
)
select_result = await hass.config_entries.flow.async_configure(
form_result["flow_id"],
user_input={CONF_HOST: "sample-host-name"},
)
with pytest.raises(InvalidData):
await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"input_sources": ["TV"]},
)
@pytest.mark.usefixtures("mock_setup_entry") @pytest.mark.usefixtures("mock_setup_entry")
async def test_configure(hass: HomeAssistant) -> None: async def test_configure(hass: HomeAssistant) -> None:
"""Test receiver configure.""" """Test receiver configure."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN, context={"source": SOURCE_USER}
context={"source": SOURCE_USER},
) )
result = await hass.config_entries.flow.async_configure( assert result["type"] is FlowResultType.MENU
result["flow_id"], assert result["step_id"] == "user"
{"next_step_id": "manual"},
)
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"], {"next_step_id": "manual"}
user_input={CONF_HOST: RECEIVER_INFO.host}, )
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO.host}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_receiver"
assert result["description_placeholders"]["name"] == _receiver_display_name(
RECEIVER_INFO
) )
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
@ -378,6 +355,8 @@ async def test_configure(hass: HomeAssistant) -> None:
OPTION_LISTENING_MODES: ["THX"], OPTION_LISTENING_MODES: ["THX"],
}, },
) )
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_receiver" assert result["step_id"] == "configure_receiver"
assert result["errors"] == {OPTION_INPUT_SOURCES: "empty_input_source_list"} assert result["errors"] == {OPTION_INPUT_SOURCES: "empty_input_source_list"}
@ -389,6 +368,8 @@ async def test_configure(hass: HomeAssistant) -> None:
OPTION_LISTENING_MODES: [], OPTION_LISTENING_MODES: [],
}, },
) )
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "configure_receiver" assert result["step_id"] == "configure_receiver"
assert result["errors"] == {OPTION_LISTENING_MODES: "empty_listening_mode_list"} assert result["errors"] == {OPTION_LISTENING_MODES: "empty_listening_mode_list"}
@ -400,6 +381,7 @@ async def test_configure(hass: HomeAssistant) -> None:
OPTION_LISTENING_MODES: ["THX"], OPTION_LISTENING_MODES: ["THX"],
}, },
) )
assert result["type"] is FlowResultType.CREATE_ENTRY assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["options"] == { assert result["options"] == {
OPTION_VOLUME_RESOLUTION: 200, OPTION_VOLUME_RESOLUTION: 200,
@ -409,36 +391,11 @@ async def test_configure(hass: HomeAssistant) -> None:
} }
async def test_configure_invalid_resolution_set(hass: HomeAssistant) -> None:
"""Test receiver configure with invalid resolution."""
init_result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
form_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
{"next_step_id": "manual"},
)
select_result = await hass.config_entries.flow.async_configure(
form_result["flow_id"],
user_input={CONF_HOST: "sample-host-name"},
)
with pytest.raises(InvalidData):
await hass.config_entries.flow.async_configure(
select_result["flow_id"],
user_input={"volume_resolution": 42, "input_sources": ["TV"]},
)
@pytest.mark.usefixtures("mock_setup_entry") @pytest.mark.usefixtures("mock_setup_entry")
async def test_reconfigure( async def test_reconfigure(
hass: HomeAssistant, mock_config_entry: MockConfigEntry hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None: ) -> None:
"""Test the reconfigure config flow.""" """Test successful reconfigure flow."""
await setup_integration(hass, mock_config_entry) await setup_integration(hass, mock_config_entry)
old_host = mock_config_entry.data[CONF_HOST] old_host = mock_config_entry.data[CONF_HOST]
@ -449,21 +406,19 @@ async def test_reconfigure(
assert result["type"] is FlowResultType.FORM assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "manual" assert result["step_id"] == "manual"
result2 = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={"host": mock_config_entry.data[CONF_HOST]} result["flow_id"], user_input={CONF_HOST: mock_config_entry.data[CONF_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={OPTION_VOLUME_RESOLUTION: 200},
) )
assert result3["type"] is FlowResultType.ABORT assert result["type"] is FlowResultType.FORM
assert result3["reason"] == "reconfigure_successful" assert result["step_id"] == "configure_receiver"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={OPTION_VOLUME_RESOLUTION: 200}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_config_entry.data[CONF_HOST] == old_host assert mock_config_entry.data[CONF_HOST] == old_host
assert mock_config_entry.options[OPTION_VOLUME_RESOLUTION] == 200 assert mock_config_entry.options[OPTION_VOLUME_RESOLUTION] == 200
@ -474,24 +429,25 @@ async def test_reconfigure(
@pytest.mark.usefixtures("mock_setup_entry") @pytest.mark.usefixtures("mock_setup_entry")
async def test_reconfigure_new_device( async def test_reconfigure_error(
hass: HomeAssistant, mock_config_entry: MockConfigEntry hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None: ) -> None:
"""Test the reconfigure config flow with new device.""" """Test reconfigure flow with an error."""
await setup_integration(hass, mock_config_entry) await setup_integration(hass, mock_config_entry)
old_unique_id = mock_config_entry.unique_id old_unique_id = mock_config_entry.unique_id
result = await mock_config_entry.start_reconfigure_flow(hass) result = await mock_config_entry.start_reconfigure_flow(hass)
with mock_discovery([RECEIVER_INFO_2]): assert result["type"] is FlowResultType.FORM
result2 = await hass.config_entries.flow.async_configure( assert result["step_id"] == "manual"
result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO_2.host}
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT result = await hass.config_entries.flow.async_configure(
assert result2["reason"] == "unique_id_mismatch" result["flow_id"], user_input={CONF_HOST: RECEIVER_INFO_2.host}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unique_id_mismatch"
# unique id should remain unchanged # unique id should remain unchanged
assert mock_config_entry.unique_id == old_unique_id assert mock_config_entry.unique_id == old_unique_id
@ -519,6 +475,9 @@ async def test_options_flow(
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure( result = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
user_input={ user_input={