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

Remove YAML configuration from upnp (#72410)

pull/72715/head
epenet 2022-05-30 17:07:18 +02:00 committed by GitHub
parent 57ed667257
commit 640f53ce21
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 11 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
homeassistant/components/upnp/__init__.py
View File

@ -5,13 +5,10 @@ import asyncio
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import timedelta
from ipaddress import ip_address
from typing import Any
from async_upnp_client.exceptions import UpnpCommunicationError, UpnpConnectionError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.components.binary_sensor import BinarySensorEntityDescription
from homeassistant.components.sensor import SensorEntityDescription
@ -19,10 +16,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
@ -30,7 +25,6 @@ from homeassistant.helpers.update_coordinator import (
)
from .const import (
CONF_LOCAL_IP,
CONFIG_ENTRY_MAC_ADDRESS,
CONFIG_ENTRY_ORIGINAL_UDN,
CONFIG_ENTRY_ST,
@ -46,43 +40,15 @@ NOTIFICATION_TITLE = "UPnP/IGD Setup"
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
CONFIG_SCHEMA = vol.Schema(
vol.All(
cv.deprecated(DOMAIN),
{
DOMAIN: vol.Schema(
vol.All(
cv.deprecated(CONF_LOCAL_IP),
{
vol.Optional(CONF_LOCAL_IP): vol.All(ip_address, cv.string),
},
)
)
},
),
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up UPnP component."""
hass.data[DOMAIN] = {}
# Only start if set up via configuration.yaml.
if DOMAIN in config:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
)
return True
CONFIG_SCHEMA = cv.removed(DOMAIN, raise_if_present=False)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up UPnP/IGD device from a config entry."""
LOGGER.debug("Setting up config entry: %s", entry.entry_id)
hass.data.setdefault(DOMAIN, {})
udn = entry.data[CONFIG_ENTRY_UDN]
st = entry.data[CONFIG_ENTRY_ST] # pylint: disable=invalid-name
usn = f"{udn}::{st}"

87
homeassistant/components/upnp/config_flow.py
View File

@ -1,7 +1,6 @@
"""Config flow for UPNP."""
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from typing import Any, cast
@ -9,7 +8,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.components.ssdp import SsdpChange, SsdpServiceInfo
from homeassistant.components.ssdp import SsdpServiceInfo
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
@ -21,7 +20,6 @@ from .const import (
CONFIG_ENTRY_UDN,
DOMAIN,
LOGGER,
SSDP_SEARCH_TIMEOUT,
ST_IGD_V1,
ST_IGD_V2,
)
@ -48,47 +46,6 @@ def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool:
)
async def _async_wait_for_discoveries(hass: HomeAssistant) -> bool:
"""Wait for a device to be discovered."""
device_discovered_event = asyncio.Event()
async def device_discovered(info: SsdpServiceInfo, change: SsdpChange) -> None:
if change != SsdpChange.BYEBYE:
LOGGER.debug(
"Device discovered: %s, at: %s",
info.ssdp_usn,
info.ssdp_location,
)
device_discovered_event.set()
cancel_discovered_callback_1 = await ssdp.async_register_callback(
hass,
device_discovered,
{
ssdp.ATTR_SSDP_ST: ST_IGD_V1,
},
)
cancel_discovered_callback_2 = await ssdp.async_register_callback(
hass,
device_discovered,
{
ssdp.ATTR_SSDP_ST: ST_IGD_V2,
},
)
try:
await asyncio.wait_for(
device_discovered_event.wait(), timeout=SSDP_SEARCH_TIMEOUT
)
except asyncio.TimeoutError:
return False
finally:
cancel_discovered_callback_1()
cancel_discovered_callback_2()
return True
async def _async_discover_igd_devices(
hass: HomeAssistant,
) -> list[ssdp.SsdpServiceInfo]:
@ -120,7 +77,9 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize the UPnP/IGD config flow."""
self._discoveries: list[SsdpServiceInfo] | None = None
async def async_step_user(self, user_input: Mapping | None = None) -> FlowResult:
async def async_step_user(
self, user_input: Mapping[str, Any] | None = None
) -> FlowResult:
"""Handle a flow start."""
LOGGER.debug("async_step_user: user_input: %s", user_input)
@ -172,42 +131,6 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
data_schema=data_schema,
)
async def async_step_import(self, import_info: Mapping | None) -> Mapping[str, Any]:
"""Import a new UPnP/IGD device as a config entry.
This flow is triggered by `async_setup`. If no device has been
configured before, find any device and create a config_entry for it.
Otherwise, do nothing.
"""
LOGGER.debug("async_step_import: import_info: %s", import_info)
# Landed here via configuration.yaml entry.
# Any device already added, then abort.
if self._async_current_entries():
LOGGER.debug("Already configured, aborting")
return self.async_abort(reason="already_configured")
# Discover devices.
await _async_wait_for_discoveries(self.hass)
discoveries = await _async_discover_igd_devices(self.hass)
# Ensure anything to add. If not, silently abort.
if not discoveries:
LOGGER.info("No UPnP devices discovered, aborting")
return self.async_abort(reason="no_devices_found")
# Ensure complete discovery.
discovery = discoveries[0]
if not _is_complete_discovery(discovery):
LOGGER.debug("Incomplete discovery, ignoring")
return self.async_abort(reason="incomplete_discovery")
# Ensure not already configuring/configured.
unique_id = discovery.ssdp_usn
await self.async_set_unique_id(unique_id)
return await self._async_create_entry_from_discovery(discovery)
async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo) -> FlowResult:
"""Handle a discovered UPnP/IGD device.
@ -275,7 +198,7 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_ssdp_confirm()
async def async_step_ssdp_confirm(
self, user_input: Mapping | None = None
self, user_input: Mapping[str, Any] | None = None
) -> FlowResult:
"""Confirm integration via SSDP."""
LOGGER.debug("async_step_ssdp_confirm: user_input: %s", user_input)

5
tests/components/upnp/conftest.py
View File

@ -156,10 +156,7 @@ async def ssdp_no_discovery():
) as mock_register, patch(
"homeassistant.components.ssdp.async_get_discovery_info_by_st",
return_value=[],
) as mock_get_info, patch(
"homeassistant.components.upnp.config_flow.SSDP_SEARCH_TIMEOUT",
0.1,
):
) as mock_get_info:
yield (mock_register, mock_get_info)

