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.
96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
"""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
|