jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-21 20:27:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Working on IGD

Browse Source
This commit is contained in:
Steven Looman 2018-08-30 16:38:43 +02:00
parent 3db766e2ec
commit 20879726b0
9 changed files with 284 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

147
homeassistant/components/igd/__init__.py
View File

@ -4,52 +4,34 @@ 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 For more details about this component, please refer to the documentation at
https://home-assistant.io/components/upnp/ https://home-assistant.io/components/upnp/
""" """
# XXX TODO:
# + flow:
# + discovery
# + adding device
# + removing device
# - configured:
# - adding
# - sensors:
# + adding
# + handle overflow
# - removing
# - port forward:
# - adding
# - removing
# - shutdown
import asyncio
from ipaddress import IPv4Address from ipaddress import IPv4Address
from ipaddress import ip_address from ipaddress import ip_address
import aiohttp
import asyncio
import aiohttp
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.const import CONF_URL
from homeassistant.exceptions import PlatformNotReady from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers import discovery from homeassistant.helpers import discovery
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from homeassistant.util import get_local_ip from homeassistant.util import get_local_ip
from homeassistant.components.discovery import DOMAIN as DISCOVERY_DOMAIN
from .config_flow import configured_udns from .const import CONF_ENABLE_PORT_MAPPING, CONF_ENABLE_SENSORS
from .const import CONF_PORT_FORWARD, CONF_SENSORS
from .const import DOMAIN from .const import DOMAIN
from .const import LOGGER as _LOGGER from .const import LOGGER as _LOGGER
REQUIREMENTS = ['async-upnp-client==0.12.4'] REQUIREMENTS = ['async-upnp-client==0.12.4']
DEPENDENCIES = ['http'] DEPENDENCIES = ['http'] # ,'discovery']
CONF_LOCAL_IP = 'local_ip' CONF_LOCAL_IP = 'local_ip'
CONF_ENABLE_PORT_MAPPING = 'port_mapping'
CONF_PORTS = 'ports' CONF_PORTS = 'ports'
CONF_UNITS = 'unit' CONF_UNITS = 'unit'
CONF_HASS = 'hass' CONF_HASS = 'hass'
@ -66,17 +48,16 @@ UNITS = {
CONFIG_SCHEMA = vol.Schema({ CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({ DOMAIN: vol.Schema({
vol.Required(CONF_URL): cv.url, vol.Optional(CONF_ENABLE_PORT_MAPPING, default=False): cv.boolean,
vol.Optional(CONF_ENABLE_PORT_MAPPING, default=True): cv.boolean, vol.Optional(CONF_ENABLE_SENSORS, default=True): cv.boolean,
vol.Optional(CONF_UNITS, default="MBytes"): vol.In(UNITS),
vol.Optional(CONF_LOCAL_IP): vol.All(ip_address, cv.string), vol.Optional(CONF_LOCAL_IP): vol.All(ip_address, cv.string),
vol.Optional(CONF_PORTS): vol.Optional(CONF_UNITS, default="MBytes"): vol.In(UNITS),
vol.Schema({vol.Any(CONF_HASS, cv.positive_int): cv.positive_int})
}), }),
}, extra=vol.ALLOW_EXTRA) }, extra=vol.ALLOW_EXTRA)
async def _async_create_igd_device(hass: HomeAssistantType, ssdp_description: str): async def _async_create_igd_device(hass: HomeAssistantType,
ssdp_description: str):
""".""" """."""
# build requester # build requester
from async_upnp_client.aiohttp import AiohttpSessionRequester from async_upnp_client.aiohttp import AiohttpSessionRequester
@ -111,19 +92,22 @@ def _get_device(hass: HomeAssistantType, udn):
return hass.data[DOMAIN]['devices'][udn] return hass.data[DOMAIN]['devices'][udn]
async def _async_create_port_forward(hass: HomeAssistantType, igd_device): async def _async_add_port_mapping(hass: HomeAssistantType,
"""Create a port forward.""" igd_device,
_LOGGER.debug('Creating port forward: %s', igd_device) local_ip=None):
"""Create a port mapping."""
# determine local ip, ensure sane IP # determine local ip, ensure sane IP
if local_ip is None:
local_ip = get_local_ip() local_ip = get_local_ip()
if local_ip == '127.0.0.1': if local_ip == '127.0.0.1':
_LOGGER.warning('Could not create port forward, our IP is 127.0.0.1') _LOGGER.warning('Could not create port mapping, our IP is 127.0.0.1')
return False return False
local_ip = IPv4Address(local_ip) local_ip = IPv4Address(local_ip)
# create port mapping # create port mapping
port = hass.http.server_port port = hass.http.server_port
_LOGGER.debug('Creating port mapping %s:%s:%s (TCP)', port, local_ip, port)
await igd_device.async_add_port_mapping(remote_host=None, await igd_device.async_add_port_mapping(remote_host=None,
external_port=port, external_port=port,
protocol='TCP', protocol='TCP',
@ -136,48 +120,52 @@ async def _async_create_port_forward(hass: HomeAssistantType, igd_device):
return True return True
async def _async_remove_port_forward(hass: HomeAssistantType, igd_device): async def _async_delete_port_mapping(hass: HomeAssistantType, igd_device):
"""Remove a port forward.""" """Remove a port mapping."""
_LOGGER.debug('Removing port forward: %s', igd_device)
# remove port mapping
port = hass.http.server_port port = hass.http.server_port
await igd_device.async_remove_port_mapping(remote_host=None, await igd_device.async_delete_port_mapping(remote_host=None,
external_port=port, external_port=port,
protocol='TCP') protocol='TCP')
# config # config
async def async_setup(hass: HomeAssistantType, config): async def async_setup(hass: HomeAssistantType, config: ConfigType):
"""Register a port mapping for Home Assistant via UPnP.""" """Register a port mapping for Home Assistant via UPnP."""
_LOGGER.debug('async_setup: config: %s', config) # defaults
conf = config.get(DOMAIN, {}) hass.data[DOMAIN] = {
'auto_config': {
hass.data[DOMAIN] = hass.data.get(DOMAIN, {}) 'active': False,
configured = configured_udns(hass) 'port_forward': False,
_LOGGER.debug('configured: %s', configured) 'sensors': False,
# if no ssdp given: take any discovered - by flow - IGD entry
# if none discovered, raise PlatformNotReady
# if ssdp given: use the SSDP
igds = [] # XXX TODO
for igd_conf in igds:
hass.async_add_job(hass.config_entries.flow.async_init(
DOMAIN, context={'source': config_entries.SOURCE_IMPORT},
data={
'ssdp_description': igd_conf['ssdp_description'],
} }
)) }
# ensure sane config
if DOMAIN not in config:
return False
if DISCOVERY_DOMAIN not in config:
_LOGGER.warning('IGD needs discovery, please enable it')
return False
igd_config = config[DOMAIN]
if CONF_LOCAL_IP in igd_config:
hass.data[DOMAIN]['local_ip'] = igd_config[CONF_LOCAL_IP]
hass.data[DOMAIN]['auto_config'] = {
'active': True,
'port_forward': igd_config[CONF_ENABLE_PORT_MAPPING],
'sensors': igd_config[CONF_ENABLE_SENSORS],
}
_LOGGER.debug('Enabled auto_config: %s', hass.data[DOMAIN]['auto_config'])
return True return True
# config flow # config flow
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry): async def async_setup_entry(hass: HomeAssistantType,
config_entry: ConfigEntry):
"""Set up a bridge from a config entry.""" """Set up a bridge from a config entry."""
_LOGGER.debug('async_setup_entry: title: %s, data: %s', config_entry.title, config_entry.data)
data = config_entry.data data = config_entry.data
ssdp_description = data['ssdp_description'] ssdp_description = data['ssdp_description']
@ -189,44 +177,49 @@ async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigEntry):
_store_device(hass, igd_device.udn, igd_device) _store_device(hass, igd_device.udn, igd_device)
# port forward # port mapping
if data.get(CONF_PORT_FORWARD): if data.get(CONF_ENABLE_PORT_MAPPING):
await _async_create_port_forward(hass, igd_device) local_ip = hass.data[DOMAIN].get('local_ip')
await _async_add_port_mapping(hass, igd_device, local_ip=local_ip)
# sensors # sensors
if data.get(CONF_SENSORS): if data.get(CONF_ENABLE_SENSORS):
discovery_info = { discovery_info = {
'unit': 'MBytes', 'unit': 'MBytes',
'udn': data['udn'], 'udn': data['udn'],
} }
hass_config = config_entry.data hass_config = config_entry.data
hass.async_create_task(discovery.async_load_platform( hass.async_create_task(
discovery.async_load_platform(
hass, 'sensor', DOMAIN, discovery_info, hass_config)) hass, 'sensor', DOMAIN, discovery_info, hass_config))
async def unload_entry(event): async def unload_entry(event):
"""Unload entry on quit.""" """Unload entry on quit."""
await async_unload_entry(hass, config_entry) await async_unload_entry(hass, config_entry)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, unload_entry) hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, unload_entry)
return True return True
async def async_unload_entry(hass: HomeAssistantType, config_entry: ConfigEntry): async def async_unload_entry(hass: HomeAssistantType,
config_entry: ConfigEntry):
"""Unload a config entry.""" """Unload a config entry."""
_LOGGER.debug('async_unload_entry: title: %s, data: %s', config_entry.title, config_entry.data)
data = config_entry.data data = config_entry.data
udn = data['udn'] udn = data['udn']
igd_device = _get_device(hass, udn)
# port forward igd_device = _get_device(hass, udn)
if data.get(CONF_PORT_FORWARD): if igd_device is None:
_LOGGER.debug('Removing port forward for: %s', igd_device) return True
_async_remove_port_forward(hass, igd_device)
# port mapping
if data.get(CONF_ENABLE_PORT_MAPPING):
await _async_delete_port_mapping(hass, igd_device)
# sensors # sensors
if data.get(CONF_SENSORS): if data.get(CONF_ENABLE_SENSORS):
# XXX TODO: remove sensors # XXX TODO: remove sensors
pass pass
_store_device(hass, udn, None)
return True return True

