Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Discovery ignores DLNA DMR devices when they are better supported by another integration (#57363)

pull/58225/head^2
Michael Chisholm 2021-10-23 07:26:33 +11:00 committed by GitHub
parent a195418dd3
commit ee087c7a05
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 365 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -14,7 +14,13 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.const import CONF_DEVICE_ID, CONF_NAME, CONF_TYPE, CONF_URL
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_HOST,
CONF_NAME,
CONF_TYPE,
CONF_URL,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import IntegrationError
@ -51,7 +57,7 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize flow."""
self._discoveries: list[Mapping[str, str]] = []
self._discoveries: dict[str, Mapping[str, Any]] = {}
self._location: str | None = None
self._udn: str | None = None
self._device_type: str | None = None
@ -67,13 +73,43 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return DlnaDmrOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input: FlowInput = None) -> FlowResult:
"""Handle a flow initialized by the user: manual URL entry.
"""Handle a flow initialized by the user.
Discovered devices will already be displayed, no need to prompt user
with them here.
Let user choose from a list of found and unconfigured devices or to
enter an URL manually.
"""
LOGGER.debug("async_step_user: user_input: %s", user_input)
if user_input is not None:
host = user_input.get(CONF_HOST)
if not host:
# No device chosen, user might want to directly enter an URL
return await self.async_step_manual()
# User has chosen a device, ask for confirmation
discovery = self._discoveries[host]
await self._async_set_info_from_discovery(discovery)
return self._create_entry()
discoveries = await self._async_get_discoveries()
if not discoveries:
# Nothing found, maybe the user knows an URL to try
return await self.async_step_manual()
self._discoveries = {
discovery.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
or urlparse(discovery[ssdp.ATTR_SSDP_LOCATION]).hostname: discovery
for discovery in discoveries
}
data_schema = vol.Schema(
{vol.Optional(CONF_HOST): vol.In(self._discoveries.keys())}
)
return self.async_show_form(step_id="user", data_schema=data_schema)
async def async_step_manual(self, user_input: FlowInput = None) -> FlowResult:
"""Manual URL entry by the user."""
LOGGER.debug("async_step_manual: user_input: %s", user_input)
# Device setup manually, assume we don't get SSDP broadcast notifications
self._options[CONF_POLL_AVAILABILITY] = True
@ -89,7 +125,7 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
data_schema = vol.Schema({CONF_URL: str})
return self.async_show_form(
step_id="user", data_schema=data_schema, errors=errors
step_id="manual", data_schema=data_schema, errors=errors
)
async def async_step_import(self, import_data: FlowInput = None) -> FlowResult:
@ -177,6 +213,9 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
await self._async_set_info_from_discovery(discovery_info)
if _is_ignored_device(discovery_info):
return self.async_abort(reason="alternative_integration")
# Abort if a migration flow for the device's location is in progress
for progress in self._async_in_progress(include_uninitialized=True):
if progress["context"].get("unique_id") == self._location:
@ -190,6 +229,29 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm()
async def async_step_unignore(self, user_input: Mapping[str, Any]) -> FlowResult:
"""Rediscover previously ignored devices by their unique_id."""
LOGGER.debug("async_step_unignore: user_input: %s", user_input)
self._udn = user_input["unique_id"]
assert self._udn
await self.async_set_unique_id(self._udn)
# Find a discovery matching the unignored unique_id for a DMR device
for dev_type in DmrDevice.DEVICE_TYPES:
discovery = await ssdp.async_get_discovery_info_by_udn_st(
self.hass, self._udn, dev_type
)
if discovery:
break
else:
return self.async_abort(reason="discovery_error")
await self._async_set_info_from_discovery(discovery, abort_if_configured=False)
self.context["title_placeholders"] = {"name": self._name}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input: FlowInput = None) -> FlowResult:
"""Allow the user to confirm adding the device."""
LOGGER.debug("async_step_confirm: %s", user_input)
@ -213,7 +275,7 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
try:
device = await domain_data.upnp_factory.async_create_device(self._location)
except UpnpError as err:
raise ConnectError("could_not_connect") from err
raise ConnectError("cannot_connect") from err
try:
device = find_device_of_type(device, DmrDevice.DEVICE_TYPES)
@ -284,12 +346,12 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
or DEFAULT_NAME
)
async def _async_get_discoveries(self) -> list[Mapping[str, str]]:
async def _async_get_discoveries(self) -> list[Mapping[str, Any]]:
"""Get list of unconfigured DLNA devices discovered by SSDP."""
LOGGER.debug("_get_discoveries")
# Get all compatible devices from ssdp's cache
discoveries: list[Mapping[str, str]] = []
discoveries: list[Mapping[str, Any]] = []
for udn_st in DmrDevice.DEVICE_TYPES:
st_discoveries = await ssdp.async_get_discovery_info_by_st(
self.hass, udn_st
@ -298,7 +360,8 @@ class DlnaDmrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
# Filter out devices already configured
current_unique_ids = {
entry.unique_id for entry in self._async_current_entries()
entry.unique_id
for entry in self._async_current_entries(include_ignore=False)
}
discoveries = [
disc
@ -374,3 +437,25 @@ class DlnaDmrOptionsFlowHandler(config_entries.OptionsFlow):
data_schema=vol.Schema(fields),
errors=errors,
)
def _is_ignored_device(discovery_info: Mapping[str, Any]) -> bool:
"""Return True if this device should be ignored for discovery.
These devices are supported better by other integrations, so don't bug
the user about them. The user can add them if desired by via the user config
flow, which will list all discovered but unconfigured devices.
"""
# Did the discovery trigger more than just this flow?
if len(discovery_info.get(ssdp.ATTR_HA_MATCHING_DOMAINS, set())) > 1:
LOGGER.debug(
"Ignoring device supported by multiple integrations: %s",
discovery_info[ssdp.ATTR_HA_MATCHING_DOMAINS],
)
return True
# Is the root device not a DMR?
if discovery_info.get(ssdp.ATTR_UPNP_DEVICE_TYPE) not in DmrDevice.DEVICE_TYPES:
return True
return False

26
homeassistant/components/dlna_dmr/manifest.json
View File

@ -5,6 +5,32 @@
"documentation": "https://www.home-assistant.io/integrations/dlna_dmr",
"requirements": ["async-upnp-client==0.22.9"],
"dependencies": ["ssdp"],
"ssdp": [
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:1",
"st": "urn:schemas-upnp-org:device:MediaRenderer:1"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:2",
"st": "urn:schemas-upnp-org:device:MediaRenderer:2"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:3",
"st": "urn:schemas-upnp-org:device:MediaRenderer:3"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:1",
"nt": "urn:schemas-upnp-org:device:MediaRenderer:1"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:2",
"nt": "urn:schemas-upnp-org:device:MediaRenderer:2"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:3",
"nt": "urn:schemas-upnp-org:device:MediaRenderer:3"
}
],
"codeowners": ["@StevenLooman", "@chishm"],
"iot_class": "local_push"
}

