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

Refactor dynalite integration for multi-platform (#32335) 

* refactoring for multi platform

* adopted test_bridge to refactoring

* refactoring tests for multi-platform
additional coverage in config and init

* comment for clarity

* more specific imports from lib

* library version bump

* removed async_update

* changed parameter order to start with hass

* removed pylint disable

* unsubscribe from signal dispatcher
inherit from Entity

* use device.unique_id

* changed hass_obj to hass

* added test for remove entity
bug fix from the test

* removed the polling try_connect. hate polling... it is now part of the async_setup()
significantly makes the code clearer and simplifies the tests

* removed leftover debug logs in the library

* changed tests to get the entry_id from hass

* changed place to assign hass.data only after success

* fixes for test_init

* removed assert

* removed device_info

* removed bridge internal from common

* modified test_bridge to work without the bridge directly

* removed bridge from test_existing_update

* changed update to not use bridge internals

* dyn_bridge fixture no longer used - removed
pull/32391/head
Ziv 2020-03-01 16:44:24 -05:00 committed by GitHub
parent fc98faa425
commit e13d5bdc10
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 348 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
homeassistant/components/dynalite/__init__.py
View File

@ -120,16 +120,11 @@ async def async_setup_entry(hass, entry):
"""Set up a bridge from a config entry."""
LOGGER.debug("Setting up entry %s", entry.data)
bridge = DynaliteBridge(hass, entry.data)
hass.data[DOMAIN][entry.entry_id] = bridge
entry.add_update_listener(async_entry_changed)
if not await bridge.async_setup():
LOGGER.error("Could not set up bridge for entry %s", entry.data)
hass.data[DOMAIN].pop(entry.entry_id)
return False
if not await bridge.try_connection():
LOGGER.error("Could not connect with entry %s", entry)
hass.data[DOMAIN].pop(entry.entry_id)
raise ConfigEntryNotReady
hass.data[DOMAIN][entry.entry_id] = bridge
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, "light")
)

48
homeassistant/components/dynalite/bridge.py
View File