60
homeassistant/components/igd/config_flow.py
View File

@ -1,11 +1,12 @@
"""Config flow for IGD.""" """Config flow for IGD."""
import voluptuous as vol
from homeassistant import config_entries, data_entry_flow from homeassistant import config_entries, data_entry_flow
from homeassistant.core import callback from homeassistant.core import callback
import voluptuous as vol from .const import CONF_ENABLE_PORT_MAPPING, CONF_ENABLE_SENSORS
from .const import DOMAIN from .const import DOMAIN
from .const import LOGGER as _LOGGER
@callback @callback
def configured_udns(hass): def configured_udns(hass):
@ -29,24 +30,23 @@ class IgdFlowHandler(data_entry_flow.FlowHandler):
@property @property
def _discovereds(self): def _discovereds(self):
"""Get all discovered entries.""" """Get all discovered entries."""
if DOMAIN not in self.hass.data:
_LOGGER.debug('DOMAIN not in hass.data')
if 'discovered' not in self.hass.data.get(DOMAIN, {}):
_LOGGER.debug('discovered not in hass.data[DOMAIN]')
return self.hass.data.get(DOMAIN, {}).get('discovered', {}) return self.hass.data.get(DOMAIN, {}).get('discovered', {})
def _store_discovery_info(self, discovery_info): def _store_discovery_info(self, discovery_info):
"""Add discovery info.""" """Add discovery info."""
udn = discovery_info['udn'] udn = discovery_info['udn']
if DOMAIN not in self.hass.data:
_LOGGER.debug('DOMAIN not in hass.data')
self.hass.data[DOMAIN] = self.hass.data.get(DOMAIN, {}) self.hass.data[DOMAIN] = self.hass.data.get(DOMAIN, {})
if 'discovered' not in self.hass.data[DOMAIN]: self.hass.data[DOMAIN]['discovered'] = \
_LOGGER.debug('Creating new discovered: %s', self.hass.data[DOMAIN]) self.hass.data[DOMAIN].get('discovered', {})
self.hass.data[DOMAIN]['discovered'] = self.hass.data[DOMAIN].get('discovered', {})
self.hass.data[DOMAIN]['discovered'][udn] = discovery_info self.hass.data[DOMAIN]['discovered'][udn] = discovery_info
def _auto_config_settings(self):
"""Check if auto_config has been enabled."""
self.hass.data[DOMAIN] = self.hass.data.get(DOMAIN, {})
return self.hass.data[DOMAIN].get('auto_config', {
'active': False,
})
async def async_step_discovery(self, discovery_info): async def async_step_discovery(self, discovery_info):
""" """
Handle a discovered IGD. Handle a discovered IGD.
@ -54,32 +54,33 @@ class IgdFlowHandler(data_entry_flow.FlowHandler):
This flow is triggered by the discovery component. It will check if the This flow is triggered by the discovery component. It will check if the
host is already configured and delegate to the import step if not. host is already configured and delegate to the import step if not.
""" """
_LOGGER.debug('async_step_discovery %s: %s', id(self), discovery_info)
# ensure not already discovered/configured # ensure not already discovered/configured
udn = discovery_info['udn'] udn = discovery_info['udn']
if udn in configured_udns(self.hass): if udn in configured_udns(self.hass):
_LOGGER.debug('Already configured: %s', discovery_info)
return self.async_abort(reason='already_configured') return self.async_abort(reason='already_configured')
# store discovered device # store discovered device
self._store_discovery_info(discovery_info) self._store_discovery_info(discovery_info)
# abort --> not showing up in discovered things # auto config?
# return self.async_abort(reason='user_input_required') auto_config = self._auto_config_settings()
if auto_config['active']:
import_info = {
'igd_host': discovery_info['host'],
'sensors': auto_config['sensors'],
'port_forward': auto_config['port_forward'],
}
return await self._async_save_entry(import_info)
# user -> showing up in discovered things
return await self.async_step_user() return await self.async_step_user()
async def async_step_user(self, user_input=None): async def async_step_user(self, user_input=None):
"""Manual set up.""" """Manual set up."""
_LOGGER.debug('async_step_user %s: %s', id(self), user_input)
# if user input given, handle it # if user input given, handle it
user_input = user_input or {} user_input = user_input or {}
if 'igd_host' in user_input: if 'igd_host' in user_input:
if not user_input['sensors'] and not user_input['port_forward']: if not user_input['sensors'] and not user_input['port_forward']:
_LOGGER.debug('Aborting, no sensors and no portforward')
return self.async_abort(reason='no_sensors_or_port_forward') return self.async_abort(reason='no_sensors_or_port_forward')
configured_hosts = [ configured_hosts = [
@ -90,10 +91,9 @@ class IgdFlowHandler(data_entry_flow.FlowHandler):
if user_input['igd_host'] in configured_hosts: if user_input['igd_host'] in configured_hosts:
return self.async_abort(reason='already_configured') return self.async_abort(reason='already_configured')
return await self._async_save(user_input) return await self._async_save_entry(user_input)
# let user choose from all discovered IGDs # let user choose from all discovered IGDs
_LOGGER.debug('Discovered devices: %s', self._discovereds)
igd_hosts = [ igd_hosts = [
entry['host'] entry['host']
for entry in self._discovereds.values() for entry in self._discovereds.values()
@ -111,10 +111,12 @@ class IgdFlowHandler(data_entry_flow.FlowHandler):
}) })
) )
async def _async_save(self, import_info): async def async_step_import(self, import_info):
"""Store IGD as new entry.""" """Import a new IGD as a config entry."""
_LOGGER.debug('async_step_import %s: %s', id(self), import_info) return await self._async_save_entry(import_info)
async def _async_save_entry(self, import_info):
"""Store IGD as new entry."""
# ensure we know the host # ensure we know the host
igd_host = import_info['igd_host'] igd_host = import_info['igd_host']
discovery_infos = [info discovery_infos = [info
@ -129,7 +131,7 @@ class IgdFlowHandler(data_entry_flow.FlowHandler):
data={ data={
'ssdp_description': discovery_info['ssdp_description'], 'ssdp_description': discovery_info['ssdp_description'],
'udn': discovery_info['udn'], 'udn': discovery_info['udn'],
'sensors': import_info['sensors'], CONF_ENABLE_SENSORS: import_info['sensors'],
'port_forward': import_info['port_forward'], CONF_ENABLE_PORT_MAPPING: import_info['port_forward'],
}, },
) )

