Rewrite zwave init tests to pytest style test function (phase 1) (#41301)
This commit is contained in:
Rob Bierbooms
GitHub
parent
c5653ec57b
commit
b2b5f2ffcb
2 changed files with 487 additions and 301 deletions
|
|
@ -1,8 +1,10 @@
|
||||||
"""Fixtures for Z-Wave tests."""
|
"""Fixtures for Z-Wave tests."""
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from tests.async_mock import MagicMock, patch
|
from homeassistant.components.zwave import const
|
||||||
from tests.mock.zwave import MockNetwork, MockOption
|
|
||||||
|
from tests.async_mock import AsyncMock, MagicMock, patch
|
||||||
|
from tests.mock.zwave import MockNetwork, MockNode, MockOption, MockValue
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
|
@ -24,3 +26,53 @@ def mock_openzwave():
|
||||||
},
|
},
|
||||||
):
|
):
|
||||||
yield base_mock
|
yield base_mock
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_discovery():
|
||||||
|
"""Mock discovery."""
|
||||||
|
discovery = MagicMock()
|
||||||
|
discovery.async_load_platform = AsyncMock(return_value=None)
|
||||||
|
yield discovery
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_import_module():
|
||||||
|
"""Mock import module."""
|
||||||
|
platform = MagicMock()
|
||||||
|
mock_device = MagicMock()
|
||||||
|
mock_device.name = "test_device"
|
||||||
|
platform.get_device.return_value = mock_device
|
||||||
|
|
||||||
|
import_module = MagicMock()
|
||||||
|
import_module.return_value = platform
|
||||||
|
yield import_module
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_values():
|
||||||
|
"""Mock values."""
|
||||||
|
node = MockNode()
|
||||||
|
mock_schema = {
|
||||||
|
const.DISC_COMPONENT: "mock_component",
|
||||||
|
const.DISC_VALUES: {
|
||||||
|
const.DISC_PRIMARY: {const.DISC_COMMAND_CLASS: ["mock_primary_class"]},
|
||||||
|
"secondary": {const.DISC_COMMAND_CLASS: ["mock_secondary_class"]},
|
||||||
|
"optional": {
|
||||||
|
const.DISC_COMMAND_CLASS: ["mock_optional_class"],
|
||||||
|
const.DISC_OPTIONAL: True,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
value_class = MagicMock()
|
||||||
|
value_class.primary = MockValue(
|
||||||
|
command_class="mock_primary_class", node=node, value_id=1000
|
||||||
|
)
|
||||||
|
value_class.secondary = MockValue(command_class="mock_secondary_class", node=node)
|
||||||
|
value_class.duplicate_secondary = MockValue(
|
||||||
|
command_class="mock_secondary_class", node=node
|
||||||
|
)
|
||||||
|
value_class.optional = MockValue(command_class="mock_optional_class", node=node)
|
||||||
|
value_class.no_match_value = MockValue(command_class="mock_bad_class", node=node)
|
||||||
|
|
||||||
|
yield (node, value_class, mock_schema)
|
||||||
|
|
|
||||||
|
|
@ -22,7 +22,7 @@ from homeassistant.helpers.device_registry import async_get_registry as get_dev_
|
||||||
from homeassistant.helpers.entity_registry import async_get_registry
|
from homeassistant.helpers.entity_registry import async_get_registry
|
||||||
from homeassistant.setup import setup_component
|
from homeassistant.setup import setup_component
|
||||||
|
|
||||||
from tests.async_mock import AsyncMock, MagicMock, patch
|
from tests.async_mock import MagicMock, patch
|
||||||
from tests.common import async_fire_time_changed, get_test_home_assistant, mock_registry
|
from tests.common import async_fire_time_changed, get_test_home_assistant, mock_registry
|
||||||
from tests.mock.zwave import MockEntityValues, MockNetwork, MockNode, MockValue
|
from tests.mock.zwave import MockEntityValues, MockNetwork, MockNode, MockValue
|
||||||
|
|
||||||
|
|
@ -817,363 +817,497 @@ async def test_network_complete_some_dead(hass, mock_openzwave):
|
||||||
assert len(events) == 1
|
assert len(events) == 1
|
||||||
|
|
||||||
|
|
||||||
class TestZWaveDeviceEntityValues(unittest.TestCase):
|
async def test_entity_discovery(
|
||||||
"""Tests for the ZWaveDeviceEntityValues helper."""
|
hass, mock_discovery, mock_import_module, mock_values, mock_openzwave
|
||||||
|
):
|
||||||
|
"""Test the creation of a new entity."""
|
||||||
|
(node, value_class, mock_schema) = mock_values
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
registry = mock_registry(hass)
|
||||||
def set_mock_openzwave(self, mock_openzwave):
|
|
||||||
"""Use the mock_openzwave fixture for this class."""
|
|
||||||
self.mock_openzwave = mock_openzwave
|
|
||||||
|
|
||||||
def setUp(self):
|
mock_receivers = []
|
||||||
"""Initialize values for this testcase class."""
|
|
||||||
self.hass = get_test_home_assistant()
|
|
||||||
self.hass.start()
|
|
||||||
self.registry = mock_registry(self.hass)
|
|
||||||
|
|
||||||
setup_component(self.hass, "zwave", {"zwave": {}})
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
self.hass.block_till_done()
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
self.node = MockNode()
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
self.mock_schema = {
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
const.DISC_COMPONENT: "mock_component",
|
await hass.async_block_till_done()
|
||||||
const.DISC_VALUES: {
|
|
||||||
const.DISC_PRIMARY: {const.DISC_COMMAND_CLASS: ["mock_primary_class"]},
|
|
||||||
"secondary": {const.DISC_COMMAND_CLASS: ["mock_secondary_class"]},
|
|
||||||
"optional": {
|
|
||||||
const.DISC_COMMAND_CLASS: ["mock_optional_class"],
|
|
||||||
const.DISC_OPTIONAL: True,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
self.primary = MockValue(
|
|
||||||
command_class="mock_primary_class", node=self.node, value_id=1000
|
|
||||||
)
|
|
||||||
self.secondary = MockValue(command_class="mock_secondary_class", node=self.node)
|
|
||||||
self.duplicate_secondary = MockValue(
|
|
||||||
command_class="mock_secondary_class", node=self.node
|
|
||||||
)
|
|
||||||
self.optional = MockValue(command_class="mock_optional_class", node=self.node)
|
|
||||||
self.no_match_value = MockValue(command_class="mock_bad_class", node=self.node)
|
|
||||||
|
|
||||||
self.entity_id = "mock_component.mock_node_mock_value"
|
assert len(mock_receivers) == 1
|
||||||
self.zwave_config = {"zwave": {}}
|
|
||||||
self.device_config = {self.entity_id: {}}
|
|
||||||
self.addCleanup(self.tear_down_cleanup)
|
|
||||||
|
|
||||||
def tear_down_cleanup(self):
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
"""Stop everything that was started."""
|
zwave_config = {"zwave": {}}
|
||||||
self.hass.stop()
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
with patch.object(zwave, "discovery", mock_discovery):
|
||||||
@patch.object(zwave, "discovery")
|
|
||||||
def test_entity_discovery(self, discovery, import_module):
|
|
||||||
"""Test the creation of a new entity."""
|
|
||||||
discovery.async_load_platform = AsyncMock(return_value=None)
|
|
||||||
mock_platform = MagicMock()
|
|
||||||
import_module.return_value = mock_platform
|
|
||||||
mock_device = MagicMock()
|
|
||||||
mock_device.name = "test_device"
|
|
||||||
mock_platform.get_device.return_value = mock_device
|
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
assert values.primary is self.primary
|
assert values.primary is value_class.primary
|
||||||
assert len(list(values)) == 3
|
assert len(list(values)) == 3
|
||||||
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
||||||
[self.primary, None, None], key=lambda a: id(a)
|
[value_class.primary, None, None], key=lambda a: id(a)
|
||||||
)
|
)
|
||||||
assert not discovery.async_load_platform.called
|
|
||||||
|
|
||||||
values.check_value(self.secondary)
|
with patch.object(zwave, "discovery", mock_discovery), patch.object(
|
||||||
self.hass.block_till_done()
|
zwave, "import_module", mock_import_module
|
||||||
|
):
|
||||||
|
values.check_value(value_class.secondary)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert values.secondary is self.secondary
|
assert mock_discovery.async_load_platform.called
|
||||||
assert len(list(values)) == 3
|
assert len(mock_discovery.async_load_platform.mock_calls) == 1
|
||||||
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
|
||||||
[self.primary, self.secondary, None], key=lambda a: id(a)
|
|
||||||
)
|
|
||||||
|
|
||||||
assert discovery.async_load_platform.called
|
args = mock_discovery.async_load_platform.mock_calls[0][1]
|
||||||
assert len(discovery.async_load_platform.mock_calls) == 1
|
assert args[0] == hass
|
||||||
args = discovery.async_load_platform.mock_calls[0][1]
|
|
||||||
assert args[0] == self.hass
|
|
||||||
assert args[1] == "mock_component"
|
assert args[1] == "mock_component"
|
||||||
assert args[2] == "zwave"
|
assert args[2] == "zwave"
|
||||||
assert args[3] == {const.DISCOVERY_DEVICE: mock_device.unique_id}
|
assert args[3] == {
|
||||||
assert args[4] == self.zwave_config
|
const.DISCOVERY_DEVICE: mock_import_module().get_device().unique_id
|
||||||
|
|
||||||
discovery.async_load_platform.reset_mock()
|
|
||||||
values.check_value(self.optional)
|
|
||||||
values.check_value(self.duplicate_secondary)
|
|
||||||
values.check_value(self.no_match_value)
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
assert values.optional is self.optional
|
|
||||||
assert len(list(values)) == 3
|
|
||||||
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
|
||||||
[self.primary, self.secondary, self.optional], key=lambda a: id(a)
|
|
||||||
)
|
|
||||||
assert not discovery.async_load_platform.called
|
|
||||||
|
|
||||||
assert values._entity.value_added.called
|
|
||||||
assert len(values._entity.value_added.mock_calls) == 1
|
|
||||||
assert values._entity.value_changed.called
|
|
||||||
assert len(values._entity.value_changed.mock_calls) == 1
|
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
|
||||||
def test_entity_existing_values(self, discovery, import_module):
|
|
||||||
"""Test the loading of already discovered values."""
|
|
||||||
discovery.async_load_platform = AsyncMock(return_value=None)
|
|
||||||
mock_platform = MagicMock()
|
|
||||||
import_module.return_value = mock_platform
|
|
||||||
mock_device = MagicMock()
|
|
||||||
mock_device.name = "test_device"
|
|
||||||
mock_platform.get_device.return_value = mock_device
|
|
||||||
self.node.values = {
|
|
||||||
self.primary.value_id: self.primary,
|
|
||||||
self.secondary.value_id: self.secondary,
|
|
||||||
self.optional.value_id: self.optional,
|
|
||||||
self.no_match_value.value_id: self.no_match_value,
|
|
||||||
}
|
}
|
||||||
|
assert args[4] == zwave_config
|
||||||
|
|
||||||
|
assert values.secondary is value_class.secondary
|
||||||
|
assert len(list(values)) == 3
|
||||||
|
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
||||||
|
[value_class.primary, value_class.secondary, None], key=lambda a: id(a)
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_discovery.async_load_platform.reset_mock()
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery):
|
||||||
|
values.check_value(value_class.optional)
|
||||||
|
values.check_value(value_class.duplicate_secondary)
|
||||||
|
values.check_value(value_class.no_match_value)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
|
assert values.optional is value_class.optional
|
||||||
|
assert len(list(values)) == 3
|
||||||
|
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
||||||
|
[value_class.primary, value_class.secondary, value_class.optional],
|
||||||
|
key=lambda a: id(a),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert values._entity.value_added.called
|
||||||
|
assert len(values._entity.value_added.mock_calls) == 1
|
||||||
|
assert values._entity.value_changed.called
|
||||||
|
assert len(values._entity.value_changed.mock_calls) == 1
|
||||||
|
|
||||||
|
|
||||||
|
async def test_entity_existing_values(
|
||||||
|
hass, mock_discovery, mock_import_module, mock_values, mock_openzwave
|
||||||
|
):
|
||||||
|
"""Test the loading of already discovered values."""
|
||||||
|
(node, value_class, mock_schema) = mock_values
|
||||||
|
|
||||||
|
registry = mock_registry(hass)
|
||||||
|
|
||||||
|
mock_receivers = []
|
||||||
|
|
||||||
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.values = {
|
||||||
|
value_class.primary.value_id: value_class.primary,
|
||||||
|
value_class.secondary.value_id: value_class.secondary,
|
||||||
|
value_class.optional.value_id: value_class.optional,
|
||||||
|
value_class.no_match_value.value_id: value_class.no_match_value,
|
||||||
|
}
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery), patch.object(
|
||||||
|
zwave, "import_module", mock_import_module
|
||||||
|
):
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert values.primary is self.primary
|
assert mock_discovery.async_load_platform.called
|
||||||
assert values.secondary is self.secondary
|
assert len(mock_discovery.async_load_platform.mock_calls) == 1
|
||||||
assert values.optional is self.optional
|
args = mock_discovery.async_load_platform.mock_calls[0][1]
|
||||||
assert len(list(values)) == 3
|
assert args[0] == hass
|
||||||
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
|
||||||
[self.primary, self.secondary, self.optional], key=lambda a: id(a)
|
|
||||||
)
|
|
||||||
|
|
||||||
assert discovery.async_load_platform.called
|
|
||||||
assert len(discovery.async_load_platform.mock_calls) == 1
|
|
||||||
args = discovery.async_load_platform.mock_calls[0][1]
|
|
||||||
assert args[0] == self.hass
|
|
||||||
assert args[1] == "mock_component"
|
assert args[1] == "mock_component"
|
||||||
assert args[2] == "zwave"
|
assert args[2] == "zwave"
|
||||||
assert args[3] == {const.DISCOVERY_DEVICE: mock_device.unique_id}
|
assert args[3] == {
|
||||||
assert args[4] == self.zwave_config
|
const.DISCOVERY_DEVICE: mock_import_module().get_device().unique_id
|
||||||
assert not self.primary.enable_poll.called
|
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
|
||||||
def test_node_schema_mismatch(self, discovery, import_module):
|
|
||||||
"""Test node schema mismatch."""
|
|
||||||
self.node.generic = "no_match"
|
|
||||||
self.node.values = {
|
|
||||||
self.primary.value_id: self.primary,
|
|
||||||
self.secondary.value_id: self.secondary,
|
|
||||||
}
|
}
|
||||||
self.mock_schema[const.DISC_GENERIC_DEVICE_CLASS] = ["generic_match"]
|
assert args[4] == zwave_config
|
||||||
|
assert not value_class.primary.enable_poll.called
|
||||||
|
|
||||||
|
assert values.primary is value_class.primary
|
||||||
|
assert values.secondary is value_class.secondary
|
||||||
|
assert values.optional is value_class.optional
|
||||||
|
assert len(list(values)) == 3
|
||||||
|
assert sorted(list(values), key=lambda a: id(a)) == sorted(
|
||||||
|
[value_class.primary, value_class.secondary, value_class.optional],
|
||||||
|
key=lambda a: id(a),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_node_schema_mismatch(hass, mock_discovery, mock_values, mock_openzwave):
|
||||||
|
"""Test node schema mismatch."""
|
||||||
|
(node, value_class, mock_schema) = mock_values
|
||||||
|
|
||||||
|
registry = mock_registry(hass)
|
||||||
|
|
||||||
|
mock_receivers = []
|
||||||
|
|
||||||
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.generic = "no_match"
|
||||||
|
node.values = {
|
||||||
|
value_class.primary.value_id: value_class.primary,
|
||||||
|
value_class.secondary.value_id: value_class.secondary,
|
||||||
|
}
|
||||||
|
mock_schema[const.DISC_GENERIC_DEVICE_CLASS] = ["generic_match"]
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery):
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
values._check_entity_ready()
|
values._check_entity_ready()
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert not discovery.async_load_platform.called
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
|
||||||
def test_entity_workaround_component(self, discovery, import_module):
|
|
||||||
"""Test component workaround."""
|
|
||||||
discovery.async_load_platform = AsyncMock(return_value=None)
|
|
||||||
mock_platform = MagicMock()
|
|
||||||
import_module.return_value = mock_platform
|
|
||||||
mock_device = MagicMock()
|
|
||||||
mock_device.name = "test_device"
|
|
||||||
mock_platform.get_device.return_value = mock_device
|
|
||||||
self.node.manufacturer_id = "010f"
|
|
||||||
self.node.product_type = "0b00"
|
|
||||||
self.primary.command_class = const.COMMAND_CLASS_SENSOR_ALARM
|
|
||||||
self.entity_id = "binary_sensor.mock_node_mock_value"
|
|
||||||
self.device_config = {self.entity_id: {}}
|
|
||||||
|
|
||||||
self.mock_schema = {
|
async def test_entity_workaround_component(
|
||||||
const.DISC_COMPONENT: "mock_component",
|
hass, mock_discovery, mock_import_module, mock_values, mock_openzwave
|
||||||
const.DISC_VALUES: {
|
):
|
||||||
const.DISC_PRIMARY: {
|
"""Test component workaround."""
|
||||||
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_BINARY]
|
(node, value_class, mock_schema) = mock_values
|
||||||
}
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
with patch.object(zwave, "async_dispatcher_send") as mock_dispatch_send:
|
registry = mock_registry(hass)
|
||||||
|
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
mock_receivers = []
|
||||||
hass=self.hass,
|
|
||||||
schema=self.mock_schema,
|
|
||||||
primary_value=self.primary,
|
|
||||||
zwave_config=self.zwave_config,
|
|
||||||
device_config=self.device_config,
|
|
||||||
registry=self.registry,
|
|
||||||
)
|
|
||||||
values._check_entity_ready()
|
|
||||||
self.hass.block_till_done()
|
|
||||||
|
|
||||||
assert mock_dispatch_send.called
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
assert len(mock_dispatch_send.mock_calls) == 1
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
args = mock_dispatch_send.mock_calls[0][1]
|
mock_receivers.append(receiver)
|
||||||
assert args[1] == "zwave_new_binary_sensor"
|
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
@patch.object(zwave, "discovery")
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
def test_entity_workaround_ignore(self, discovery, import_module):
|
await hass.async_block_till_done()
|
||||||
"""Test ignore workaround."""
|
|
||||||
self.node.manufacturer_id = "010f"
|
|
||||||
self.node.product_type = "0301"
|
|
||||||
self.primary.command_class = const.COMMAND_CLASS_SWITCH_BINARY
|
|
||||||
|
|
||||||
self.mock_schema = {
|
node.manufacturer_id = "010f"
|
||||||
const.DISC_COMPONENT: "mock_component",
|
node.product_type = "0b00"
|
||||||
const.DISC_VALUES: {
|
value_class.primary.command_class = const.COMMAND_CLASS_SENSOR_ALARM
|
||||||
const.DISC_PRIMARY: {
|
|
||||||
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_BINARY]
|
entity_id = "binary_sensor.mock_node_mock_value"
|
||||||
}
|
zwave_config = {"zwave": {}}
|
||||||
},
|
device_config = {entity_id: {}}
|
||||||
}
|
|
||||||
|
mock_schema = {
|
||||||
|
const.DISC_COMPONENT: "mock_component",
|
||||||
|
const.DISC_VALUES: {
|
||||||
|
const.DISC_PRIMARY: {
|
||||||
|
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_BINARY]
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
with patch.object(
|
||||||
|
zwave, "async_dispatcher_send"
|
||||||
|
) as mock_dispatch_send, patch.object(
|
||||||
|
zwave, "discovery", mock_discovery
|
||||||
|
), patch.object(
|
||||||
|
zwave, "import_module", mock_import_module
|
||||||
|
):
|
||||||
|
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
values._check_entity_ready()
|
values._check_entity_ready()
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert not discovery.async_load_platform.called
|
assert mock_dispatch_send.called
|
||||||
|
assert len(mock_dispatch_send.mock_calls) == 1
|
||||||
|
args = mock_dispatch_send.mock_calls[0][1]
|
||||||
|
assert args[1] == "zwave_new_binary_sensor"
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
async def test_entity_workaround_ignore(
|
||||||
def test_entity_config_ignore(self, discovery, import_module):
|
hass, mock_discovery, mock_values, mock_openzwave
|
||||||
"""Test ignore config."""
|
):
|
||||||
self.node.values = {
|
"""Test ignore workaround."""
|
||||||
self.primary.value_id: self.primary,
|
(node, value_class, mock_schema) = mock_values
|
||||||
self.secondary.value_id: self.secondary,
|
|
||||||
}
|
registry = mock_registry(hass)
|
||||||
self.device_config = {self.entity_id: {zwave.CONF_IGNORED: True}}
|
|
||||||
|
mock_receivers = []
|
||||||
|
|
||||||
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.manufacturer_id = "010f"
|
||||||
|
node.product_type = "0301"
|
||||||
|
value_class.primary.command_class = const.COMMAND_CLASS_SWITCH_BINARY
|
||||||
|
|
||||||
|
mock_schema = {
|
||||||
|
const.DISC_COMPONENT: "mock_component",
|
||||||
|
const.DISC_VALUES: {
|
||||||
|
const.DISC_PRIMARY: {
|
||||||
|
const.DISC_COMMAND_CLASS: [const.COMMAND_CLASS_SWITCH_BINARY]
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery):
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
values._check_entity_ready()
|
values._check_entity_ready()
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert not discovery.async_load_platform.called
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
|
||||||
def test_entity_config_ignore_with_registry(self, discovery, import_module):
|
|
||||||
"""Test ignore config.
|
|
||||||
|
|
||||||
The case when the device is in entity registry.
|
async def test_entity_config_ignore(hass, mock_discovery, mock_values, mock_openzwave):
|
||||||
"""
|
"""Test ignore config."""
|
||||||
self.node.values = {
|
(node, value_class, mock_schema) = mock_values
|
||||||
self.primary.value_id: self.primary,
|
|
||||||
self.secondary.value_id: self.secondary,
|
|
||||||
}
|
|
||||||
self.device_config = {"mock_component.registry_id": {zwave.CONF_IGNORED: True}}
|
|
||||||
with patch.object(self.registry, "async_schedule_save"):
|
|
||||||
self.registry.async_get_or_create(
|
|
||||||
"mock_component",
|
|
||||||
zwave.DOMAIN,
|
|
||||||
"567-1000",
|
|
||||||
suggested_object_id="registry_id",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
registry = mock_registry(hass)
|
||||||
|
|
||||||
|
mock_receivers = []
|
||||||
|
|
||||||
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.values = {
|
||||||
|
value_class.primary.value_id: value_class.primary,
|
||||||
|
value_class.secondary.value_id: value_class.secondary,
|
||||||
|
}
|
||||||
|
device_config = {entity_id: {zwave.CONF_IGNORED: True}}
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery):
|
||||||
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
|
hass=hass,
|
||||||
|
schema=mock_schema,
|
||||||
|
primary_value=value_class.primary,
|
||||||
|
zwave_config=zwave_config,
|
||||||
|
device_config=device_config,
|
||||||
|
registry=registry,
|
||||||
|
)
|
||||||
|
values._check_entity_ready()
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
|
|
||||||
|
async def test_entity_config_ignore_with_registry(
|
||||||
|
hass, mock_discovery, mock_values, mock_openzwave
|
||||||
|
):
|
||||||
|
"""Test ignore config.
|
||||||
|
|
||||||
|
The case when the device is in entity registry.
|
||||||
|
"""
|
||||||
|
(node, value_class, mock_schema) = mock_values
|
||||||
|
|
||||||
|
registry = mock_registry(hass)
|
||||||
|
|
||||||
|
mock_receivers = []
|
||||||
|
|
||||||
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.values = {
|
||||||
|
value_class.primary.value_id: value_class.primary,
|
||||||
|
value_class.secondary.value_id: value_class.secondary,
|
||||||
|
}
|
||||||
|
device_config = {"mock_component.registry_id": {zwave.CONF_IGNORED: True}}
|
||||||
|
with patch.object(registry, "async_schedule_save"):
|
||||||
|
registry.async_get_or_create(
|
||||||
|
"mock_component",
|
||||||
|
zwave.DOMAIN,
|
||||||
|
"567-1000",
|
||||||
|
suggested_object_id="registry_id",
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery):
|
||||||
zwave.ZWaveDeviceEntityValues(
|
zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert not discovery.async_load_platform.called
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
async def test_entity_platform_ignore(
|
||||||
def test_entity_platform_ignore(self, discovery, import_module):
|
hass, mock_discovery, mock_values, mock_openzwave
|
||||||
"""Test platform ignore device."""
|
):
|
||||||
self.node.values = {
|
"""Test platform ignore device."""
|
||||||
self.primary.value_id: self.primary,
|
(node, value_class, mock_schema) = mock_values
|
||||||
self.secondary.value_id: self.secondary,
|
|
||||||
}
|
registry = mock_registry(hass)
|
||||||
platform = MagicMock()
|
|
||||||
import_module.return_value = platform
|
mock_receivers = []
|
||||||
platform.get_device.return_value = None
|
|
||||||
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.values = {
|
||||||
|
value_class.primary.value_id: value_class.primary,
|
||||||
|
value_class.secondary.value_id: value_class.secondary,
|
||||||
|
}
|
||||||
|
|
||||||
|
import_module = MagicMock()
|
||||||
|
platform = MagicMock()
|
||||||
|
import_module.return_value = platform
|
||||||
|
platform.get_device.return_value = None
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery), patch.object(
|
||||||
|
zwave, "import_module", import_module
|
||||||
|
):
|
||||||
zwave.ZWaveDeviceEntityValues(
|
zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert not discovery.async_load_platform.called
|
assert not mock_discovery.async_load_platform.called
|
||||||
|
|
||||||
@patch.object(zwave, "import_module")
|
|
||||||
@patch.object(zwave, "discovery")
|
async def test_config_polling_intensity(
|
||||||
def test_config_polling_intensity(self, discovery, import_module):
|
hass, mock_discovery, mock_import_module, mock_values, mock_openzwave
|
||||||
"""Test polling intensity."""
|
):
|
||||||
mock_platform = MagicMock()
|
"""Test polling intensity."""
|
||||||
import_module.return_value = mock_platform
|
(node, value_class, mock_schema) = mock_values
|
||||||
mock_device = MagicMock()
|
|
||||||
mock_device.name = "test_device"
|
registry = mock_registry(hass)
|
||||||
mock_platform.get_device.return_value = mock_device
|
|
||||||
self.node.values = {
|
mock_receivers = []
|
||||||
self.primary.value_id: self.primary,
|
|
||||||
self.secondary.value_id: self.secondary,
|
def mock_connect(receiver, signal, *args, **kwargs):
|
||||||
}
|
if signal == MockNetwork.SIGNAL_VALUE_ADDED:
|
||||||
self.device_config = {self.entity_id: {zwave.CONF_POLLING_INTENSITY: 123}}
|
mock_receivers.append(receiver)
|
||||||
|
|
||||||
|
with patch("pydispatch.dispatcher.connect", new=mock_connect):
|
||||||
|
await async_setup_component(hass, "zwave", {"zwave": {}})
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity_id = "mock_component.mock_node_mock_value"
|
||||||
|
zwave_config = {"zwave": {}}
|
||||||
|
device_config = {entity_id: {}}
|
||||||
|
|
||||||
|
node.values = {
|
||||||
|
value_class.primary.value_id: value_class.primary,
|
||||||
|
value_class.secondary.value_id: value_class.secondary,
|
||||||
|
}
|
||||||
|
device_config = {entity_id: {zwave.CONF_POLLING_INTENSITY: 123}}
|
||||||
|
|
||||||
|
with patch.object(zwave, "discovery", mock_discovery), patch.object(
|
||||||
|
zwave, "import_module", mock_import_module
|
||||||
|
):
|
||||||
values = zwave.ZWaveDeviceEntityValues(
|
values = zwave.ZWaveDeviceEntityValues(
|
||||||
hass=self.hass,
|
hass=hass,
|
||||||
schema=self.mock_schema,
|
schema=mock_schema,
|
||||||
primary_value=self.primary,
|
primary_value=value_class.primary,
|
||||||
zwave_config=self.zwave_config,
|
zwave_config=zwave_config,
|
||||||
device_config=self.device_config,
|
device_config=device_config,
|
||||||
registry=self.registry,
|
registry=registry,
|
||||||
)
|
)
|
||||||
values._check_entity_ready()
|
values._check_entity_ready()
|
||||||
self.hass.block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
assert discovery.async_load_platform.called
|
assert mock_discovery.async_load_platform.called
|
||||||
assert self.primary.enable_poll.called
|
|
||||||
assert len(self.primary.enable_poll.mock_calls) == 1
|
assert value_class.primary.enable_poll.called
|
||||||
assert self.primary.enable_poll.mock_calls[0][1][0] == 123
|
assert len(value_class.primary.enable_poll.mock_calls) == 1
|
||||||
|
assert value_class.primary.enable_poll.mock_calls[0][1][0] == 123
|
||||||
|
|
||||||
|
|
||||||
class TestZwave(unittest.TestCase):
|
class TestZwave(unittest.TestCase):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue