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

Merge pull request #24636 from home-assistant/rc 

0.94.4
pull/24637/head 0.94.4
Paulus Schoutsen 2019-06-19 16:36:04 -07:00 committed by GitHub
parent aa91211229 4e0683565d
commit d85ae5dcae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 576 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
homeassistant/components/esphome/config_flow.py
View File

@ -56,6 +56,7 @@ class EsphomeFlowHandler(config_entries.ConfigFlow):
self.context['title_placeholders'] = {
'name': self._name
}
self.context['name'] = self._name
# Only show authentication step if device uses password
if device_info.uses_password:
@ -98,9 +99,11 @@ class EsphomeFlowHandler(config_entries.ConfigFlow):
already_configured = data.device_info.name == node_name
if already_configured:
return self.async_abort(
reason='already_configured'
)
return self.async_abort(reason='already_configured')
for flow in self._async_in_progress():
if flow['context']['name'] == node_name:
return self.async_abort(reason='already_configured')
return await self._async_authenticate_or_add(user_input={
'host': address,

1
homeassistant/components/sun/__init__.py
View File

@ -170,6 +170,7 @@ class Sun(Entity):
utc_point_in_time, 'dusk', PHASE_ASTRONOMICAL_TWILIGHT)
self.next_midnight = self._check_event(
utc_point_in_time, 'solar_midnight', None)
self.location.solar_depression = 'civil'
# if the event was solar midday or midnight, phase will now
# be None. Solar noon doesn't always happen when the sun is

116
homeassistant/components/tplink/__init__.py
View File

@ -6,28 +6,43 @@ import voluptuous as vol
from homeassistant.const import CONF_HOST
from homeassistant import config_entries
import homeassistant.helpers.config_validation as cv
from .config_flow import async_get_devices
from .const import DOMAIN
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from .common import (
async_discover_devices,
get_static_devices,
ATTR_CONFIG,
CONF_DIMMER,
CONF_DISCOVERY,
CONF_LIGHT,
CONF_SWITCH,
SmartDevices
)
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'tplink'
TPLINK_HOST_SCHEMA = vol.Schema({
vol.Required(CONF_HOST): cv.string
})
CONF_LIGHT = 'light'
CONF_SWITCH = 'switch'
CONF_DISCOVERY = 'discovery'
ATTR_CONFIG = 'config'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional('light', default=[]): vol.All(cv.ensure_list,
[TPLINK_HOST_SCHEMA]),
vol.Optional('switch', default=[]): vol.All(cv.ensure_list,
[TPLINK_HOST_SCHEMA]),
vol.Optional('discovery', default=True): cv.boolean,
vol.Optional(CONF_LIGHT, default=[]): vol.All(
cv.ensure_list,
[TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_SWITCH, default=[]): vol.All(
cv.ensure_list,
[TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_DIMMER, default=[]): vol.All(
cv.ensure_list,
[TPLINK_HOST_SCHEMA]
),
vol.Optional(CONF_DISCOVERY, default=True): cv.boolean,
}),
}, extra=vol.ALLOW_EXTRA)
@ -46,76 +61,45 @@ async def async_setup(hass, config):
return True
async def async_setup_entry(hass, config_entry):
async def async_setup_entry(hass: HomeAssistantType, config_entry: ConfigType):
"""Set up TPLink from a config entry."""
from pyHS100 import SmartBulb, SmartPlug, SmartDeviceException
devices = {}
config_data = hass.data[DOMAIN].get(ATTR_CONFIG)
# These will contain the initialized devices
lights = hass.data[DOMAIN][CONF_LIGHT] = []
switches = hass.data[DOMAIN][CONF_SWITCH] = []
# If discovery is defined and not disabled, discover devices
# If initialized from configure integrations, there's no config
# so we default here to True
if config_data is None or config_data[CONF_DISCOVERY]:
devs = await async_get_devices(hass)
_LOGGER.info("Discovered %s TP-Link smart home device(s)", len(devs))
devices.update(devs)
def _device_for_type(host, type_):
dev = None
if type_ == CONF_LIGHT:
dev = SmartBulb(host)
elif type_ == CONF_SWITCH:
dev = SmartPlug(host)
return dev
# When arriving from configure integrations, we have no config data.
# Add static devices
static_devices = SmartDevices()
if config_data is not None:
for type_ in [CONF_LIGHT, CONF_SWITCH]:
for entry in config_data[type_]:
try:
host = entry['host']
dev = _device_for_type(host, type_)
devices[host] = dev
_LOGGER.debug("Succesfully added %s %s: %s",
type_, host, dev)
except SmartDeviceException as ex:
_LOGGER.error("Unable to initialize %s %s: %s",
type_, host, ex)
static_devices = get_static_devices(
config_data,
)
# This is necessary to avoid I/O blocking on is_dimmable
def _fill_device_lists():
for dev in devices.values():
if isinstance(dev, SmartPlug):
try:
if dev.is_dimmable: # Dimmers act as lights
lights.append(dev)
else:
switches.append(dev)
except SmartDeviceException as ex:
_LOGGER.error("Unable to connect to device %s: %s",
dev.host, ex)
lights.extend(static_devices.lights)
switches.extend(static_devices.switches)
elif isinstance(dev, SmartBulb):
lights.append(dev)
else:
_LOGGER.error("Unknown smart device type: %s", type(dev))
# Add discovered devices
if config_data is None or config_data[CONF_DISCOVERY]:
discovered_devices = await async_discover_devices(hass, static_devices)
# Avoid blocking on is_dimmable
await hass.async_add_executor_job(_fill_device_lists)
lights.extend(discovered_devices.lights)
switches.extend(discovered_devices.switches)
forward_setup = hass.config_entries.async_forward_entry_setup
if lights:
_LOGGER.debug("Got %s lights: %s", len(lights), lights)
_LOGGER.debug(
"Got %s lights: %s",
len(lights),
", ".join([d.host for d in lights])
)
hass.async_create_task(forward_setup(config_entry, 'light'))
if switches:
_LOGGER.debug("Got %s switches: %s", len(switches), switches)
_LOGGER.debug(
"Got %s switches: %s",
len(switches),
", ".join([d.host for d in switches])
)
hass.async_create_task(forward_setup(config_entry, 'switch'))
return True