4
homeassistant/components/igd/const.py
View File

@ -3,5 +3,5 @@ import logging
DOMAIN = 'igd' DOMAIN = 'igd'
LOGGER = logging.getLogger('homeassistant.components.igd') LOGGER = logging.getLogger('homeassistant.components.igd')
CONF_PORT_FORWARD = 'port_forward' CONF_ENABLE_PORT_MAPPING = 'port_forward'
CONF_SENSORS = 'sensors' CONF_ENABLE_SENSORS = 'sensors'

18
homeassistant/components/sensor/igd.py
View File

@ -1,15 +1,17 @@
""" """
Support for UPnP Sensors (IGD). Support for IGD Sensors.
For more details about this platform, please refer to the documentation at For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/sensor.upnp/ https://home-assistant.io/components/sensor.igd/
""" """
# pylint: disable=invalid-name
import logging import logging
from homeassistant.components import history from homeassistant.components import history
from homeassistant.components.igd import DOMAIN, UNITS from homeassistant.components.igd import DOMAIN, UNITS
from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity import Entity
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['igd', 'history'] DEPENDENCIES = ['igd', 'history']
@ -98,9 +100,6 @@ class IGDSensor(Entity):
self._handle_new_value(new_value) self._handle_new_value(new_value)
# _LOGGER.debug('Removing self: %s', self)
# await self.async_remove() # XXX TODO: does not remove from the UI
@property @property
def _last_state(self): def _last_state(self):
"""Get the last state reported to hass.""" """Get the last state reported to hass."""
@ -126,17 +125,11 @@ class IGDSensor(Entity):
try: try:
state = coercer(float(last_state.state)) * self.unit_factor state = coercer(float(last_state.state)) * self.unit_factor
except ValueError: except ValueError:
_LOGGER.debug('%s: value error, coercer: %s, state: %s', self.entity_id, coercer, last_state.state)
raise
state = coercer(0.0) state = coercer(0.0)
return state return state
def _handle_new_value(self, new_value): def _handle_new_value(self, new_value):
_LOGGER.debug('%s: handle_new_value: state: %s, new_value: %s, last_value: %s',
self.entity_id, self._state, new_value, self._last_value)
# ❯❯❯ upnp-client --debug --pprint --device http://192.168.178.1/RootDevice.xml call-action WANCIFC/GetTotalBytesReceived
if self.entity_id is None: if self.entity_id is None:
# don't know our entity ID yet, do nothing but store value # don't know our entity ID yet, do nothing but store value
self._last_value = new_value self._last_value = new_value
@ -161,7 +154,8 @@ class IGDSensor(Entity):
if new_value >= 0: if new_value >= 0:
diff += new_value diff += new_value
else: else:
# some devices don't overflow and start at 0, but somewhere to -2**32 # some devices don't overflow and start at 0,
# but somewhere to -2**32
diff += new_value - -OVERFLOW_AT diff += new_value - -OVERFLOW_AT
self._state += diff self._state += diff

