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

Add zwave per-node entity. (#6690) 

* Add zwave per-node entity.

* Disable lint import error

* Add tests for base class

* More tests

* More tests

* Sort .coveragerc

* more tests

* Move location, battery and wakeup to node entity

* More tests

* Cleanup

* Make zwave node entity visible by default

* Remove battery sensor

* Fix tests
pull/6757/head
Andrey 2017-03-23 17:37:20 +02:00 committed by Paulus Schoutsen
parent 20c5f9de4b
commit 8a86ec5b74
11 changed files with 408 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.coveragerc
View File

@ -118,8 +118,6 @@ omit =
homeassistant/components/zigbee.py
homeassistant/components/*/zigbee.py
homeassistant/components/zwave/*
homeassistant/components/enocean.py
homeassistant/components/*/enocean.py
@ -436,6 +434,9 @@ omit =
homeassistant/components/weather/openweathermap.py
homeassistant/components/weather/zamg.py
homeassistant/components/zeroconf.py
homeassistant/components/zwave/__init__.py
homeassistant/components/zwave/util.py
homeassistant/components/zwave/workaround.py
[report]

2
homeassistant/components/sensor/zwave.py
View File

@ -18,8 +18,6 @@ _LOGGER = logging.getLogger(__name__)
def get_device(node, values, **kwargs):
"""Create zwave entity device."""
# Generic Device mappings
if values.primary.command_class == zwave.const.COMMAND_CLASS_BATTERY:
return ZWaveSensor(values)
if node.has_command_class(zwave.const.COMMAND_CLASS_SENSOR_MULTILEVEL):
return ZWaveMultilevelSensor(values)
if node.has_command_class(zwave.const.COMMAND_CLASS_METER) and \

96
homeassistant/components/zwave/__init__.py
View File

@ -13,13 +13,11 @@ from pprint import pprint
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.loader import get_platform
from homeassistant.helpers import discovery
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.const import (
ATTR_BATTERY_LEVEL, ATTR_LOCATION, ATTR_ENTITY_ID, ATTR_WAKEUP,
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from homeassistant.helpers.entity import Entity
ATTR_ENTITY_ID, EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from homeassistant.helpers.entity_values import EntityValues
from homeassistant.helpers.event import track_time_change
from homeassistant.util import convert, slugify
@ -29,9 +27,11 @@ from homeassistant.helpers.dispatcher import (
async_dispatcher_connect, async_dispatcher_send)
from . import const
from .const import DOMAIN
from .node_entity import ZWaveBaseEntity, ZWaveNodeEntity
from . import workaround
from .discovery_schemas import DISCOVERY_SCHEMAS
from .util import check_node_schema, check_value_schema
from .util import check_node_schema, check_value_schema, node_name
REQUIREMENTS = ['pydispatcher==2.0.5']
@ -60,7 +60,6 @@ DEFAULT_DEBUG = False
DEFAULT_CONF_IGNORED = False
DEFAULT_CONF_REFRESH_VALUE = False
DEFAULT_CONF_REFRESH_DELAY = 5
DOMAIN = 'zwave'
DATA_ZWAVE_DICT = 'zwave_devices'
@ -140,20 +139,14 @@ def _obj_to_dict(obj):
if key[0] != '_' and not hasattr(getattr(obj, key), '__call__')}
def _node_name(node):
"""Return the name of the node."""
return node.name or '{} {}'.format(
node.manufacturer_name, node.product_name)
def _value_name(value):
"""Return the name of the value."""
return '{} {}'.format(_node_name(value.node), value.label)
return '{} {}'.format(node_name(value.node), value.label)
def _node_object_id(node):
"""Return the object_id of the node."""
node_object_id = '{}_{}'.format(slugify(_node_name(node)), node.node_id)
node_object_id = '{}_{}'.format(slugify(node_name(node)), node.node_id)
return node_object_id
@ -291,13 +284,26 @@ def setup(hass, config):
continue
if not check_value_schema(
value,
schema[const.DISC_INSTANCE_VALUES][const.DISC_PRIMARY]):
schema[const.DISC_VALUES][const.DISC_PRIMARY]):
continue
values = ZWaveDeviceEntityValues(
hass, schema, value, config, device_config)
discovered_values.append(values)
component = EntityComponent(_LOGGER, DOMAIN, hass)
def node_added(node):
"""Called when a node is added on the network."""
entity = ZWaveNodeEntity(node)
node_config = device_config.get(entity.entity_id)
if node_config.get(CONF_IGNORED):
_LOGGER.info(
"Ignoring node entity %s due to device settings.",
entity.entity_id)
return
component.add_entities([entity])
def scene_activated(node, scene_id):
"""Called when a scene is activated on any node in the network."""
hass.bus.fire(const.EVENT_SCENE_ACTIVATED, {
@ -328,6 +334,8 @@ def setup(hass, config):
dispatcher.connect(
value_added, ZWaveNetwork.SIGNAL_VALUE_ADDED, weak=False)
dispatcher.connect(
node_added, ZWaveNetwork.SIGNAL_NODE_ADDED, weak=False)
dispatcher.connect(
scene_activated, ZWaveNetwork.SIGNAL_SCENE_EVENT, weak=False)
dispatcher.connect(
@ -619,18 +627,8 @@ class ZWaveDeviceEntityValues():
self._entity = None
self._workaround_ignore = False
# Combine node value schemas and instance value schemas into one
# values schema mapping.
self._schema[const.DISC_VALUES] = {}
for name in self._schema[const.DISC_NODE_VALUES].keys():
for name in self._schema[const.DISC_VALUES].keys():
self._values[name] = None
self._schema[const.DISC_VALUES][name] = \
self._schema[const.DISC_NODE_VALUES][name]
for name in self._schema[const.DISC_INSTANCE_VALUES].keys():
self._values[name] = None
self._schema[const.DISC_VALUES][name] = \
self._schema[const.DISC_INSTANCE_VALUES][name]
self._schema[const.DISC_VALUES][name][const.DISC_INSTANCE] = \
[primary_value.instance]
@ -714,7 +712,7 @@ class ZWaveDeviceEntityValues():
if node_config.get(CONF_IGNORED):
_LOGGER.info(
"Ignoring node %s due to device settings.", self._node.node_id)
"Ignoring entity %s due to device settings.", name)
# No entity will be created for this value
self._workaround_ignore = True
return
@ -749,12 +747,13 @@ class ZWaveDeviceEntityValues():
self._hass.add_job(discover_device, component, device, dict_id)
class ZWaveDeviceEntity(Entity):
class ZWaveDeviceEntity(ZWaveBaseEntity):
"""Representation of a Z-Wave node entity."""
def __init__(self, values, domain):
"""Initialize the z-Wave device."""
# pylint: disable=import-error
super().__init__()
from openzwave.network import ZWaveNetwork
from pydispatch import dispatcher
self.values = values
@ -765,7 +764,6 @@ class ZWaveDeviceEntity(Entity):
self._name = _value_name(self.values.primary)
self._unique_id = "ZWAVE-{}-{}".format(self.node.node_id,
self.values.primary.object_id)
self._update_scheduled = False
self._update_attributes()
dispatcher.connect(
@ -780,10 +778,7 @@ class ZWaveDeviceEntity(Entity):
"""Called when a value for this entity's node has changed."""
self._update_attributes()
self.update_properties()
# If value changed after device was created but before setup_platform
# was called - skip updating state.
if self.hass and not self._update_scheduled:
self.hass.add_job(self._schedule_update)
self.maybe_schedule_update()
@asyncio.coroutine
def async_added_to_hass(self):
@ -796,17 +791,6 @@ class ZWaveDeviceEntity(Entity):
def _update_attributes(self):
"""Update the node attributes. May only be used inside callback."""
self.node_id = self.node.node_id
self.location = self.node.location
if self.values.battery:
self.battery_level = self.values.battery.data
else:
self.battery_level = None
if self.values.wakeup:
self.wakeup_interval = self.values.wakeup.data
else:
self.wakeup_interval = None
if self.values.power:
self.power_consumption = round(
@ -840,15 +824,6 @@ class ZWaveDeviceEntity(Entity):
const.ATTR_NODE_ID: self.node_id,
}
if self.battery_level is not None:
attrs[ATTR_BATTERY_LEVEL] = self.battery_level
if self.location:
attrs[ATTR_LOCATION] = self.location
if self.wakeup_interval is not None:
attrs[ATTR_WAKEUP] = self.wakeup_interval
if self.power_consumption is not None:
attrs[ATTR_POWER] = self.power_consumption
@ -858,18 +833,3 @@ class ZWaveDeviceEntity(Entity):
"""Refresh all dependent values from zwave network."""
for value in self.values:
self.node.refresh_value(value.value_id)
@callback
def _schedule_update(self):
"""Schedule delayed update."""
if self._update_scheduled:
return
@callback
def do_update():
"""Really update."""
self.hass.async_add_job(self.async_update_ha_state)
self._update_scheduled = False
self._update_scheduled = True
self.hass.loop.call_later(0.1, do_update)

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

@ -1,4 +1,5 @@
"""Z-Wave Constants."""
DOMAIN = "zwave"
ATTR_NODE_ID = "node_id"
ATTR_TARGET_NODE_ID = "target_node_id"
@ -318,11 +319,9 @@ DISC_COMPONENT = "component"
DISC_GENERIC_DEVICE_CLASS = "generic_device_class"
DISC_GENRE = "genre"
DISC_INDEX = "index"
DISC_INSTANCE_VALUES = "instance_values"
DISC_INSTANCE = "instance"
DISC_LABEL = "label"
DISC_NODE_ID = "node_id"
DISC_NODE_VALUES = "node_values"
DISC_OPTIONAL = "optional"
DISC_PRIMARY = "primary"
DISC_READONLY = "readonly"

40
homeassistant/components/zwave/discovery_schemas.py
View File

@ -1,18 +1,7 @@
"""Zwave discovery schemas."""
from . import const
DEFAULT_NODE_VALUES_SCHEMA = {
'wakeup': {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_WAKE_UP],
const.DISC_GENRE: const.GENRE_USER,
const.DISC_OPTIONAL: True,
},
'battery': {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_BATTERY],
const.DISC_OPTIONAL: True,
},
}
DEFAULT_INSTANCE_VALUES_SCHEMA = {
DEFAULT_VALUES_SCHEMA = {
'power': {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SENSOR_MULTILEVEL,
const.COMMAND_CLASS_METER],
@ -32,8 +21,7 @@ DISCOVERY_SCHEMAS = [
const.GENERIC_TYPE_SWITCH_MULTILEVEL,
const.GENERIC_TYPE_SENSOR_NOTIFICATION,
const.GENERIC_TYPE_THERMOSTAT],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SENSOR_BINARY],
const.DISC_TYPE: const.TYPE_BOOL,
@ -41,8 +29,7 @@ DISCOVERY_SCHEMAS = [
}})},
{const.DISC_COMPONENT: 'climate',
const.DISC_GENERIC_DEVICE_CLASS: [const.GENERIC_TYPE_THERMOSTAT],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [
const.COMMAND_CLASS_THERMOSTAT_SETPOINT],
@ -87,8 +74,7 @@ DISCOVERY_SCHEMAS = [
const.SPECIFIC_TYPE_MOTOR_MULTIPOSITION,
const.SPECIFIC_TYPE_SECURE_BARRIER_ADDON,
const.SPECIFIC_TYPE_SECURE_DOOR],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_MULTILEVEL],
const.DISC_GENRE: const.GENRE_USER,
@ -114,8 +100,7 @@ DISCOVERY_SCHEMAS = [
const.SPECIFIC_TYPE_MOTOR_MULTIPOSITION,
const.SPECIFIC_TYPE_SECURE_BARRIER_ADDON,
const.SPECIFIC_TYPE_SECURE_DOOR],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [
const.COMMAND_CLASS_BARRIER_OPERATOR,
@ -130,8 +115,7 @@ DISCOVERY_SCHEMAS = [
const.SPECIFIC_TYPE_POWER_SWITCH_MULTILEVEL,
const.SPECIFIC_TYPE_SCENE_SWITCH_MULTILEVEL,
const.SPECIFIC_TYPE_NOT_USED],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_MULTILEVEL],
const.DISC_GENRE: const.GENRE_USER,
@ -156,8 +140,7 @@ DISCOVERY_SCHEMAS = [
const.DISC_SPECIFIC_DEVICE_CLASS: [
const.SPECIFIC_TYPE_ADVANCED_DOOR_LOCK,
const.SPECIFIC_TYPE_SECURE_KEYPAD_DOOR_LOCK],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_DOOR_LOCK],
const.DISC_TYPE: const.TYPE_BOOL,
@ -185,15 +168,13 @@ DISCOVERY_SCHEMAS = [
const.DISC_OPTIONAL: True,
}})},
{const.DISC_COMPONENT: 'sensor',
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [
const.COMMAND_CLASS_SENSOR_MULTILEVEL,
const.COMMAND_CLASS_METER,
const.COMMAND_CLASS_ALARM,
const.COMMAND_CLASS_SENSOR_ALARM,
const.COMMAND_CLASS_BATTERY],
const.COMMAND_CLASS_SENSOR_ALARM],
const.DISC_GENRE: const.GENRE_USER,
}})},
{const.DISC_COMPONENT: 'switch',
@ -210,8 +191,7 @@ DISCOVERY_SCHEMAS = [
const.GENERIC_TYPE_REPEATER_SLAVE,
const.GENERIC_TYPE_THERMOSTAT,
const.GENERIC_TYPE_WALL_CONTROLLER],
const.DISC_NODE_VALUES: dict(DEFAULT_NODE_VALUES_SCHEMA),
const.DISC_INSTANCE_VALUES: dict(DEFAULT_INSTANCE_VALUES_SCHEMA, **{
const.DISC_VALUES: dict(DEFAULT_VALUES_SCHEMA, **{
const.DISC_PRIMARY: {
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_BINARY],
const.DISC_TYPE: const.TYPE_BOOL,

155
homeassistant/components/zwave/node_entity.py Normal file
View File

@ -0,0 +1,155 @@
"""Entity class that represents Z-Wave node."""
import logging
from homeassistant.core import callback
from homeassistant.const import ATTR_BATTERY_LEVEL, ATTR_WAKEUP
from homeassistant.helpers.entity import Entity
from homeassistant.util import slugify
from .const import ATTR_NODE_ID, DOMAIN, COMMAND_CLASS_WAKE_UP
from .util import node_name
_LOGGER = logging.getLogger(__name__)
ATTR_QUERY_STAGE = 'query_stage'
ATTR_AWAKE = 'is_awake'
ATTR_READY = 'is_ready'
ATTR_FAILED = 'is_failed'
STAGE_COMPLETE = 'Complete'
_REQUIRED_ATTRIBUTES = [
ATTR_QUERY_STAGE, ATTR_AWAKE, ATTR_READY, ATTR_FAILED,
'is_info_received', 'max_baud_rate', 'is_zwave_plus']
_OPTIONAL_ATTRIBUTES = ['capabilities', 'neighbors', 'location']
ATTRIBUTES = _REQUIRED_ATTRIBUTES + _OPTIONAL_ATTRIBUTES
class ZWaveBaseEntity(Entity):
"""Base class for Z-Wave Node and Value entities."""
def __init__(self):
"""Initialize the base Z-Wave class."""
self._update_scheduled = False
def maybe_schedule_update(self):
"""Maybe schedule state update.
If value changed after device was created but before setup_platform
was called - skip updating state.
"""
if self.hass and not self._update_scheduled:
self.hass.add_job(self._schedule_update)
@callback
def _schedule_update(self):
"""Schedule delayed update."""
if self._update_scheduled:
return
@callback
def do_update():
"""Really update."""
self.hass.async_add_job(self.async_update_ha_state)
self._update_scheduled = False
self._update_scheduled = True
self.hass.loop.call_later(0.1, do_update)
def sub_status(status, stage):
"""Format sub-status."""
return '{} ({})'.format(status, stage) if stage else status
class ZWaveNodeEntity(ZWaveBaseEntity):
"""Representation of a Z-Wave node."""
def __init__(self, node):
"""Initialize node."""
# pylint: disable=import-error
super().__init__()
from openzwave.network import ZWaveNetwork
from pydispatch import dispatcher
self.node = node
self.node_id = self.node.node_id
self._name = node_name(self.node)
self.entity_id = "{}.{}_{}".format(
DOMAIN, slugify(self._name), self.node_id)
self._attributes = {}
self.wakeup_interval = None
self.location = None
self.battery_level = None
dispatcher.connect(
self.network_node_changed, ZWaveNetwork.SIGNAL_VALUE_CHANGED)
dispatcher.connect(self.network_node_changed, ZWaveNetwork.SIGNAL_NODE)
dispatcher.connect(
self.network_node_changed, ZWaveNetwork.SIGNAL_NOTIFICATION)
def network_node_changed(self, node=None, args=None):
"""Called when node has changed on the network."""
if node and node.node_id != self.node_id:
return
if args is not None and 'nodeId' in args and \
args['nodeId'] != self.node_id:
return
self.node_changed()
def node_changed(self):
"""Update node properties."""
self._attributes = {}
for attr in ATTRIBUTES:
value = getattr(self.node, attr)
if attr in _REQUIRED_ATTRIBUTES or value:
self._attributes[attr] = value
if self.node.can_wake_up():
for value in self.node.get_values(COMMAND_CLASS_WAKE_UP).values():
self.wakeup_interval = value.data
break
else:
self.wakeup_interval = None
self.battery_level = self.node.get_battery_level()
self.maybe_schedule_update()
@property
def state(self):
"""Return the state."""
if ATTR_READY not in self._attributes:
return None
stage = ''
if not self._attributes[ATTR_READY]:
# If node is not ready use stage as sub-status.
stage = self._attributes[ATTR_QUERY_STAGE]
if self._attributes[ATTR_FAILED]:
return sub_status('Dead', stage)
if not self._attributes[ATTR_AWAKE]:
return sub_status('Sleeping', stage)
if self._attributes[ATTR_READY]:
return sub_status('Ready', stage)
return stage
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def device_state_attributes(self):
"""Return the device specific state attributes."""
attrs = {
ATTR_NODE_ID: self.node_id,
}
attrs.update(self._attributes)
if self.battery_level is not None:
attrs[ATTR_BATTERY_LEVEL] = self.battery_level
if self.wakeup_interval is not None:
attrs[ATTR_WAKEUP] = self.wakeup_interval
return attrs

6
homeassistant/components/zwave/util.py
View File

@ -69,3 +69,9 @@ def check_value_schema(value, schema):
value.instance, schema[const.DISC_INSTANCE])
return False
return True
def node_name(node):
"""Return the name of the node."""
return node.name or '{} {}'.format(
node.manufacturer_name, node.product_name)

27
tests/components/sensor/test_zwave.py
View File

@ -17,18 +17,6 @@ def test_get_device_detects_none(mock_openzwave):
assert device is None
def test_get_device_detects_sensor(mock_openzwave):
"""Test get_device returns a Z-Wave Sensor."""
node = MockNode(command_classes=[const.COMMAND_CLASS_BATTERY])
value = MockValue(data=0, command_class=const.COMMAND_CLASS_BATTERY,
node=node)
values = MockEntityValues(primary=value)
device = zwave.get_device(node=node, values=values, node_config={})
assert isinstance(device, zwave.ZWaveSensor)
assert device.force_update
def test_get_device_detects_alarmsensor(mock_openzwave):
"""Test get_device returns a Z-Wave alarmsensor."""
node = MockNode(command_classes=[const.COMMAND_CLASS_ALARM,
@ -61,21 +49,6 @@ def test_get_device_detects_multilevel_meter(mock_openzwave):
assert isinstance(device, zwave.ZWaveMultilevelSensor)
def test_sensor_value_changed(mock_openzwave):
"""Test value changed for Z-Wave sensor."""
node = MockNode(command_classes=[const.COMMAND_CLASS_BATTERY])
value = MockValue(data=12.34, command_class=const.COMMAND_CLASS_BATTERY,
node=node, units='%')
values = MockEntityValues(primary=value)
device = zwave.get_device(node=node, values=values, node_config={})
assert device.state == 12.34
assert device.unit_of_measurement == '%'
value.data = 45.67
value_changed(value)
assert device.state == 45.67
def test_multilevelsensor_value_changed_temp_fahrenheit(mock_openzwave):
"""Test value changed for Z-Wave multilevel sensor for temperature."""
node = MockNode(command_classes=[const.COMMAND_CLASS_SENSOR_MULTILEVEL,

168
tests/components/zwave/test_node_entity.py Normal file
View File

@ -0,0 +1,168 @@
"""Test Z-Wave node entity."""
import unittest
from unittest.mock import patch, Mock
from tests.common import get_test_home_assistant
import tests.mock.zwave as mock_zwave
import pytest
from homeassistant.components.zwave import node_entity
@pytest.mark.usefixtures('mock_openzwave')
class TestZWaveBaseEntity(unittest.TestCase):
"""Class to test ZWaveBaseEntity."""
def setUp(self):
"""Initialize values for this testcase class."""
self.hass = get_test_home_assistant()
def call_soon(time, func, *args):
"""Replace call_later by call_soon."""
return self.hass.loop.call_soon(func, *args)
self.hass.loop.call_later = call_soon
self.base_entity = node_entity.ZWaveBaseEntity()
self.base_entity.hass = self.hass
self.hass.start()
def tearDown(self): # pylint: disable=invalid-name
"""Stop everything that was started."""
self.hass.stop()
def test_maybe_schedule_update(self):
"""Test maybe_schedule_update."""
with patch.object(self.base_entity, 'async_update_ha_state',
Mock()) as mock_update:
self.base_entity.maybe_schedule_update()
self.hass.block_till_done()
mock_update.assert_called_once_with()
def test_maybe_schedule_update_called_twice(self):
"""Test maybe_schedule_update called twice."""
with patch.object(self.base_entity, 'async_update_ha_state',
Mock()) as mock_update:
self.base_entity.maybe_schedule_update()
self.base_entity.maybe_schedule_update()
self.hass.block_till_done()
mock_update.assert_called_once_with()
@pytest.mark.usefixtures('mock_openzwave')
class TestZWaveNodeEntity(unittest.TestCase):
"""Class to test ZWaveNodeEntity."""
def setUp(self):
"""Initialize values for this testcase class."""
self.node = mock_zwave.MockNode(
query_stage='Dynamic', is_awake=True, is_ready=False,
is_failed=False, is_info_received=True, max_baud_rate=40000,
is_zwave_plus=False, capabilities=[], neighbors=[], location=None)
self.entity = node_entity.ZWaveNodeEntity(self.node)
def test_network_node_changed_from_value(self):
"""Test for network_node_changed."""
value = mock_zwave.MockValue(node=self.node)
with patch.object(self.entity, 'maybe_schedule_update') as mock:
mock_zwave.value_changed(value)
mock.assert_called_once_with()
def test_network_node_changed_from_node(self):
"""Test for network_node_changed."""
with patch.object(self.entity, 'maybe_schedule_update') as mock:
mock_zwave.node_changed(self.node)
mock.assert_called_once_with()
def test_network_node_changed_from_another_node(self):
"""Test for network_node_changed."""
with patch.object(self.entity, 'maybe_schedule_update') as mock:
node = mock_zwave.MockNode(node_id=1024)
mock_zwave.node_changed(node)
self.assertFalse(mock.called)
def test_network_node_changed_from_notification(self):
"""Test for network_node_changed."""
with patch.object(self.entity, 'maybe_schedule_update') as mock:
mock_zwave.notification(node_id=self.node.node_id)
mock.assert_called_once_with()
def test_network_node_changed_from_another_notification(self):
"""Test for network_node_changed."""
with patch.object(self.entity, 'maybe_schedule_update') as mock:
mock_zwave.notification(node_id=1024)
self.assertFalse(mock.called)
def test_node_changed(self):
"""Test node_changed function."""
self.assertEqual({'node_id': self.node.node_id},
self.entity.device_state_attributes)
self.node.get_values.return_value = {
1: mock_zwave.MockValue(data=1800)
}
self.entity.node_changed()
self.assertEqual(
{'node_id': self.node.node_id,
'query_stage': 'Dynamic',
'is_awake': True,
'is_ready': False,
'is_failed': False,
'is_info_received': True,
'max_baud_rate': 40000,
'is_zwave_plus': False,
'battery_level': 42,
'wake_up_interval': 1800},
self.entity.device_state_attributes)
self.node.can_wake_up_value = False
self.entity.node_changed()
self.assertNotIn(
'wake_up_interval', self.entity.device_state_attributes)
def test_name(self):
"""Test name property."""
self.assertEqual('Mock Node', self.entity.name)
def test_state_before_update(self):
"""Test state before update was called."""
self.assertIsNone(self.entity.state)
def test_state_not_ready(self):
"""Test state property."""
self.node.is_ready = False
self.entity.node_changed()
self.assertEqual('Dynamic', self.entity.state)
self.node.is_failed = True
self.entity.node_changed()
self.assertEqual('Dead (Dynamic)', self.entity.state)
self.node.is_failed = False
self.node.is_awake = False
self.entity.node_changed()
self.assertEqual('Sleeping (Dynamic)', self.entity.state)
def test_state_ready(self):
"""Test state property."""
self.node.is_ready = True
self.entity.node_changed()
self.assertEqual('Ready', self.entity.state)
self.node.is_failed = True
self.entity.node_changed()
self.assertEqual('Dead', self.entity.state)
self.node.is_failed = False
self.node.is_awake = False
self.entity.node_changed()
self.assertEqual('Sleeping', self.entity.state)
def test_not_polled(self):
"""Test should_poll property."""
self.assertFalse(self.entity.should_poll)
def test_sub_status():
"""Test sub_status function."""
assert node_entity.sub_status('Status', 'Stage') == 'Status (Stage)'
assert node_entity.sub_status('Status', '') == 'Status'

4
tests/conftest.py
View File

@ -14,7 +14,7 @@ from homeassistant.components import mqtt
from .common import async_test_home_assistant, mock_coro
from .test_util.aiohttp import mock_aiohttp_client
from .mock.zwave import SIGNAL_VALUE_CHANGED
from .mock.zwave import SIGNAL_VALUE_CHANGED, SIGNAL_NODE, SIGNAL_NOTIFICATION
if os.environ.get('UVLOOP') == '1':
import uvloop
@ -101,6 +101,8 @@ def mock_openzwave():
libopenzwave = base_mock.libopenzwave
libopenzwave.__file__ = 'test'
base_mock.network.ZWaveNetwork.SIGNAL_VALUE_CHANGED = SIGNAL_VALUE_CHANGED
base_mock.network.ZWaveNetwork.SIGNAL_NODE = SIGNAL_NODE
base_mock.network.ZWaveNetwork.SIGNAL_NOTIFICATION = SIGNAL_NOTIFICATION
with patch.dict('sys.modules', {
'libopenzwave': libopenzwave,

35
tests/mock/zwave.py
View File

@ -4,6 +4,8 @@ from unittest.mock import MagicMock
from pydispatch import dispatcher
SIGNAL_VALUE_CHANGED = 'mock_value_changed'
SIGNAL_NODE = 'mock_node'
SIGNAL_NOTIFICATION = 'mock_notification'
def value_changed(value):
@ -16,6 +18,24 @@ def value_changed(value):
)
def node_changed(node):
"""Fire a node changed."""
dispatcher.send(
SIGNAL_NODE,
node=node,
network=node._network
)
def notification(node_id, network=None):
"""Fire a notification."""
dispatcher.send(
SIGNAL_NOTIFICATION,
args={'nodeId': node_id},
network=network
)
class MockNode(MagicMock):
"""Mock Z-Wave node."""
@ -25,7 +45,9 @@ class MockNode(MagicMock):
manufacturer_id='ABCD',
product_id='123',
product_type='678',
command_classes=None):
command_classes=None,
can_wake_up_value=True,
**kwargs):
"""Initialize a Z-Wave mock node."""
super().__init__()
self.node_id = node_id
@ -33,12 +55,23 @@ class MockNode(MagicMock):
self.manufacturer_id = manufacturer_id
self.product_id = product_id
self.product_type = product_type
self.can_wake_up_value = can_wake_up_value
self._command_classes = command_classes or []
for attr_name in kwargs:
setattr(self, attr_name, kwargs[attr_name])
def has_command_class(self, command_class):
"""Test if mock has a command class."""
return command_class in self._command_classes
def get_battery_level(self):
"""Return mock battery level."""
return 42
def can_wake_up(self):
"""Return whether the node can wake up."""
return self.can_wake_up_value
def _get_child_mock(self, **kw):
"""Create child mocks with right MagicMock class."""
return MagicMock(**kw)