Simplification of upnp component

homeassistant/components/upnp/__init__.py

@@ -1,24 +1,18 @@
 """Open ports in your router for Home Assistant and provide statistics."""
 from ipaddress import ip_address
 from operator import itemgetter
-from typing import Mapping
 
 import voluptuous as vol
 
 from homeassistant import config_entries
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import EVENT_HOMEASSISTANT_STOP
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import config_validation as cv, device_registry as dr
 from homeassistant.helpers.typing import ConfigType, HomeAssistantType
 from homeassistant.util import get_local_ip
 
 from .const import (
-    CONF_ENABLE_PORT_MAPPING,
-    CONF_ENABLE_SENSORS,
-    CONF_HASS,
     CONF_LOCAL_IP,
-    CONF_PORTS,
     CONFIG_ENTRY_ST,
     CONFIG_ENTRY_UDN,
     DISCOVERY_LOCATION,
@@ -34,61 +28,11 @@
 NOTIFICATION_TITLE = "UPnP/IGD Setup"
 
 CONFIG_SCHEMA = vol.Schema(
-    {
-        DOMAIN: vol.Schema(
-            {
-                vol.Optional(CONF_ENABLE_PORT_MAPPING, default=False): cv.boolean,
-                vol.Optional(CONF_ENABLE_SENSORS, default=True): cv.boolean,
-                vol.Optional(CONF_LOCAL_IP): vol.All(ip_address, cv.string),
-                vol.Optional(CONF_PORTS, default={}): vol.Schema(
-                    {vol.Any(CONF_HASS, cv.port): vol.Any(CONF_HASS, cv.port)}
-                ),
-            }
-        )
-    },
+    {DOMAIN: vol.Schema({vol.Optional(CONF_LOCAL_IP): vol.All(ip_address, cv.string)})},
     extra=vol.ALLOW_EXTRA,
 )
 
 
-def _substitute_hass_ports(ports: Mapping, hass_port: int = None) -> Mapping:
-    """
-    Substitute 'hass' for the hass_port.
-
-    This triggers a warning when hass_port is None.
-    """
-    ports = ports.copy()
-
-    # substitute 'hass' for hass_port, both keys and values
-    if CONF_HASS in ports:
-        if hass_port is None:
-            _LOGGER.warning(
-                "Could not determine Home Assistant http port, "
-                "not setting up port mapping from %s to %s. "
-                "Enable the http-component.",
-                CONF_HASS,
-                ports[CONF_HASS],
-            )
-        else:
-            ports[hass_port] = ports[CONF_HASS]
-        del ports[CONF_HASS]
-
-    for port in ports:
-        if ports[port] == CONF_HASS:
-            if hass_port is None:
-                _LOGGER.warning(
-                    "Could not determine Home Assistant http port, "
-                    "not setting up port mapping from %s to %s. "
-                    "Enable the http-component.",
-                    port,
-                    ports[port],
-                )
-                del ports[port]
-            else:
-                ports[port] = hass_port
-
-    return ports
-
-
 async def async_discover_and_construct(
     hass: HomeAssistantType, udn: str = None, st: str = None
 ) -> Device:
@@ -137,7 +81,6 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
         "config": conf,
         "devices": {},
         "local_ip": conf.get(CONF_LOCAL_IP, local_ip),
-        "ports": conf.get(CONF_PORTS),
     }
 
     # Only start if set up via configuration.yaml.
@@ -154,8 +97,6 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType):
 async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry) -> bool:
     """Set up UPnP/IGD device from a config entry."""
     _LOGGER.debug("async_setup_entry, config_entry: %s", config_entry.data)
-    domain_data = hass.data[DOMAIN]
-    conf = domain_data["config"]
 
     # discover and construct
     udn = config_entry.data.get(CONFIG_ENTRY_UDN)
@@ -165,7 +106,7 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry)
         _LOGGER.info("Unable to create UPnP/IGD, aborting")
         raise ConfigEntryNotReady
 
-    # 'register'/save device
+    # Save device
     hass.data[DOMAIN]["devices"][device.udn] = device
 
     # Ensure entry has proper unique_id.
