Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove base_test from modbus test harness (#55018) 

* Remove base_test.

* Review comments.
pull/53813/head
jan iversen 2021-08-22 19:59:38 +02:00 committed by GitHub
parent 6d049c724c
commit da20552cd8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 263 additions and 342 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

189
tests/components/modbus/conftest.py
View File

@ -1,4 +1,5 @@
"""The tests for the Modbus sensor component."""
import copy
from dataclasses import dataclass
from datetime import timedelta
import logging
@ -7,19 +8,8 @@ from unittest import mock
from pymodbus.exceptions import ModbusException
import pytest
from homeassistant.components.modbus.const import (
DEFAULT_HUB,
MODBUS_DOMAIN as DOMAIN,
TCP,
)
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PLATFORM,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_TYPE,
)
from homeassistant.components.modbus.const import MODBUS_DOMAIN as DOMAIN, TCP
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, CONF_TYPE
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -59,14 +49,33 @@ def mock_pymodbus():
yield mock_pb
@pytest.fixture(
params=[
{"testLoad": True},
],
)
async def mock_modbus(hass, caplog, request, do_config):
"""Load integration modbus using mocked pymodbus."""
@pytest.fixture
def check_config_loaded():
"""Set default for check_config_loaded."""
return True
@pytest.fixture
def register_words():
"""Set default for register_words."""
return [0x00, 0x00]
@pytest.fixture
def config_addon():
"""Add entra configuration items."""
return None
@pytest.fixture
async def mock_modbus(
hass, caplog, register_words, check_config_loaded, config_addon, do_config
):
"""Load integration modbus using mocked pymodbus."""
conf = copy.deepcopy(do_config)
if config_addon:
for key in conf.keys():
conf[key][0].update(config_addon)
caplog.set_level(logging.WARNING)
config = {
DOMAIN: [
@ -75,7 +84,7 @@ async def mock_modbus(hass, caplog, request, do_config):
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_NAME: TEST_MODBUS_NAME,
**do_config,
**conf,
}
]
}
@ -83,19 +92,35 @@ async def mock_modbus(hass, caplog, request, do_config):
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient", return_value=mock_pb
):
mock_pb.read_coils.return_value = ReadResult([0x00])
read_result = ReadResult([0x00, 0x00])
mock_pb.read_discrete_inputs.return_value = read_result
mock_pb.read_input_registers.return_value = read_result
mock_pb.read_holding_registers.return_value = read_result
if request.param["testLoad"]:
assert await async_setup_component(hass, DOMAIN, config) is True
if register_words is None:
exc = ModbusException("fail read_coils")
mock_pb.read_coils.side_effect = exc
mock_pb.read_discrete_inputs.side_effect = exc
mock_pb.read_input_registers.side_effect = exc
mock_pb.read_holding_registers.side_effect = exc
else:
await async_setup_component(hass, DOMAIN, config)
read_result = ReadResult(register_words)
mock_pb.read_coils.return_value = read_result
mock_pb.read_discrete_inputs.return_value = read_result
mock_pb.read_input_registers.return_value = read_result
mock_pb.read_holding_registers.return_value = read_result
now = dt_util.utcnow()
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
result = await async_setup_component(hass, DOMAIN, config)
assert result or not check_config_loaded
await hass.async_block_till_done()
yield mock_pb
@pytest.fixture
async def mock_do_cycle(hass):
"""Trigger update call with time_changed event."""
now = dt_util.utcnow() + timedelta(seconds=90)
with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
@pytest.fixture
async def mock_test_state(hass, request):
"""Mock restore cache."""
@ -108,109 +133,3 @@ async def mock_ha(hass):
"""Load homeassistant to allow service calls."""
assert await async_setup_component(hass, "homeassistant", {})
await hass.async_block_till_done()
async def base_test(
hass,
config_device,
device_name,
entity_domain,
array_name_discovery,
array_name_old_config,
register_words,
expected,
method_discovery=False,
config_modbus=None,
scan_interval=None,
expect_init_to_fail=False,
expect_setup_to_fail=False,
):
"""Run test on device for given config."""
if config_modbus is None:
config_modbus = {
DOMAIN: {
CONF_NAME: DEFAULT_HUB,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
},
}
mock_sync = mock.MagicMock()
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient",
autospec=True,
return_value=mock_sync,
):
# Setup inputs for the sensor
if register_words is None:
mock_sync.read_coils.side_effect = ModbusException("fail read_coils")
mock_sync.read_discrete_inputs.side_effect = ModbusException(
"fail read_coils"
)
mock_sync.read_input_registers.side_effect = ModbusException(
"fail read_coils"
)
mock_sync.read_holding_registers.side_effect = ModbusException(
"fail read_coils"
)
else:
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)
is not expect_setup_to_fail
)
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.config.components) is not expect_setup_to_fail
if config_device is not None:
entity_id = f"{entity_domain}.{device_name}"
device = hass.states.get(entity_id)
if expect_init_to_fail:
assert device is None
elif device is None:
pytest.fail("CONFIG failed, see output")
# 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

