jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 09:47:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate ssdp to config_flow for frontier_silicon (#89496)

Browse Source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Thijs W 2023-03-29 04:06:21 +02:00 committed by GitHub
parent 47a2598b66
commit e3cad8baac
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 221 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
homeassistant/components/discovery/__init__.py
View File

@ -60,7 +60,6 @@ class ServiceDetails(NamedTuple):
SERVICE_HANDLERS = {
SERVICE_ENIGMA2: ServiceDetails("media_player", "enigma2"),
"yamaha": ServiceDetails("media_player", "yamaha"),
"frontier_silicon": ServiceDetails("media_player", "frontier_silicon"),
"openhome": ServiceDetails("media_player", "openhome"),
"bluesound": ServiceDetails("media_player", "bluesound"),
}

97
homeassistant/components/frontier_silicon/config_flow.py
View File

@ -3,15 +3,24 @@ from __future__ import annotations
import logging
from typing import Any
from urllib.parse import urlparse
from afsapi import AFSAPI, ConnectionError as FSConnectionError, InvalidPinException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
from homeassistant.data_entry_flow import FlowResult
from .const import CONF_PIN, CONF_WEBFSAPI_URL, DEFAULT_PIN, DEFAULT_PORT, DOMAIN
from .const import (
CONF_PIN,
CONF_WEBFSAPI_URL,
DEFAULT_PIN,
DEFAULT_PORT,
DOMAIN,
SSDP_ATTR_SPEAKER_NAME,
)
_LOGGER = logging.getLogger(__name__)
@ -32,11 +41,17 @@ STEP_DEVICE_CONFIG_DATA_SCHEMA = vol.Schema(
)
def hostname_from_url(url: str) -> str:
"""Return the hostname from a url."""
return str(urlparse(url).hostname)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Frontier Silicon Media Player."""
VERSION = 1
_name: str
_webfsapi_url: str
async def async_step_import(self, import_info: dict[str, Any]) -> FlowResult:
@ -101,6 +116,46 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=data_schema, errors=errors
)
async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo) -> FlowResult:
"""Process entity discovered via SSDP."""
device_url = discovery_info.ssdp_location
if device_url is None:
return self.async_abort(reason="cannot_connect")
device_hostname = hostname_from_url(device_url)
for entry in self._async_current_entries(include_ignore=False):
if device_hostname == hostname_from_url(entry.data[CONF_WEBFSAPI_URL]):
return self.async_abort(reason="already_configured")
speaker_name = discovery_info.ssdp_headers.get(SSDP_ATTR_SPEAKER_NAME)
self.context["title_placeholders"] = {"name": speaker_name}
try:
self._webfsapi_url = await AFSAPI.get_webfsapi_endpoint(device_url)
except FSConnectionError:
return self.async_abort(reason="cannot_connect")
except Exception as exception: # pylint: disable=broad-except
_LOGGER.debug(exception)
return self.async_abort(reason="unknown")
try:
# try to login with default pin
afsapi = AFSAPI(self._webfsapi_url, DEFAULT_PIN)
unique_id = await afsapi.get_radio_id()
except InvalidPinException:
return self.async_abort(reason="invalid_auth")
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured(
updates={CONF_WEBFSAPI_URL: self._webfsapi_url}, reload_on_update=True
)
self._name = await afsapi.get_friendly_name()
return await self.async_step_confirm()
async def _async_step_device_config_if_needed(self) -> FlowResult:
"""Most users will not have changed the default PIN on their radio.
@ -111,21 +166,29 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# try to login with default pin
afsapi = AFSAPI(self._webfsapi_url, DEFAULT_PIN)
name = await afsapi.get_friendly_name()
self._name = await afsapi.get_friendly_name()
except InvalidPinException:
# Ask for a PIN
return await self.async_step_device_config()
self.context["title_placeholders"] = {"name": name}
self.context["title_placeholders"] = {"name": self._name}
unique_id = await afsapi.get_radio_id()
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=name,
data={CONF_WEBFSAPI_URL: self._webfsapi_url, CONF_PIN: DEFAULT_PIN},
)
return await self._async_create_entry()
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Allow the user to confirm adding the device. Used when the default PIN could successfully be used."""
if user_input is not None:
return await self._async_create_entry()
self._set_confirm_only()
return self.async_show_form(step_id="confirm")
async def async_step_device_config(
self, user_input: dict[str, Any] | None = None
@ -145,7 +208,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
try:
afsapi = AFSAPI(self._webfsapi_url, user_input[CONF_PIN])
name = await afsapi.get_friendly_name()
self._name = await afsapi.get_friendly_name()
except FSConnectionError:
errors["base"] = "cannot_connect"
@ -156,15 +219,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "unknown"
else:
unique_id = await afsapi.get_radio_id()
await self.async_set_unique_id(unique_id)
await self.async_set_unique_id(unique_id, raise_on_progress=False)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=name,
data={
CONF_WEBFSAPI_URL: self._webfsapi_url,
CONF_PIN: user_input[CONF_PIN],
},
)
return await self._async_create_entry(user_input[CONF_PIN])
data_schema = self.add_suggested_values_to_schema(
STEP_DEVICE_CONFIG_DATA_SCHEMA, user_input
@ -174,3 +231,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=data_schema,
errors=errors,
)
async def _async_create_entry(self, pin: str | None = None):
"""Create the entry."""
return self.async_create_entry(
title=self._name,
data={CONF_WEBFSAPI_URL: self._webfsapi_url, CONF_PIN: pin or DEFAULT_PIN},
)

3
homeassistant/components/frontier_silicon/const.py
View File

@ -4,6 +4,9 @@ DOMAIN = "frontier_silicon"
CONF_WEBFSAPI_URL = "webfsapi_url"
CONF_PIN = "pin"
SSDP_ST = "urn:schemas-frontier-silicon-com:undok:fsapi:1"
SSDP_ATTR_SPEAKER_NAME = "SPEAKER-NAME"
DEFAULT_PIN = "1234"
DEFAULT_PORT = 80

3
homeassistant/components/frontier_silicon/manifest.json
View File

@ -5,5 +5,6 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/frontier_silicon",
"iot_class": "local_polling",
"requirements": ["afsapi==0.2.7"]
"requirements": ["afsapi==0.2.7"],
"ssdp": [{ "st": "urn:schemas-frontier-silicon-com:undok:fsapi:1" }]
}

14
homeassistant/components/frontier_silicon/media_player.py
View File

@ -54,21 +54,7 @@ async def async_setup_platform(
"""Set up the Frontier Silicon platform.
YAML is deprecated, and imported automatically.
SSDP discovery is temporarily retained - to be refactor subsequently.
"""
if discovery_info is not None:
webfsapi_url = await AFSAPI.get_webfsapi_endpoint(
discovery_info["ssdp_description"]
)
afsapi = AFSAPI(webfsapi_url, DEFAULT_PIN)
name = await afsapi.get_friendly_name()
async_add_entities(
[AFSAPIDevice(name, afsapi)],
True,
)
return
ir.async_create_issue(
hass,

5
homeassistant/generated/ssdp.py
View File

@ -130,6 +130,11 @@ SSDP = {
"st": "urn:schemas-upnp-org:device:fritzbox:1",
},
],
"frontier_silicon": [
{
"st": "urn:schemas-frontier-silicon-com:undok:fsapi:1",
},
],
"harmony": [
{
"deviceType": "urn:myharmony-com:device:harmony:1",

143
tests/components/frontier_silicon/test_config_flow.py
View File

@ -5,7 +5,12 @@ from afsapi import ConnectionError, InvalidPinException
import pytest
from homeassistant import config_entries
from homeassistant.components.frontier_silicon.const import CONF_WEBFSAPI_URL, DOMAIN
from homeassistant.components import ssdp
from homeassistant.components.frontier_silicon.const import (
CONF_WEBFSAPI_URL,
DEFAULT_PIN,
DOMAIN,
)
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PIN, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -15,6 +20,23 @@ from tests.common import MockConfigEntry
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
MOCK_DISCOVERY = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_udn="uuid:3dcc7100-f76c-11dd-87af-00226124ca30",
ssdp_st="mock_st",
ssdp_location="http://1.1.1.1/device",
upnp={"SPEAKER-NAME": "Speaker Name"},
)
INVALID_MOCK_DISCOVERY = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_udn="uuid:3dcc7100-f76c-11dd-87af-00226124ca30",
ssdp_st="mock_st",
ssdp_location=None,
upnp={"SPEAKER-NAME": "Speaker Name"},
)
async def test_import_success(hass: HomeAssistant) -> None:
"""Test successful import."""
@ -49,7 +71,7 @@ async def test_import_webfsapi_endpoint_failures(
) -> None:
"""Test various failure of get_webfsapi_endpoint."""
with patch(
"afsapi.AFSAPI.get_webfsapi_endpoint",
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_webfsapi_endpoint",
side_effect=webfsapi_endpoint_error,
):
result = await hass.config_entries.flow.async_init(
@ -80,7 +102,7 @@ async def test_import_radio_id_failures(
) -> None:
"""Test various failure of get_radio_id."""
with patch(
"afsapi.AFSAPI.get_radio_id",
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_radio_id",
side_effect=radio_id_error,
):
result = await hass.config_entries.flow.async_init(
@ -157,7 +179,7 @@ async def test_form_nondefault_pin(
assert result["errors"] == {}
with patch(
"afsapi.AFSAPI.get_friendly_name",
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_friendly_name",
side_effect=InvalidPinException,
):
result2 = await hass.config_entries.flow.async_configure(
@ -179,8 +201,8 @@ async def test_form_nondefault_pin(
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Name of the device"
assert result3["data"] == {
"webfsapi_url": "http://1.1.1.1:80/webfsapi",
"pin": "4321",
CONF_WEBFSAPI_URL: "http://1.1.1.1:80/webfsapi",
CONF_PIN: "4321",
}
mock_setup_entry.assert_called_once()
@ -208,7 +230,7 @@ async def test_form_nondefault_pin_invalid(
assert result["errors"] == {}
with patch(
"afsapi.AFSAPI.get_friendly_name",
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_friendly_name",
side_effect=InvalidPinException,
):
result2 = await hass.config_entries.flow.async_configure(
@ -222,7 +244,7 @@ async def test_form_nondefault_pin_invalid(
assert result2["errors"] is None
with patch(
"afsapi.AFSAPI.get_friendly_name",
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_friendly_name",
side_effect=friendly_name_error,
):
result3 = await hass.config_entries.flow.async_configure(
@ -244,8 +266,8 @@ async def test_form_nondefault_pin_invalid(
assert result4["type"] == FlowResultType.CREATE_ENTRY
assert result4["title"] == "Name of the device"
assert result4["data"] == {
"webfsapi_url": "http://1.1.1.1:80/webfsapi",
"pin": "4321",
CONF_WEBFSAPI_URL: "http://1.1.1.1:80/webfsapi",
CONF_PIN: "4321",
}
mock_setup_entry.assert_called_once()
@ -272,7 +294,7 @@ async def test_invalid_device_url(
assert result["errors"] == {}
with patch(
"afsapi.AFSAPI.get_webfsapi_endpoint",
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_webfsapi_endpoint",
side_effect=webfsapi_endpoint_error,
):
result2 = await hass.config_entries.flow.async_configure(
@ -294,7 +316,102 @@ async def test_invalid_device_url(
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Name of the device"
assert result3["data"] == {
"webfsapi_url": "http://1.1.1.1:80/webfsapi",
"pin": "1234",
CONF_WEBFSAPI_URL: "http://1.1.1.1:80/webfsapi",
CONF_PIN: "1234",
}
mock_setup_entry.assert_called_once()
async def test_ssdp(hass: HomeAssistant, mock_setup_entry: MockConfigEntry) -> None:
"""Test a device being discovered."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_DISCOVERY,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "confirm"
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Name of the device"
assert result2["data"] == {
CONF_WEBFSAPI_URL: "http://1.1.1.1:80/webfsapi",
CONF_PIN: DEFAULT_PIN,
}
mock_setup_entry.assert_called_once()
async def test_ssdp_invalid_location(hass: HomeAssistant) -> None:
"""Test a device being discovered."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=INVALID_MOCK_DISCOVERY,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_ssdp_already_configured(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test an already known device being discovered."""
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_DISCOVERY,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
("webfsapi_endpoint_error", "result_error"),
[(ValueError, "unknown"), (ConnectionError, "cannot_connect")],
)
async def test_ssdp_fail(
hass: HomeAssistant, webfsapi_endpoint_error: Exception, result_error: str
) -> None:
"""Test a device being discovered but failing to reply."""
with patch(
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_webfsapi_endpoint",
side_effect=webfsapi_endpoint_error,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_DISCOVERY,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == result_error
async def test_ssdp_nondefault_pin(hass: HomeAssistant) -> None:
"""Test a device being discovered."""
with patch(
"homeassistant.components.frontier_silicon.config_flow.AFSAPI.get_radio_id",
side_effect=InvalidPinException,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=MOCK_DISCOVERY,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "invalid_auth"