202
homeassistant/components/tplink/common.py Normal file
View File

@ -0,0 +1,202 @@
"""Common code for tplink."""
import asyncio
import logging
from datetime import timedelta
from typing import Any, Callable, List
from pyHS100 import (
SmartBulb,
SmartDevice,
SmartPlug,
SmartDeviceException
)
from homeassistant.helpers.typing import HomeAssistantType
_LOGGER = logging.getLogger(__name__)
ATTR_CONFIG = 'config'
CONF_DIMMER = 'dimmer'
CONF_DISCOVERY = 'discovery'
CONF_LIGHT = 'light'
CONF_SWITCH = 'switch'
class SmartDevices:
"""Hold different kinds of devices."""
def __init__(
self,
lights: List[SmartDevice] = None,
switches: List[SmartDevice] = None
):
"""Constructor."""
self._lights = lights or []
self._switches = switches or []
@property
def lights(self):
"""Get the lights."""
return self._lights
@property
def switches(self):
"""Get the switches."""
return self._switches
def has_device_with_host(self, host):
"""Check if a devices exists with a specific host."""
for device in self.lights + self.switches:
if device.host == host:
return True
return False
async def async_get_discoverable_devices(hass):
"""Return if there are devices that can be discovered."""
from pyHS100 import Discover
def discover():
devs = Discover.discover()
return devs
return await hass.async_add_executor_job(discover)
async def async_discover_devices(
hass: HomeAssistantType,
existing_devices: SmartDevices
) -> SmartDevices:
"""Get devices through discovery."""
_LOGGER.debug("Discovering devices")
devices = await async_get_discoverable_devices(hass)
_LOGGER.info(
"Discovered %s TP-Link smart home device(s)",
len(devices)
)
lights = []
switches = []
def process_devices():
for dev in devices.values():
# If this device already exists, ignore dynamic setup.
if existing_devices.has_device_with_host(dev.host):
continue
if isinstance(dev, SmartPlug):
try:
if dev.is_dimmable: # Dimmers act as lights
lights.append(dev)
else:
switches.append(dev)
except SmartDeviceException as ex:
_LOGGER.error("Unable to connect to device %s: %s",
dev.host, ex)
elif isinstance(dev, SmartBulb):
lights.append(dev)
else:
_LOGGER.error("Unknown smart device type: %s", type(dev))
await hass.async_add_executor_job(process_devices)
return SmartDevices(lights, switches)
def get_static_devices(config_data) -> SmartDevices:
"""Get statically defined devices in the config."""
_LOGGER.debug("Getting static devices")
lights = []
switches = []
for type_ in [CONF_LIGHT, CONF_SWITCH, CONF_DIMMER]:
for entry in config_data[type_]:
host = entry['host']
if type_ == CONF_LIGHT:
lights.append(SmartBulb(host))
elif type_ == CONF_SWITCH:
switches.append(SmartPlug(host))
# Dimmers need to be defined as smart plugs to work correctly.
elif type_ == CONF_DIMMER:
lights.append(SmartPlug(host))
return SmartDevices(
lights,
switches
)
async def async_add_entities_retry(
hass: HomeAssistantType,
async_add_entities: Callable[[List[Any], bool], None],
objects: List[Any],
callback: Callable[[Any, Callable], None],
interval: timedelta = timedelta(seconds=60)
):
"""
Add entities now and retry later if issues are encountered.
If the callback throws an exception or returns false, that
object will try again a while later.
This is useful for devices that are not online when hass starts.
:param hass:
:param async_add_entities: The callback provided to a
platform's async_setup.
:param objects: The objects to create as entities.
:param callback: The callback that will perform the add.
:param interval: THe time between attempts to add.
:return: A callback to cancel the retries.
"""
add_objects = objects.copy()
is_cancelled = False
def cancel_interval_callback():
nonlocal is_cancelled
is_cancelled = True
async def process_objects_loop(delay: int):
if is_cancelled:
return
await process_objects()
if not add_objects:
return
await asyncio.sleep(delay)
hass.async_create_task(process_objects_loop(delay))
async def process_objects(*args):
# Process each object.
for add_object in list(add_objects):
# Call the individual item callback.
try:
_LOGGER.debug(
"Attempting to add object of type %s",
type(add_object)
)
result = await hass.async_add_job(
callback,
add_object,
async_add_entities
)
except SmartDeviceException as ex:
_LOGGER.debug(
str(ex)
)
result = False
if result is True or result is None:
_LOGGER.debug("Added object.")
add_objects.remove(add_object)
else:
_LOGGER.debug("Failed to add object, will try again later")
await process_objects_loop(interval.seconds)
return cancel_interval_callback