46
tests/components/modbus/test_binary_sensor.py
View File

@ -1,4 +1,4 @@
"""The tests for the Modbus sensor component."""
"""Thetests for the Modbus sensor component."""
import pytest
from homeassistant.components.binary_sensor import DOMAIN as SENSOR_DOMAIN
@ -21,7 +21,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import TEST_ENTITY_NAME, ReadResult, base_test
from .conftest import TEST_ENTITY_NAME, ReadResult
ENTITY_ID = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}"
@ -56,9 +56,31 @@ async def test_config_binary_sensor(hass, mock_modbus):
assert SENSOR_DOMAIN in hass.config.components
@pytest.mark.parametrize("do_type", [CALL_TYPE_COIL, CALL_TYPE_DISCRETE])
@pytest.mark.parametrize(
"regs,expected",
"do_config",
[
{
CONF_BINARY_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
},
],
},
{
CONF_BINARY_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,expected",
[
(
[0xFF],
@ -86,21 +108,9 @@ async def test_config_binary_sensor(hass, mock_modbus):
),
],
)
async def test_all_binary_sensor(hass, do_type, regs, expected):
async def test_all_binary_sensor(hass, expected, mock_modbus, mock_do_cycle):
"""Run test for given config."""
state = await base_test(
hass,
{CONF_NAME: TEST_ENTITY_NAME, CONF_ADDRESS: 1234, CONF_INPUT_TYPE: do_type},
TEST_ENTITY_NAME,
SENSOR_DOMAIN,
CONF_BINARY_SENSORS,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(

45
tests/components/modbus/test_climate.py
View File

@ -23,7 +23,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import TEST_ENTITY_NAME, ReadResult, base_test
from .conftest import TEST_ENTITY_NAME, ReadResult
ENTITY_ID = f"{CLIMATE_DOMAIN}.{TEST_ENTITY_NAME}"
@ -62,36 +62,33 @@ async def test_config_climate(hass, mock_modbus):
@pytest.mark.parametrize(
"regs,expected",
"do_config",
[
{
CONF_CLIMATES: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_SLAVE: 1,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_COUNT: 2,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,expected",
[
(
[0x00],
[0x00, 0x00],
"auto",
),
],
)
async def test_temperature_climate(hass, regs, expected):
async def test_temperature_climate(hass, expected, mock_modbus, mock_do_cycle):
"""Run test for given config."""
return
state = await base_test(
hass,
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_SLAVE: 1,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_COUNT: 2,
},
TEST_ENTITY_NAME,
CLIMATE_DOMAIN,
CONF_CLIMATES,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(

76
tests/components/modbus/test_cover.py
View File

@ -30,7 +30,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import TEST_ENTITY_NAME, ReadResult, base_test
from .conftest import TEST_ENTITY_NAME, ReadResult
ENTITY_ID = f"{COVER_DOMAIN}.{TEST_ENTITY_NAME}"
@ -67,7 +67,22 @@ async def test_config_cover(hass, mock_modbus):
@pytest.mark.parametrize(
"regs,expected",
"do_config",
[
{
CONF_COVERS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,expected",
[
(
[0x00],
@ -91,30 +106,27 @@ async def test_config_cover(hass, mock_modbus):
),
],
)
async def test_coil_cover(hass, regs, expected):
async def test_coil_cover(hass, expected, mock_modbus, mock_do_cycle):
"""Run test for given config."""
state = await base_test(
hass,
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
},
TEST_ENTITY_NAME,
COVER_DOMAIN,
CONF_COVERS,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(
"regs,expected",
"do_config",
[
{
CONF_COVERS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,expected",
[
(
[0x00],
@ -138,25 +150,9 @@ async def test_coil_cover(hass, regs, expected):
),
],
)
async def test_register_cover(hass, regs, expected):
async def test_register_cover(hass, expected, mock_modbus, mock_do_cycle):
"""Run test for given config."""
state = await base_test(
hass,
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
},
TEST_ENTITY_NAME,
COVER_DOMAIN,
CONF_COVERS,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(

60
tests/components/modbus/test_fan.py
View File

@ -35,13 +35,7 @@ from homeassistant.const import (
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from .conftest import (
TEST_ENTITY_NAME,
TEST_MODBUS_HOST,
TEST_PORT_TCP,
ReadResult,
base_test,
)
from .conftest import TEST_ENTITY_NAME, TEST_MODBUS_HOST, TEST_PORT_TCP, ReadResult
ENTITY_ID = f"{FAN_DOMAIN}.{TEST_ENTITY_NAME}"
@ -137,9 +131,33 @@ async def test_config_fan(hass, mock_modbus):
assert FAN_DOMAIN in hass.config.components
@pytest.mark.parametrize("call_type", [CALL_TYPE_COIL, CALL_TYPE_REGISTER_HOLDING])
@pytest.mark.parametrize(
"regs,verify,expected",
"do_config",
[
{
CONF_FANS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
},
],
},
{
CONF_FANS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,config_addon,expected",
[
(
[0x00],
@ -163,32 +181,14 @@ async def test_config_fan(hass, mock_modbus):
),
(
None,
{},
None,
STATE_OFF,
),
],
)
async def test_all_fan(hass, call_type, regs, verify, expected):
async def test_all_fan(hass, mock_modbus, mock_do_cycle, expected):
"""Run test for given config."""
state = await base_test(
hass,
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: call_type,
**verify,
},
TEST_ENTITY_NAME,
FAN_DOMAIN,
CONF_FANS,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(

60
tests/components/modbus/test_light.py
View File

@ -35,13 +35,7 @@ from homeassistant.const import (
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from .conftest import (
TEST_ENTITY_NAME,
TEST_MODBUS_HOST,
TEST_PORT_TCP,
ReadResult,
base_test,
)
from .conftest import TEST_ENTITY_NAME, TEST_MODBUS_HOST, TEST_PORT_TCP, ReadResult
ENTITY_ID = f"{LIGHT_DOMAIN}.{TEST_ENTITY_NAME}"
@ -137,9 +131,33 @@ async def test_config_light(hass, mock_modbus):
assert LIGHT_DOMAIN in hass.config.components
@pytest.mark.parametrize("call_type", [CALL_TYPE_COIL, CALL_TYPE_REGISTER_HOLDING])
@pytest.mark.parametrize(
"regs,verify,expected",
"do_config",
[
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
},
],
},
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,config_addon,expected",
[
(
[0x00],
@ -163,32 +181,14 @@ async def test_config_light(hass, mock_modbus):
),
(
None,
{},
None,
STATE_OFF,
),
],
)
async def test_all_light(hass, call_type, regs, verify, expected):
async def test_all_light(hass, mock_modbus, mock_do_cycle, expected):
"""Run test for given config."""
state = await base_test(
hass,
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: call_type,
**verify,
},
TEST_ENTITY_NAME,
LIGHT_DOMAIN,
CONF_LIGHTS,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(

69
tests/components/modbus/test_sensor.py
View File

@ -8,7 +8,6 @@ from homeassistant.components.modbus.const import (
CONF_INPUT_TYPE,
CONF_LAZY_ERROR,
CONF_PRECISION,
CONF_REGISTERS,
CONF_SCALE,
CONF_SWAP,
CONF_SWAP_BYTE,
@ -36,7 +35,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import TEST_ENTITY_NAME, ReadResult, base_test
from .conftest import TEST_ENTITY_NAME, ReadResult
ENTITY_ID = f"{SENSOR_DOMAIN}.{TEST_ENTITY_NAME}"
@ -132,7 +131,7 @@ async def test_config_sensor(hass, mock_modbus):
assert SENSOR_DOMAIN in hass.config.components
@pytest.mark.parametrize("mock_modbus", [{"testLoad": False}], indirect=True)
@pytest.mark.parametrize("check_config_loaded", [False])
@pytest.mark.parametrize(
"do_config,error_message",
[
@ -235,7 +234,20 @@ async def test_config_wrong_struct_sensor(hass, error_message, mock_modbus, capl
@pytest.mark.parametrize(
"cfg,regs,expected",
"do_config",
[
{
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
},
],
},
],
)
@pytest.mark.parametrize(
"config_addon,register_words,expected",
[
(
{
@ -504,26 +516,26 @@ async def test_config_wrong_struct_sensor(hass, error_message, mock_modbus, capl
),
],
)
async def test_all_sensor(hass, cfg, regs, expected):
async def test_all_sensor(hass, mock_modbus, mock_do_cycle, expected):
"""Run test for sensor."""
state = await base_test(
hass,
{CONF_NAME: TEST_ENTITY_NAME, CONF_ADDRESS: 1234, **cfg},
TEST_ENTITY_NAME,
SENSOR_DOMAIN,
CONF_SENSORS,
CONF_REGISTERS,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(
"cfg,regs,expected",
"do_config",
[
{
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 51,
},
],
},
],
)
@pytest.mark.parametrize(
"config_addon,register_words,expected",
[
(
{
@ -558,22 +570,9 @@ async def test_all_sensor(hass, cfg, regs, expected):
),
],
)
async def test_struct_sensor(hass, cfg, regs, expected):
async def test_struct_sensor(hass, mock_modbus, mock_do_cycle, expected):
"""Run test for sensor struct."""
state = await base_test(
hass,
{CONF_NAME: TEST_ENTITY_NAME, CONF_ADDRESS: 1234, **cfg},
TEST_ENTITY_NAME,
SENSOR_DOMAIN,
CONF_SENSORS,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(

60
tests/components/modbus/test_switch.py
View File

@ -41,13 +41,7 @@ from homeassistant.core import State
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from .conftest import (
TEST_ENTITY_NAME,
TEST_MODBUS_HOST,
TEST_PORT_TCP,
ReadResult,
base_test,
)
from .conftest import TEST_ENTITY_NAME, TEST_MODBUS_HOST, TEST_PORT_TCP, ReadResult
from tests.common import async_fire_time_changed
@ -151,9 +145,33 @@ async def test_config_switch(hass, mock_modbus):
assert SWITCH_DOMAIN in hass.config.components
@pytest.mark.parametrize("call_type", [CALL_TYPE_COIL, CALL_TYPE_REGISTER_HOLDING])
@pytest.mark.parametrize(
"regs,verify,expected",
"do_config",
[
{
CONF_SWITCHES: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
},
],
},
{
CONF_SWITCHES: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
],
},
],
)
@pytest.mark.parametrize(
"register_words,config_addon,expected",
[
(
[0x00],
@ -177,32 +195,14 @@ async def test_config_switch(hass, mock_modbus):
),
(
None,
{},
None,
STATE_OFF,
),
],
)
async def test_all_switch(hass, call_type, regs, verify, expected):
async def test_all_switch(hass, mock_modbus, mock_do_cycle, expected):
"""Run test for given config."""
state = await base_test(
hass,
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_WRITE_TYPE: call_type,
**verify,
},
TEST_ENTITY_NAME,
SWITCH_DOMAIN,
CONF_SWITCHES,
None,
regs,
expected,
method_discovery=True,
scan_interval=5,
)
assert state == expected
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(