@@ -174,7 +115,7 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry)
             entry=config_entry, unique_id=device.unique_id,
         )
 
-    # create device registry entry
+    # Create device registry entry.
     device_registry = await dr.async_get_registry(hass)
     device_registry.async_get_or_create(
         config_entry_id=config_entry.entry_id,
@@ -185,35 +126,11 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry)
         model=device.model_name,
     )
 
-    # set up sensors
-    if conf.get(CONF_ENABLE_SENSORS):
-        _LOGGER.debug("Enabling sensors")
-
-        # register sensor setup handlers
-        hass.async_create_task(
-            hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
-        )
-
-    # set up port mapping
-    if conf.get(CONF_ENABLE_PORT_MAPPING):
-        _LOGGER.debug("Enabling port mapping")
-        local_ip = domain_data[CONF_LOCAL_IP]
-        ports = conf.get(CONF_PORTS, {})
-
-        hass_port = None
-        if hasattr(hass, "http"):
-            hass_port = hass.http.server_port
-
-        ports = _substitute_hass_ports(ports, hass_port=hass_port)
-        await device.async_add_port_mappings(ports, local_ip)
-
-    # set up port mapping deletion on stop-hook
-    async def delete_port_mapping(event):
-        """Delete port mapping on quit."""
-        _LOGGER.debug("Deleting port mappings")
-        await device.async_delete_port_mappings()
-
-    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, delete_port_mapping)
+    # Create sensors.
+    _LOGGER.debug("Enabling sensors")
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(config_entry, "sensor")
+    )
 
     return True
 
@@ -222,13 +139,5 @@ async def async_unload_entry(
     hass: HomeAssistantType, config_entry: ConfigEntry
 ) -> bool:
     """Unload a UPnP/IGD device from a config entry."""
-    udn = config_entry.data[CONFIG_ENTRY_UDN]
-    device = hass.data[DOMAIN]["devices"][udn]
-
-    # remove port mapping
-    _LOGGER.debug("Deleting port mappings")
-    await device.async_delete_port_mappings()
-
-    # remove sensors
     _LOGGER.debug("Deleting sensors")
     return await hass.config_entries.async_forward_entry_unload(config_entry, "sensor")

homeassistant/components/upnp/const.py

@@ -4,11 +4,7 @@
 
 from homeassistant.const import TIME_SECONDS
 
-CONF_ENABLE_PORT_MAPPING = "port_mapping"
-CONF_ENABLE_SENSORS = "sensors"
-CONF_HASS = "hass"
 CONF_LOCAL_IP = "local_ip"
-CONF_PORTS = "ports"
 DOMAIN = "upnp"
 LOGGER = logging.getLogger(__package__)
 BYTES_RECEIVED = "bytes_received"

homeassistant/components/upnp/device.py

@@ -3,8 +3,7 @@
 from ipaddress import IPv4Address
 from typing import List, Mapping
 
-import aiohttp
-from async_upnp_client import UpnpError, UpnpFactory
+from async_upnp_client import UpnpFactory
 from async_upnp_client.aiohttp import AiohttpSessionRequester
 from async_upnp_client.profiles.igd import IgdDevice
 
@@ -111,70 +110,6 @@ def __str__(self) -> str:
         """Get string representation."""
         return f"IGD Device: {self.name}/{self.udn}"
 