13
homeassistant/components/tplink/config_flow.py
View File

@ -2,19 +2,10 @@
from homeassistant.helpers import config_entry_flow
from homeassistant import config_entries
from .const import DOMAIN
async def async_get_devices(hass):
"""Return if there are devices that can be discovered."""
from pyHS100 import Discover
def discover():
devs = Discover.discover()
return devs
return await hass.async_add_executor_job(discover)
from .common import async_get_discoverable_devices
config_entry_flow.register_discovery_flow(DOMAIN,
'TP-Link Smart Home',
async_get_devices,
async_get_discoverable_devices,
config_entries.CONN_CLASS_LOCAL_POLL)

58
homeassistant/components/tplink/light.py
View File

@ -2,15 +2,19 @@
import logging
import time
from pyHS100 import SmartBulb, SmartDeviceException
from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_COLOR_TEMP, ATTR_HS_COLOR, SUPPORT_BRIGHTNESS,
SUPPORT_COLOR, SUPPORT_COLOR_TEMP, Light)
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util.color import (
color_temperature_kelvin_to_mired as kelvin_to_mired,
color_temperature_mired_to_kelvin as mired_to_kelvin)
from . import CONF_LIGHT, DOMAIN as TPLINK_DOMAIN
from .common import async_add_entities_retry
PARALLEL_UPDATES = 0
@ -31,17 +35,35 @@ async def async_setup_platform(hass, config, add_entities,
'convert to use the tplink component.')
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up discovered switches."""
devs = []
for dev in hass.data[TPLINK_DOMAIN][CONF_LIGHT]:
devs.append(TPLinkSmartBulb(dev))
async_add_entities(devs, True)
async def async_setup_entry(
hass: HomeAssistantType,
config_entry,
async_add_entities
):
"""Set up switches."""
await async_add_entities_retry(
hass,
async_add_entities,
hass.data[TPLINK_DOMAIN][CONF_LIGHT],
add_entity
)
return True
def add_entity(device: SmartBulb, async_add_entities):
"""Check if device is online and add the entity."""
# Attempt to get the sysinfo. If it fails, it will raise an
# exception that is caught by async_add_entities_retry which
# will try again later.
device.get_sysinfo()
async_add_entities(
[TPLinkSmartBulb(device)],
update_before_add=True
)
def brightness_to_percentage(byt):
"""Convert brightness from absolute 0..255 to percentage."""
return int((byt*100.0)/255.0)
@ -55,7 +77,7 @@ def brightness_from_percentage(percent):
class TPLinkSmartBulb(Light):
"""Representation of a TPLink Smart Bulb."""
def __init__(self, smartbulb) -> None:
def __init__(self, smartbulb: SmartBulb) -> None:
"""Initialize the bulb."""
self.smartbulb = smartbulb
self._sysinfo = None
@ -69,25 +91,29 @@ class TPLinkSmartBulb(Light):
self._max_mireds = None
self._emeter_params = {}
self._mac = None
self._alias = None
self._model = None
@property
def unique_id(self):
"""Return a unique ID."""
return self._sysinfo["mac"]
return self._mac
@property
def name(self):
"""Return the name of the Smart Bulb."""
return self._sysinfo["alias"]
return self._alias
@property
def device_info(self):
"""Return information about the device."""
return {
"name": self.name,
"model": self._sysinfo["model"],
"name": self._alias,
"model": self._model,
"manufacturer": 'TP-Link',
"connections": {
(dr.CONNECTION_NETWORK_MAC, self._sysinfo["mac"])
(dr.CONNECTION_NETWORK_MAC, self._mac)
},
"sw_version": self._sysinfo["sw_ver"],
}
@ -104,7 +130,6 @@ class TPLinkSmartBulb(Light):
def turn_on(self, **kwargs):
"""Turn the light on."""
from pyHS100 import SmartBulb
self.smartbulb.state = SmartBulb.BULB_STATE_ON
if ATTR_COLOR_TEMP in kwargs:
@ -122,7 +147,6 @@ class TPLinkSmartBulb(Light):
def turn_off(self, **kwargs):
"""Turn the light off."""
from pyHS100 import SmartBulb
self.smartbulb.state = SmartBulb.BULB_STATE_OFF
@property
@ -157,7 +181,6 @@ class TPLinkSmartBulb(Light):
def update(self):
"""Update the TP-Link Bulb's state."""
from pyHS100 import SmartDeviceException, SmartBulb
try:
if self._supported_features is None:
self.get_features()
@ -212,6 +235,9 @@ class TPLinkSmartBulb(Light):
"""Determine all supported features in one go."""
self._sysinfo = self.smartbulb.sys_info
self._supported_features = 0
self._mac = self.smartbulb.mac
self._alias = self.smartbulb.alias
self._model = self.smartbulb.model
if self.smartbulb.is_dimmable:
self._supported_features += SUPPORT_BRIGHTNESS