14
homeassistant/components/dlna_dmr/strings.json
View File

@ -3,7 +3,14 @@
"flow_title": "{name}",
"step": {
"user": {
"title": "DLNA Digital Media Renderer",
"title": "Discovered DLNA DMR devices",
"description": "Choose a device to configure or leave blank to enter a URL",
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
},
"manual": {
"title": "Manual DLNA DMR device connection",
"description": "URL to a device description XML file",
"data": {
"url": "[%key:common::config_flow::data::url%]"
@ -18,14 +25,15 @@
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"could_not_connect": "Failed to connect to DLNA device",
"alternative_integration": "Device is better supported by another integration",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"discovery_error": "Failed to discover a matching DLNA device",
"incomplete_config": "Configuration is missing a required variable",
"non_unique_id": "Multiple devices found with the same unique ID",
"not_dmr": "Device is not a Digital Media Renderer"
},
"error": {
"could_not_connect": "Failed to connect to DLNA device",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"not_dmr": "Device is not a Digital Media Renderer"
}
},

16
homeassistant/components/dlna_dmr/translations/en.json
View File

@ -2,14 +2,15 @@
"config": {
"abort": {
"already_configured": "Device is already configured",
"could_not_connect": "Failed to connect to DLNA device",
"alternative_integration": "Device is better supported by another integration",
"cannot_connect": "Failed to connect",
"discovery_error": "Failed to discover a matching DLNA device",
"incomplete_config": "Configuration is missing a required variable",
"non_unique_id": "Multiple devices found with the same unique ID",
"not_dmr": "Device is not a Digital Media Renderer"
},
"error": {
"could_not_connect": "Failed to connect to DLNA device",
"cannot_connect": "Failed to connect",
"not_dmr": "Device is not a Digital Media Renderer"
},
"flow_title": "{name}",
@ -20,12 +21,19 @@
"import_turn_on": {
"description": "Please turn on the device and click submit to continue migration"
},
"user": {
"manual": {
"data": {
"url": "URL"
},
"description": "URL to a device description XML file",
"title": "DLNA Digital Media Renderer"
"title": "Manual DLNA DMR device connection"
},
"user": {
"data": {
"host": "Host"
},
"description": "Choose a device to configure or leave blank to enter a URL",
"title": "Discovered DLNA DMR devices"
}
}
},

3
homeassistant/components/ssdp/__init__.py
View File

@ -54,6 +54,8 @@ ATTR_UPNP_SERIAL = "serialNumber"
ATTR_UPNP_UDN = "UDN"
ATTR_UPNP_UPC = "UPC"
ATTR_UPNP_PRESENTATION_URL = "presentationURL"
# Attributes for accessing info added by Home Assistant
ATTR_HA_MATCHING_DOMAINS = "x-homeassistant-matching-domains"
PRIMARY_MATCH_KEYS = [ATTR_UPNP_MANUFACTURER, "st", ATTR_UPNP_DEVICE_TYPE, "nt"]
@ -398,6 +400,7 @@ class Scanner:
return
discovery_info = discovery_info_from_headers_and_description(info_with_desc)
discovery_info[ATTR_HA_MATCHING_DOMAINS] = matching_domains
ssdp_change = SSDP_SOURCE_SSDP_CHANGE_MAPPING[source]
await _async_process_callbacks(callbacks, discovery_info, ssdp_change)

26
homeassistant/generated/ssdp.py
View File

@ -83,6 +83,32 @@ SSDP = {
"manufacturer": "DIRECTV"
}
],
"dlna_dmr": [
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:1",
"st": "urn:schemas-upnp-org:device:MediaRenderer:1"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:2",
"st": "urn:schemas-upnp-org:device:MediaRenderer:2"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:3",
"st": "urn:schemas-upnp-org:device:MediaRenderer:3"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:1",
"nt": "urn:schemas-upnp-org:device:MediaRenderer:1"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:2",
"nt": "urn:schemas-upnp-org:device:MediaRenderer:2"
},
{
"deviceType": "urn:schemas-upnp-org:device:MediaRenderer:3",
"nt": "urn:schemas-upnp-org:device:MediaRenderer:3"
}
],
"fritz": [
{
"st": "urn:schemas-upnp-org:device:fritzbox:1"

204
tests/components/dlna_dmr/test_config_flow.py
View File

@ -16,6 +16,7 @@ from homeassistant.components.dlna_dmr.const import (
)
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_HOST,
CONF_NAME,
CONF_PLATFORM,
CONF_TYPE,
@ -49,26 +50,26 @@ MOCK_CONFIG_IMPORT_DATA = {
}
MOCK_ROOT_DEVICE_UDN = "ROOT_DEVICE"
MOCK_ROOT_DEVICE_TYPE = "ROOT_DEVICE_TYPE"
MOCK_DISCOVERY = {
ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION,
ssdp.ATTR_SSDP_UDN: MOCK_DEVICE_UDN,
ssdp.ATTR_SSDP_ST: MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_UDN: MOCK_ROOT_DEVICE_UDN,
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_ROOT_DEVICE_TYPE,
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME,
ssdp.ATTR_HA_MATCHING_DOMAINS: {DLNA_DOMAIN},
}
async def test_user_flow(hass: HomeAssistant) -> None:
"""Test user-init'd config flow with user entering a valid URL."""
async def test_user_flow_undiscovered_manual(hass: HomeAssistant) -> None:
"""Test user-init'd flow, no discovered devices, user entering a valid URL."""
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
assert result["step_id"] == "user"
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_URL: MOCK_DEVICE_LOCATION}
@ -87,6 +88,79 @@ async def test_user_flow(hass: HomeAssistant) -> None:
await hass.async_block_till_done()
async def test_user_flow_discovered_manual(
hass: HomeAssistant, ssdp_scanner_mock: Mock
) -> None:
"""Test user-init'd flow, with discovered devices, user entering a valid URL."""
ssdp_scanner_mock.async_get_discovery_info_by_st.side_effect = [
[MOCK_DISCOVERY],
[],
[],
]
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_URL: MOCK_DEVICE_LOCATION}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == MOCK_DEVICE_NAME
assert result["data"] == {
CONF_URL: MOCK_DEVICE_LOCATION,
CONF_DEVICE_ID: MOCK_DEVICE_UDN,
CONF_TYPE: MOCK_DEVICE_TYPE,
}
assert result["options"] == {CONF_POLL_AVAILABILITY: True}
# Wait for platform to be fully setup
await hass.async_block_till_done()
async def test_user_flow_selected(hass: HomeAssistant, ssdp_scanner_mock: Mock) -> None:
"""Test user-init'd flow, user selects discovered device."""
ssdp_scanner_mock.async_get_discovery_info_by_st.side_effect = [
[MOCK_DISCOVERY],
[],
[],
]
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: MOCK_DEVICE_NAME}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == MOCK_DEVICE_NAME
assert result["data"] == {
CONF_URL: MOCK_DEVICE_LOCATION,
CONF_DEVICE_ID: MOCK_DEVICE_UDN,
CONF_TYPE: MOCK_DEVICE_TYPE,
}
assert result["options"] == {}
await hass.async_block_till_done()
async def test_user_flow_uncontactable(
hass: HomeAssistant, domain_data_mock: Mock
) -> None:
@ -99,15 +173,15 @@ async def test_user_flow_uncontactable(
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
assert result["step_id"] == "user"
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_URL: MOCK_DEVICE_LOCATION}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "could_not_connect"}
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
assert result["step_id"] == "manual"
async def test_user_flow_embedded_st(
@ -117,7 +191,7 @@ async def test_user_flow_embedded_st(
# Device is the wrong type
upnp_device = domain_data_mock.upnp_factory.async_create_device.return_value
upnp_device.udn = MOCK_ROOT_DEVICE_UDN
upnp_device.device_type = MOCK_ROOT_DEVICE_TYPE
upnp_device.device_type = "ROOT_DEVICE_TYPE"
upnp_device.name = "ROOT_DEVICE_NAME"
embedded_device = Mock(spec=UpnpDevice)
embedded_device.udn = MOCK_DEVICE_UDN
@ -130,7 +204,7 @@ async def test_user_flow_embedded_st(
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
assert result["step_id"] == "user"
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_URL: MOCK_DEVICE_LOCATION}
@ -160,7 +234,7 @@ async def test_user_flow_wrong_st(hass: HomeAssistant, domain_data_mock: Mock) -
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {}
assert result["step_id"] == "user"
assert result["step_id"] == "manual"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_URL: MOCK_DEVICE_LOCATION}
@ -168,7 +242,7 @@ async def test_user_flow_wrong_st(hass: HomeAssistant, domain_data_mock: Mock) -
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "not_dmr"}
assert result["step_id"] == "user"
assert result["step_id"] == "manual"
async def test_import_flow_invalid(hass: HomeAssistant, domain_data_mock: Mock) -> None:
@ -298,7 +372,7 @@ async def test_import_flow_offline(
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "could_not_connect"}
assert result["errors"] == {"base": "cannot_connect"}
assert result["step_id"] == "import_turn_on"
# Device is discovered via SSDP, new flow should not be initialized
@ -469,7 +543,7 @@ async def test_ssdp_flow_upnp_udn(
ssdp.ATTR_SSDP_UDN: MOCK_DEVICE_UDN,
ssdp.ATTR_SSDP_ST: MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_UDN: "DIFFERENT_ROOT_DEVICE",
ssdp.ATTR_UPNP_DEVICE_TYPE: "DIFFERENT_ROOT_DEVICE_TYPE",
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME,
},
)
@ -478,6 +552,108 @@ async def test_ssdp_flow_upnp_udn(
assert config_entry_mock.data[CONF_URL] == NEW_DEVICE_LOCATION
async def test_ssdp_ignore_device(hass: HomeAssistant) -> None:
"""Test SSDP discovery ignores certain devices."""
discovery = MOCK_DISCOVERY.copy()
discovery[ssdp.ATTR_HA_MATCHING_DOMAINS] = {DLNA_DOMAIN, "other_domain"}
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=discovery,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "alternative_integration"
discovery = MOCK_DISCOVERY.copy()
discovery[ssdp.ATTR_UPNP_DEVICE_TYPE] = "urn:schemas-upnp-org:device:ZonePlayer:1"
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=discovery,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "alternative_integration"
async def test_unignore_flow(hass: HomeAssistant, ssdp_scanner_mock: Mock) -> None:
"""Test a config flow started by unignoring a device."""
# Create ignored entry
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN,
context={"source": config_entries.SOURCE_IGNORE},
data={"unique_id": MOCK_DEVICE_UDN, "title": MOCK_DEVICE_NAME},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == MOCK_DEVICE_NAME
assert result["data"] == {}
# Device was found via SSDP, matching the 2nd device type tried
ssdp_scanner_mock.async_get_discovery_info_by_udn_st.side_effect = [
None,
MOCK_DISCOVERY,
None,
None,
None,
]
# Unignore it and expect config flow to start
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN,
context={"source": config_entries.SOURCE_UNIGNORE},
data={"unique_id": MOCK_DEVICE_UDN},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == MOCK_DEVICE_NAME
assert result["data"] == {
CONF_URL: MOCK_DEVICE_LOCATION,
CONF_DEVICE_ID: MOCK_DEVICE_UDN,
CONF_TYPE: MOCK_DEVICE_TYPE,
}
assert result["options"] == {}
# Wait for platform to be fully setup
await hass.async_block_till_done()
async def test_unignore_flow_offline(
hass: HomeAssistant, ssdp_scanner_mock: Mock
) -> None:
"""Test a config flow started by unignoring a device, but the device is offline."""
# Create ignored entry
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN,
context={"source": config_entries.SOURCE_IGNORE},
data={"unique_id": MOCK_DEVICE_UDN, "title": MOCK_DEVICE_NAME},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == MOCK_DEVICE_NAME
assert result["data"] == {}
# Device is not in the SSDP discoveries (perhaps HA restarted between ignore and unignore)
ssdp_scanner_mock.async_get_discovery_info_by_udn_st.return_value = None
# Unignore it and expect config flow to start then abort
result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN,
context={"source": config_entries.SOURCE_UNIGNORE},
data={"unique_id": MOCK_DEVICE_UDN},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "discovery_error"
async def test_options_flow(
hass: HomeAssistant, config_entry_mock: MockConfigEntry
) -> None:

2
tests/components/ssdp/test_init.py
View File

@ -71,6 +71,7 @@ async def test_ssdp_flow_dispatched_on_st(mock_get_ssdp, hass, caplog, mock_flow
ssdp.ATTR_UPNP_UDN: "uuid:mock-udn",
ssdp.ATTR_SSDP_UDN: ANY,
"_timestamp": ANY,
ssdp.ATTR_HA_MATCHING_DOMAINS: {"mock-domain"},
}
assert "Failed to fetch ssdp data" not in caplog.text
@ -463,6 +464,7 @@ async def test_scan_with_registered_callback(
"x-rincon-bootseq": "55",
ssdp.ATTR_SSDP_UDN: ANY,
"_timestamp": ANY,
ssdp.ATTR_HA_MATCHING_DOMAINS: set(),
},
ssdp.SsdpChange.ALIVE,
)