Convert syncthru to config flow and native SSDP discovery (#36690)
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>pull/37669/head
Ville Skyttä
GitHub
parent
fb3049d6a2
commit
a077c280c8
|
|
@ -60,7 +60,7 @@ SERVICE_HANDLERS = {
|
|||
SERVICE_ENIGMA2: ("media_player", "enigma2"),
|
||||
SERVICE_WINK: ("wink", None),
|
||||
SERVICE_SABNZBD: ("sabnzbd", None),
|
||||
SERVICE_SAMSUNG_PRINTER: ("sensor", "syncthru"),
|
||||
SERVICE_SAMSUNG_PRINTER: ("sensor", None),
|
||||
SERVICE_KONNECTED: ("konnected", None),
|
||||
SERVICE_OCTOPRINT: ("octoprint", None),
|
||||
SERVICE_FREEBOX: ("freebox", None),
|
||||
|
|
|
|||
|
|
@ -1 +1,23 @@
|
|||
"""The syncthru component."""
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
|
||||
"""Set up."""
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Set up config entry."""
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, SENSOR_DOMAIN)
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass, entry):
|
||||
"""Unload the config entry."""
|
||||
return await hass.config_entries.async_forward_entry_unload(entry, SENSOR_DOMAIN)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,137 @@
|
|||
"""Config flow for Samsung SyncThru."""
|
||||
|
||||
import re
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pysyncthru import SyncThru
|
||||
from url_normalize import url_normalize
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components import ssdp
|
||||
from homeassistant.const import CONF_NAME, CONF_URL
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
|
||||
# pylint: disable=unused-import # for DOMAIN https://github.com/PyCQA/pylint/issues/3202
|
||||
from .const import DEFAULT_MODEL, DEFAULT_NAME_TEMPLATE, DOMAIN
|
||||
|
||||
|
||||
class SyncThruConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Samsung SyncThru config flow."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
|
||||
|
||||
url: str
|
||||
name: str
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle user initiated flow."""
|
||||
if user_input is None:
|
||||
return await self._async_show_form(step_id="user")
|
||||
return await self._async_check_and_create("user", user_input)
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
"""Handle import initiated flow."""
|
||||
return await self.async_step_user(user_input=user_input)
|
||||
|
||||
async def async_step_ssdp(self, discovery_info):
|
||||
"""Handle SSDP initiated flow."""
|
||||
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_UDN])
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
self.url = url_normalize(
|
||||
discovery_info.get(
|
||||
ssdp.ATTR_UPNP_PRESENTATION_URL,
|
||||
f"http://{urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION]).hostname}/",
|
||||
)
|
||||
)
|
||||
|
||||
for existing_entry in (
|
||||
x for x in self._async_current_entries() if x.data[CONF_URL] == self.url
|
||||
):
|
||||
# Update unique id of entry with the same URL
|
||||
if not existing_entry.unique_id:
|
||||
await self.hass.config_entries.async_update_entry(
|
||||
existing_entry, unique_id=discovery_info[ssdp.ATTR_UPNP_UDN]
|
||||
)
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
self.name = discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
|
||||
if self.name:
|
||||
# Remove trailing " (ip)" if present for consistency with user driven config
|
||||
self.name = re.sub(r"\s+\([\d.]+\)\s*$", "", self.name)
|
||||
|
||||
# https://github.com/PyCQA/pylint/issues/3167
|
||||
self.context["title_placeholders"] = { # pylint: disable=no-member
|
||||
CONF_NAME: self.name
|
||||
}
|
||||
return await self.async_step_confirm()
|
||||
|
||||
async def async_step_confirm(self, user_input=None):
|
||||
"""Handle discovery confirmation by user."""
|
||||
if user_input is not None:
|
||||
return await self._async_check_and_create("confirm", user_input)
|
||||
|
||||
return await self._async_show_form(
|
||||
step_id="confirm", user_input={CONF_URL: self.url, CONF_NAME: self.name},
|
||||
)
|
||||
|
||||
async def _async_show_form(self, step_id, user_input=None, errors=None):
|
||||
"""Show our form."""
|
||||
if user_input is None:
|
||||
user_input = {}
|
||||
return self.async_show_form(
|
||||
step_id=step_id,
|
||||
data_schema=vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_URL, default=user_input.get(CONF_URL, "")): str,
|
||||
vol.Optional(CONF_NAME, default=user_input.get(CONF_NAME, "")): str,
|
||||
}
|
||||
),
|
||||
errors=errors or {},
|
||||
)
|
||||
|
||||
async def _async_check_and_create(self, step_id, user_input):
|
||||
"""Validate input, proceed to create."""
|
||||
user_input[CONF_URL] = url_normalize(
|
||||
user_input[CONF_URL], default_scheme="http"
|
||||
)
|
||||
if "://" not in user_input[CONF_URL]:
|
||||
return await self._async_show_form(
|
||||
step_id=step_id, user_input=user_input, errors={CONF_URL: "invalid_url"}
|
||||
)
|
||||
|
||||
# If we don't have a unique id, copy one from existing entry with same URL
|
||||
if not self.unique_id:
|
||||
for existing_entry in (
|
||||
x
|
||||
for x in self._async_current_entries()
|
||||
if x.data[CONF_URL] == user_input[CONF_URL] and x.unique_id
|
||||
):
|
||||
await self.async_set_unique_id(existing_entry.unique_id)
|
||||
break
|
||||
|
||||
session = aiohttp_client.async_get_clientsession(self.hass)
|
||||
printer = SyncThru(user_input[CONF_URL], session)
|
||||
errors = {}
|
||||
try:
|
||||
await printer.update()
|
||||
if not user_input.get(CONF_NAME):
|
||||
user_input[CONF_NAME] = DEFAULT_NAME_TEMPLATE.format(
|
||||
printer.model() or DEFAULT_MODEL
|
||||
)
|
||||
except ValueError:
|
||||
errors[CONF_URL] = "syncthru_not_supported"
|
||||
else:
|
||||
if printer.is_unknown_state():
|
||||
errors[CONF_URL] = "unknown_state"
|
||||
|
||||
if errors:
|
||||
return await self._async_show_form(
|
||||
step_id=step_id, user_input=user_input, errors=errors
|
||||
)
|
||||
|
||||
return self.async_create_entry(
|
||||
title=user_input.get(CONF_NAME), data=user_input,
|
||||
)
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
"""Samsung SyncThru constants."""
|
||||
|
||||
DEFAULT_NAME_TEMPLATE = "Samsung {}"
|
||||
DEFAULT_MODEL = "Printer"
|
||||
|
||||
DOMAIN = "syncthru"
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
"""Samsung SyncThru exceptions."""
|
||||
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
|
||||
class SyncThruNotSupported(HomeAssistantError):
|
||||
"""Error to indicate SyncThru is not supported."""
|
||||
|
|
@ -2,6 +2,13 @@
|
|||
"domain": "syncthru",
|
||||
"name": "Samsung SyncThru Printer",
|
||||
"documentation": "https://www.home-assistant.io/integrations/syncthru",
|
||||
"requirements": ["pysyncthru==0.5.0"],
|
||||
"config_flow": true,
|
||||
"requirements": ["pysyncthru==0.5.0", "url-normalize==1.4.1"],
|
||||
"ssdp": [
|
||||
{
|
||||
"deviceType": "urn:schemas-upnp-org:device:Printer:1",
|
||||
"manufacturer": "Samsung Electronics"
|
||||
}
|
||||
],
|
||||
"codeowners": ["@nielstron"]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,14 +6,18 @@ from pysyncthru import SyncThru
|
|||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_RESOURCE, UNIT_PERCENTAGE
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import CONF_NAME, CONF_RESOURCE, CONF_URL, UNIT_PERCENTAGE
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from .const import DEFAULT_MODEL, DEFAULT_NAME_TEMPLATE, DOMAIN
|
||||
from .exceptions import SyncThruNotSupported
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_NAME = "Samsung Printer"
|
||||
COLORS = ["black", "cyan", "magenta", "yellow"]
|
||||
DRUM_COLORS = COLORS
|
||||
TONER_COLORS = COLORS
|
||||
|
|
@ -28,30 +32,38 @@ DEFAULT_MONITORED_CONDITIONS.extend([f"output_tray_{key}" for key in OUTPUT_TRAY
|
|||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_RESOURCE): cv.url,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(
|
||||
CONF_NAME, default=DEFAULT_NAME_TEMPLATE.format(DEFAULT_MODEL)
|
||||
): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
||||
"""Set up the SyncThru component."""
|
||||
|
||||
if discovery_info is not None:
|
||||
_LOGGER.info(
|
||||
"Discovered a new Samsung Printer at %s", discovery_info.get(CONF_HOST)
|
||||
_LOGGER.warning(
|
||||
"Loading syncthru via platform config is deprecated and no longer "
|
||||
"necessary as of 0.113. Please remove it from your configuration YAML."
|
||||
)
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={
|
||||
CONF_URL: config.get(CONF_RESOURCE),
|
||||
CONF_NAME: config.get(CONF_NAME),
|
||||
},
|
||||
)
|
||||
host = discovery_info.get(CONF_HOST)
|
||||
name = discovery_info.get(CONF_NAME, DEFAULT_NAME)
|
||||
# Main device, always added
|
||||
else:
|
||||
host = config.get(CONF_RESOURCE)
|
||||
name = config.get(CONF_NAME)
|
||||
# always pass through all of the obtained information
|
||||
monitored = DEFAULT_MONITORED_CONDITIONS
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up from config entry."""
|
||||
|
||||
session = aiohttp_client.async_get_clientsession(hass)
|
||||
|
||||
printer = SyncThru(host, session)
|
||||
printer = SyncThru(config_entry.data[CONF_URL], session)
|
||||
# Test if the discovered device actually is a syncthru printer
|
||||
# and fetch the available toner/drum/etc
|
||||
try:
|
||||
|
|
@ -63,34 +75,23 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
|
|||
supp_drum = printer.drum_status(filter_supported=True)
|
||||
supp_tray = printer.input_tray_status(filter_supported=True)
|
||||
supp_output_tray = printer.output_tray_status()
|
||||
except ValueError:
|
||||
# if an exception is thrown, printer does not support syncthru
|
||||
# and should not be set up
|
||||
# If the printer was discovered automatically, no warning or error
|
||||
# should be issued and printer should not be set up
|
||||
if discovery_info is not None:
|
||||
_LOGGER.info("Samsung printer at %s does not support SyncThru", host)
|
||||
return
|
||||
# Otherwise, emulate printer that supports everything
|
||||
supp_toner = TONER_COLORS
|
||||
supp_drum = DRUM_COLORS
|
||||
supp_tray = TRAYS
|
||||
supp_output_tray = OUTPUT_TRAYS
|
||||
except ValueError as ex:
|
||||
raise SyncThruNotSupported from ex
|
||||
else:
|
||||
if printer.is_unknown_state():
|
||||
raise PlatformNotReady
|
||||
|
||||
name = config_entry.data[CONF_NAME]
|
||||
devices = [SyncThruMainSensor(printer, name)]
|
||||
|
||||
for key in supp_toner:
|
||||
if f"toner_{key}" in monitored:
|
||||
devices.append(SyncThruTonerSensor(printer, name, key))
|
||||
devices.append(SyncThruTonerSensor(printer, name, key))
|
||||
for key in supp_drum:
|
||||
if f"drum_{key}" in monitored:
|
||||
devices.append(SyncThruDrumSensor(printer, name, key))
|
||||
devices.append(SyncThruDrumSensor(printer, name, key))
|
||||
for key in supp_tray:
|
||||
if f"tray_{key}" in monitored:
|
||||
devices.append(SyncThruInputTraySensor(printer, name, key))
|
||||
devices.append(SyncThruInputTraySensor(printer, name, key))
|
||||
for key in supp_output_tray:
|
||||
if f"output_tray_{key}" in monitored:
|
||||
devices.append(SyncThruOutputTraySensor(printer, name, key))
|
||||
devices.append(SyncThruOutputTraySensor(printer, name, key))
|
||||
|
||||
async_add_entities(devices, True)
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
},
|
||||
"error": {
|
||||
"invalid_url": "Invalid URL",
|
||||
"syncthru_not_supported": "Device does not support SyncThru",
|
||||
"unknown_state": "Printer state unknown, verify URL and network connectivity"
|
||||
},
|
||||
"flow_title": "Samsung SyncThru Printer: {name}",
|
||||
"step": {
|
||||
"confirm": {
|
||||
"data": {
|
||||
"name": "[%key:component::syncthru::config::step::user::data::name%]",
|
||||
"url": "[%key:component::syncthru::config::step::user::data::url%]"
|
||||
}
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"name": "Name",
|
||||
"url": "Web interface URL"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
},
|
||||
"error": {
|
||||
"invalid_url": "Invalid URL",
|
||||
"syncthru_not_supported": "Device does not support SyncThru",
|
||||
"unknown_state": "Printer state unknown, verify URL and network connectivity"
|
||||
},
|
||||
"flow_title": "Samsung SyncThru Printer: {name}",
|
||||
"step": {
|
||||
"confirm": {
|
||||
"data": {
|
||||
"name": "[%key:component::syncthru::config::step::user::data::name%]",
|
||||
"url": "[%key:component::syncthru::config::step::user::data::url%]"
|
||||
}
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"name": "Name",
|
||||
"url": "Web interface URL"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -156,6 +156,7 @@ FLOWS = [
|
|||
"spotify",
|
||||
"squeezebox",
|
||||
"starline",
|
||||
"syncthru",
|
||||
"synology_dsm",
|
||||
"tado",
|
||||
"tellduslive",
|
||||
|
|
|
|||
|
|
@ -143,6 +143,12 @@ SSDP = {
|
|||
"st": "urn:schemas-upnp-org:device:ZonePlayer:1"
|
||||
}
|
||||
],
|
||||
"syncthru": [
|
||||
{
|
||||
"deviceType": "urn:schemas-upnp-org:device:Printer:1",
|
||||
"manufacturer": "Samsung Electronics"
|
||||
}
|
||||
],
|
||||
"synology_dsm": [
|
||||
{
|
||||
"deviceType": "urn:schemas-upnp-org:device:Basic:1",
|
||||
|
|
|
|||
|
|
@ -2155,6 +2155,7 @@ upb_lib==0.4.11
|
|||
upcloud-api==0.4.5
|
||||
|
||||
# homeassistant.components.huawei_lte
|
||||
# homeassistant.components.syncthru
|
||||
url-normalize==1.4.1
|
||||
|
||||
# homeassistant.components.uscis
|
||||
|
|
|
|||
|
|
@ -747,6 +747,9 @@ pyspcwebgw==0.4.0
|
|||
# homeassistant.components.squeezebox
|
||||
pysqueezebox==0.2.4
|
||||
|
||||
# homeassistant.components.syncthru
|
||||
pysyncthru==0.5.0
|
||||
|
||||
# homeassistant.components.ecobee
|
||||
python-ecobee-api==0.2.7
|
||||
|
||||
|
|
@ -929,6 +932,7 @@ twilio==6.32.0
|
|||
upb_lib==0.4.11
|
||||
|
||||
# homeassistant.components.huawei_lte
|
||||
# homeassistant.components.syncthru
|
||||
url-normalize==1.4.1
|
||||
|
||||
# homeassistant.components.uvc
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the syncthru integration."""
|
||||
|
|
@ -0,0 +1,143 @@
|
|||
"""Tests for syncthru config flow."""
|
||||
|
||||
import re
|
||||
|
||||
from homeassistant import config_entries, data_entry_flow, setup
|
||||
from homeassistant.components import ssdp
|
||||
from homeassistant.components.syncthru.config_flow import SyncThru
|
||||
from homeassistant.components.syncthru.const import DOMAIN
|
||||
from homeassistant.const import CONF_NAME, CONF_URL
|
||||
|
||||
from tests.async_mock import patch
|
||||
from tests.common import MockConfigEntry, mock_coro
|
||||
|
||||
FIXTURE_USER_INPUT = {
|
||||
CONF_URL: "http://192.168.1.2/",
|
||||
CONF_NAME: "My Printer",
|
||||
}
|
||||
|
||||
|
||||
def mock_connection(aioclient_mock):
|
||||
"""Mock syncthru connection."""
|
||||
aioclient_mock.get(
|
||||
re.compile("."),
|
||||
text="""
|
||||
{
|
||||
\tstatus: {
|
||||
\tstatus1: " Sleeping... "
|
||||
\t},
|
||||
\tidentity: {
|
||||
\tserial_num: "000000000000000",
|
||||
\t}
|
||||
}
|
||||
""",
|
||||
)
|
||||
|
||||
|
||||
async def test_show_setup_form(hass):
|
||||
"""Test that the setup form is served."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=None
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_already_configured_by_url(hass, aioclient_mock):
|
||||
"""Test we match and update already configured devices by URL."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
udn = "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX"
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={**FIXTURE_USER_INPUT, CONF_NAME: "Already configured"},
|
||||
title="Already configured",
|
||||
unique_id=udn,
|
||||
).add_to_hass(hass)
|
||||
mock_connection(aioclient_mock)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=FIXTURE_USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"][CONF_URL] == FIXTURE_USER_INPUT[CONF_URL]
|
||||
assert result["data"][CONF_NAME] == FIXTURE_USER_INPUT[CONF_NAME]
|
||||
assert result["result"].unique_id == udn
|
||||
|
||||
|
||||
async def test_syncthru_not_supported(hass):
|
||||
"""Test we show user form on unsupported device."""
|
||||
with patch.object(SyncThru, "update", side_effect=ValueError):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=FIXTURE_USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {CONF_URL: "syncthru_not_supported"}
|
||||
|
||||
|
||||
async def test_unknown_state(hass):
|
||||
"""Test we show user form on unsupported device."""
|
||||
with patch.object(SyncThru, "update", return_value=mock_coro()), patch.object(
|
||||
SyncThru, "is_unknown_state", return_value=True
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=FIXTURE_USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {CONF_URL: "unknown_state"}
|
||||
|
||||
|
||||
async def test_success(hass, aioclient_mock):
|
||||
"""Test successful flow provides entry creation data."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
mock_connection(aioclient_mock)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.syncthru.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=FIXTURE_USER_INPUT,
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["data"][CONF_URL] == FIXTURE_USER_INPUT[CONF_URL]
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_ssdp(hass, aioclient_mock):
|
||||
"""Test SSDP discovery initiates config properly."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
mock_connection(aioclient_mock)
|
||||
|
||||
url = "http://192.168.1.2/"
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_SSDP},
|
||||
data={
|
||||
ssdp.ATTR_SSDP_LOCATION: "http://192.168.1.2:5200/Printer.xml",
|
||||
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:Printer:1",
|
||||
ssdp.ATTR_UPNP_MANUFACTURER: "Samsung Electronics",
|
||||
ssdp.ATTR_UPNP_PRESENTATION_URL: url,
|
||||
ssdp.ATTR_UPNP_SERIAL: "00000000",
|
||||
ssdp.ATTR_UPNP_UDN: "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "confirm"
|
||||
assert CONF_URL in result["data_schema"].schema
|
||||
for k in result["data_schema"].schema:
|
||||
if k == CONF_URL:
|
||||
assert k.default() == url
|
||||
Loading…
Reference in New Issue