starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-08 23:36:09 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
21dec15cd3e7aed1d2affaed816c822fa367ff24
core/tests/components/modbus/test_init.py
T

1869 lines
55 KiB
Python
Raw Blame History

"""The tests for the Modbus init.
This file is responsible for testing:
- pymodbus API
- Functionality of class ModbusHub
- Coverage 100%:
__init__.py
const.py
modbus.py
validators.py
entity.py (only ModbusBaseEntity)
It uses binary_sensors/sensors to do black box testing of the read calls.
"""
import asyncio
from collections.abc import Awaitable, Callable
from datetime import timedelta
import logging
from unittest import mock
from freezegun.api import FrozenDateTimeFactory
import probatio
from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse
import pytest
from homeassistant import config as hass_config
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.modbus.const import (
ATTR_ADDRESS,
ATTR_HUB,
ATTR_SLAVE,
ATTR_UNIT,
ATTR_VALUE,
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
CALL_TYPE_REGISTER_HOLDING,
CALL_TYPE_REGISTER_INPUT,
CALL_TYPE_WRITE_COIL,
CALL_TYPE_WRITE_COILS,
CALL_TYPE_WRITE_REGISTER,
CALL_TYPE_WRITE_REGISTERS,
CONF_BAUDRATE,
CONF_BYTESIZE,
CONF_CLIMATES,
CONF_CURRENT_TEMP_OFFSET,
CONF_CURRENT_TEMP_SCALE,
CONF_DATA_TYPE,
CONF_DEVICE_ADDRESS,
CONF_FAN_MODE_HIGH,
CONF_FAN_MODE_OFF,
CONF_FAN_MODE_ON,
CONF_FAN_MODE_VALUES,
CONF_INPUT_TYPE,
CONF_MSG_WAIT,
CONF_PARITY,
CONF_SCALE,
CONF_SLAVE_COUNT,
CONF_STOPBITS,
CONF_SWAP,
CONF_SWAP_BYTE,
CONF_SWAP_WORD,
CONF_SWAP_WORD_BYTE,
CONF_SWING_MODE_SWING_BOTH,
CONF_SWING_MODE_SWING_OFF,
CONF_SWING_MODE_SWING_ON,
CONF_SWING_MODE_VALUES,
CONF_TARGET_TEMP,
CONF_TARGET_TEMP_OFFSET,
CONF_TARGET_TEMP_SCALE,
CONF_VIRTUAL_COUNT,
DEFAULT_SCAN_INTERVAL,
DEVICE_ID,
DOMAIN,
RTUOVERTCP,
SERIAL,
SERVICE_STOP,
SERVICE_WRITE_COIL,
SERVICE_WRITE_REGISTER,
TCP,
UDP,
DataType,
)
from homeassistant.components.modbus.validators import (
check_config,
duplicate_fan_mode_validator,
duplicate_swing_mode_validator,
ensure_and_check_conflicting_scales_and_offsets,
hvac_fixedsize_reglist_validator,
nan_validator,
not_zero_value,
register_int_list_validator,
struct_validator,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_STATE,
CONF_ADDRESS,
CONF_BINARY_SENSORS,
CONF_COUNT,
CONF_DELAY,
CONF_HOST,
CONF_METHOD,
CONF_NAME,
CONF_OFFSET,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_SENSORS,
CONF_SLAVE,
CONF_STRUCTURE,
CONF_TIMEOUT,
CONF_TYPE,
CONF_UNIQUE_ID,
EVENT_HOMEASSISTANT_STOP,
SERVICE_RELOAD,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .conftest import (
TEST_ENTITY_NAME,
TEST_MODBUS_HOST,
TEST_MODBUS_NAME,
TEST_PORT_SERIAL,
TEST_PORT_TCP,
ReadResult,
)
from tests.common import async_fire_time_changed, get_fixture_path
@pytest.fixture(name="mock_modbus_with_pymodbus")
async def mock_modbus_with_pymodbus_fixture(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, do_config, mock_pymodbus
):
"""Load integration modbus using mocked pymodbus."""
caplog.clear()
caplog.set_level(logging.ERROR)
config = {DOMAIN: do_config}
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
assert DOMAIN in hass.config.components
assert caplog.text == ""
return mock_pymodbus
async def test_fixedRegList_validator() -> None:
"""Test fixed temp registers validator."""
for value in (
15,
[30, 31, 32, 33, 34, 35, 36],
):
assert isinstance(hvac_fixedsize_reglist_validator(value), list)
with pytest.raises(probatio.Invalid):
hvac_fixedsize_reglist_validator([15, "ab", 17, 18, 19, 20, 21])
with pytest.raises(probatio.Invalid):
hvac_fixedsize_reglist_validator([15, 17])
async def test_register_int_list_validator() -> None:
"""Test conf address register validator."""
for value, vtype in (
(15, int),
([15], list),
):
assert isinstance(register_int_list_validator(value), vtype)
with pytest.raises(probatio.Invalid):
register_int_list_validator([15, 16])
with pytest.raises(probatio.Invalid):
register_int_list_validator(-15)
with pytest.raises(probatio.Invalid):
register_int_list_validator(["aq"])
async def test_nan_validator() -> None:
"""Test number validator."""
for value, value_type in (
(15, int),
("15", int),
("abcdef", int),
("0xabcdef", int),
):
assert isinstance(nan_validator(value), value_type)
with pytest.raises(probatio.Invalid):
nan_validator("x15")
with pytest.raises(probatio.Invalid):
nan_validator("not a hex string")
@pytest.mark.parametrize(
"do_config",
[
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 2,
CONF_DATA_TYPE: DataType.STRING,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_DATA_TYPE: DataType.INT32,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_DATA_TYPE: DataType.INT32,
CONF_SWAP: CONF_SWAP_BYTE,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 2,
CONF_DATA_TYPE: DataType.CUSTOM,
CONF_STRUCTURE: ">i",
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_SLAVE: 5,
CONF_DATA_TYPE: DataType.INT32,
CONF_SWAP: CONF_SWAP_BYTE,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_SLAVE: 5,
CONF_DATA_TYPE: DataType.INT32,
CONF_SWAP: CONF_SWAP_WORD,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_SLAVE: 5,
CONF_DATA_TYPE: DataType.INT32,
CONF_SWAP: CONF_SWAP_WORD_BYTE,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_SLAVE: 5,
CONF_DATA_TYPE: DataType.INT16,
CONF_SWAP: CONF_SWAP_BYTE,
},
],
)
async def test_ok_struct_validator(do_config) -> None:
"""Test struct validator."""
try:
struct_validator(do_config)
except probatio.Invalid:
pytest.fail("struct_validator unexpected exception")
@pytest.mark.parametrize(
"do_config",
[
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 8,
CONF_DATA_TYPE: "int",
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 8,
CONF_DATA_TYPE: DataType.CUSTOM,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 8,
CONF_DATA_TYPE: DataType.CUSTOM,
CONF_STRUCTURE: "no good",
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 20,
CONF_DATA_TYPE: DataType.CUSTOM,
CONF_STRUCTURE: ">f",
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 1,
CONF_DATA_TYPE: DataType.CUSTOM,
CONF_STRUCTURE: ">f",
CONF_SWAP: CONF_SWAP_WORD,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 1,
CONF_DATA_TYPE: DataType.STRING,
CONF_STRUCTURE: ">f",
CONF_SWAP: CONF_SWAP_WORD,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 2,
CONF_DATA_TYPE: DataType.CUSTOM,
CONF_STRUCTURE: ">f",
CONF_SLAVE_COUNT: 5,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 2,
CONF_DATA_TYPE: DataType.CUSTOM,
CONF_STRUCTURE: ">f",
CONF_VIRTUAL_COUNT: 5,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_DATA_TYPE: DataType.STRING,
CONF_SLAVE_COUNT: 2,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_DATA_TYPE: DataType.STRING,
CONF_VIRTUAL_COUNT: 2,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_DATA_TYPE: DataType.INT16,
CONF_SWAP: CONF_SWAP_WORD,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 2,
CONF_SLAVE_COUNT: 2,
CONF_DATA_TYPE: DataType.INT32,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_COUNT: 2,
CONF_VIRTUAL_COUNT: 2,
CONF_DATA_TYPE: DataType.INT32,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_DATA_TYPE: DataType.INT16,
CONF_SWAP: CONF_SWAP_WORD_BYTE,
},
],
)
async def test_exception_struct_validator(do_config) -> None:
"""Test struct validator."""
try:
struct_validator(do_config)
except probatio.Invalid:
return
pytest.fail("struct_validator missing exception")
@pytest.mark.parametrize(
"do_config",
[
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST + " 2",
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_NAME: TEST_MODBUS_NAME + "2",
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
[
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_NAME: TEST_MODBUS_NAME + " 2",
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
],
)
async def test_check_config(hass: HomeAssistant, do_config) -> None:
"""Test duplicate modbus validator."""
check_config(hass, do_config)
assert len(do_config) == 1
@pytest.mark.parametrize(
"do_config",
[
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
CONF_SLAVE: 0,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 119,
CONF_SLAVE: 0,
},
],
}
],
],
)
async def test_check_config_sensor(hass: HomeAssistant, do_config) -> None:
"""Test duplicate entity validator."""
check_config(hass, do_config)
assert len(do_config[0][CONF_SENSORS]) == 1
@pytest.mark.parametrize(
"do_config",
[
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_CLIMATES: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
CONF_SLAVE: 0,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 119,
CONF_SLAVE: 0,
},
],
}
],
],
)
async def test_check_config_climate(hass: HomeAssistant, do_config) -> None:
"""Test duplicate entity validator."""
check_config(hass, do_config)
assert len(do_config[0][CONF_CLIMATES]) == 1
@pytest.mark.parametrize(
"do_config",
[
{
CONF_ADDRESS: 11,
CONF_FAN_MODE_VALUES: {
CONF_FAN_MODE_ON: 7,
CONF_FAN_MODE_OFF: 9,
CONF_FAN_MODE_HIGH: 9,
},
}
],
)
async def test_duplicate_fan_mode_validator(do_config) -> None:
"""Test duplicate modbus validator."""
duplicate_fan_mode_validator(do_config)
assert len(do_config[CONF_FAN_MODE_VALUES]) == 2
@pytest.mark.parametrize(
"do_config",
[
{
CONF_ADDRESS: 11,
CONF_SWING_MODE_VALUES: {
CONF_SWING_MODE_SWING_ON: 7,
CONF_SWING_MODE_SWING_OFF: 9,
CONF_SWING_MODE_SWING_BOTH: 9,
},
}
],
)
async def test_duplicate_swing_mode_validator(do_config) -> None:
"""Test duplicate modbus validator."""
duplicate_swing_mode_validator(do_config)
assert len(do_config[CONF_SWING_MODE_VALUES]) == 2
@pytest.mark.parametrize(
"do_config",
[
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_TIMEOUT: 3,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
CONF_SLAVE: 0,
},
],
CONF_BINARY_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME + "1",
CONF_ADDRESS: 1179,
CONF_SLAVE: 0,
},
],
},
],
],
)
async def test_no_duplicate_names(hass: HomeAssistant, do_config) -> None:
"""Test duplicate entity validator."""
check_config(hass, do_config)
assert len(do_config[0][CONF_SENSORS]) == 1
assert len(do_config[0][CONF_BINARY_SENSORS]) == 1
@pytest.mark.parametrize(
"do_config",
[
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_TIMEOUT: 30,
CONF_DELAY: 10,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: UDP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: UDP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_TIMEOUT: 30,
CONF_DELAY: 10,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: RTUOVERTCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: RTUOVERTCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_TIMEOUT: 30,
CONF_DELAY: 10,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_MSG_WAIT: 100,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "ascii",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_NAME: TEST_MODBUS_NAME,
CONF_TIMEOUT: 30,
CONF_DELAY: 10,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_DELAY: 5,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
[
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST + "_1",
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: f"{TEST_MODBUS_NAME} 2",
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_NAME: f"{TEST_MODBUS_NAME} 3",
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
[
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP + 10,
CONF_NAME: f"{TEST_MODBUS_NAME} 2",
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
{
# Special test for scan_interval validator with scan_interval: 0
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
CONF_SLAVE: 0,
CONF_SCAN_INTERVAL: 0,
}
],
},
{
# Special test for scan_interval validator with scan_interval: 0
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
CONF_DEVICE_ADDRESS: 0,
CONF_SCAN_INTERVAL: 0,
}
],
},
],
)
async def test_config_modbus(hass: HomeAssistant, mock_modbus_with_pymodbus) -> None:
"""Run configuration test for modbus."""
assert len(hass.data[DOMAIN])
@pytest.mark.parametrize(
"do_config",
[
[
# Duplicate CONF_NAME
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
[
# Duplicate CONF_HOST+CONF_PORT (for type != SERIAL)
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME + "_1",
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
[
# Duplicate CONF_PORT (for type == SERIAL)
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
{
CONF_NAME: TEST_MODBUS_NAME + "_1",
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
],
)
async def test_config_wrong_modbus(
hass: HomeAssistant, mock_modbus_with_pymodbus, issue_registry: ir.IssueRegistry
) -> None:
"""Run configuration test for modbus."""
assert len(hass.data[DOMAIN]) == 1
assert len(issue_registry.issues) == 1
assert (DOMAIN, "duplicate_modbus_entry") in issue_registry.issues
VALUE = "value"
FUNC = "func"
DATA = "data"
SERVICE = "service"
@pytest.mark.parametrize(
"do_config",
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
},
],
)
@pytest.mark.parametrize(
"do_write",
[
{
DATA: ATTR_VALUE,
VALUE: 15,
SERVICE: SERVICE_WRITE_REGISTER,
FUNC: CALL_TYPE_WRITE_REGISTER,
},
{
DATA: ATTR_VALUE,
VALUE: [1, 2, 3],
SERVICE: SERVICE_WRITE_REGISTER,
FUNC: CALL_TYPE_WRITE_REGISTERS,
},
{
DATA: ATTR_STATE,
VALUE: False,
SERVICE: SERVICE_WRITE_COIL,
FUNC: CALL_TYPE_WRITE_COIL,
},
{
DATA: ATTR_STATE,
VALUE: [True, False, True],
SERVICE: SERVICE_WRITE_COIL,
FUNC: CALL_TYPE_WRITE_COILS,
},
],
)
@pytest.mark.parametrize(
"do_return",
[
{VALUE: ReadResult([0x0001]), DATA: ""},
{VALUE: ExceptionResponse(0x06), DATA: "Pymodbus:"},
{VALUE: ModbusException("fail write_"), DATA: "Pymodbus:"},
],
)
@pytest.mark.parametrize(
"do_slave",
[
ATTR_SLAVE,
ATTR_UNIT,
],
)
async def test_pb_service_write(
hass: HomeAssistant,
do_write,
do_return,
do_slave,
caplog: pytest.LogCaptureFixture,
mock_modbus_with_pymodbus,
) -> None:
"""Run test for service write_register."""
func_name = {
CALL_TYPE_WRITE_COIL: mock_modbus_with_pymodbus.write_coil,
CALL_TYPE_WRITE_COILS: mock_modbus_with_pymodbus.write_coils,
CALL_TYPE_WRITE_REGISTER: mock_modbus_with_pymodbus.write_register,
CALL_TYPE_WRITE_REGISTERS: mock_modbus_with_pymodbus.write_registers,
}
value_arg_name = {
CALL_TYPE_WRITE_COIL: "value",
CALL_TYPE_WRITE_COILS: "values",
CALL_TYPE_WRITE_REGISTER: "value",
CALL_TYPE_WRITE_REGISTERS: "values",
}
data = {
ATTR_HUB: TEST_MODBUS_NAME,
do_slave: 17,
ATTR_ADDRESS: 16,
do_write[DATA]: do_write[VALUE],
}
mock_modbus_with_pymodbus.reset_mock()
caplog.clear()
caplog.set_level(logging.DEBUG)
func_name[do_write[FUNC]].return_value = do_return[VALUE]
await hass.services.async_call(DOMAIN, do_write[SERVICE], data, blocking=True)
assert func_name[do_write[FUNC]].called
assert func_name[do_write[FUNC]].call_args.args == (data[ATTR_ADDRESS],)
assert func_name[do_write[FUNC]].call_args.kwargs == {
DEVICE_ID: 17,
value_arg_name[do_write[FUNC]]: data[do_write[DATA]],
}
if do_return[DATA]:
assert any(message.startswith("Pymodbus:") for message in caplog.messages)
@pytest.fixture(name="mock_modbus_read_pymodbus")
async def mock_modbus_read_pymodbus_fixture(
hass: HomeAssistant,
do_group,
do_type,
do_scan_interval,
do_return,
caplog: pytest.LogCaptureFixture,
mock_pymodbus,
freezer: FrozenDateTimeFactory,
):
"""Load integration modbus using mocked pymodbus."""
caplog.clear()
caplog.set_level(logging.ERROR)
mock_pymodbus.read_coils.return_value = do_return
mock_pymodbus.read_discrete_inputs.return_value = do_return
mock_pymodbus.read_input_registers.return_value = do_return
mock_pymodbus.read_holding_registers.return_value = do_return
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
do_group: [
{
CONF_INPUT_TYPE: do_type,
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
CONF_SLAVE: 0,
CONF_SCAN_INTERVAL: do_scan_interval,
}
],
}
],
}
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
assert DOMAIN in hass.config.components
assert caplog.text == ""
freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL + 60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL + 60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
return mock_pymodbus
@pytest.mark.parametrize(
("do_domain", "do_group", "do_type", "do_scan_interval"),
[
(SENSOR_DOMAIN, CONF_SENSORS, CALL_TYPE_REGISTER_HOLDING, 10),
(SENSOR_DOMAIN, CONF_SENSORS, CALL_TYPE_REGISTER_INPUT, 10),
(BINARY_SENSOR_DOMAIN, CONF_BINARY_SENSORS, CALL_TYPE_DISCRETE, 10),
(BINARY_SENSOR_DOMAIN, CONF_BINARY_SENSORS, CALL_TYPE_COIL, 1),
],
)
@pytest.mark.parametrize(
("do_return", "do_exception", "do_expect_state", "do_expect_value"),
[
(ReadResult([1]), None, STATE_ON, "1"),
(ExceptionResponse(0x99), None, STATE_UNAVAILABLE, STATE_UNAVAILABLE),
(
ReadResult([1]),
ModbusException("fail read_"),
STATE_UNAVAILABLE,
STATE_UNAVAILABLE,
),
],
)
async def test_pb_read(
hass: HomeAssistant,
do_domain,
do_expect_state,
do_expect_value,
caplog: pytest.LogCaptureFixture,
mock_modbus_read_pymodbus,
) -> None:
"""Run test for different read."""
# Check state
entity_id = f"{do_domain}.{TEST_ENTITY_NAME}".replace(" ", "_")
state = hass.states.get(entity_id).state
assert hass.states.get(entity_id).state
# this if is needed to avoid explode the
if do_domain == SENSOR_DOMAIN:
do_expect = do_expect_value
else:
do_expect = do_expect_state
assert state == do_expect
async def test_pymodbus_constructor_fail(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Run test for failing pymodbus constructor."""
config = {
DOMAIN: [
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
},
],
}
]
}
with mock.patch(
"homeassistant.components.modbus.modbus.AsyncModbusTcpClient", autospec=True
) 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 False
await hass.async_block_till_done()
message = f"Pymodbus: {TEST_MODBUS_NAME}: Modbus Error: test"
assert caplog.messages[0].startswith(message)
assert caplog.records[0].levelname == "ERROR"
assert mock_pb.called
async def test_pymodbus_close_fail(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, mock_pymodbus
) -> None:
"""Run test for failing pymodbus close."""
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
}
]
}
caplog.set_level(logging.ERROR)
mock_pymodbus.connect.return_value = True
mock_pymodbus.close.side_effect = ModbusException("close fail")
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
# Close() is called as part of teardown
async def test_unreachable_device_does_not_hold_startup(
hass: HomeAssistant, mock_pymodbus: mock.AsyncMock
) -> None:
"""Test entities waiting for a device that never connects do not hold up startup."""
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
}
],
}
]
}
mock_pymodbus.connect.return_value = False
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
# the entities start waiting for the first connection
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
# startup wraps up by waiting for foreground tasks, that wait must not be one
async with asyncio.timeout(1):
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNKNOWN
async def test_aborted_add_cancels_first_update(
hass: HomeAssistant, mock_pymodbus: mock.AsyncMock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test an entity whose add is aborted does not run its first update later."""
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
}
],
}
]
}
# the platform aborts the add when restoring the last state fails, which
# happens after the first update has been scheduled
with mock.patch(
"homeassistant.components.modbus.sensor.ModbusRegisterSensor.async_get_last_sensor_data",
side_effect=ValueError("restore failed"),
):
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
assert hass.states.get(entity_id) is None
caplog.clear()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
await hass.async_block_till_done(wait_background_tasks=True)
assert not [record for record in caplog.records if record.levelno >= logging.ERROR]
assert hass.states.get(entity_id) is None
async def test_renamed_entity_keeps_polling(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_pymodbus: mock.AsyncMock,
) -> None:
"""Test an entity polls again after a rename, which removes and re-adds it."""
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
CONF_UNIQUE_ID: "renamed_sensor",
}
],
}
]
}
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(entity_id).state == "0"
new_entity_id = f"{SENSOR_DOMAIN}.renamed"
entity_registry.async_update_entity(entity_id, new_entity_id=new_entity_id)
await hass.async_block_till_done()
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x2A])
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=2))
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(new_entity_id).state == "42"
async def test_overlapping_updates_keep_one_poll_timer(
hass: HomeAssistant, mock_pymodbus: mock.AsyncMock
) -> None:
"""Test an update overlapping a poll does not start a second polling loop."""
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
CONF_SCAN_INTERVAL: 10,
}
],
}
]
}
assert await async_setup_component(hass, "homeassistant", {})
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
start = dt_util.utcnow()
async_fire_time_changed(hass, start + timedelta(seconds=1))
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(entity_id).state == "0"
# hold the next poll on the device, so an update requested now overlaps it
release_poll = asyncio.Event()
result = mock_pymodbus.read_holding_registers.return_value
async def _held_read(*args: object, **kwargs: object) -> ReadResult:
await release_poll.wait()
return result
mock_pymodbus.read_holding_registers.side_effect = _held_read
async_fire_time_changed(hass, start + timedelta(seconds=12))
await asyncio.sleep(0)
await hass.services.async_call(
"homeassistant", "update_entity", {ATTR_ENTITY_ID: entity_id}
)
await asyncio.sleep(0)
release_poll.set()
await hass.async_block_till_done(wait_background_tasks=True)
mock_pymodbus.read_holding_registers.reset_mock()
async_fire_time_changed(hass, start + timedelta(seconds=25))
await hass.async_block_till_done(wait_background_tasks=True)
assert mock_pymodbus.read_holding_registers.call_count == 1
async def _fire_first_connect_timer(hass: HomeAssistant) -> None:
"""Let the entities start waiting for the first connection."""
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
await hass.async_block_till_done()
async def _stop_hub(hass: HomeAssistant) -> None:
"""Stop the hub, which also releases every wait for its connection."""
await hass.services.async_call(
DOMAIN, SERVICE_STOP, {ATTR_HUB: TEST_MODBUS_NAME}, blocking=True
)
await hass.async_block_till_done()
@pytest.mark.parametrize(
"steps",
[
pytest.param(
[_stop_hub, _fire_first_connect_timer], id="stopped_before_the_wait"
),
pytest.param(
[_fire_first_connect_timer, _stop_hub], id="stopped_while_waiting"
),
],
)
async def test_stop_cancels_pending_first_update(
hass: HomeAssistant,
mock_pymodbus: mock.AsyncMock,
steps: list[Callable[[HomeAssistant], Awaitable[None]]],
) -> None:
"""Test stopping the hub cancels an update still waiting for the first connection."""
entity_id = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
}
],
}
]
}
mock_pymodbus.connect.return_value = False
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
for step in steps:
await step(hass)
# an update that survived the stop would run against the closed hub and
# write the entity unavailable
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(entity_id).state == STATE_UNKNOWN
async def test_pymodbus_connect_fail(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, mock_pymodbus
) -> None:
"""Run test for failing pymodbus constructor."""
config = {
DOMAIN: [
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 9999,
}
],
}
]
}
caplog.set_level(logging.WARNING)
ExceptionMessage = "test connect exception"
mock_pymodbus.connect.side_effect = ModbusException(ExceptionMessage)
assert await async_setup_component(hass, DOMAIN, config) is True
async def test_delay(
hass: HomeAssistant, mock_pymodbus, freezer: FrozenDateTimeFactory
) -> None:
"""Run test for startup delay."""
# the purpose of this test is to test startup delay
# We "hijiack" a binary_sensor to make a proper blackbox test.
set_delay = 15
set_scan_interval = 5
entity_id = f"{BINARY_SENSOR_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
config = {
DOMAIN: [
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
CONF_DELAY: set_delay,
CONF_BINARY_SENSORS: [
{
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 52,
CONF_SLAVE: 0,
CONF_SCAN_INTERVAL: set_scan_interval,
},
],
}
]
}
mock_pymodbus.read_coils.return_value = ReadResult([0x01])
start_time = dt_util.utcnow()
assert await async_setup_component(hass, DOMAIN, config) is True
await hass.async_block_till_done()
assert hass.states.get(entity_id).state in (STATE_UNKNOWN, STATE_UNAVAILABLE)
time_sensor_active = start_time + timedelta(seconds=2)
time_after_delay = start_time + timedelta(seconds=(set_delay))
time_after_scan = time_after_delay + timedelta(seconds=(set_scan_interval))
time_stop = time_after_scan + timedelta(seconds=10)
now = start_time
while now < time_stop:
# This test assumed listeners are always fired at 0
# microseconds which is impossible in production so
# we use 999999 microseconds to simulate the real world.
freezer.tick(timedelta(seconds=1, microseconds=999999))
now = dt_util.utcnow()
async_fire_time_changed(hass, now)
# the hub sleeps the delay in a background task, so waiting for those
# before the delay has passed would wait forever
await hass.async_block_till_done(wait_background_tasks=now > time_after_delay)
if now > time_sensor_active:
if now <= time_after_delay:
assert hass.states.get(entity_id).state in (
STATE_UNKNOWN,
STATE_UNAVAILABLE,
)
if now <= time_after_delay + timedelta(seconds=2):
continue
if now > time_after_scan + timedelta(seconds=2):
assert hass.states.get(entity_id).state == STATE_ON
@pytest.mark.parametrize(
"do_config",
[
{
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
CONF_SLAVE: 0,
CONF_SCAN_INTERVAL: 0,
}
],
},
],
)
async def test_shutdown(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mock_pymodbus,
mock_modbus_with_pymodbus,
) -> None:
"""Run test for shutdown."""
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert mock_pymodbus.close.called
assert caplog.text == ""
@pytest.mark.parametrize("do_config", [{}])
async def test_write_no_client(hass: HomeAssistant, mock_modbus) -> None:
"""Run test for service stop and write without client."""
await mock_modbus.reset()
data = {
ATTR_HUB: TEST_MODBUS_NAME,
}
await hass.services.async_call(DOMAIN, SERVICE_STOP, data, blocking=True)
await hass.async_block_till_done()
assert mock_modbus.close.called
data = {
ATTR_HUB: TEST_MODBUS_NAME,
ATTR_SLAVE: 17,
ATTR_ADDRESS: 16,
ATTR_STATE: True,
}
await hass.services.async_call(DOMAIN, SERVICE_WRITE_COIL, data, blocking=True)
@pytest.mark.parametrize("do_config", [{}])
async def test_integration_reload(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mock_modbus,
) -> None:
"""Run test for integration reload."""
caplog.set_level(logging.DEBUG)
caplog.clear()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=10))
await hass.async_block_till_done(wait_background_tasks=True)
yaml_path = get_fixture_path("configuration.yaml", DOMAIN)
with mock.patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
assert "Modbus reloading" in caplog.text
state_sensor_1 = hass.states.get("sensor.dummy")
state_sensor_2 = hass.states.get("sensor.dummy_2")
assert state_sensor_1
assert not state_sensor_2
caplog.clear()
yaml_path = get_fixture_path("configuration_2.yaml", DOMAIN)
with mock.patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
assert "Modbus reloading" in caplog.text
state_sensor_1 = hass.states.get("sensor.dummy")
state_sensor_2 = hass.states.get("sensor.dummy_2")
assert state_sensor_1
assert state_sensor_2
caplog.clear()
yaml_path = get_fixture_path("configuration_empty.yaml", DOMAIN)
with mock.patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
assert "Modbus not present anymore" in caplog.text
state_sensor_1 = hass.states.get("sensor.dummy")
state_sensor_2 = hass.states.get("sensor.dummy_2")
assert not state_sensor_1
assert not state_sensor_2
@pytest.mark.skip
@pytest.mark.parametrize("do_config", [{}])
async def test_integration_reload_failed(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, mock_modbus
) -> None:
"""Run test for integration connect failure on reload."""
caplog.set_level(logging.DEBUG)
caplog.clear()
yaml_path = get_fixture_path("configuration.yaml", "modbus")
with (
mock.patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path),
mock.patch.object(mock_modbus, "connect", side_effect=ModbusException("error")),
):
await hass.services.async_call(DOMAIN, SERVICE_RELOAD, blocking=True)
await hass.async_block_till_done()
assert "Modbus reloading" in caplog.text
assert "connect failed, retry in pymodbus" in caplog.text
@pytest.mark.parametrize("do_config", [{}])
async def test_integration_setup_failed(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, mock_modbus
) -> None:
"""Run test for integration setup on reload."""
with mock.patch.object(
hass_config,
"YAML_CONFIG_FILE",
get_fixture_path("configuration.yaml", "modbus"),
):
hass.data[DOMAIN][TEST_MODBUS_NAME].async_setup = mock.AsyncMock(
return_value=False
)
await hass.services.async_call(DOMAIN, SERVICE_RELOAD, blocking=True)
await hass.async_block_till_done()
async def test_no_entities(hass: HomeAssistant) -> None:
"""Run test for failing pymodbus constructor."""
config = {
DOMAIN: [
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
}
]
}
assert await async_setup_component(hass, DOMAIN, config) is False
@pytest.mark.parametrize(
("do_config", "expected_slave_value"),
[
(
{
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 1234,
},
],
},
1,
),
(
{
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 1234,
CONF_SLAVE: 0,
},
],
},
0,
),
(
{
CONF_SENSORS: [
{
CONF_NAME: "dummy",
CONF_ADDRESS: 1234,
CONF_DEVICE_ADDRESS: 6,
},
],
},
6,
),
],
)
async def test_check_default_slave(
hass: HomeAssistant,
mock_modbus,
do_config,
mock_do_cycle,
expected_slave_value: int,
) -> None:
"""Test default slave."""
assert mock_modbus.read_holding_registers.mock_calls
first_call = mock_modbus.read_holding_registers.mock_calls[0]
assert first_call.kwargs[DEVICE_ID] == expected_slave_value
@pytest.mark.parametrize(
"do_config",
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: SERIAL,
CONF_BAUDRATE: 9600,
CONF_BYTESIZE: 8,
CONF_METHOD: "rtu",
CONF_PORT: TEST_PORT_SERIAL,
CONF_PARITY: "E",
CONF_STOPBITS: 1,
CONF_SENSORS: [
{
CONF_NAME: "dummy_noslave",
CONF_ADDRESS: 8888,
}
],
},
],
)
@pytest.mark.parametrize(
"do_write",
[
{
DATA: ATTR_VALUE,
VALUE: 15,
SERVICE: SERVICE_WRITE_REGISTER,
FUNC: CALL_TYPE_WRITE_REGISTER,
},
{
DATA: ATTR_STATE,
VALUE: False,
SERVICE: SERVICE_WRITE_COIL,
FUNC: CALL_TYPE_WRITE_COIL,
},
],
)
@pytest.mark.parametrize(
"do_return",
[
{VALUE: ReadResult([0x0001]), DATA: ""},
{VALUE: ExceptionResponse(0x06), DATA: "Pymodbus:"},
{VALUE: ModbusException("fail write_"), DATA: "Pymodbus:"},
],
)
async def test_pb_service_write_no_slave(
hass: HomeAssistant,
do_write,
do_return,
caplog: pytest.LogCaptureFixture,
mock_modbus_with_pymodbus,
) -> None:
"""Run test for service write_register in case of missing slave/unit parameter."""
func_name = {
CALL_TYPE_WRITE_COIL: mock_modbus_with_pymodbus.write_coil,
CALL_TYPE_WRITE_REGISTER: mock_modbus_with_pymodbus.write_register,
}
value_arg_name = {
CALL_TYPE_WRITE_COIL: "value",
CALL_TYPE_WRITE_REGISTER: "value",
}
data = {
ATTR_HUB: TEST_MODBUS_NAME,
ATTR_ADDRESS: 16,
do_write[DATA]: do_write[VALUE],
}
mock_modbus_with_pymodbus.reset_mock()
caplog.clear()
caplog.set_level(logging.DEBUG)
func_name[do_write[FUNC]].return_value = do_return[VALUE]
await hass.services.async_call(DOMAIN, do_write[SERVICE], data, blocking=True)
assert func_name[do_write[FUNC]].called
assert func_name[do_write[FUNC]].call_args.args == (data[ATTR_ADDRESS],)
assert func_name[do_write[FUNC]].call_args.kwargs == {
DEVICE_ID: 1,
value_arg_name[do_write[FUNC]]: data[do_write[DATA]],
}
if do_return[DATA]:
assert any(message.startswith("Pymodbus:") for message in caplog.messages)
@pytest.mark.parametrize(
"do_config",
[
(
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_TARGET_TEMP: 120,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_SCAN_INTERVAL: 0,
CONF_SCALE: 10,
CONF_OFFSET: 20,
CONF_CURRENT_TEMP_SCALE: 1,
},
),
(
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_TARGET_TEMP: 120,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_SCAN_INTERVAL: 0,
CONF_SCALE: 1,
CONF_OFFSET: 20,
CONF_CURRENT_TEMP_OFFSET: 0,
},
),
(
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_TARGET_TEMP: 120,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_SCAN_INTERVAL: 0,
CONF_SCALE: 1,
CONF_OFFSET: 20,
CONF_TARGET_TEMP_SCALE: 20,
},
),
(
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_TARGET_TEMP: 120,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_SCAN_INTERVAL: 0,
CONF_SCALE: 10,
CONF_OFFSET: 20,
CONF_TARGET_TEMP_OFFSET: 30,
},
),
],
)
async def test_ensure_and_check_conflicting_scales_and_offsets(do_config) -> None:
"""Test ensure_and_check_conflicting_scales_and_offsets."""
with pytest.raises(probatio.Invalid):
ensure_and_check_conflicting_scales_and_offsets(do_config[0])
async def test_not_zero_value() -> None:
"""Test not 0 validator validator."""
with pytest.raises(probatio.Invalid):
not_zero_value(0, "Value cannot be zero.")
Reference in New Issue View Git Blame Copy Permalink