core/tests/components/modbus/test_light.py

319 lines
9.1 KiB
Python

"""The tests for the Modbus light component."""
from pymodbus.exceptions import ModbusException
import pytest
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.components.modbus.const import (
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
CALL_TYPE_REGISTER_HOLDING,
CALL_TYPE_REGISTER_INPUT,
CONF_INPUT_TYPE,
CONF_LAZY_ERROR,
CONF_STATE_OFF,
CONF_STATE_ON,
CONF_VERIFY,
CONF_WRITE_TYPE,
MODBUS_DOMAIN,
TCP,
)
from homeassistant.const import (
CONF_ADDRESS,
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_HOST,
CONF_LIGHTS,
CONF_NAME,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
CONF_TYPE,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
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
ENTITY_ID = f"{LIGHT_DOMAIN}.{TEST_ENTITY_NAME}"
ENTITY_ID2 = f"{ENTITY_ID}2"
@pytest.mark.parametrize(
"do_config",
[
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
}
]
},
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_LAZY_ERROR: 10,
}
]
},
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_ADDRESS: 1235,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
},
}
]
},
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
CONF_ADDRESS: 1235,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
},
}
]
},
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
CONF_ADDRESS: 1235,
CONF_STATE_OFF: 0,
CONF_STATE_ON: 1,
},
}
]
},
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_VERIFY: None,
}
]
},
],
)
async def test_config_light(hass, mock_modbus):
"""Run configuration test for light."""
assert LIGHT_DOMAIN in hass.config.components
@pytest.mark.parametrize(
"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,do_exception,config_addon,expected",
[
(
[0x00],
False,
{CONF_VERIFY: {}},
STATE_OFF,
),
(
[0x01],
False,
{CONF_VERIFY: {}},
STATE_ON,
),
(
[0xFE],
False,
{CONF_VERIFY: {}},
STATE_OFF,
),
(
[0x00],
True,
{CONF_VERIFY: {}},
STATE_UNAVAILABLE,
),
(
[0x00],
True,
None,
STATE_OFF,
),
],
)
async def test_all_light(hass, mock_do_cycle, expected):
"""Run test for given config."""
assert hass.states.get(ENTITY_ID).state == expected
@pytest.mark.parametrize(
"mock_test_state",
[(State(ENTITY_ID, STATE_ON),)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_SCAN_INTERVAL: 0,
}
]
},
],
)
async def test_restore_state_light(hass, mock_test_state, mock_modbus):
"""Run test for sensor restore state."""
assert hass.states.get(ENTITY_ID).state == mock_test_state[0].state
async def test_light_service_turn(hass, caplog, mock_pymodbus):
"""Run test for service turn_on/turn_off."""
config = {
MODBUS_DOMAIN: {
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_SCAN_INTERVAL: 0,
},
{
CONF_NAME: f"{TEST_ENTITY_NAME}2",
CONF_ADDRESS: 18,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_SCAN_INTERVAL: 0,
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_ID).state == STATE_OFF
await hass.services.async_call(
"light", "turn_on", service_data={"entity_id": ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_ON
await hass.services.async_call(
"light", "turn_off", service_data={"entity_id": ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).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(
"light", "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(
"light", "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(
"light", "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(
"light", "turn_off", service_data={"entity_id": ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
"do_config",
[
{
CONF_LIGHTS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},
}
]
},
],
)
async def test_service_light_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_modbus.read_coils.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == STATE_ON