diff --git a/.coveragerc b/.coveragerc
index 60375fbb97e..52ffc3da56a 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -52,7 +52,7 @@ omit =
homeassistant/components/digital_ocean.py
homeassistant/components/*/digital_ocean.py
-
+
homeassistant/components/doorbird.py
homeassistant/components/*/doorbird.py
@@ -581,7 +581,6 @@ omit =
homeassistant/components/thingspeak.py
homeassistant/components/tts/amazon_polly.py
homeassistant/components/tts/picotts.py
- homeassistant/components/upnp.py
homeassistant/components/vacuum/roomba.py
homeassistant/components/weather/bom.py
homeassistant/components/weather/buienradar.py
diff --git a/homeassistant/components/upnp.py b/homeassistant/components/upnp.py
index 9e45def63db..87990495cf4 100644
--- a/homeassistant/components/upnp.py
+++ b/homeassistant/components/upnp.py
@@ -4,16 +4,18 @@ Will open a port in your router for Home Assistant and provide statistics.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/upnp/
"""
+from ipaddress import ip_address
import logging
-from urllib.parse import urlsplit
import voluptuous as vol
from homeassistant.const import (EVENT_HOMEASSISTANT_STOP)
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import discovery
+from homeassistant.util import get_local_ip
-REQUIREMENTS = ['miniupnpc==1.9']
+REQUIREMENTS = ['miniupnpc==2.0.2']
+DEPENDENCIES = ['http']
_LOGGER = logging.getLogger(__name__)
@@ -22,9 +24,11 @@ DOMAIN = 'upnp'
DATA_UPNP = 'UPNP'
+CONF_LOCAL_IP = 'local_ip'
CONF_ENABLE_PORT_MAPPING = 'port_mapping'
-CONF_EXTERNAL_PORT = 'external_port'
+CONF_PORTS = 'ports'
CONF_UNITS = 'unit'
+CONF_HASS = 'hass'
NOTIFICATION_ID = 'upnp_notification'
NOTIFICATION_TITLE = 'UPnP Setup'
@@ -39,8 +43,10 @@ UNITS = {
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_ENABLE_PORT_MAPPING, default=True): cv.boolean,
- vol.Optional(CONF_EXTERNAL_PORT, default=0): cv.positive_int,
vol.Optional(CONF_UNITS, default="MBytes"): vol.In(UNITS),
+ vol.Optional(CONF_LOCAL_IP): ip_address,
+ vol.Optional(CONF_PORTS):
+ vol.Schema({vol.Any(CONF_HASS, cv.positive_int): cv.positive_int})
}),
}, extra=vol.ALLOW_EXTRA)
@@ -48,6 +54,19 @@ CONFIG_SCHEMA = vol.Schema({
# pylint: disable=import-error, no-member, broad-except
def setup(hass, config):
"""Register a port mapping for Home Assistant via UPnP."""
+ config = config[DOMAIN]
+ host = config.get(CONF_LOCAL_IP)
+
+ if host is not None:
+ host = str(host)
+ else:
+ host = get_local_ip()
+
+ if host == '127.0.0.1':
+ _LOGGER.error(
+ 'Unable to determine local IP. Add it to your configuration.')
+ return False
+
import miniupnpc
upnp = miniupnpc.UPnP()
@@ -61,40 +80,44 @@ def setup(hass, config):
_LOGGER.exception("Error when attempting to discover an UPnP IGD")
return False
- unit = config[DOMAIN].get(CONF_UNITS)
+ unit = config.get(CONF_UNITS)
discovery.load_platform(hass, 'sensor', DOMAIN, {'unit': unit}, config)
- port_mapping = config[DOMAIN].get(CONF_ENABLE_PORT_MAPPING)
+ port_mapping = config.get(CONF_ENABLE_PORT_MAPPING)
if not port_mapping:
return True
- base_url = urlsplit(hass.config.api.base_url)
- host = base_url.hostname
- internal_port = base_url.port
- external_port = int(config[DOMAIN].get(CONF_EXTERNAL_PORT))
+ internal_port = hass.http.server_port
- if external_port == 0:
- external_port = internal_port
+ ports = config.get(CONF_PORTS)
+ if ports is None:
+ ports = {CONF_HASS: internal_port}
- try:
- upnp.addportmapping(
- external_port, 'TCP', host, internal_port, 'Home Assistant', '')
+ registered = []
+ for internal, external in ports.items():
+ if internal == CONF_HASS:
+ internal = internal_port
+ try:
+ upnp.addportmapping(
+ external, 'TCP', host, internal, 'Home Assistant', '')
+ registered.append(external)
+ except Exception:
+ _LOGGER.exception("UPnP failed to configure port mapping for %s",
+ external)
+ hass.components.persistent_notification.create(
+ 'ERROR: tcp port {} is already mapped in your router.'
+ '
Please disable port_mapping in the upnp '
+ 'configuration section.
'
+ 'You will need to restart hass after fixing.'
+ ''.format(external),
+ title=NOTIFICATION_TITLE,
+ notification_id=NOTIFICATION_ID)
- def deregister_port(event):
- """De-register the UPnP port mapping."""
- upnp.deleteportmapping(external_port, 'TCP')
+ def deregister_port(event):
+ """De-register the UPnP port mapping."""
+ for external in registered:
+ upnp.deleteportmapping(external, 'TCP')
- hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, deregister_port)
+ hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, deregister_port)
- except Exception as ex:
- _LOGGER.error("UPnP failed to configure port mapping: %s", str(ex))
- hass.components.persistent_notification.create(
- 'ERROR: tcp port {} is already mapped in your router.'
- '
Please disable port_mapping in the upnp '
- 'configuration section.
'
- 'You will need to restart hass after fixing.'
- ''.format(external_port),
- title=NOTIFICATION_TITLE,
- notification_id=NOTIFICATION_ID)
- return False
return True
diff --git a/requirements_all.txt b/requirements_all.txt
index e6ea4b71f96..d94c6059a5b 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -420,7 +420,7 @@ mficlient==0.3.0
miflora==0.1.16
# homeassistant.components.upnp
-miniupnpc==1.9
+miniupnpc==2.0.2
# homeassistant.components.sensor.mopar
motorparts==1.0.2
diff --git a/tests/components/test_upnp.py b/tests/components/test_upnp.py
new file mode 100644
index 00000000000..e2096d28e58
--- /dev/null
+++ b/tests/components/test_upnp.py
@@ -0,0 +1,142 @@
+"""Test the UPNP component."""
+import asyncio
+from collections import OrderedDict
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from homeassistant.const import EVENT_HOMEASSISTANT_STOP
+from homeassistant.setup import async_setup_component
+
+
+@pytest.fixture
+def mock_miniupnpc():
+ """Mock miniupnpc."""
+ mock = MagicMock()
+
+ with patch.dict('sys.modules', {'miniupnpc': mock}):
+ yield mock.UPnP()
+
+
+@pytest.fixture
+def mock_local_ip():
+ """Mock get_local_ip."""
+ with patch('homeassistant.components.upnp.get_local_ip',
+ return_value='192.168.0.10'):
+ yield
+
+
+@pytest.fixture(autouse=True)
+def mock_discovery():
+ """Mock discovery of upnp sensor."""
+ with patch('homeassistant.components.upnp.discovery'):
+ yield
+
+
+@asyncio.coroutine
+def test_setup_fail_if_no_ip(hass):
+ """Test setup fails if we can't find a local IP."""
+ with patch('homeassistant.components.upnp.get_local_ip',
+ return_value='127.0.0.1'):
+ result = yield from async_setup_component(hass, 'upnp', {
+ 'upnp': {}
+ })
+
+ assert not result
+
+
+@asyncio.coroutine
+def test_setup_fail_if_cannot_select_igd(hass, mock_local_ip, mock_miniupnpc):
+ """Test setup fails if we can't find an UPnP IGD."""
+ mock_miniupnpc.selectigd.side_effect = Exception
+
+ result = yield from async_setup_component(hass, 'upnp', {
+ 'upnp': {}
+ })
+
+ assert not result
+
+
+@asyncio.coroutine
+def test_setup_succeeds_if_specify_ip(hass, mock_miniupnpc):
+ """Test setup succeeds if we specify IP and can't find a local IP."""
+ with patch('homeassistant.components.upnp.get_local_ip',
+ return_value='127.0.0.1'):
+ result = yield from async_setup_component(hass, 'upnp', {
+ 'upnp': {
+ 'local_ip': '192.168.0.10'
+ }
+ })
+
+ assert result
+
+
+@asyncio.coroutine
+def test_no_config_maps_hass_local_to_remote_port(hass, mock_miniupnpc):
+ """Test by default we map local to remote port."""
+ result = yield from async_setup_component(hass, 'upnp', {
+ 'upnp': {
+ 'local_ip': '192.168.0.10'
+ }
+ })
+
+ assert result
+ assert len(mock_miniupnpc.addportmapping.mock_calls) == 1
+ external, _, host, internal, _, _ = \
+ mock_miniupnpc.addportmapping.mock_calls[0][1]
+ assert host == '192.168.0.10'
+ assert external == 8123
+ assert internal == 8123
+
+
+@asyncio.coroutine
+def test_map_hass_to_remote_port(hass, mock_miniupnpc):
+ """Test mapping hass to remote port."""
+ result = yield from async_setup_component(hass, 'upnp', {
+ 'upnp': {
+ 'local_ip': '192.168.0.10',
+ 'ports': {
+ 'hass': 1000
+ }
+ }
+ })
+
+ assert result
+ assert len(mock_miniupnpc.addportmapping.mock_calls) == 1
+ external, _, host, internal, _, _ = \
+ mock_miniupnpc.addportmapping.mock_calls[0][1]
+ assert external == 1000
+ assert internal == 8123
+
+
+@asyncio.coroutine
+def test_map_internal_to_remote_ports(hass, mock_miniupnpc):
+ """Test mapping local to remote ports."""
+ ports = OrderedDict()
+ ports['hass'] = 1000
+ ports[1883] = 3883
+
+ result = yield from async_setup_component(hass, 'upnp', {
+ 'upnp': {
+ 'local_ip': '192.168.0.10',
+ 'ports': ports
+ }
+ })
+
+ assert result
+ assert len(mock_miniupnpc.addportmapping.mock_calls) == 2
+ external, _, host, internal, _, _ = \
+ mock_miniupnpc.addportmapping.mock_calls[0][1]
+ assert external == 1000
+ assert internal == 8123
+
+ external, _, host, internal, _, _ = \
+ mock_miniupnpc.addportmapping.mock_calls[1][1]
+ assert external == 3883
+ assert internal == 1883
+
+ hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
+ yield from hass.async_block_till_done()
+ assert len(mock_miniupnpc.deleteportmapping.mock_calls) == 2
+ assert mock_miniupnpc.deleteportmapping.mock_calls[0][1][0] == 1000
+ assert mock_miniupnpc.deleteportmapping.mock_calls[1][1][0] == 3883