4
requirements_all.txt
View File

@ -141,6 +141,7 @@ apns2==0.3.0
# homeassistant.components.asterisk_mbox # homeassistant.components.asterisk_mbox
asterisk_mbox==0.4.0 asterisk_mbox==0.4.0
# homeassistant.components.igd
# homeassistant.components.media_player.dlna_dmr # homeassistant.components.media_player.dlna_dmr
async-upnp-client==0.12.4 async-upnp-client==0.12.4
@ -1183,9 +1184,6 @@ pytrafikverket==0.1.5.8
# homeassistant.components.device_tracker.unifi # homeassistant.components.device_tracker.unifi
pyunifi==2.13 pyunifi==2.13
# homeassistant.components.upnp
pyupnp-async==0.1.1.1
# homeassistant.components.binary_sensor.uptimerobot # homeassistant.components.binary_sensor.uptimerobot
pyuptimerobot==0.0.5 pyuptimerobot==0.0.5

3
requirements_test_all.txt
View File

@ -177,9 +177,6 @@ pytradfri[async]==5.5.1
# homeassistant.components.device_tracker.unifi # homeassistant.components.device_tracker.unifi
pyunifi==2.13 pyunifi==2.13
# homeassistant.components.upnp
pyupnp-async==0.1.1.1
# homeassistant.components.notify.html5 # homeassistant.components.notify.html5
pywebpush==1.6.0 pywebpush==1.6.0

