Tests for zwave services (#6937)
* Initial tests for zwave services * Fix linter issues * Complete zwave service tests
This commit is contained in:
parent
8cff98d07b
commit
df77529bfe
6 changed files with 452 additions and 36 deletions
|
@ -175,7 +175,6 @@ omit =
|
|||
homeassistant/components/climate/oem.py
|
||||
homeassistant/components/climate/proliphix.py
|
||||
homeassistant/components/climate/radiotherm.py
|
||||
homeassistant/components/config/zwave.py
|
||||
homeassistant/components/cover/garadget.py
|
||||
homeassistant/components/cover/homematic.py
|
||||
homeassistant/components/cover/myq.py
|
||||
|
@ -442,7 +441,6 @@ 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
|
||||
|
||||
|
||||
|
|
|
@ -185,8 +185,8 @@ def get_config_value(node, value_index, tries=5):
|
|||
"""Return the current configuration value for a specific index."""
|
||||
try:
|
||||
for value in node.values.values():
|
||||
# 112 == config command class
|
||||
if value.command_class == 112 and value.index == value_index:
|
||||
if (value.command_class == const.COMMAND_CLASS_CONFIGURATION
|
||||
and value.index == value_index):
|
||||
return value.data
|
||||
except RuntimeError:
|
||||
# If we get an runtime error the dict has changed while
|
||||
|
@ -384,7 +384,7 @@ def setup(hass, config):
|
|||
_LOGGER.info("Zwave test_network have been initialized.")
|
||||
NETWORK.test()
|
||||
|
||||
def stop_zwave(_service_or_event):
|
||||
def stop_network(_service_or_event):
|
||||
"""Stop Z-Wave network."""
|
||||
_LOGGER.info("Stopping ZWave network.")
|
||||
NETWORK.stop()
|
||||
|
@ -532,7 +532,7 @@ def setup(hass, config):
|
|||
poll_interval = NETWORK.get_poll_interval()
|
||||
_LOGGER.info("zwave polling interval set to %d ms", poll_interval)
|
||||
|
||||
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_zwave)
|
||||
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_network)
|
||||
|
||||
# Register node services for Z-Wave network
|
||||
hass.services.register(DOMAIN, const.SERVICE_ADD_NODE, add_node,
|
||||
|
@ -553,7 +553,8 @@ def setup(hass, config):
|
|||
hass.services.register(DOMAIN, const.SERVICE_TEST_NETWORK,
|
||||
test_network,
|
||||
descriptions[const.SERVICE_TEST_NETWORK])
|
||||
hass.services.register(DOMAIN, const.SERVICE_STOP_NETWORK, stop_zwave,
|
||||
hass.services.register(DOMAIN, const.SERVICE_STOP_NETWORK,
|
||||
stop_network,
|
||||
descriptions[const.SERVICE_STOP_NETWORK])
|
||||
hass.services.register(DOMAIN, const.SERVICE_START_NETWORK,
|
||||
start_zwave,
|
||||
|
@ -840,4 +841,5 @@ class ZWaveDeviceEntity(ZWaveBaseEntity):
|
|||
def refresh_from_network(self):
|
||||
"""Refresh all dependent values from zwave network."""
|
||||
for value in self.values:
|
||||
self.node.refresh_value(value.value_id)
|
||||
if value is not None:
|
||||
self.node.refresh_value(value.value_id)
|
||||
|
|
|
@ -15,7 +15,7 @@ def device(hass, mock_openzwave):
|
|||
node = MockNode()
|
||||
values = MockEntityValues(
|
||||
primary=MockValue(data=1, node=node),
|
||||
temperature=MockValue(data=5, node=node),
|
||||
temperature=MockValue(data=5, node=node, units=None),
|
||||
mode=MockValue(data=b'test1', data_items=[0, 1, 2], node=node),
|
||||
fan_mode=MockValue(data=b'test2', data_items=[3, 4, 5], node=node),
|
||||
operating_state=MockValue(data=6, node=node),
|
||||
|
@ -30,9 +30,10 @@ def device(hass, mock_openzwave):
|
|||
def device_zxt_120(hass, mock_openzwave):
|
||||
"""Fixture to provide a precreated climate device."""
|
||||
node = MockNode(manufacturer_id='5254', product_id='8377')
|
||||
|
||||
values = MockEntityValues(
|
||||
primary=MockValue(data=1, node=node),
|
||||
temperature=MockValue(data=5, node=node),
|
||||
temperature=MockValue(data=5, node=node, units=None),
|
||||
mode=MockValue(data=b'test1', data_items=[0, 1, 2], node=node),
|
||||
fan_mode=MockValue(data=b'test2', data_items=[3, 4, 5], node=node),
|
||||
operating_state=MockValue(data=6, node=node),
|
||||
|
|
|
@ -1,11 +1,21 @@
|
|||
"""Tests for the Z-Wave init."""
|
||||
import asyncio
|
||||
import unittest
|
||||
from collections import OrderedDict
|
||||
|
||||
from homeassistant.bootstrap import async_setup_component
|
||||
from homeassistant.const import ATTR_ENTITY_ID, EVENT_HOMEASSISTANT_START
|
||||
from homeassistant.components import zwave
|
||||
from homeassistant.components.binary_sensor.zwave import get_device
|
||||
from homeassistant.components.zwave import (
|
||||
CONFIG_SCHEMA, CONF_DEVICE_CONFIG_GLOB)
|
||||
const, CONFIG_SCHEMA, CONF_DEVICE_CONFIG_GLOB)
|
||||
from homeassistant.setup import setup_component
|
||||
|
||||
import pytest
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
from tests.mock.zwave import MockNetwork, MockNode, MockValue, MockEntityValues
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
|
@ -49,3 +59,360 @@ class TestZwave(unittest.TestCase):
|
|||
{'zwave': {CONF_DEVICE_CONFIG_GLOB: OrderedDict()}})
|
||||
self.assertIsInstance(
|
||||
conf['zwave'][CONF_DEVICE_CONFIG_GLOB], OrderedDict)
|
||||
|
||||
|
||||
class TestZWaveServices(unittest.TestCase):
|
||||
"""Tests for zwave services."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def set_mock_openzwave(self, mock_openzwave):
|
||||
"""Use the mock_openzwave fixture for this class."""
|
||||
self.mock_openzwave = mock_openzwave
|
||||
|
||||
def setUp(self):
|
||||
"""Initialize values for this testcase class."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self.hass.start()
|
||||
|
||||
# Initialize zwave
|
||||
setup_component(self.hass, 'zwave', {'zwave': {}})
|
||||
self.hass.block_till_done()
|
||||
zwave.NETWORK.state = MockNetwork.STATE_READY
|
||||
self.hass.bus.fire(EVENT_HOMEASSISTANT_START)
|
||||
self.hass.block_till_done()
|
||||
|
||||
def tearDown(self): # pylint: disable=invalid-name
|
||||
"""Stop everything that was started."""
|
||||
self.hass.stop()
|
||||
|
||||
def test_add_node(self):
|
||||
"""Test zwave add_node service."""
|
||||
self.hass.services.call('zwave', 'add_node', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.controller.add_node.called
|
||||
assert len(zwave.NETWORK.controller.add_node.mock_calls) == 1
|
||||
assert len(zwave.NETWORK.controller.add_node.mock_calls[0][1]) == 0
|
||||
|
||||
def test_add_node_secure(self):
|
||||
"""Test zwave add_node_secure service."""
|
||||
self.hass.services.call('zwave', 'add_node_secure', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.controller.add_node.called
|
||||
assert len(zwave.NETWORK.controller.add_node.mock_calls) == 1
|
||||
assert zwave.NETWORK.controller.add_node.mock_calls[0][1][0] is True
|
||||
|
||||
def test_remove_node(self):
|
||||
"""Test zwave remove_node service."""
|
||||
self.hass.services.call('zwave', 'remove_node', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.controller.remove_node.called
|
||||
assert len(zwave.NETWORK.controller.remove_node.mock_calls) == 1
|
||||
|
||||
def test_cancel_command(self):
|
||||
"""Test zwave cancel_command service."""
|
||||
self.hass.services.call('zwave', 'cancel_command', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.controller.cancel_command.called
|
||||
assert len(zwave.NETWORK.controller.cancel_command.mock_calls) == 1
|
||||
|
||||
def test_heal_network(self):
|
||||
"""Test zwave heal_network service."""
|
||||
self.hass.services.call('zwave', 'heal_network', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.heal.called
|
||||
assert len(zwave.NETWORK.heal.mock_calls) == 1
|
||||
|
||||
def test_soft_reset(self):
|
||||
"""Test zwave soft_reset service."""
|
||||
self.hass.services.call('zwave', 'soft_reset', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.controller.soft_reset.called
|
||||
assert len(zwave.NETWORK.controller.soft_reset.mock_calls) == 1
|
||||
|
||||
def test_test_network(self):
|
||||
"""Test zwave test_network service."""
|
||||
self.hass.services.call('zwave', 'test_network', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.test.called
|
||||
assert len(zwave.NETWORK.test.mock_calls) == 1
|
||||
|
||||
def test_stop_network(self):
|
||||
"""Test zwave stop_network service."""
|
||||
self.hass.services.call('zwave', 'stop_network', {})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.stop.called
|
||||
assert len(zwave.NETWORK.stop.mock_calls) == 1
|
||||
|
||||
def test_rename_node(self):
|
||||
"""Test zwave rename_node service."""
|
||||
zwave.NETWORK.nodes = {11: MagicMock()}
|
||||
self.hass.services.call('zwave', 'rename_node', {
|
||||
const.ATTR_NODE_ID: 11,
|
||||
const.ATTR_NAME: 'test_name',
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert zwave.NETWORK.nodes[11].name == 'test_name'
|
||||
|
||||
def test_remove_failed_node(self):
|
||||
"""Test zwave remove_failed_node service."""
|
||||
self.hass.services.call('zwave', 'remove_failed_node', {
|
||||
const.ATTR_NODE_ID: 12,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
remove_failed_node = zwave.NETWORK.controller.remove_failed_node
|
||||
assert remove_failed_node.called
|
||||
assert len(remove_failed_node.mock_calls) == 1
|
||||
assert remove_failed_node.mock_calls[0][1][0] == 12
|
||||
|
||||
def test_replace_failed_node(self):
|
||||
"""Test zwave replace_failed_node service."""
|
||||
self.hass.services.call('zwave', 'replace_failed_node', {
|
||||
const.ATTR_NODE_ID: 13,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
replace_failed_node = zwave.NETWORK.controller.replace_failed_node
|
||||
assert replace_failed_node.called
|
||||
assert len(replace_failed_node.mock_calls) == 1
|
||||
assert replace_failed_node.mock_calls[0][1][0] == 13
|
||||
|
||||
def test_set_config_parameter(self):
|
||||
"""Test zwave set_config_parameter service."""
|
||||
value = MockValue(
|
||||
index=12,
|
||||
command_class=const.COMMAND_CLASS_CONFIGURATION,
|
||||
)
|
||||
value_list = MockValue(
|
||||
index=13,
|
||||
command_class=const.COMMAND_CLASS_CONFIGURATION,
|
||||
type=const.TYPE_LIST,
|
||||
data_items=['item1', 'item2', 'item3'],
|
||||
)
|
||||
node = MockNode(node_id=14)
|
||||
node.get_values.return_value = {12: value, 13: value_list}
|
||||
zwave.NETWORK.nodes = {14: node}
|
||||
|
||||
self.hass.services.call('zwave', 'set_config_parameter', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_CONFIG_PARAMETER: 13,
|
||||
const.ATTR_CONFIG_VALUE: 1,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert node.set_config_param.called
|
||||
assert len(node.set_config_param.mock_calls) == 1
|
||||
assert node.set_config_param.mock_calls[0][1][0] == 13
|
||||
assert node.set_config_param.mock_calls[0][1][1] == 1
|
||||
assert node.set_config_param.mock_calls[0][1][2] == 2
|
||||
node.set_config_param.reset_mock()
|
||||
|
||||
self.hass.services.call('zwave', 'set_config_parameter', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_CONFIG_PARAMETER: 13,
|
||||
const.ATTR_CONFIG_VALUE: 7,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert not node.set_config_param.called
|
||||
node.set_config_param.reset_mock()
|
||||
|
||||
self.hass.services.call('zwave', 'set_config_parameter', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_CONFIG_PARAMETER: 12,
|
||||
const.ATTR_CONFIG_VALUE: 0x01020304,
|
||||
const.ATTR_CONFIG_SIZE: 4,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert node.set_config_param.called
|
||||
assert len(node.set_config_param.mock_calls) == 1
|
||||
assert node.set_config_param.mock_calls[0][1][0] == 12
|
||||
assert node.set_config_param.mock_calls[0][1][1] == 0x01020304
|
||||
assert node.set_config_param.mock_calls[0][1][2] == 4
|
||||
node.set_config_param.reset_mock()
|
||||
|
||||
def test_print_config_parameter(self):
|
||||
"""Test zwave print_config_parameter service."""
|
||||
value1 = MockValue(
|
||||
index=12,
|
||||
command_class=const.COMMAND_CLASS_CONFIGURATION,
|
||||
data=1234,
|
||||
)
|
||||
value2 = MockValue(
|
||||
index=13,
|
||||
command_class=const.COMMAND_CLASS_CONFIGURATION,
|
||||
data=2345,
|
||||
)
|
||||
node = MockNode(node_id=14)
|
||||
node.values = {12: value1, 13: value2}
|
||||
zwave.NETWORK.nodes = {14: node}
|
||||
|
||||
with patch.object(zwave, '_LOGGER') as mock_logger:
|
||||
self.hass.services.call('zwave', 'print_config_parameter', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_CONFIG_PARAMETER: 13,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert mock_logger.info.called
|
||||
assert len(mock_logger.info.mock_calls) == 1
|
||||
assert mock_logger.info.mock_calls[0][1][1] == 13
|
||||
assert mock_logger.info.mock_calls[0][1][2] == 14
|
||||
assert mock_logger.info.mock_calls[0][1][3] == 2345
|
||||
|
||||
def test_print_node(self):
|
||||
"""Test zwave print_config_parameter service."""
|
||||
node1 = MockNode(node_id=14)
|
||||
node2 = MockNode(node_id=15)
|
||||
zwave.NETWORK.nodes = {14: node1, 15: node2}
|
||||
|
||||
with patch.object(zwave, 'pprint') as mock_pprint:
|
||||
self.hass.services.call('zwave', 'print_node', {
|
||||
const.ATTR_NODE_ID: 15,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert mock_pprint.called
|
||||
assert len(mock_pprint.mock_calls) == 1
|
||||
assert mock_pprint.mock_calls[0][1][0]['node_id'] == 15
|
||||
|
||||
def test_set_wakeup(self):
|
||||
"""Test zwave set_wakeup service."""
|
||||
value = MockValue(
|
||||
index=12,
|
||||
command_class=const.COMMAND_CLASS_WAKE_UP,
|
||||
)
|
||||
node = MockNode(node_id=14)
|
||||
node.values = {12: value}
|
||||
node.get_values.return_value = node.values
|
||||
zwave.NETWORK.nodes = {14: node}
|
||||
|
||||
self.hass.services.call('zwave', 'set_wakeup', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_CONFIG_VALUE: 15,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert value.data == 15
|
||||
|
||||
node.can_wake_up_value = False
|
||||
self.hass.services.call('zwave', 'set_wakeup', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_CONFIG_VALUE: 20,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert value.data == 15
|
||||
|
||||
def test_add_association(self):
|
||||
"""Test zwave change_association service."""
|
||||
ZWaveGroup = self.mock_openzwave.group.ZWaveGroup
|
||||
group = MagicMock()
|
||||
ZWaveGroup.return_value = group
|
||||
|
||||
value = MockValue(
|
||||
index=12,
|
||||
command_class=const.COMMAND_CLASS_WAKE_UP,
|
||||
)
|
||||
node = MockNode(node_id=14)
|
||||
node.values = {12: value}
|
||||
node.get_values.return_value = node.values
|
||||
zwave.NETWORK.nodes = {14: node}
|
||||
|
||||
self.hass.services.call('zwave', 'change_association', {
|
||||
const.ATTR_ASSOCIATION: 'add',
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_TARGET_NODE_ID: 24,
|
||||
const.ATTR_GROUP: 3,
|
||||
const.ATTR_INSTANCE: 5,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert ZWaveGroup.called
|
||||
assert len(ZWaveGroup.mock_calls) == 2
|
||||
assert ZWaveGroup.mock_calls[0][1][0] == 3
|
||||
assert ZWaveGroup.mock_calls[0][1][2] == 14
|
||||
assert group.add_association.called
|
||||
assert len(group.add_association.mock_calls) == 1
|
||||
assert group.add_association.mock_calls[0][1][0] == 24
|
||||
assert group.add_association.mock_calls[0][1][1] == 5
|
||||
|
||||
def test_remove_association(self):
|
||||
"""Test zwave change_association service."""
|
||||
ZWaveGroup = self.mock_openzwave.group.ZWaveGroup
|
||||
group = MagicMock()
|
||||
ZWaveGroup.return_value = group
|
||||
|
||||
value = MockValue(
|
||||
index=12,
|
||||
command_class=const.COMMAND_CLASS_WAKE_UP,
|
||||
)
|
||||
node = MockNode(node_id=14)
|
||||
node.values = {12: value}
|
||||
node.get_values.return_value = node.values
|
||||
zwave.NETWORK.nodes = {14: node}
|
||||
|
||||
self.hass.services.call('zwave', 'change_association', {
|
||||
const.ATTR_ASSOCIATION: 'remove',
|
||||
const.ATTR_NODE_ID: 14,
|
||||
const.ATTR_TARGET_NODE_ID: 24,
|
||||
const.ATTR_GROUP: 3,
|
||||
const.ATTR_INSTANCE: 5,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert ZWaveGroup.called
|
||||
assert len(ZWaveGroup.mock_calls) == 2
|
||||
assert ZWaveGroup.mock_calls[0][1][0] == 3
|
||||
assert ZWaveGroup.mock_calls[0][1][2] == 14
|
||||
assert group.remove_association.called
|
||||
assert len(group.remove_association.mock_calls) == 1
|
||||
assert group.remove_association.mock_calls[0][1][0] == 24
|
||||
assert group.remove_association.mock_calls[0][1][1] == 5
|
||||
|
||||
def test_refresh_entity(self):
|
||||
"""Test zwave refresh_entity service."""
|
||||
node = MockNode()
|
||||
value = MockValue(data=False, node=node,
|
||||
command_class=const.COMMAND_CLASS_SENSOR_BINARY)
|
||||
power_value = MockValue(data=50, node=node,
|
||||
command_class=const.COMMAND_CLASS_METER)
|
||||
values = MockEntityValues(primary=value, power=power_value)
|
||||
device = get_device(node=node, values=values, node_config={})
|
||||
device.hass = self.hass
|
||||
device.entity_id = 'binary_sensor.mock_entity_id'
|
||||
self.hass.add_job(device.async_added_to_hass())
|
||||
self.hass.block_till_done()
|
||||
|
||||
self.hass.services.call('zwave', 'refresh_entity', {
|
||||
ATTR_ENTITY_ID: 'binary_sensor.mock_entity_id',
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert node.refresh_value.called
|
||||
assert len(node.refresh_value.mock_calls) == 2
|
||||
self.assertEqual(sorted([node.refresh_value.mock_calls[0][1][0],
|
||||
node.refresh_value.mock_calls[1][1][0]]),
|
||||
sorted([value.value_id, power_value.value_id]))
|
||||
|
||||
def test_refresh_node(self):
|
||||
"""Test zwave refresh_node service."""
|
||||
node = MockNode(node_id=14)
|
||||
zwave.NETWORK.nodes = {14: node}
|
||||
self.hass.services.call('zwave', 'refresh_node', {
|
||||
const.ATTR_NODE_ID: 14,
|
||||
})
|
||||
self.hass.block_till_done()
|
||||
|
||||
assert node.refresh_info.called
|
||||
assert len(node.refresh_info.mock_calls) == 1
|
||||
|
|
|
@ -12,9 +12,9 @@ from homeassistant import util, setup
|
|||
from homeassistant.util import location
|
||||
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, SIGNAL_NODE, SIGNAL_NOTIFICATION
|
||||
from tests.common import async_test_home_assistant, mock_coro
|
||||
from tests.test_util.aiohttp import mock_aiohttp_client
|
||||
from tests.mock.zwave import MockNetwork
|
||||
|
||||
if os.environ.get('UVLOOP') == '1':
|
||||
import uvloop
|
||||
|
@ -100,9 +100,7 @@ def mock_openzwave():
|
|||
base_mock = MagicMock()
|
||||
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
|
||||
base_mock.network.ZWaveNetwork = MockNetwork
|
||||
|
||||
with patch.dict('sys.modules', {
|
||||
'libopenzwave': libopenzwave,
|
||||
|
|
|
@ -3,15 +3,11 @@ 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):
|
||||
"""Fire a value changed."""
|
||||
dispatcher.send(
|
||||
SIGNAL_VALUE_CHANGED,
|
||||
MockNetwork.SIGNAL_VALUE_CHANGED,
|
||||
value=value,
|
||||
node=value.node,
|
||||
network=value.node._network
|
||||
|
@ -21,7 +17,7 @@ def value_changed(value):
|
|||
def node_changed(node):
|
||||
"""Fire a node changed."""
|
||||
dispatcher.send(
|
||||
SIGNAL_NODE,
|
||||
MockNetwork.SIGNAL_NODE,
|
||||
node=node,
|
||||
network=node._network
|
||||
)
|
||||
|
@ -30,12 +26,70 @@ def node_changed(node):
|
|||
def notification(node_id, network=None):
|
||||
"""Fire a notification."""
|
||||
dispatcher.send(
|
||||
SIGNAL_NOTIFICATION,
|
||||
MockNetwork.SIGNAL_NOTIFICATION,
|
||||
args={'nodeId': node_id},
|
||||
network=network
|
||||
)
|
||||
|
||||
|
||||
class MockNetwork(MagicMock):
|
||||
"""Mock Z-Wave network."""
|
||||
|
||||
SIGNAL_NETWORK_FAILED = 'mock_NetworkFailed'
|
||||
SIGNAL_NETWORK_STARTED = 'mock_NetworkStarted'
|
||||
SIGNAL_NETWORK_READY = 'mock_NetworkReady'
|
||||
SIGNAL_NETWORK_STOPPED = 'mock_NetworkStopped'
|
||||
SIGNAL_NETWORK_RESETTED = 'mock_DriverResetted'
|
||||
SIGNAL_NETWORK_AWAKED = 'mock_DriverAwaked'
|
||||
SIGNAL_DRIVER_FAILED = 'mock_DriverFailed'
|
||||
SIGNAL_DRIVER_READY = 'mock_DriverReady'
|
||||
SIGNAL_DRIVER_RESET = 'mock_DriverReset'
|
||||
SIGNAL_DRIVER_REMOVED = 'mock_DriverRemoved'
|
||||
SIGNAL_GROUP = 'mock_Group'
|
||||
SIGNAL_NODE = 'mock_Node'
|
||||
SIGNAL_NODE_ADDED = 'mock_NodeAdded'
|
||||
SIGNAL_NODE_EVENT = 'mock_NodeEvent'
|
||||
SIGNAL_NODE_NAMING = 'mock_NodeNaming'
|
||||
SIGNAL_NODE_NEW = 'mock_NodeNew'
|
||||
SIGNAL_NODE_PROTOCOL_INFO = 'mock_NodeProtocolInfo'
|
||||
SIGNAL_NODE_READY = 'mock_NodeReady'
|
||||
SIGNAL_NODE_REMOVED = 'mock_NodeRemoved'
|
||||
SIGNAL_SCENE_EVENT = 'mock_SceneEvent'
|
||||
SIGNAL_VALUE = 'mock_Value'
|
||||
SIGNAL_VALUE_ADDED = 'mock_ValueAdded'
|
||||
SIGNAL_VALUE_CHANGED = 'mock_ValueChanged'
|
||||
SIGNAL_VALUE_REFRESHED = 'mock_ValueRefreshed'
|
||||
SIGNAL_VALUE_REMOVED = 'mock_ValueRemoved'
|
||||
SIGNAL_POLLING_ENABLED = 'mock_PollingEnabled'
|
||||
SIGNAL_POLLING_DISABLED = 'mock_PollingDisabled'
|
||||
SIGNAL_CREATE_BUTTON = 'mock_CreateButton'
|
||||
SIGNAL_DELETE_BUTTON = 'mock_DeleteButton'
|
||||
SIGNAL_BUTTON_ON = 'mock_ButtonOn'
|
||||
SIGNAL_BUTTON_OFF = 'mock_ButtonOff'
|
||||
SIGNAL_ESSENTIAL_NODE_QUERIES_COMPLETE = \
|
||||
'mock_EssentialNodeQueriesComplete'
|
||||
SIGNAL_NODE_QUERIES_COMPLETE = 'mock_NodeQueriesComplete'
|
||||
SIGNAL_AWAKE_NODES_QUERIED = 'mock_AwakeNodesQueried'
|
||||
SIGNAL_ALL_NODES_QUERIED = 'mock_AllNodesQueried'
|
||||
SIGNAL_ALL_NODES_QUERIED_SOME_DEAD = 'mock_AllNodesQueriedSomeDead'
|
||||
SIGNAL_MSG_COMPLETE = 'mock_MsgComplete'
|
||||
SIGNAL_NOTIFICATION = 'mock_Notification'
|
||||
SIGNAL_CONTROLLER_COMMAND = 'mock_ControllerCommand'
|
||||
SIGNAL_CONTROLLER_WAITING = 'mock_ControllerWaiting'
|
||||
|
||||
STATE_STOPPED = 0
|
||||
STATE_FAILED = 1
|
||||
STATE_RESETTED = 3
|
||||
STATE_STARTED = 5
|
||||
STATE_AWAKED = 7
|
||||
STATE_READY = 10
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Initialize a Z-Wave mock network."""
|
||||
super().__init__()
|
||||
self.state = MockNetwork.STATE_STOPPED
|
||||
|
||||
|
||||
class MockNode(MagicMock):
|
||||
"""Mock Z-Wave node."""
|
||||
|
||||
|
@ -47,6 +101,7 @@ class MockNode(MagicMock):
|
|||
product_type='678',
|
||||
command_classes=None,
|
||||
can_wake_up_value=True,
|
||||
network=None,
|
||||
**kwargs):
|
||||
"""Initialize a Z-Wave mock node."""
|
||||
super().__init__()
|
||||
|
@ -57,6 +112,8 @@ class MockNode(MagicMock):
|
|||
self.product_type = product_type
|
||||
self.can_wake_up_value = can_wake_up_value
|
||||
self._command_classes = command_classes or []
|
||||
if network is not None:
|
||||
self._network = network
|
||||
for attr_name in kwargs:
|
||||
setattr(self, attr_name, kwargs[attr_name])
|
||||
|
||||
|
@ -84,30 +141,23 @@ class MockValue(MagicMock):
|
|||
|
||||
def __init__(self, *,
|
||||
label='Mock Value',
|
||||
data=None,
|
||||
data_items=None,
|
||||
node=None,
|
||||
instance=0,
|
||||
index=0,
|
||||
command_class=None,
|
||||
units=None,
|
||||
type=None,
|
||||
value_id=None):
|
||||
value_id=None,
|
||||
**kwargs):
|
||||
"""Initialize a Z-Wave mock value."""
|
||||
super().__init__()
|
||||
self.label = label
|
||||
self.data = data
|
||||
self.data_items = data_items
|
||||
self.node = node
|
||||
self.instance = instance
|
||||
self.index = index
|
||||
self.command_class = command_class
|
||||
self.units = units
|
||||
self.type = type
|
||||
if value_id is None:
|
||||
MockValue._mock_value_id += 1
|
||||
value_id = MockValue._mock_value_id
|
||||
self.value_id = value_id
|
||||
for attr_name in kwargs:
|
||||
setattr(self, attr_name, kwargs[attr_name])
|
||||
|
||||
def _get_child_mock(self, **kw):
|
||||
"""Create child mocks with right MagicMock class."""
|
||||
|
|
Loading…
Add table
Reference in a new issue