855bd653b4
* Update test harness to allow discovery_info testing. This is a needed update, before the whole config is converted to the new structure, because it allows test of both the old and new configuration style. The following changes have been made: - Combine test functions into a single base_test. - Prepare for new mock by combining the 2 common test functions into one. - Change mock to be only modbusHub - Do not replace whole modbus class, but only the class that connects to pymodbus (modbusHub). This allows test of modbus configurations and not only component configurations, and is needed when testing discovery activated platform. - Add test of climate. Warning this is merely test of discovery, the real cover tests still needs to be added. - Add test of cover. Warning this is merely test of discovery, the real cover tests still needs to be added. * Update comment for old config * Do not use hass.data, but instead patch pymodbus library. * Add test of configuration (old/new way as available). * Move assert to test function. Make assert more understandable by removing it from the helper. add base_config_test (check just calls base_test) to make it clear if test is wanted or just controlling the config. * use setup_component to load Modbus since it is an integration. * Optimized flow in test helper.
149 lines
4.5 KiB
Python
149 lines
4.5 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 DEFAULT_HUB, MODBUS_DOMAIN as DOMAIN
|
|
from homeassistant.const import (
|
|
CONF_HOST,
|
|
CONF_NAME,
|
|
CONF_PLATFORM,
|
|
CONF_PORT,
|
|
CONF_SCAN_INTERVAL,
|
|
CONF_TYPE,
|
|
)
|
|
from homeassistant.setup import async_setup_component
|
|
import homeassistant.util.dt as dt_util
|
|
|
|
from tests.common import async_fire_time_changed
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class ReadResult:
|
|
"""Storage class for register read results."""
|
|
|
|
def __init__(self, register_words):
|
|
"""Init."""
|
|
self.registers = register_words
|
|
self.bits = register_words
|
|
|
|
|
|
async def base_test(
|
|
hass,
|
|
config_device,
|
|
device_name,
|
|
entity_domain,
|
|
array_name_discovery,
|
|
array_name_old_config,
|
|
register_words,
|
|
expected,
|
|
method_discovery=False,
|
|
check_config_only=False,
|
|
config_modbus=None,
|
|
scan_interval=None,
|
|
):
|
|
"""Run test on device for given config."""
|
|
|
|
if config_modbus is None:
|
|
config_modbus = {
|
|
DOMAIN: {
|
|
CONF_NAME: DEFAULT_HUB,
|
|
CONF_TYPE: "tcp",
|
|
CONF_HOST: "modbusTest",
|
|
CONF_PORT: 5001,
|
|
},
|
|
}
|
|
|
|
mock_sync = mock.MagicMock()
|
|
with mock.patch(
|
|
"homeassistant.components.modbus.modbus.ModbusTcpClient", return_value=mock_sync
|
|
), mock.patch(
|
|
"homeassistant.components.modbus.modbus.ModbusSerialClient",
|
|
return_value=mock_sync,
|
|
), mock.patch(
|
|
"homeassistant.components.modbus.modbus.ModbusUdpClient", return_value=mock_sync
|
|
):
|
|
|
|
# Setup inputs for the sensor
|
|
read_result = ReadResult(register_words)
|
|
mock_sync.read_coils.return_value = read_result
|
|
mock_sync.read_discrete_inputs.return_value = read_result
|
|
mock_sync.read_input_registers.return_value = read_result
|
|
mock_sync.read_holding_registers.return_value = read_result
|
|
|
|
# mock timer and add old/new config
|
|
now = dt_util.utcnow()
|
|
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
|
|
if method_discovery and config_device is not None:
|
|
# setup modbus which in turn does setup for the devices
|
|
config_modbus[DOMAIN].update(
|
|
{array_name_discovery: [{**config_device}]}
|
|
)
|
|
config_device = None
|
|
assert await async_setup_component(hass, DOMAIN, config_modbus)
|
|
await hass.async_block_till_done()
|
|
|
|
# setup platform old style
|
|
if config_device is not None:
|
|
config_device = {
|
|
entity_domain: {
|
|
CONF_PLATFORM: DOMAIN,
|
|
array_name_old_config: [
|
|
{
|
|
**config_device,
|
|
}
|
|
],
|
|
}
|
|
}
|
|
if scan_interval is not None:
|
|
config_device[entity_domain][CONF_SCAN_INTERVAL] = scan_interval
|
|
assert await async_setup_component(hass, entity_domain, config_device)
|
|
await hass.async_block_till_done()
|
|
|
|
assert DOMAIN in hass.data
|
|
if config_device is not None:
|
|
entity_id = f"{entity_domain}.{device_name}"
|
|
device = hass.states.get(entity_id)
|
|
if device is None:
|
|
pytest.fail("CONFIG failed, see output")
|
|
if check_config_only:
|
|
return
|
|
|
|
# Trigger update call with time_changed event
|
|
now = now + timedelta(seconds=scan_interval + 60)
|
|
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}.{device_name}"
|
|
return hass.states.get(entity_id).state
|
|
|
|
|
|
async def base_config_test(
|
|
hass,
|
|
config_device,
|
|
device_name,
|
|
entity_domain,
|
|
array_name_discovery,
|
|
array_name_old_config,
|
|
method_discovery=False,
|
|
config_modbus=None,
|
|
):
|
|
"""Check config of device for given config."""
|
|
|
|
await base_test(
|
|
hass,
|
|
config_device,
|
|
device_name,
|
|
entity_domain,
|
|
array_name_discovery,
|
|
array_name_old_config,
|
|
None,
|
|
None,
|
|
method_discovery=method_discovery,
|
|
check_config_only=True,
|
|
)
|