Modbus: isolate common test functions (#33447)
Since all entity test functions are going to use the modbus class, isolate the common parts in conftest.py, and thereby make it simpler to write additional test cases. cleaned up test_modbus_sensor.py while splitting the code.
This commit is contained in:
parent
c3ac8869b0
commit
64bdf2d35b
2 changed files with 177 additions and 76 deletions
96
tests/components/modbus/conftest.py
Normal file
96
tests/components/modbus/conftest.py
Normal file
|
@ -0,0 +1,96 @@
|
|||
"""The tests for the Modbus sensor component."""
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.modbus.const import (
|
||||
CALL_TYPE_REGISTER_INPUT,
|
||||
CONF_REGISTER,
|
||||
CONF_REGISTER_TYPE,
|
||||
CONF_REGISTERS,
|
||||
DEFAULT_HUB,
|
||||
MODBUS_DOMAIN,
|
||||
)
|
||||
from homeassistant.const import CONF_NAME, CONF_PLATFORM, CONF_SCAN_INTERVAL
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from tests.common import MockModule, async_fire_time_changed, mock_integration
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_hub(hass):
|
||||
"""Mock hub."""
|
||||
mock_integration(hass, MockModule(MODBUS_DOMAIN))
|
||||
hub = mock.MagicMock()
|
||||
hub.name = "hub"
|
||||
hass.data[MODBUS_DOMAIN] = {DEFAULT_HUB: hub}
|
||||
return hub
|
||||
|
||||
|
||||
common_register_config = {CONF_NAME: "test-config", CONF_REGISTER: 1234}
|
||||
|
||||
|
||||
class ReadResult:
|
||||
"""Storage class for register read results."""
|
||||
|
||||
def __init__(self, register_words):
|
||||
"""Init."""
|
||||
self.registers = register_words
|
||||
|
||||
|
||||
read_result = None
|
||||
|
||||
|
||||
async def simulate_read_registers(unit, address, count):
|
||||
"""Simulate modbus register read."""
|
||||
del unit, address, count # not used in simulation, but in real connection
|
||||
global read_result
|
||||
return read_result
|
||||
|
||||
|
||||
async def run_test(
|
||||
hass, mock_hub, register_config, entity_domain, register_words, expected
|
||||
):
|
||||
"""Run test for given config and check that sensor outputs expected result."""
|
||||
|
||||
# Full sensor configuration
|
||||
sensor_name = "modbus_test_sensor"
|
||||
scan_interval = 5
|
||||
config = {
|
||||
entity_domain: {
|
||||
CONF_PLATFORM: "modbus",
|
||||
CONF_SCAN_INTERVAL: scan_interval,
|
||||
CONF_REGISTERS: [
|
||||
dict(**{CONF_NAME: sensor_name, CONF_REGISTER: 1234}, **register_config)
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
# Setup inputs for the sensor
|
||||
global read_result
|
||||
read_result = ReadResult(register_words)
|
||||
if register_config.get(CONF_REGISTER_TYPE) == CALL_TYPE_REGISTER_INPUT:
|
||||
mock_hub.read_input_registers = simulate_read_registers
|
||||
else:
|
||||
mock_hub.read_holding_registers = simulate_read_registers
|
||||
|
||||
# Initialize sensor
|
||||
now = dt_util.utcnow()
|
||||
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
|
||||
assert await async_setup_component(hass, entity_domain, config)
|
||||
|
||||
# Trigger update call with time_changed event
|
||||
now += timedelta(seconds=scan_interval + 1)
|
||||
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
|
||||
async_fire_time_changed(hass, now)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Check state
|
||||
entity_id = f"{entity_domain}.{sensor_name}"
|
||||
state = hass.states.get(entity_id).state
|
||||
assert state == expected
|
|
@ -1,8 +1,5 @@
|
|||
"""The tests for the Modbus sensor component."""
|
||||
from datetime import timedelta
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
import logging
|
||||
|
||||
from homeassistant.components.modbus.const import (
|
||||
CALL_TYPE_REGISTER_HOLDING,
|
||||
|
@ -11,78 +8,18 @@ from homeassistant.components.modbus.const import (
|
|||
CONF_DATA_TYPE,
|
||||
CONF_OFFSET,
|
||||
CONF_PRECISION,
|
||||
CONF_REGISTER,
|
||||
CONF_REGISTER_TYPE,
|
||||
CONF_REGISTERS,
|
||||
CONF_REVERSE_ORDER,
|
||||
CONF_SCALE,
|
||||
DATA_TYPE_FLOAT,
|
||||
DATA_TYPE_INT,
|
||||
DATA_TYPE_UINT,
|
||||
DEFAULT_HUB,
|
||||
MODBUS_DOMAIN,
|
||||
)
|
||||
from homeassistant.const import CONF_NAME, CONF_PLATFORM, CONF_SCAN_INTERVAL
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.util.dt as dt_util
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
|
||||
from tests.common import MockModule, async_fire_time_changed, mock_integration
|
||||
from .conftest import run_test
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_hub(hass):
|
||||
"""Mock hub."""
|
||||
mock_integration(hass, MockModule(MODBUS_DOMAIN))
|
||||
hub = mock.MagicMock()
|
||||
hub.name = "hub"
|
||||
hass.data[MODBUS_DOMAIN] = {DEFAULT_HUB: hub}
|
||||
return hub
|
||||
|
||||
|
||||
common_register_config = {CONF_NAME: "test-config", CONF_REGISTER: 1234}
|
||||
|
||||
|
||||
class ReadResult:
|
||||
"""Storage class for register read results."""
|
||||
|
||||
def __init__(self, register_words):
|
||||
"""Init."""
|
||||
self.registers = register_words
|
||||
|
||||
|
||||
async def run_test(hass, mock_hub, register_config, register_words, expected):
|
||||
"""Run test for given config and check that sensor outputs expected result."""
|
||||
|
||||
# Full sensor configuration
|
||||
sensor_name = "modbus_test_sensor"
|
||||
scan_interval = 5
|
||||
config = {
|
||||
MODBUS_DOMAIN: {
|
||||
CONF_PLATFORM: "modbus",
|
||||
CONF_SCAN_INTERVAL: scan_interval,
|
||||
CONF_REGISTERS: [
|
||||
dict(**{CONF_NAME: sensor_name, CONF_REGISTER: 1234}, **register_config)
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
# Setup inputs for the sensor
|
||||
read_result = ReadResult(register_words)
|
||||
if register_config.get(CONF_REGISTER_TYPE) == CALL_TYPE_REGISTER_INPUT:
|
||||
mock_hub.read_input_registers.return_value = read_result
|
||||
else:
|
||||
mock_hub.read_holding_registers.return_value = read_result
|
||||
|
||||
# Initialize sensor
|
||||
now = dt_util.utcnow()
|
||||
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
|
||||
assert await async_setup_component(hass, MODBUS_DOMAIN, config)
|
||||
|
||||
# Trigger update call with time_changed event
|
||||
now += timedelta(seconds=scan_interval + 1)
|
||||
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
|
||||
async_fire_time_changed(hass, now)
|
||||
await hass.async_block_till_done()
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def test_simple_word_register(hass, mock_hub):
|
||||
|
@ -94,14 +31,26 @@ async def test_simple_word_register(hass, mock_hub):
|
|||
CONF_OFFSET: 0,
|
||||
CONF_PRECISION: 0,
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[0], expected="0")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0],
|
||||
expected="0",
|
||||
)
|
||||
|
||||
|
||||
async def test_optional_conf_keys(hass, mock_hub):
|
||||
"""Test handling of optional configuration keys."""
|
||||
register_config = {}
|
||||
await run_test(
|
||||
hass, mock_hub, register_config, register_words=[0x8000], expected="-32768"
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x8000],
|
||||
expected="-32768",
|
||||
)
|
||||
|
||||
|
||||
|
@ -114,7 +63,14 @@ async def test_offset(hass, mock_hub):
|
|||
CONF_OFFSET: 13,
|
||||
CONF_PRECISION: 0,
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[7], expected="20")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[7],
|
||||
expected="20",
|
||||
)
|
||||
|
||||
|
||||
async def test_scale_and_offset(hass, mock_hub):
|
||||
|
@ -126,7 +82,14 @@ async def test_scale_and_offset(hass, mock_hub):
|
|||
CONF_OFFSET: 13,
|
||||
CONF_PRECISION: 0,
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[7], expected="34")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[7],
|
||||
expected="34",
|
||||
)
|
||||
|
||||
|
||||
async def test_ints_can_have_precision(hass, mock_hub):
|
||||
|
@ -139,7 +102,12 @@ async def test_ints_can_have_precision(hass, mock_hub):
|
|||
CONF_PRECISION: 4,
|
||||
}
|
||||
await run_test(
|
||||
hass, mock_hub, register_config, register_words=[7], expected="34.0000"
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[7],
|
||||
expected="34.0000",
|
||||
)
|
||||
|
||||
|
||||
|
@ -152,7 +120,14 @@ async def test_floats_get_rounded_correctly(hass, mock_hub):
|
|||
CONF_OFFSET: 0,
|
||||
CONF_PRECISION: 0,
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[1], expected="2")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[1],
|
||||
expected="2",
|
||||
)
|
||||
|
||||
|
||||
async def test_parameters_as_strings(hass, mock_hub):
|
||||
|
@ -164,7 +139,14 @@ async def test_parameters_as_strings(hass, mock_hub):
|
|||
CONF_OFFSET: "5",
|
||||
CONF_PRECISION: "1",
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[9], expected="18.5")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[9],
|
||||
expected="18.5",
|
||||
)
|
||||
|
||||
|
||||
async def test_floating_point_scale(hass, mock_hub):
|
||||
|
@ -176,7 +158,14 @@ async def test_floating_point_scale(hass, mock_hub):
|
|||
CONF_OFFSET: 0,
|
||||
CONF_PRECISION: 2,
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[1], expected="2.40")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[1],
|
||||
expected="2.40",
|
||||
)
|
||||
|
||||
|
||||
async def test_floating_point_offset(hass, mock_hub):
|
||||
|
@ -188,7 +177,14 @@ async def test_floating_point_offset(hass, mock_hub):
|
|||
CONF_OFFSET: -10.3,
|
||||
CONF_PRECISION: 1,
|
||||
}
|
||||
await run_test(hass, mock_hub, register_config, register_words=[2], expected="-8.3")
|
||||
await run_test(
|
||||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[2],
|
||||
expected="-8.3",
|
||||
)
|
||||
|
||||
|
||||
async def test_signed_two_word_register(hass, mock_hub):
|
||||
|
@ -204,6 +200,7 @@ async def test_signed_two_word_register(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x89AB, 0xCDEF],
|
||||
expected="-1985229329",
|
||||
)
|
||||
|
@ -222,6 +219,7 @@ async def test_unsigned_two_word_register(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x89AB, 0xCDEF],
|
||||
expected=str(0x89ABCDEF),
|
||||
)
|
||||
|
@ -238,6 +236,7 @@ async def test_reversed(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x89AB, 0xCDEF],
|
||||
expected=str(0xCDEF89AB),
|
||||
)
|
||||
|
@ -256,6 +255,7 @@ async def test_four_word_register(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x89AB, 0xCDEF, 0x0123, 0x4567],
|
||||
expected="9920249030613615975",
|
||||
)
|
||||
|
@ -274,6 +274,7 @@ async def test_four_word_register_precision_is_intact_with_int_params(hass, mock
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x0123, 0x4567, 0x89AB, 0xCDEF],
|
||||
expected="163971058432973793",
|
||||
)
|
||||
|
@ -292,6 +293,7 @@ async def test_four_word_register_precision_is_lost_with_float_params(hass, mock
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x0123, 0x4567, 0x89AB, 0xCDEF],
|
||||
expected="163971058432973792",
|
||||
)
|
||||
|
@ -311,6 +313,7 @@ async def test_two_word_input_register(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x89AB, 0xCDEF],
|
||||
expected=str(0x89ABCDEF),
|
||||
)
|
||||
|
@ -330,6 +333,7 @@ async def test_two_word_holding_register(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[0x89AB, 0xCDEF],
|
||||
expected=str(0x89ABCDEF),
|
||||
)
|
||||
|
@ -349,6 +353,7 @@ async def test_float_data_type(hass, mock_hub):
|
|||
hass,
|
||||
mock_hub,
|
||||
register_config,
|
||||
SENSOR_DOMAIN,
|
||||
register_words=[16286, 1617],
|
||||
expected="1.23457",
|
||||
)
|
||||
|
|
Loading…
Add table
Reference in a new issue