@ -1,16 +1,11 @@
"""Code to handle a Dynalite bridge."""
import asyncio
from dynalite_devices_lib import DynaliteDevices
from dynalite_devices_lib.dynalite_devices import DynaliteDevices
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import CONF_ALL, CONF_HOST, LOGGER
CONNECT_TIMEOUT = 30
CONNECT_INTERVAL = 1
from .const import CONF_ALL, CONF_HOST, ENTITY_PLATFORMS, LOGGER
class DynaliteBridge:
@ -20,8 +15,8 @@ class DynaliteBridge:
"""Initialize the system based on host parameter."""
self.hass = hass
self.area = {}
self.async_add_devices = None
self.waiting_devices = []
self.async_add_devices = {}
self.waiting_devices = {}
self.host = config[CONF_HOST]
# Configure the dynalite devices
self.dynalite_devices = DynaliteDevices(
@ -38,7 +33,7 @@ class DynaliteBridge:
async def reload_config(self, config):
"""Reconfigure a bridge when config changes."""
LOGGER.debug("Setting up bridge - host %s, config %s", self.host, config)
LOGGER.debug("Reloading bridge - host %s, config %s", self.host, config)
self.dynalite_devices.configure(config)
def update_signal(self, device=None):
@ -62,27 +57,22 @@ class DynaliteBridge:
else:
async_dispatcher_send(self.hass, self.update_signal(device))
async def try_connection(self):
"""Try to connect to dynalite with timeout."""
# Currently by polling. Future - will need to change the library to be proactive
for _ in range(0, CONNECT_TIMEOUT):
if self.dynalite_devices.available:
return True
await asyncio.sleep(CONNECT_INTERVAL)
return False
@callback
def register_add_devices(self, async_add_devices):
def register_add_devices(self, platform, async_add_devices):
"""Add an async_add_entities for a category."""
self.async_add_devices = async_add_devices
if self.waiting_devices:
self.async_add_devices(self.waiting_devices)
self.async_add_devices[platform] = async_add_devices
if platform in self.waiting_devices:
self.async_add_devices[platform](self.waiting_devices[platform])
def add_devices_when_registered(self, devices):
"""Add the devices to HA if the add devices callback was registered, otherwise queue until it is."""
if not devices:
return
if self.async_add_devices:
self.async_add_devices(devices)
else: # handle it later when it is registered
self.waiting_devices.extend(devices)
for platform in ENTITY_PLATFORMS:
platform_devices = [
device for device in devices if device.category == platform
]
if platform in self.async_add_devices:
self.async_add_devices[platform](platform_devices)
else: # handle it later when it is registered
if platform not in self.waiting_devices:
self.waiting_devices[platform] = []
self.waiting_devices[platform].extend(platform_devices)

2
homeassistant/components/dynalite/config_flow.py
View File

@ -31,8 +31,6 @@ class DynaliteFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
bridge = DynaliteBridge(self.hass, import_info)
if not await bridge.async_setup():
LOGGER.error("Unable to setup bridge - import info=%s", import_info)
return self.async_abort(reason="bridge_setup_failed")
if not await bridge.try_connection():
return self.async_abort(reason="no_connection")
LOGGER.debug("Creating entry for the bridge - %s", import_info)
return self.async_create_entry(title=host, data=import_info)

3
homeassistant/components/dynalite/const.py
View File

@ -4,6 +4,8 @@ import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = "dynalite"
ENTITY_PLATFORMS = ["light"]
CONF_ACTIVE = "active"
CONF_ALL = "ALL"
CONF_AREA = "area"
@ -17,5 +19,6 @@ CONF_NAME = "name"
CONF_POLLTIMER = "polltimer"
CONF_PORT = "port"
DEFAULT_NAME = "dynalite"
DEFAULT_PORT = 12345

85
homeassistant/components/dynalite/dynalitebase.py Executable file
View File

@ -0,0 +1,85 @@
"""Support for the Dynalite devices as entities."""
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from .const import DOMAIN, LOGGER
def async_setup_entry_base(
hass, config_entry, async_add_entities, platform, entity_from_device
):
"""Record the async_add_entities function to add them later when received from Dynalite."""
LOGGER.debug("Setting up %s entry = %s", platform, config_entry.data)
bridge = hass.data[DOMAIN][config_entry.entry_id]
@callback
def async_add_entities_platform(devices):
# assumes it is called with a single platform
added_entities = []
for device in devices:
if device.category == platform:
added_entities.append(entity_from_device(device, bridge))
if added_entities:
async_add_entities(added_entities)
bridge.register_add_devices(platform, async_add_entities_platform)
class DynaliteBase(Entity):
"""Base class for the Dynalite entities."""
def __init__(self, device, bridge):
"""Initialize the base class."""
self._device = device
self._bridge = bridge
self._unsub_dispatchers = []
@property
def name(self):
"""Return the name of the entity."""
return self._device.name
@property
def unique_id(self):
"""Return the unique ID of the entity."""
return self._device.unique_id
@property
def available(self):
"""Return if entity is available."""
return self._device.available
@property
def device_info(self):
"""Device info for this entity."""
return {
"identifiers": {(DOMAIN, self._device.unique_id)},
"name": self.name,
"manufacturer": "Dynalite",
}
async def async_added_to_hass(self):
"""Added to hass so need to register to dispatch."""
# register for device specific update
self._unsub_dispatchers.append(
async_dispatcher_connect(
self.hass,
self._bridge.update_signal(self._device),
self.async_schedule_update_ha_state,
)
)
# register for wide update
self._unsub_dispatchers.append(
async_dispatcher_connect(
self.hass,
self._bridge.update_signal(),
self.async_schedule_update_ha_state,
)
)
async def async_will_remove_from_hass(self):
"""Unregister signal dispatch listeners when being removed."""
for unsub in self._unsub_dispatchers:
unsub()
self._unsub_dispatchers = []

66
homeassistant/components/dynalite/light.py
View File

@ -1,64 +1,25 @@
"""Support for Dynalite channels as lights."""
from homeassistant.components.light import SUPPORT_BRIGHTNESS, Light
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DOMAIN, LOGGER
from .dynalitebase import DynaliteBase, async_setup_entry_base
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Record the async_add_entities function to add them later when received from Dynalite."""
LOGGER.debug("Setting up light entry = %s", config_entry.data)
bridge = hass.data[DOMAIN][config_entry.entry_id]
@callback
def async_add_lights(devices):
added_lights = []
for device in devices:
if device.category == "light":
added_lights.append(DynaliteLight(device, bridge))
if added_lights:
async_add_entities(added_lights)
def light_from_device(device, bridge):
return DynaliteLight(device, bridge)
bridge.register_add_devices(async_add_lights)
async_setup_entry_base(
hass, config_entry, async_add_entities, "light", light_from_device
)
class DynaliteLight(Light):
class DynaliteLight(DynaliteBase, Light):
"""Representation of a Dynalite Channel as a Home Assistant Light."""
def __init__(self, device, bridge):
"""Initialize the base class."""
self._device = device
self._bridge = bridge
@property
def name(self):
"""Return the name of the entity."""
return self._device.name
@property
def unique_id(self):
"""Return the unique ID of the entity."""
return self._device.unique_id
@property
def available(self):
"""Return if entity is available."""
return self._device.available
async def async_update(self):
"""Update the entity."""
return
@property
def device_info(self):
"""Device info for this entity."""
return {
"identifiers": {(DOMAIN, self.unique_id)},
"name": self.name,
"manufacturer": "Dynalite",
}
@property
def brightness(self):
"""Return the brightness of this light between 0..255."""
@ -81,16 +42,3 @@ class DynaliteLight(Light):
def supported_features(self):
"""Flag supported features."""
return SUPPORT_BRIGHTNESS
async def async_added_to_hass(self):
"""Added to hass so need to register to dispatch."""
# register for device specific update
async_dispatcher_connect(
self.hass,
self._bridge.update_signal(self._device),
self.async_schedule_update_ha_state,
)
# register for wide update
async_dispatcher_connect(
self.hass, self._bridge.update_signal(), self.async_schedule_update_ha_state
)

2
homeassistant/components/dynalite/manifest.json
View File

@ -5,5 +5,5 @@
"documentation": "https://www.home-assistant.io/integrations/dynalite",
"dependencies": [],
"codeowners": ["@ziv1234"],
"requirements": ["dynalite_devices==0.1.26"]
"requirements": ["dynalite_devices==0.1.30"]
}

2
requirements_all.txt
View File

@ -463,7 +463,7 @@ dsmr_parser==0.18
dweepy==0.3.0
# homeassistant.components.dynalite
dynalite_devices==0.1.26
dynalite_devices==0.1.30
# homeassistant.components.rainforest_eagle
eagle200_reader==0.2.1

2
requirements_test_all.txt
View File

@ -171,7 +171,7 @@ distro==1.4.0
dsmr_parser==0.18
# homeassistant.components.dynalite
dynalite_devices==0.1.26
dynalite_devices==0.1.30
# homeassistant.components.ee_brightbox
eebrightbox==0.0.4

65
tests/components/dynalite/common.py
View File

@ -1,9 +1,64 @@
"""Common functions for the Dynalite tests."""
"""Common functions for tests."""
from asynctest import CoroutineMock, Mock, call, patch
from homeassistant.components import dynalite
from homeassistant.helpers import entity_registry
from tests.common import MockConfigEntry
ATTR_SERVICE = "service"
ATTR_METHOD = "method"
ATTR_ARGS = "args"
def get_bridge_from_hass(hass_obj):
"""Get the bridge from hass.data."""
key = next(iter(hass_obj.data[dynalite.DOMAIN]))
return hass_obj.data[dynalite.DOMAIN][key]
def create_mock_device(platform, spec):
"""Create a dynalite mock device for a platform according to a spec."""
device = Mock(spec=spec)
device.category = platform
device.unique_id = "UNIQUE"
device.name = "NAME"
device.device_class = "Device Class"
return device
async def get_entry_id_from_hass(hass):
"""Get the config entry id from hass."""
ent_reg = await entity_registry.async_get_registry(hass)
assert ent_reg
conf_entries = hass.config_entries.async_entries(dynalite.DOMAIN)
assert len(conf_entries) == 1
return conf_entries[0].entry_id
async def create_entity_from_device(hass, device):
"""Set up the component and platform and create a light based on the device provided."""
host = "1.2.3.4"
entry = MockConfigEntry(domain=dynalite.DOMAIN, data={dynalite.CONF_HOST: host})
entry.add_to_hass(hass)
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices"
) as mock_dyn_dev:
mock_dyn_dev().async_setup = CoroutineMock(return_value=True)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
new_device_func = mock_dyn_dev.mock_calls[1][2]["newDeviceFunc"]
new_device_func([device])
await hass.async_block_till_done()
async def run_service_tests(hass, device, platform, services):
"""Run a series of service calls and check that the entity and device behave correctly."""
for cur_item in services:
service = cur_item[ATTR_SERVICE]
args = cur_item.get(ATTR_ARGS, {})
service_data = {"entity_id": f"{platform}.name", **args}
await hass.services.async_call(platform, service, service_data, blocking=True)
await hass.async_block_till_done()
for check_item in services:
check_method = getattr(device, check_item[ATTR_METHOD])
if check_item[ATTR_SERVICE] == service:
check_method.assert_called_once()
assert check_method.mock_calls == [call(**args)]
check_method.reset_mock()
else:
check_method.assert_not_called()

120
tests/components/dynalite/test_bridge.py
View File

@ -1,81 +1,85 @@
"""Test Dynalite bridge."""
from unittest.mock import Mock, call
from asynctest import patch
from dynalite_devices_lib import CONF_ALL
import pytest
from asynctest import CoroutineMock, Mock, patch
from dynalite_devices_lib.const import CONF_ALL
from homeassistant.components import dynalite
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from tests.common import MockConfigEntry
@pytest.fixture
def dyn_bridge():
"""Define a basic mock bridge."""
hass = Mock()
async def test_update_device(hass):
"""Test that update works."""
host = "1.2.3.4"
bridge = dynalite.DynaliteBridge(hass, {dynalite.CONF_HOST: host})
return bridge
async def test_update_device(dyn_bridge):
"""Test a successful setup."""
async_dispatch = Mock()
entry = MockConfigEntry(domain=dynalite.DOMAIN, data={dynalite.CONF_HOST: host})
entry.add_to_hass(hass)
with patch(
"homeassistant.components.dynalite.bridge.async_dispatcher_send", async_dispatch
):
dyn_bridge.update_device(CONF_ALL)
async_dispatch.assert_called_once()
assert async_dispatch.mock_calls[0] == call(
dyn_bridge.hass, f"dynalite-update-{dyn_bridge.host}"
)
async_dispatch.reset_mock()
device = Mock
device.unique_id = "abcdef"
dyn_bridge.update_device(device)
async_dispatch.assert_called_once()
assert async_dispatch.mock_calls[0] == call(
dyn_bridge.hass, f"dynalite-update-{dyn_bridge.host}-{device.unique_id}"
)
"homeassistant.components.dynalite.bridge.DynaliteDevices"
) as mock_dyn_dev:
mock_dyn_dev().async_setup = CoroutineMock(return_value=True)
assert await hass.config_entries.async_setup(entry.entry_id)
# Not waiting so it add the devices before registration
update_device_func = mock_dyn_dev.mock_calls[1][2]["updateDeviceFunc"]
device = Mock()
device.unique_id = "abcdef"
wide_func = Mock()
async_dispatcher_connect(hass, f"dynalite-update-{host}", wide_func)
specific_func = Mock()
async_dispatcher_connect(
hass, f"dynalite-update-{host}-{device.unique_id}", specific_func
)
update_device_func(CONF_ALL)
await hass.async_block_till_done()
wide_func.assert_called_once()
specific_func.assert_not_called()
update_device_func(device)
await hass.async_block_till_done()
wide_func.assert_called_once()
specific_func.assert_called_once()
async def test_add_devices_then_register(dyn_bridge):
async def test_add_devices_then_register(hass):
"""Test that add_devices work."""
# First test empty
dyn_bridge.add_devices_when_registered([])
assert not dyn_bridge.waiting_devices
host = "1.2.3.4"
entry = MockConfigEntry(domain=dynalite.DOMAIN, data={dynalite.CONF_HOST: host})
entry.add_to_hass(hass)
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices"
) as mock_dyn_dev:
mock_dyn_dev().async_setup = CoroutineMock(return_value=True)
assert await hass.config_entries.async_setup(entry.entry_id)
# Not waiting so it add the devices before registration
new_device_func = mock_dyn_dev.mock_calls[1][2]["newDeviceFunc"]
# Now with devices
device1 = Mock()
device1.category = "light"
device1.name = "NAME"
device2 = Mock()
device2.category = "switch"
dyn_bridge.add_devices_when_registered([device1, device2])
reg_func = Mock()
dyn_bridge.register_add_devices(reg_func)
reg_func.assert_called_once()
assert reg_func.mock_calls[0][1][0][0] is device1
new_device_func([device1, device2])
await hass.async_block_till_done()
assert hass.states.get("light.name")
async def test_register_then_add_devices(dyn_bridge):
async def test_register_then_add_devices(hass):
"""Test that add_devices work after register_add_entities."""
host = "1.2.3.4"
entry = MockConfigEntry(domain=dynalite.DOMAIN, data={dynalite.CONF_HOST: host})
entry.add_to_hass(hass)
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices"
) as mock_dyn_dev:
mock_dyn_dev().async_setup = CoroutineMock(return_value=True)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
new_device_func = mock_dyn_dev.mock_calls[1][2]["newDeviceFunc"]
# Now with devices
device1 = Mock()
device1.category = "light"
device1.name = "NAME"
device2 = Mock()
device2.category = "switch"
reg_func = Mock()
dyn_bridge.register_add_devices(reg_func)
dyn_bridge.add_devices_when_registered([device1, device2])
reg_func.assert_called_once()
assert reg_func.mock_calls[0][1][0][0] is device1
async def test_try_connection(dyn_bridge):
"""Test that try connection works."""
# successful
with patch.object(dyn_bridge.dynalite_devices, "connected", True):
assert await dyn_bridge.try_connection()
# unsuccessful
with patch.object(dyn_bridge.dynalite_devices, "connected", False), patch(
"homeassistant.components.dynalite.bridge.CONNECT_INTERVAL", 0
):
assert not await dyn_bridge.try_connection()
new_device_func([device1, device2])
await hass.async_block_till_done()
assert hass.states.get("light.name")

64
tests/components/dynalite/test_config_flow.py
View File

@ -1,53 +1,50 @@
"""Test Dynalite config flow."""
from asynctest import patch
from asynctest import CoroutineMock, patch
from homeassistant import config_entries
from homeassistant.components import dynalite
from .common import get_bridge_from_hass
from tests.common import MockConfigEntry
async def run_flow(hass, setup, connection):
async def run_flow(hass, connection):
"""Run a flow with or without errors and return result."""
host = "1.2.3.4"
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=setup,
), patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.available", connection
), patch(
"homeassistant.components.dynalite.bridge.CONNECT_INTERVAL", 0
side_effect=connection,
):
result = await hass.config_entries.flow.async_init(
dynalite.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={dynalite.CONF_HOST: host},
)
await hass.async_block_till_done()
return result
async def test_flow_works(hass):
"""Test a successful config flow."""
result = await run_flow(hass, True, True)
result = await run_flow(hass, [True, True])
assert result["type"] == "create_entry"
assert result["result"].state == "loaded"
async def test_flow_setup_fails(hass):
"""Test a flow where async_setup fails."""
result = await run_flow(hass, False, True)
assert result["type"] == "abort"
assert result["reason"] == "bridge_setup_failed"
async def test_flow_no_connection(hass):
"""Test a flow where connection times out."""
result = await run_flow(hass, True, False)
result = await run_flow(hass, [False])
assert result["type"] == "abort"
assert result["reason"] == "no_connection"
async def test_flow_setup_fails_in_setup_entry(hass):
"""Test a flow where the initial check works but inside setup_entry, the bridge setup fails."""
result = await run_flow(hass, [True, False])
assert result["type"] == "create_entry"
assert result["result"].state == "setup_retry"
async def test_existing(hass):
"""Test when the entry exists with the same config."""
host = "1.2.3.4"
@ -57,8 +54,6 @@ async def test_existing(hass):
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=True,
), patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.available", True
):
result = await hass.config_entries.flow.async_init(
dynalite.DOMAIN,
@ -70,31 +65,30 @@ async def test_existing(hass):
async def test_existing_update(hass):
"""Test when the entry exists with the same config."""
"""Test when the entry exists with a different config."""
host = "1.2.3.4"
port1 = 7777
port2 = 8888
entry = MockConfigEntry(
domain=dynalite.DOMAIN,
data={dynalite.CONF_HOST: host, dynalite.CONF_PORT: port1},
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=True,
), patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.available", True
):
assert await hass.config_entries.flow.async_init(
dynalite.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={dynalite.CONF_HOST: host, dynalite.CONF_PORT: port1},
)
"homeassistant.components.dynalite.bridge.DynaliteDevices"
) as mock_dyn_dev:
mock_dyn_dev().async_setup = CoroutineMock(return_value=True)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
old_bridge = get_bridge_from_hass(hass)
assert old_bridge.dynalite_devices.port == port1
mock_dyn_dev().configure.assert_called_once()
assert mock_dyn_dev().configure.mock_calls[0][1][0][dynalite.CONF_PORT] == port1
result = await hass.config_entries.flow.async_init(
dynalite.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={dynalite.CONF_HOST: host, dynalite.CONF_PORT: port2},
)
await hass.async_block_till_done()
assert mock_dyn_dev().configure.call_count == 2
assert mock_dyn_dev().configure.mock_calls[1][1][0][dynalite.CONF_PORT] == port2
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
bridge = get_bridge_from_hass(hass)
assert bridge.dynalite_devices.port == port2

77
tests/components/dynalite/test_init.py
View File

@ -1,6 +1,7 @@
"""Test Dynalite __init__."""
from asynctest import patch
from asynctest import call, patch
from homeassistant.components import dynalite
from homeassistant.setup import async_setup_component
@ -12,51 +13,75 @@ async def test_empty_config(hass):
"""Test with an empty config."""
assert await async_setup_component(hass, dynalite.DOMAIN, {}) is True
assert len(hass.config_entries.flow.async_progress()) == 0
assert hass.data[dynalite.DOMAIN] == {}
assert len(hass.config_entries.async_entries(dynalite.DOMAIN)) == 0
async def test_async_setup(hass):
"""Test a successful setup."""
host = "1.2.3.4"
with patch(
"dynalite_devices_lib.DynaliteDevices.async_setup", return_value=True
), patch("dynalite_devices_lib.DynaliteDevices.available", True):
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=True,
):
assert await async_setup_component(
hass,
dynalite.DOMAIN,
{dynalite.DOMAIN: {dynalite.CONF_BRIDGES: [{dynalite.CONF_HOST: host}]}},
{
dynalite.DOMAIN: {
dynalite.CONF_BRIDGES: [
{
dynalite.CONF_HOST: host,
dynalite.CONF_AREA: {"1": {dynalite.CONF_NAME: "Name"}},
}
]
}
},
)
assert len(hass.data[dynalite.DOMAIN]) == 1
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(dynalite.DOMAIN)) == 1
async def test_async_setup_failed(hass):
"""Test a setup when DynaliteBridge.async_setup fails."""
async def test_async_setup_bad_config2(hass):
"""Test a successful with bad config on numbers."""
host = "1.2.3.4"
with patch("dynalite_devices_lib.DynaliteDevices.async_setup", return_value=False):
assert await async_setup_component(
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=True,
):
assert not await async_setup_component(
hass,
dynalite.DOMAIN,
{dynalite.DOMAIN: {dynalite.CONF_BRIDGES: [{dynalite.CONF_HOST: host}]}},
{
dynalite.DOMAIN: {
dynalite.CONF_BRIDGES: [
{
dynalite.CONF_HOST: host,
dynalite.CONF_AREA: {"WRONG": {dynalite.CONF_NAME: "Name"}},
}
]
}
},
)
assert hass.data[dynalite.DOMAIN] == {}
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(dynalite.DOMAIN)) == 0
async def test_unload_entry(hass):
"""Test being able to unload an entry."""
host = "1.2.3.4"
entry = MockConfigEntry(domain=dynalite.DOMAIN, data={"host": host})
entry = MockConfigEntry(domain=dynalite.DOMAIN, data={dynalite.CONF_HOST: host})
entry.add_to_hass(hass)
with patch(
"dynalite_devices_lib.DynaliteDevices.async_setup", return_value=True
), patch("dynalite_devices_lib.DynaliteDevices.available", True):
assert await async_setup_component(
hass,
dynalite.DOMAIN,
{dynalite.DOMAIN: {dynalite.CONF_BRIDGES: [{dynalite.CONF_HOST: host}]}},
)
assert hass.data[dynalite.DOMAIN].get(entry.entry_id)
assert await hass.config_entries.async_unload(entry.entry_id)
assert not hass.data[dynalite.DOMAIN].get(entry.entry_id)
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=True,
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(dynalite.DOMAIN)) == 1
with patch.object(
hass.config_entries, "async_forward_entry_unload", return_value=True
) as mock_unload:
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
mock_unload.assert_called_once()
assert mock_unload.mock_calls == [call(entry, "light")]

85
tests/components/dynalite/test_light.py
View File

@ -1,78 +1,49 @@
"""Test Dynalite light."""
from unittest.mock import Mock
from asynctest import CoroutineMock, patch
from dynalite_devices_lib.light import DynaliteChannelLightDevice
import pytest
from homeassistant.components import dynalite
from homeassistant.components.light import SUPPORT_BRIGHTNESS
from homeassistant.setup import async_setup_component
from .common import (
ATTR_METHOD,
ATTR_SERVICE,
create_entity_from_device,
create_mock_device,
get_entry_id_from_hass,
run_service_tests,
)
@pytest.fixture
def mock_device():
"""Mock a Dynalite device."""
device = Mock()
device.category = "light"
device.unique_id = "UNIQUE"
device.name = "NAME"
device.device_info = {
"identifiers": {(dynalite.DOMAIN, device.unique_id)},
"name": device.name,
"manufacturer": "Dynalite",
}
return device
async def create_light_from_device(hass, device):
"""Set up the component and platform and create a light based on the device provided."""
host = "1.2.3.4"
with patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.async_setup",
return_value=True,
), patch(
"homeassistant.components.dynalite.bridge.DynaliteDevices.available", True
):
assert await async_setup_component(
hass,
dynalite.DOMAIN,
{dynalite.DOMAIN: {dynalite.CONF_BRIDGES: [{dynalite.CONF_HOST: host}]}},
)
await hass.async_block_till_done()
# Find the bridge
bridge = None
assert len(hass.data[dynalite.DOMAIN]) == 1
key = next(iter(hass.data[dynalite.DOMAIN]))
bridge = hass.data[dynalite.DOMAIN][key]
bridge.dynalite_devices.newDeviceFunc([device])
await hass.async_block_till_done()
return create_mock_device("light", DynaliteChannelLightDevice)
async def test_light_setup(hass, mock_device):
"""Test a successful setup."""
await create_light_from_device(hass, mock_device)
await create_entity_from_device(hass, mock_device)
entity_state = hass.states.get("light.name")
assert entity_state.attributes["friendly_name"] == mock_device.name
assert entity_state.attributes["brightness"] == mock_device.brightness
assert entity_state.attributes["supported_features"] == SUPPORT_BRIGHTNESS
async def test_turn_on(hass, mock_device):
"""Test turning a light on."""
mock_device.async_turn_on = CoroutineMock(return_value=True)
await create_light_from_device(hass, mock_device)
await hass.services.async_call(
"light", "turn_on", {"entity_id": "light.name"}, blocking=True
await run_service_tests(
hass,
mock_device,
"light",
[
{ATTR_SERVICE: "turn_on", ATTR_METHOD: "async_turn_on"},
{ATTR_SERVICE: "turn_off", ATTR_METHOD: "async_turn_off"},
],
)
await hass.async_block_till_done()
mock_device.async_turn_on.assert_awaited_once()
async def test_turn_off(hass, mock_device):
"""Test turning a light off."""
mock_device.async_turn_off = CoroutineMock(return_value=True)
await create_light_from_device(hass, mock_device)
await hass.services.async_call(
"light", "turn_off", {"entity_id": "light.name"}, blocking=True
)
async def test_remove_entity(hass, mock_device):
"""Test when an entity is removed from HA."""
await create_entity_from_device(hass, mock_device)
assert hass.states.get("light.name")
entry_id = await get_entry_id_from_hass(hass)
assert await hass.config_entries.async_unload(entry_id)
await hass.async_block_till_done()
mock_device.async_turn_off.assert_awaited_once()
assert not hass.states.get("light.name")