mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 19:27:45 +00:00
Push modbus switch to 100% test coverage (#50324)
push modbus switch to 100% test coverage.
This commit is contained in:
jan iversen
GitHub
parent
7c7432a582
commit
892a2a0372
@ -631,7 +631,6 @@ omit =
|
||||
homeassistant/components/mochad/*
|
||||
homeassistant/components/modbus/climate.py
|
||||
homeassistant/components/modbus/cover.py
|
||||
homeassistant/components/modbus/switch.py
|
||||
homeassistant/components/modem_callerid/sensor.py
|
||||
homeassistant/components/motion_blinds/__init__.py
|
||||
homeassistant/components/motion_blinds/const.py
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
"""The tests for the Modbus switch component."""
|
||||
from pymodbus.exceptions import ModbusException
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.modbus.const import (
|
||||
@ -6,12 +7,12 @@ from homeassistant.components.modbus.const import (
|
||||
CALL_TYPE_DISCRETE,
|
||||
CALL_TYPE_REGISTER_HOLDING,
|
||||
CALL_TYPE_REGISTER_INPUT,
|
||||
CONF_COILS,
|
||||
CONF_INPUT_TYPE,
|
||||
CONF_STATE_OFF,
|
||||
CONF_STATE_ON,
|
||||
CONF_VERIFY,
|
||||
CONF_WRITE_TYPE,
|
||||
MODBUS_DOMAIN,
|
||||
)
|
||||
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
|
||||
from homeassistant.const import (
|
||||
@ -19,15 +20,23 @@ from homeassistant.const import (
|
||||
CONF_COMMAND_OFF,
|
||||
CONF_COMMAND_ON,
|
||||
CONF_DEVICE_CLASS,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PORT,
|
||||
CONF_SLAVE,
|
||||
CONF_SWITCHES,
|
||||
CONF_TYPE,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
STATE_UNAVAILABLE,
|
||||
)
|
||||
from homeassistant.core import State
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .conftest import ReadResult, base_config_test, base_test, prepare_service_update
|
||||
|
||||
from tests.common import mock_restore_cache
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"do_config",
|
||||
@ -65,6 +74,27 @@ from .conftest import ReadResult, base_config_test, base_test, prepare_service_u
|
||||
CONF_STATE_ON: 1,
|
||||
},
|
||||
},
|
||||
{
|
||||
CONF_ADDRESS: 1234,
|
||||
CONF_SLAVE: 1,
|
||||
CONF_COMMAND_OFF: 0x00,
|
||||
CONF_COMMAND_ON: 0x01,
|
||||
CONF_DEVICE_CLASS: "switch",
|
||||
CONF_VERIFY: {
|
||||
CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
|
||||
CONF_ADDRESS: 1235,
|
||||
CONF_STATE_OFF: 0,
|
||||
CONF_STATE_ON: 1,
|
||||
},
|
||||
},
|
||||
{
|
||||
CONF_ADDRESS: 1234,
|
||||
CONF_SLAVE: 1,
|
||||
CONF_COMMAND_OFF: 0x00,
|
||||
CONF_COMMAND_ON: 0x01,
|
||||
CONF_DEVICE_CLASS: "switch",
|
||||
CONF_VERIFY: None,
|
||||
},
|
||||
],
|
||||
)
|
||||
async def test_config_switch(hass, do_config):
|
||||
@ -87,80 +117,38 @@ async def test_config_switch(hass, do_config):
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("call_type", [CALL_TYPE_COIL, CALL_TYPE_REGISTER_HOLDING])
|
||||
@pytest.mark.parametrize(
|
||||
"regs,expected",
|
||||
"regs,verify,expected",
|
||||
[
|
||||
(
|
||||
[0x00],
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0x80],
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0xFE],
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0xFF],
|
||||
STATE_ON,
|
||||
),
|
||||
(
|
||||
[0x01],
|
||||
STATE_ON,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_coil_switch(hass, regs, expected):
|
||||
"""Run test for given config."""
|
||||
switch_name = "modbus_test_switch"
|
||||
state = await base_test(
|
||||
hass,
|
||||
{
|
||||
CONF_NAME: switch_name,
|
||||
CONF_ADDRESS: 1234,
|
||||
CONF_WRITE_TYPE: CALL_TYPE_COIL,
|
||||
CONF_VERIFY: {},
|
||||
},
|
||||
switch_name,
|
||||
SWITCH_DOMAIN,
|
||||
CONF_SWITCHES,
|
||||
CONF_COILS,
|
||||
regs,
|
||||
expected,
|
||||
method_discovery=True,
|
||||
scan_interval=5,
|
||||
)
|
||||
assert state == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"regs,expected",
|
||||
[
|
||||
(
|
||||
[0x00],
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0x80],
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0xFE],
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0xFF],
|
||||
{CONF_VERIFY: {}},
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
[0x01],
|
||||
{CONF_VERIFY: {}},
|
||||
STATE_ON,
|
||||
),
|
||||
(
|
||||
[0xFE],
|
||||
{CONF_VERIFY: {}},
|
||||
STATE_OFF,
|
||||
),
|
||||
(
|
||||
None,
|
||||
{CONF_VERIFY: {}},
|
||||
STATE_UNAVAILABLE,
|
||||
),
|
||||
(
|
||||
None,
|
||||
{},
|
||||
STATE_OFF,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_register_switch(hass, regs, expected):
|
||||
async def test_all_switch(hass, call_type, regs, verify, expected):
|
||||
"""Run test for given config."""
|
||||
switch_name = "modbus_test_switch"
|
||||
state = await base_test(
|
||||
@ -169,9 +157,8 @@ async def test_register_switch(hass, regs, expected):
|
||||
CONF_NAME: switch_name,
|
||||
CONF_ADDRESS: 1234,
|
||||
CONF_SLAVE: 1,
|
||||
CONF_COMMAND_OFF: 0x00,
|
||||
CONF_COMMAND_ON: 0x01,
|
||||
CONF_VERIFY: {},
|
||||
CONF_WRITE_TYPE: call_type,
|
||||
**verify,
|
||||
},
|
||||
switch_name,
|
||||
SWITCH_DOMAIN,
|
||||
@ -185,42 +172,96 @@ async def test_register_switch(hass, regs, expected):
|
||||
assert state == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"regs,expected",
|
||||
[
|
||||
(
|
||||
[0x40],
|
||||
STATE_ON,
|
||||
),
|
||||
(
|
||||
[0x04],
|
||||
STATE_OFF,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_register_state_switch(hass, regs, expected):
|
||||
"""Run test for given config."""
|
||||
switch_name = "modbus_test_switch"
|
||||
state = await base_test(
|
||||
async def test_restore_state_switch(hass):
|
||||
"""Run test for sensor restore state."""
|
||||
|
||||
switch_name = "test_switch"
|
||||
entity_id = f"{SWITCH_DOMAIN}.{switch_name}"
|
||||
test_value = STATE_ON
|
||||
config_switch = {CONF_NAME: switch_name, CONF_ADDRESS: 17}
|
||||
mock_restore_cache(
|
||||
hass,
|
||||
{
|
||||
CONF_NAME: switch_name,
|
||||
CONF_ADDRESS: 1234,
|
||||
CONF_SLAVE: 1,
|
||||
CONF_COMMAND_OFF: 0x04,
|
||||
CONF_COMMAND_ON: 0x40,
|
||||
CONF_VERIFY: {},
|
||||
},
|
||||
(State(f"{entity_id}", test_value),),
|
||||
)
|
||||
await base_config_test(
|
||||
hass,
|
||||
config_switch,
|
||||
switch_name,
|
||||
SWITCH_DOMAIN,
|
||||
CONF_SWITCHES,
|
||||
None,
|
||||
regs,
|
||||
expected,
|
||||
method_discovery=True,
|
||||
scan_interval=5,
|
||||
)
|
||||
assert state == expected
|
||||
assert hass.states.get(entity_id).state == test_value
|
||||
|
||||
|
||||
async def test_switch_service_turn(hass, caplog, mock_pymodbus):
|
||||
"""Run test for service turn_on/turn_off."""
|
||||
|
||||
entity_id1 = f"{SWITCH_DOMAIN}.switch1"
|
||||
entity_id2 = f"{SWITCH_DOMAIN}.switch2"
|
||||
config = {
|
||||
MODBUS_DOMAIN: {
|
||||
CONF_TYPE: "tcp",
|
||||
CONF_HOST: "modbusTestHost",
|
||||
CONF_PORT: 5501,
|
||||
CONF_SWITCHES: [
|
||||
{
|
||||
CONF_NAME: "switch1",
|
||||
CONF_ADDRESS: 17,
|
||||
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
|
||||
},
|
||||
{
|
||||
CONF_NAME: "switch2",
|
||||
CONF_ADDRESS: 17,
|
||||
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
|
||||
CONF_VERIFY: {},
|
||||
},
|
||||
],
|
||||
},
|
||||
}
|
||||
assert await async_setup_component(hass, MODBUS_DOMAIN, config) is True
|
||||
await hass.async_block_till_done()
|
||||
assert MODBUS_DOMAIN in hass.config.components
|
||||
|
||||
assert hass.states.get(entity_id1).state == STATE_OFF
|
||||
await hass.services.async_call(
|
||||
"switch", "turn_on", service_data={"entity_id": entity_id1}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id1).state == STATE_ON
|
||||
await hass.services.async_call(
|
||||
"switch", "turn_off", service_data={"entity_id": entity_id1}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id1).state == STATE_OFF
|
||||
|
||||
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x01])
|
||||
assert hass.states.get(entity_id2).state == STATE_OFF
|
||||
await hass.services.async_call(
|
||||
"switch", "turn_on", service_data={"entity_id": entity_id2}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id2).state == STATE_ON
|
||||
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x00])
|
||||
await hass.services.async_call(
|
||||
"switch", "turn_off", service_data={"entity_id": entity_id2}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id2).state == STATE_OFF
|
||||
|
||||
mock_pymodbus.write_register.side_effect = ModbusException("fail write_")
|
||||
await hass.services.async_call(
|
||||
"switch", "turn_on", service_data={"entity_id": entity_id2}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id2).state == STATE_UNAVAILABLE
|
||||
mock_pymodbus.write_coil.side_effect = ModbusException("fail write_")
|
||||
await hass.services.async_call(
|
||||
"switch", "turn_off", service_data={"entity_id": entity_id1}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id1).state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
async def test_service_switch_update(hass, mock_pymodbus):
|
||||
@ -233,7 +274,7 @@ async def test_service_switch_update(hass, mock_pymodbus):
|
||||
CONF_NAME: "test",
|
||||
CONF_ADDRESS: 1234,
|
||||
CONF_WRITE_TYPE: CALL_TYPE_COIL,
|
||||
CONF_VERIFY: {CONF_INPUT_TYPE: CALL_TYPE_DISCRETE},
|
||||
CONF_VERIFY: {},
|
||||
}
|
||||
]
|
||||
}
|
||||
@ -246,7 +287,7 @@ async def test_service_switch_update(hass, mock_pymodbus):
|
||||
"homeassistant", "update_entity", {"entity_id": entity_id}, blocking=True
|
||||
)
|
||||
assert hass.states.get(entity_id).state == STATE_ON
|
||||
mock_pymodbus.read_discrete_inputs.return_value = ReadResult([0x00])
|
||||
mock_pymodbus.read_coils.return_value = ReadResult([0x00])
|
||||
await hass.services.async_call(
|
||||
"homeassistant", "update_entity", {"entity_id": entity_id}, blocking=True
|
||||
)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user