84
tests/components/igd/test_config_flow.py
View File

@ -1,13 +1,14 @@
"""Tests for IGD config flow.""" """Tests for IGD config flow."""
from homeassistant.components import igd from homeassistant.components import igd
from homeassistant.components.igd import config_flow as igd_config_flow
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
async def test_flow_none_discovered(hass): async def test_flow_none_discovered(hass):
"""Test no device discovered flow.""" """Test no device discovered flow."""
flow = igd.config_flow.IgdFlowHandler() flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass flow.hass = hass
result = await flow.async_step_user() result = await flow.async_step_user()
@ -17,7 +18,7 @@ async def test_flow_none_discovered(hass):
async def test_flow_already_configured(hass): async def test_flow_already_configured(hass):
"""Test device already configured flow.""" """Test device already configured flow."""
flow = igd.config_flow.IgdFlowHandler() flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass flow.hass = hass
# discovered device # discovered device
@ -48,7 +49,7 @@ async def test_flow_already_configured(hass):
async def test_flow_no_sensors_no_port_forward(hass): async def test_flow_no_sensors_no_port_forward(hass):
"""Test single device, no sensors, no port_forward.""" """Test single device, no sensors, no port_forward."""
flow = igd.config_flow.IgdFlowHandler() flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass flow.hass = hass
# discovered device # discovered device
@ -79,7 +80,7 @@ async def test_flow_no_sensors_no_port_forward(hass):
async def test_flow_discovered_form(hass): async def test_flow_discovered_form(hass):
"""Test single device discovered, show form flow.""" """Test single device discovered, show form flow."""
flow = igd.config_flow.IgdFlowHandler() flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass flow.hass = hass
# discovered device # discovered device
@ -100,7 +101,7 @@ async def test_flow_discovered_form(hass):
async def test_flow_two_discovered_form(hass): async def test_flow_two_discovered_form(hass):
"""Test single device discovered, show form flow.""" """Test single device discovered, show form flow."""
flow = igd.config_flow.IgdFlowHandler() flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass flow.hass = hass
# discovered device # discovered device
@ -135,18 +136,18 @@ async def test_flow_two_discovered_form(hass):
async def test_config_entry_created(hass): async def test_config_entry_created(hass):
flow = igd.config_flow.IgdFlowHandler() """Test config entry is created."""
flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass flow.hass = hass
# discovered device # discovered device
udn = 'uuid:device_1'
hass.data[igd.DOMAIN] = { hass.data[igd.DOMAIN] = {
'discovered': { 'discovered': {
udn: { 'uuid:device_1': {
'name': 'Test device 1', 'name': 'Test device 1',
'host': '192.168.1.1', 'host': '192.168.1.1',
'ssdp_description': 'http://192.168.1.1/desc.xml', 'ssdp_description': 'http://192.168.1.1/desc.xml',
'udn': udn, 'udn': 'uuid:device_1',
}, },
}, },
} }
@ -156,6 +157,7 @@ async def test_config_entry_created(hass):
'sensors': True, 'sensors': True,
'port_forward': False, 'port_forward': False,
}) })
assert result['type'] == 'create_entry'
assert result['data'] == { assert result['data'] == {
'port_forward': False, 'port_forward': False,
'sensors': True, 'sensors': True,
@ -163,3 +165,67 @@ async def test_config_entry_created(hass):
'udn': 'uuid:device_1', 'udn': 'uuid:device_1',
} }
assert result['title'] == 'Test device 1' assert result['title'] == 'Test device 1'
async def test_flow_discovery_auto_config_sensors(hass):
"""Test creation of device with auto_config."""
flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass
# auto_config active
hass.data[igd.DOMAIN] = {
'auto_config': {
'active': True,
'port_forward': False,
'sensors': True,
},
}
# discovered device
result = await flow.async_step_discovery({
'name': 'Test device 1',
'host': '192.168.1.1',
'ssdp_description': 'http://192.168.1.1/desc.xml',
'udn': 'uuid:device_1',
})
assert result['type'] == 'create_entry'
assert result['data'] == {
'port_forward': False,
'sensors': True,
'ssdp_description': 'http://192.168.1.1/desc.xml',
'udn': 'uuid:device_1',
}
assert result['title'] == 'Test device 1'
async def test_flow_discovery_auto_config_sensors_port_forward(hass):
"""Test creation of device with auto_config, with port forward."""
flow = igd_config_flow.IgdFlowHandler()
flow.hass = hass
# auto_config active, with port_forward
hass.data[igd.DOMAIN] = {
'auto_config': {
'active': True,
'port_forward': True,
'sensors': True,
},
}
# discovered device
result = await flow.async_step_discovery({
'name': 'Test device 1',
'host': '192.168.1.1',
'ssdp_description': 'http://192.168.1.1/desc.xml',
'udn': 'uuid:device_1',
})
assert result['type'] == 'create_entry'
assert result['data'] == {
'port_forward': True,
'sensors': True,
'ssdp_description': 'http://192.168.1.1/desc.xml',
'udn': 'uuid:device_1',
}
assert result['title'] == 'Test device 1'