54
homeassistant/components/tplink/switch.py
View File

@ -2,12 +2,16 @@
import logging
import time
from pyHS100 import SmartDeviceException, SmartPlug
from homeassistant.components.switch import (
ATTR_CURRENT_POWER_W, ATTR_TODAY_ENERGY_KWH, SwitchDevice)
from homeassistant.const import ATTR_VOLTAGE
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.typing import HomeAssistantType
from . import CONF_SWITCH, DOMAIN as TPLINK_DOMAIN
from .common import async_add_entities_retry
PARALLEL_UPDATES = 0
@ -27,13 +31,31 @@ async def async_setup_platform(hass, config, add_entities,
'convert to use the tplink component.')
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up discovered switches."""
devs = []
for dev in hass.data[TPLINK_DOMAIN][CONF_SWITCH]:
devs.append(SmartPlugSwitch(dev))
def add_entity(device: SmartPlug, async_add_entities):
"""Check if device is online and add the entity."""
# Attempt to get the sysinfo. If it fails, it will raise an
# exception that is caught by async_add_entities_retry which
# will try again later.
device.get_sysinfo()
async_add_entities(devs, True)
async_add_entities(
[SmartPlugSwitch(device)],
update_before_add=True
)
async def async_setup_entry(
hass: HomeAssistantType,
config_entry,
async_add_entities
):
"""Set up switches."""
await async_add_entities_retry(
hass,
async_add_entities,
hass.data[TPLINK_DOMAIN][CONF_SWITCH],
add_entity
)
return True
@ -41,7 +63,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class SmartPlugSwitch(SwitchDevice):
"""Representation of a TPLink Smart Plug switch."""
def __init__(self, smartplug):
def __init__(self, smartplug: SmartPlug):
"""Initialize the switch."""
self.smartplug = smartplug
self._sysinfo = None
@ -50,25 +72,29 @@ class SmartPlugSwitch(SwitchDevice):
# Set up emeter cache
self._emeter_params = {}
self._mac = None
self._alias = None
self._model = None
@property
def unique_id(self):
"""Return a unique ID."""
return self._sysinfo["mac"]
return self._mac
@property
def name(self):
"""Return the name of the Smart Plug."""
return self._sysinfo["alias"]
return self._alias
@property
def device_info(self):
"""Return information about the device."""
return {
"name": self.name,
"model": self._sysinfo["model"],
"name": self._alias,
"model": self._model,
"manufacturer": 'TP-Link',
"connections": {
(dr.CONNECTION_NETWORK_MAC, self._sysinfo["mac"])
(dr.CONNECTION_NETWORK_MAC, self._mac)
},
"sw_version": self._sysinfo["sw_ver"],
}
@ -98,10 +124,12 @@ class SmartPlugSwitch(SwitchDevice):
def update(self):
"""Update the TP-Link switch's state."""
from pyHS100 import SmartDeviceException
try:
if not self._sysinfo:
self._sysinfo = self.smartplug.sys_info
self._mac = self.smartplug.mac
self._alias = self.smartplug.alias
self._model = self.smartplug.model
self._state = self.smartplug.state == \
self.smartplug.SWITCH_STATE_ON