96
tests/components/upnp/test_config_flow.py
View File

@ -1,7 +1,7 @@
"""Test UPnP/IGD config flow."""
from copy import deepcopy
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
@ -341,97 +341,3 @@ async def test_flow_user_no_discovery(hass: HomeAssistant):
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_devices_found"
@pytest.mark.usefixtures(
"ssdp_instant_discovery",
"mock_setup_entry",
"mock_get_source_ip",
"mock_mac_address_from_host",
)
async def test_flow_import(hass: HomeAssistant):
"""Test config flow: configured through configuration.yaml."""
# Discovered via step import.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == TEST_FRIENDLY_NAME
assert result["data"] == {
CONFIG_ENTRY_ST: TEST_ST,
CONFIG_ENTRY_UDN: TEST_UDN,
CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN,
CONFIG_ENTRY_LOCATION: TEST_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
}
@pytest.mark.usefixtures(
"mock_get_source_ip",
)
async def test_flow_import_incomplete_discovery(hass: HomeAssistant):
"""Test config flow: configured through configuration.yaml, but incomplete discovery."""
incomplete_discovery = ssdp.SsdpServiceInfo(
ssdp_usn=TEST_USN,
ssdp_st=TEST_ST,
ssdp_location=TEST_LOCATION,
upnp={
# ssdp.ATTR_UPNP_UDN: TEST_UDN, # Not provided.
},
)
async def register_callback(hass, callback, match_dict):
"""Immediately do callback."""
await callback(incomplete_discovery, ssdp.SsdpChange.ALIVE)
return MagicMock()
with patch(
"homeassistant.components.ssdp.async_register_callback",
side_effect=register_callback,
), patch(
"homeassistant.components.upnp.ssdp.async_get_discovery_info_by_st",
return_value=[incomplete_discovery],
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "incomplete_discovery"
@pytest.mark.usefixtures("ssdp_instant_discovery", "mock_get_source_ip")
async def test_flow_import_already_configured(hass: HomeAssistant):
"""Test config flow: configured through configuration.yaml, but existing config entry."""
# Existing entry.
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_USN,
data={
CONFIG_ENTRY_ST: TEST_ST,
CONFIG_ENTRY_UDN: TEST_UDN,
CONFIG_ENTRY_ORIGINAL_UDN: TEST_UDN,
CONFIG_ENTRY_LOCATION: TEST_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
},
state=config_entries.ConfigEntryState.LOADED,
)
entry.add_to_hass(hass)
# Discovered via step import.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("ssdp_no_discovery", "mock_get_source_ip")
async def test_flow_import_no_devices_found(hass: HomeAssistant):
"""Test config flow: no devices found, configured through configuration.yaml."""
# Discovered via step import.
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_devices_found"