core/tests/components/modbus/test_switch.py

"""The tests for the Modbus switch component."""
from datetime import timedelta
from unittest import mock

from pymodbus.exceptions import ModbusException
import pytest

from homeassistant.components.modbus.const import (
    CALL_TYPE_COIL,
    CALL_TYPE_DISCRETE,
    CALL_TYPE_REGISTER_HOLDING,
    CALL_TYPE_REGISTER_INPUT,
    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 (
    CONF_ADDRESS,
    CONF_COMMAND_OFF,
    CONF_COMMAND_ON,
    CONF_DELAY,
    CONF_DEVICE_CLASS,
    CONF_HOST,
    CONF_NAME,
    CONF_PORT,
    CONF_SCAN_INTERVAL,
    CONF_SLAVE,
    CONF_SWITCHES,
    CONF_TYPE,
    STATE_OFF,
    STATE_ON,
    STATE_UNAVAILABLE,
)
from homeassistant.core import State
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util

from .conftest import ReadResult, base_config_test, base_test, prepare_service_update

from tests.common import async_fire_time_changed, mock_restore_cache


@pytest.mark.parametrize(
    "do_config",
    [
        {
            CONF_ADDRESS: 1234,
        },
        {
            CONF_ADDRESS: 1234,
            CONF_WRITE_TYPE: CALL_TYPE_COIL,
        },
        {
            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_REGISTER_HOLDING,
                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: {
                CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
                CONF_ADDRESS: 1235,
                CONF_STATE_OFF: 0,
                CONF_STATE_ON: 1,
                CONF_DELAY: 10,
            },
        },
        {
            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_SCAN_INTERVAL: 0,
            CONF_VERIFY: None,
        },
    ],
)
async def test_config_switch(hass, do_config):
    """Run test for switch."""
    device_name = "test_switch"

    device_config = {
        CONF_NAME: device_name,
        **do_config,
    }

    await base_config_test(
        hass,
        device_config,
        device_name,
        SWITCH_DOMAIN,
        CONF_SWITCHES,
        None,
        method_discovery=True,
    )


@pytest.mark.parametrize("call_type", [CALL_TYPE_COIL, CALL_TYPE_REGISTER_HOLDING])
@pytest.mark.parametrize(
    "regs,verify,expected",
    [
        (
            [0x00],
            {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_all_switch(hass, call_type, regs, verify, expected):
    """Run test for given config."""
    switch_name = "modbus_test_switch"
    state = await base_test(
        hass,
        {
            CONF_NAME: switch_name,
            CONF_ADDRESS: 1234,
            CONF_SLAVE: 1,
            CONF_WRITE_TYPE: call_type,
            **verify,
        },
        switch_name,
        SWITCH_DOMAIN,
        CONF_SWITCHES,
        None,
        regs,
        expected,
        method_discovery=True,
        scan_interval=5,
    )
    assert state == expected


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,
        (State(f"{entity_id}", test_value),),
    )
    await base_config_test(
        hass,
        config_switch,
        switch_name,
        SWITCH_DOMAIN,
        CONF_SWITCHES,
        None,
        method_discovery=True,
    )
    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):
    """Run test for service homeassistant.update_entity."""

    entity_id = "switch.test"
    config = {
        CONF_SWITCHES: [
            {
                CONF_NAME: "test",
                CONF_ADDRESS: 1234,
                CONF_WRITE_TYPE: CALL_TYPE_COIL,
                CONF_VERIFY: {},
            }
        ]
    }
    mock_pymodbus.read_discrete_inputs.return_value = ReadResult([0x01])
    await prepare_service_update(
        hass,
        config,
    )
    await hass.services.async_call(
        "homeassistant", "update_entity", {"entity_id": entity_id}, blocking=True
    )
    assert hass.states.get(entity_id).state == STATE_ON
    mock_pymodbus.read_coils.return_value = ReadResult([0x00])
    await hass.services.async_call(
        "homeassistant", "update_entity", {"entity_id": entity_id}, blocking=True
    )
    assert hass.states.get(entity_id).state == STATE_OFF


async def test_delay_switch(hass, mock_pymodbus):
    """Run test for switch verify delay."""

    switch_name = "test_switch"
    entity_id = f"{SWITCH_DOMAIN}.{switch_name}"

    config = {
        MODBUS_DOMAIN: [
            {
                CONF_TYPE: "tcp",
                CONF_HOST: "modbusTestHost",
                CONF_PORT: 5501,
                CONF_SWITCHES: [
                    {
                        CONF_NAME: switch_name,
                        CONF_ADDRESS: 51,
                        CONF_SCAN_INTERVAL: 0,
                        CONF_VERIFY: {
                            CONF_DELAY: 1,
                            CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
                        },
                    }
                ],
            }
        ]
    }
    mock_pymodbus.read_holding_registers.return_value = ReadResult([0x01])
    now = dt_util.utcnow()
    with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
        assert await async_setup_component(hass, MODBUS_DOMAIN, config) is True
        await hass.async_block_till_done()
    await hass.services.async_call(
        "switch", "turn_on", service_data={"entity_id": entity_id}
    )
    await hass.async_block_till_done()
    assert hass.states.get(entity_id).state == STATE_OFF
    now = now + timedelta(seconds=2)
    with mock.patch("homeassistant.helpers.event.dt_util.utcnow", return_value=now):
        async_fire_time_changed(hass, now)
        await hass.async_block_till_done()
    assert hass.states.get(entity_id).state == STATE_ON