2
homeassistant/const.py
View File

@ -2,7 +2,7 @@
"""Constants used by Home Assistant components."""
MAJOR_VERSION = 0
MINOR_VERSION = 94
PATCH_VERSION = '3'
PATCH_VERSION = '4'
__short_version__ = '{}.{}'.format(MAJOR_VERSION, MINOR_VERSION)
__version__ = '{}.{}'.format(__short_version__, PATCH_VERSION)
REQUIRED_PYTHON_VER = (3, 5, 3)

27
tests/components/esphome/test_config_flow.py
View File

@ -281,3 +281,30 @@ async def test_discovery_already_configured_name(hass, mock_client):
result = await flow.async_step_zeroconf(user_input=service_info)
assert result['type'] == 'abort'
assert result['reason'] == 'already_configured'
async def test_discovery_duplicate_data(hass, mock_client):
"""Test discovery aborts if same mDNS packet arrives."""
service_info = {
'host': '192.168.43.183',
'port': 6053,
'hostname': 'test8266.local.',
'properties': {
"address": "test8266.local"
}
}
mock_client.device_info.return_value = mock_coro(
MockDeviceInfo(False, "test8266"))
result = await hass.config_entries.flow.async_init(
'esphome', data=service_info, context={'source': 'zeroconf'}
)
assert result['type'] == 'form'
assert result['step_id'] == 'discovery_confirm'
result = await hass.config_entries.flow.async_init(
'esphome', data=service_info, context={'source': 'zeroconf'}
)
assert result['type'] == 'abort'
assert result['reason'] == 'already_configured'