104
tests/components/igd/test_init.py
View File

@ -10,9 +10,96 @@ from tests.common import MockConfigEntry
from tests.common import mock_coro from tests.common import mock_coro
async def test_async_setup_entry_port_forward_created(hass): async def test_async_setup_no_auto_config(hass):
"""Test async_setup_entry.""" """Test async_setup."""
# setup component, enable auto_config
await async_setup_component(hass, 'igd')
assert hass.data[igd.DOMAIN]['auto_config'] == {
'active': False,
'port_forward': False,
'sensors': False,
}
async def test_async_setup_auto_config(hass):
"""Test async_setup."""
# setup component, enable auto_config
await async_setup_component(hass, 'igd', {'igd': {}, 'discovery': {}})
assert hass.data[igd.DOMAIN]['auto_config'] == {
'active': True,
'port_forward': False,
'sensors': True,
}
async def test_async_setup_auto_config_port_forward(hass):
"""Test async_setup."""
# setup component, enable auto_config
await async_setup_component(hass, 'igd', {
'igd': {'port_forward': True},
'discovery': {}})
assert hass.data[igd.DOMAIN]['auto_config'] == {
'active': True,
'port_forward': True,
'sensors': True,
}
async def test_async_setup_auto_config_no_sensors(hass):
"""Test async_setup."""
# setup component, enable auto_config
await async_setup_component(hass, 'igd', {
'igd': {'sensors': False},
'discovery': {}})
assert hass.data[igd.DOMAIN]['auto_config'] == {
'active': True,
'port_forward': False,
'sensors': False,
}
async def test_async_setup_entry_default(hass):
"""Test async_setup_entry."""
udn = 'uuid:device_1'
entry = MockConfigEntry(domain=igd.DOMAIN, data={
'ssdp_description': 'http://192.168.1.1/desc.xml',
'udn': udn,
'sensors': True,
'port_forward': False,
})
# ensure hass.http is available
await async_setup_component(hass, 'igd')
# mock async_upnp_client.igd.IgdDevice
mock_igd_device = MagicMock()
mock_igd_device.udn = udn
mock_igd_device.async_add_port_mapping.return_value = mock_coro()
mock_igd_device.async_delete_port_mapping.return_value = mock_coro()
with patch.object(igd, '_async_create_igd_device') as mock_create_device:
mock_create_device.return_value = mock_coro(
return_value=mock_igd_device)
with patch('homeassistant.components.igd.get_local_ip',
return_value='192.168.1.10'):
assert await igd.async_setup_entry(hass, entry) is True
# ensure device is stored/used
assert hass.data[igd.DOMAIN]['devices'][udn] == mock_igd_device
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert hass.data[igd.DOMAIN]['devices'][udn] is None
assert len(mock_igd_device.async_add_port_mapping.mock_calls) == 0
assert len(mock_igd_device.async_delete_port_mapping.mock_calls) == 0
async def test_async_setup_entry_port_forward(hass):
"""Test async_setup_entry."""
udn = 'uuid:device_1' udn = 'uuid:device_1'
entry = MockConfigEntry(domain=igd.DOMAIN, data={ entry = MockConfigEntry(domain=igd.DOMAIN, data={
'ssdp_description': 'http://192.168.1.1/desc.xml', 'ssdp_description': 'http://192.168.1.1/desc.xml',
@ -27,15 +114,20 @@ async def test_async_setup_entry_port_forward_created(hass):
mock_igd_device = MagicMock() mock_igd_device = MagicMock()
mock_igd_device.udn = udn mock_igd_device.udn = udn
mock_igd_device.async_add_port_mapping.return_value = mock_coro() mock_igd_device.async_add_port_mapping.return_value = mock_coro()
mock_igd_device.async_remove_port_mapping.return_value = mock_coro() mock_igd_device.async_delete_port_mapping.return_value = mock_coro()
with patch.object(igd, '_async_create_igd_device') as mock_create_device: with patch.object(igd, '_async_create_igd_device') as mock_create_device:
mock_create_device.return_value = mock_coro(return_value=mock_igd_device) mock_create_device.return_value = mock_coro(
with patch('homeassistant.components.igd.get_local_ip', return_value='192.168.1.10'): return_value=mock_igd_device)
with patch('homeassistant.components.igd.get_local_ip',
return_value='192.168.1.10'):
assert await igd.async_setup_entry(hass, entry) is True assert await igd.async_setup_entry(hass, entry) is True
# ensure device is stored/used
assert hass.data[igd.DOMAIN]['devices'][udn] == mock_igd_device
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP) hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.data[igd.DOMAIN]['devices'][udn] == mock_igd_device assert hass.data[igd.DOMAIN]['devices'][udn] is None
assert len(mock_igd_device.async_add_port_mapping.mock_calls) > 0 assert len(mock_igd_device.async_add_port_mapping.mock_calls) > 0
assert len(mock_igd_device.async_delete_port_mapping.mock_calls) > 0 assert len(mock_igd_device.async_delete_port_mapping.mock_calls) > 0