evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Secure 100% test coverage for modbus, binary_sensor and sensor (#49521)

Browse source 
* Secure 100% test coverage for modbus/binary_sensor.

* Test that class constructor is called.
This commit is contained in:
jan iversen 2021-04-22 11:54:40 +02:00 • committed by GitHub
parent 8b08134850
commit f67c0ce8bb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 290 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

155
tests/components/modbus/test_init.py
View file

@ -2,16 +2,24 @@
import logging
from unittest import mock
from pymodbus.exceptions import ModbusException
import pytest
import voluptuous as vol
from homeassistant.components.modbus import number
from homeassistant.components.modbus.const import (
ATTR_ADDRESS,
ATTR_HUB,
ATTR_STATE,
ATTR_UNIT,
ATTR_VALUE,
CONF_BAUDRATE,
CONF_BYTESIZE,
CONF_PARITY,
CONF_STOPBITS,
MODBUS_DOMAIN as DOMAIN,
SERVICE_WRITE_COIL,
SERVICE_WRITE_REGISTER,
)
from homeassistant.const import (
CONF_DELAY,
@ -177,3 +185,150 @@ async def test_config_multiple_modbus(hass, caplog):
await _config_helper(hass, do_config)
assert DOMAIN in hass.config.components
assert len(caplog.records) == 0
async def test_pb_service_write_register(hass):
"""Run test for service write_register."""
conf_name = "myModbus"
config = {
DOMAIN: [
{
CONF_TYPE: "tcp",
CONF_HOST: "modbusTestHost",
CONF_PORT: 5501,
CONF_NAME: conf_name,
}
]
}
mock_pb = mock.MagicMock()
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient", return_value=mock_pb
):
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
data = {ATTR_HUB: conf_name, ATTR_UNIT: 17, ATTR_ADDRESS: 16, ATTR_VALUE: 15}
await hass.services.async_call(
DOMAIN, SERVICE_WRITE_REGISTER, data, blocking=True
)
assert mock_pb.write_register.called
assert mock_pb.write_register.call_args[0] == (
data[ATTR_ADDRESS],
data[ATTR_VALUE],
)
mock_pb.write_register.side_effect = ModbusException("fail write_")
await hass.services.async_call(
DOMAIN, SERVICE_WRITE_REGISTER, data, blocking=True
)
data[ATTR_VALUE] = [1, 2, 3]
await hass.services.async_call(
DOMAIN, SERVICE_WRITE_REGISTER, data, blocking=True
)
assert mock_pb.write_registers.called
assert mock_pb.write_registers.call_args[0] == (
data[ATTR_ADDRESS],
data[ATTR_VALUE],
)
mock_pb.write_registers.side_effect = ModbusException("fail write_")
await hass.services.async_call(
DOMAIN, SERVICE_WRITE_REGISTER, data, blocking=True
)
async def test_pb_service_write_coil(hass, caplog):
"""Run test for service write_coil."""
conf_name = "myModbus"
config = {
DOMAIN: [
{
CONF_TYPE: "tcp",
CONF_HOST: "modbusTestHost",
CONF_PORT: 5501,
CONF_NAME: conf_name,
}
]
}
mock_pb = mock.MagicMock()
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient", return_value=mock_pb
):
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
data = {ATTR_HUB: conf_name, ATTR_UNIT: 17, ATTR_ADDRESS: 16, ATTR_STATE: False}
await hass.services.async_call(DOMAIN, SERVICE_WRITE_COIL, data, blocking=True)
assert mock_pb.write_coil.called
assert mock_pb.write_coil.call_args[0] == (
data[ATTR_ADDRESS],
data[ATTR_STATE],
)
mock_pb.write_coil.side_effect = ModbusException("fail write_")
await hass.services.async_call(DOMAIN, SERVICE_WRITE_COIL, data, blocking=True)
data[ATTR_STATE] = [True, False, True]
await hass.services.async_call(DOMAIN, SERVICE_WRITE_COIL, data, blocking=True)
assert mock_pb.write_coils.called
assert mock_pb.write_coils.call_args[0] == (
data[ATTR_ADDRESS],
data[ATTR_STATE],
)
caplog.set_level(logging.DEBUG)
caplog.clear
mock_pb.write_coils.side_effect = ModbusException("fail write_")
await hass.services.async_call(DOMAIN, SERVICE_WRITE_COIL, data, blocking=True)
assert caplog.records[-1].levelname == "ERROR"
await hass.services.async_call(DOMAIN, SERVICE_WRITE_COIL, data, blocking=True)
assert caplog.records[-1].levelname == "DEBUG"
async def test_pymodbus_constructor_fail(hass, caplog):
"""Run test for failing pymodbus constructor."""
config = {
DOMAIN: [
{
CONF_TYPE: "tcp",
CONF_HOST: "modbusTestHost",
CONF_PORT: 5501,
}
]
}
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient"
) as mock_pb:
caplog.set_level(logging.ERROR)
mock_pb.side_effect = ModbusException("test no class")
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
assert len(caplog.records) == 1
assert caplog.records[0].levelname == "ERROR"
assert mock_pb.called
async def test_pymodbus_connect_fail(hass, caplog):
"""Run test for failing pymodbus constructor."""
config = {
DOMAIN: [
{
CONF_TYPE: "tcp",
CONF_HOST: "modbusTestHost",
CONF_PORT: 5501,
}
]
}
mock_pb = mock.MagicMock()
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient", return_value=mock_pb
):
caplog.set_level(logging.ERROR)
mock_pb.connect.side_effect = ModbusException("test connect fail")
mock_pb.close.side_effect = ModbusException("test connect fail")
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
assert len(caplog.records) == 1
assert caplog.records[0].levelname == "ERROR"