97
tests/components/tplink/test_common.py Normal file
View File

@ -0,0 +1,97 @@
"""Common code tests."""
from datetime import timedelta
from unittest.mock import MagicMock
from pyHS100 import SmartDeviceException
from homeassistant.components.tplink.common import async_add_entities_retry
from homeassistant.helpers.typing import HomeAssistantType
async def test_async_add_entities_retry(
hass: HomeAssistantType
):
"""Test interval callback."""
async_add_entities_callback = MagicMock()
# The objects that will be passed to async_add_entities_callback.
objects = [
"Object 1",
"Object 2",
"Object 3",
"Object 4",
]
# For each call to async_add_entities_callback, the following side effects
# will be triggered in order. This set of side effects accuratley simulates
# 3 attempts to add all entities while also handling several return types.
# To help understand what's going on, a comment exists describing what the
# object list looks like throughout the iterations.
callback_side_effects = [
# OB1, OB2, OB3, OB4
False,
False,
True, # Object 3
False,
# OB1, OB2, OB4
True, # Object 1
SmartDeviceException("My error"),
False,
# OB2, OB4
True, # Object 2
True, # Object 4
]
callback = MagicMock(side_effect=callback_side_effects)
await async_add_entities_retry(
hass,
async_add_entities_callback,
objects,
callback,
interval=timedelta(milliseconds=100)
)
await hass.async_block_till_done()
assert callback.call_count == len(callback_side_effects)
async def test_async_add_entities_retry_cancel(
hass: HomeAssistantType
):
"""Test interval callback."""
async_add_entities_callback = MagicMock()
callback_side_effects = [
False,
False,
True, # Object 1
False,
True, # Object 2
SmartDeviceException("My error"),
False,
True, # Object 3
True, # Object 4
]
callback = MagicMock(side_effect=callback_side_effects)
objects = [
"Object 1",
"Object 2",
"Object 3",
"Object 4",
]
cancel = await async_add_entities_retry(
hass,
async_add_entities_callback,
objects,
callback,
interval=timedelta(milliseconds=100)
)
cancel()
await hass.async_block_till_done()
assert callback.call_count == 4

140
tests/components/tplink/test_init.py
View File