-    async def async_add_port_mappings(
-        self, ports: Mapping[int, int], local_ip: str
-    ) -> None:
-        """Add port mappings."""
-        if local_ip == "127.0.0.1":
-            _LOGGER.error("Could not create port mapping, our IP is 127.0.0.1")
-
-        # determine local ip, ensure sane IP
-        local_ip = IPv4Address(local_ip)
-
-        # create port mappings
-        for external_port, internal_port in ports.items():
-            await self._async_add_port_mapping(external_port, local_ip, internal_port)
-            self._mapped_ports.append(external_port)
-
-    async def _async_add_port_mapping(
-        self, external_port: int, local_ip: str, internal_port: int
-    ) -> None:
-        """Add a port mapping."""
-        # create port mapping
-        _LOGGER.info(
-            "Creating port mapping %s:%s:%s (TCP)",
-            external_port,
-            local_ip,
-            internal_port,
-        )
-        try:
-            await self._igd_device.async_add_port_mapping(
-                remote_host=None,
-                external_port=external_port,
-                protocol="TCP",
-                internal_port=internal_port,
-                internal_client=local_ip,
-                enabled=True,
-                description="Home Assistant",
-                lease_duration=None,
-            )
-
-            self._mapped_ports.append(external_port)
-        except (asyncio.TimeoutError, aiohttp.ClientError, UpnpError):
-            _LOGGER.error(
-                "Could not add port mapping: %s:%s:%s",
-                external_port,
-                local_ip,
-                internal_port,
-            )
-
-    async def async_delete_port_mappings(self) -> None:
-        """Remove port mappings."""
-        for port in self._mapped_ports:
-            await self._async_delete_port_mapping(port)
-
-    async def _async_delete_port_mapping(self, external_port: int) -> None:
-        """Remove a port mapping."""
-        _LOGGER.info("Deleting port mapping %s (TCP)", external_port)
-        try:
-            await self._igd_device.async_delete_port_mapping(
-                remote_host=None, external_port=external_port, protocol="TCP"
-            )
-
-            self._mapped_ports.remove(external_port)
-        except (asyncio.TimeoutError, aiohttp.ClientError, UpnpError):
-            _LOGGER.error("Could not delete port mapping")
-
     async def async_get_traffic_data(self) -> Mapping[str, any]:
         """
         Get all traffic data in one go.

tests/components/upnp/test_init.py

@@ -1,7 +1,5 @@
 """Test UPnP/IGD setup process."""
 
-from ipaddress import IPv4Address
-
 from homeassistant.components import upnp
 from homeassistant.components.upnp.const import (
     DISCOVERY_LOCATION,
@@ -53,59 +51,3 @@ async def test_async_setup_entry_default(hass):
 
         hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
         await hass.async_block_till_done()
-
-    # ensure no port-mappings created or removed
-    assert not mock_device.added_port_mappings
-    assert not mock_device.removed_port_mappings
-
-
-async def test_async_setup_entry_port_mapping(hass):
-    """Test async_setup_entry."""
-    # pylint: disable=invalid-name
-    udn = "uuid:device_1"
-    mock_device = MockDevice(udn)
-    discovery_infos = [
-        {
-            DISCOVERY_UDN: mock_device.udn,
-            DISCOVERY_ST: mock_device.device_type,
-            DISCOVERY_LOCATION: "http://192.168.1.1/desc.xml",
-        }
-    ]
-    entry = MockConfigEntry(
-        domain=upnp.DOMAIN, data={"udn": mock_device.udn, "st": mock_device.device_type}
-    )
-
-    config = {
-        "http": {},
-        "upnp": {
-            "local_ip": "192.168.1.10",
-            "port_mapping": True,
-            "ports": {"hass": "hass"},
-        },
-    }
-    async_discover = AsyncMock(return_value=[])
-    with patch.object(
-        Device, "async_create_device", AsyncMock(return_value=mock_device)
-    ), patch.object(Device, "async_discover", async_discover):
-        # initialisation of component, no device discovered
-        await async_setup_component(hass, "http", config)
-        await async_setup_component(hass, "upnp", config)
-        await hass.async_block_till_done()
-
-        # loading of config_entry, device discovered
-        async_discover.return_value = discovery_infos
-        assert await upnp.async_setup_entry(hass, entry) is True
-
-        # ensure device is stored/used
-        assert hass.data[upnp.DOMAIN]["devices"][udn] == mock_device
-
-        # ensure add-port-mapping-methods called
-        assert mock_device.added_port_mappings == [
-            [8123, IPv4Address("192.168.1.10"), 8123]
-        ]
-
-        hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
-        await hass.async_block_till_done()
-
-    # ensure delete-port-mapping-methods called
-    assert mock_device.removed_port_mappings == [8123]