@ -1,12 +1,20 @@
"""Tests for the TP-Link component."""
from unittest.mock import patch
from typing import Dict, Any
from unittest.mock import MagicMock, patch
import pytest
from pyHS100 import SmartPlug, SmartBulb, SmartDevice, SmartDeviceException
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import tplink
from homeassistant.components.tplink.common import (
CONF_DISCOVERY,
CONF_DIMMER,
CONF_LIGHT,
CONF_SWITCH,
)
from homeassistant.const import CONF_HOST
from homeassistant.setup import async_setup_component
from pyHS100 import SmartPlug, SmartBulb
from tests.common import MockDependency, MockConfigEntry, mock_coro
MOCK_PYHS100 = MockDependency("pyHS100")
@ -15,8 +23,8 @@ MOCK_PYHS100 = MockDependency("pyHS100")
async def test_creating_entry_tries_discover(hass):
"""Test setting up does discovery."""
with MOCK_PYHS100, patch(
"homeassistant.components.tplink.async_setup_entry",
return_value=mock_coro(True),
"homeassistant.components.tplink.async_setup_entry",
return_value=mock_coro(True),
) as mock_setup, patch(
"pyHS100.Discover.discover", return_value={"host": 1234}
):
@ -41,7 +49,7 @@ async def test_configuring_tplink_causes_discovery(hass):
"""Test that specifying empty config does discovery."""
with MOCK_PYHS100, patch("pyHS100.Discover.discover") as discover:
discover.return_value = {"host": 1234}
await async_setup_component(hass, tplink.DOMAIN, {"tplink": {}})
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
assert len(discover.mock_calls) == 1
@ -58,45 +66,111 @@ async def test_configuring_tplink_causes_discovery(hass):
async def test_configuring_device_types(hass, name, cls, platform, count):
"""Test that light or switch platform list is filled correctly."""
with patch("pyHS100.Discover.discover") as discover, patch(
"pyHS100.SmartDevice._query_helper"
"pyHS100.SmartDevice._query_helper"
):
discovery_data = {
"123.123.123.{}".format(c): cls("123.123.123.123")
for c in range(count)
}
discover.return_value = discovery_data
await async_setup_component(hass, tplink.DOMAIN, {"tplink": {}})
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
assert len(discover.mock_calls) == 1
assert len(hass.data[tplink.DOMAIN][platform]) == count
class UnknownSmartDevice(SmartDevice):
"""Dummy class for testing."""
@property
def has_emeter(self) -> bool:
"""Do nothing."""
pass
def turn_off(self) -> None:
"""Do nothing."""
pass
def turn_on(self) -> None:
"""Do nothing."""
pass
@property
def is_on(self) -> bool:
"""Do nothing."""
pass
@property
def state_information(self) -> Dict[str, Any]:
"""Do nothing."""
pass
async def test_configuring_devices_from_multiple_sources(hass):
"""Test static and discover devices are not duplicated."""
with patch("pyHS100.Discover.discover") as discover, patch(
"pyHS100.SmartDevice._query_helper"
):
discover_device_fail = SmartPlug("123.123.123.123")
discover_device_fail.get_sysinfo = MagicMock(
side_effect=SmartDeviceException()
)
discover.return_value = {
"123.123.123.1": SmartBulb("123.123.123.1"),
"123.123.123.2": SmartPlug("123.123.123.2"),
"123.123.123.3": SmartBulb("123.123.123.3"),
"123.123.123.4": SmartPlug("123.123.123.4"),
"123.123.123.123": discover_device_fail,
"123.123.123.124": UnknownSmartDevice("123.123.123.124")
}
await async_setup_component(hass, tplink.DOMAIN, {
tplink.DOMAIN: {
CONF_LIGHT: [
{CONF_HOST: "123.123.123.1"},
],
CONF_SWITCH: [
{CONF_HOST: "123.123.123.2"},
],
CONF_DIMMER: [
{CONF_HOST: "123.123.123.22"},
],
}
})
await hass.async_block_till_done()
assert len(discover.mock_calls) == 1
assert len(hass.data[tplink.DOMAIN][CONF_LIGHT]) == 3
assert len(hass.data[tplink.DOMAIN][CONF_SWITCH]) == 2
async def test_is_dimmable(hass):
"""Test that is_dimmable switches are correctly added as lights."""
with patch("pyHS100.Discover.discover") as discover, patch(
"homeassistant.components.tplink.light.async_setup_entry",
return_value=mock_coro(True),
"homeassistant.components.tplink.light.async_setup_entry",
return_value=mock_coro(True),
) as setup, patch("pyHS100.SmartDevice._query_helper"), patch(
"pyHS100.SmartPlug.is_dimmable", True
):
dimmable_switch = SmartPlug("123.123.123.123")
discover.return_value = {"host": dimmable_switch}
await async_setup_component(hass, tplink.DOMAIN, {"tplink": {}})
await async_setup_component(hass, tplink.DOMAIN, {tplink.DOMAIN: {}})
await hass.async_block_till_done()
assert len(discover.mock_calls) == 1
assert len(setup.mock_calls) == 1
assert len(hass.data[tplink.DOMAIN]["light"]) == 1
assert len(hass.data[tplink.DOMAIN]["switch"]) == 0
assert len(hass.data[tplink.DOMAIN][CONF_LIGHT]) == 1
assert not hass.data[tplink.DOMAIN][CONF_SWITCH]
async def test_configuring_discovery_disabled(hass):
"""Test that discover does not get called when disabled."""
with MOCK_PYHS100, patch(
"homeassistant.components.tplink.async_setup_entry",
return_value=mock_coro(True),
"homeassistant.components.tplink.async_setup_entry",
return_value=mock_coro(True),
) as mock_setup, patch(
"pyHS100.Discover.discover", return_value=[]
) as discover:
@ -107,22 +181,22 @@ async def test_configuring_discovery_disabled(hass):
)
await hass.async_block_till_done()
assert len(discover.mock_calls) == 0
assert len(mock_setup.mock_calls) == 1
assert discover.call_count == 0
assert mock_setup.call_count == 1
async def test_platforms_are_initialized(hass):
"""Test that platforms are initialized per configuration array."""
config = {
"tplink": {
"discovery": False,
"light": [{"host": "123.123.123.123"}],
"switch": [{"host": "321.321.321.321"}],
tplink.DOMAIN: {
CONF_DISCOVERY: False,
CONF_LIGHT: [{CONF_HOST: "123.123.123.123"}],
CONF_SWITCH: [{CONF_HOST: "321.321.321.321"}],
}
}
with patch("pyHS100.Discover.discover") as discover, patch(
"pyHS100.SmartDevice._query_helper"
"pyHS100.SmartDevice._query_helper"
), patch(
"homeassistant.components.tplink.light.async_setup_entry",
return_value=mock_coro(True),
@ -136,21 +210,21 @@ async def test_platforms_are_initialized(hass):
await async_setup_component(hass, tplink.DOMAIN, config)
await hass.async_block_till_done()
assert len(discover.mock_calls) == 0
assert len(light_setup.mock_calls) == 1
assert len(switch_setup.mock_calls) == 1
assert discover.call_count == 0
assert light_setup.call_count == 1
assert switch_setup.call_count == 1
async def test_no_config_creates_no_entry(hass):
"""Test for when there is no tplink in config."""
with MOCK_PYHS100, patch(
"homeassistant.components.tplink.async_setup_entry",
return_value=mock_coro(True),
"homeassistant.components.tplink.async_setup_entry",
return_value=mock_coro(True),
) as mock_setup:
await async_setup_component(hass, tplink.DOMAIN, {})
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 0
assert mock_setup.call_count == 0
@pytest.mark.parametrize("platform", ["switch", "light"])
@ -161,14 +235,14 @@ async def test_unload(hass, platform):
entry.add_to_hass(hass)
with patch("pyHS100.SmartDevice._query_helper"), patch(
"homeassistant.components.tplink.{}"
".async_setup_entry".format(platform),
return_value=mock_coro(True),
"homeassistant.components.tplink.{}"
".async_setup_entry".format(platform),
return_value=mock_coro(True),
) as light_setup:
config = {
"tplink": {
platform: [{"host": "123.123.123.123"}],
"discovery": False,
tplink.DOMAIN: {
platform: [{CONF_HOST: "123.123.123.123"}],
CONF_DISCOVERY: False,
}
}
assert await async_setup_component(hass, tplink.DOMAIN, config)