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

Remove string literals from modbus component tests (#128899)

pull/129143/head
Claudio Ruggeri - CR-Tech 2024-10-25 12:01:42 +02:00 committed by GitHub
parent da9749ecce
commit f1bef1e7e6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 161 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,6 +3,7 @@
import pytest
from homeassistant.components.binary_sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.modbus.const import (
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
@ -15,10 +16,12 @@ from homeassistant.components.modbus.const import (
MODBUS_DOMAIN,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ADDRESS,
CONF_BINARY_SENSORS,
CONF_DEVICE_CLASS,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
CONF_UNIQUE_ID,
@ -26,7 +29,7 @@ from homeassistant.const import (
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
@ -212,14 +215,20 @@ async def test_service_binary_sensor_update(
"""Run test for service homeassistant.update_entity."""
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_modbus_ha.read_coils.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_ON
@ -428,7 +437,7 @@ async def test_no_discovery_info_binary_sensor(
assert await async_setup_component(
hass,
SENSOR_DOMAIN,
{SENSOR_DOMAIN: {"platform": MODBUS_DOMAIN}},
{SENSOR_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert SENSOR_DOMAIN in hass.config.components

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

@ -20,6 +20,10 @@ from homeassistant.components.climate import (
FAN_OFF,
FAN_ON,
FAN_TOP,
SERVICE_SET_FAN_MODE,
SERVICE_SET_HVAC_MODE,
SERVICE_SET_SWING_MODE,
SERVICE_SET_TEMPERATURE,
SWING_BOTH,
SWING_HORIZONTAL,
SWING_OFF,
@ -27,6 +31,7 @@ from homeassistant.components.climate import (
SWING_VERTICAL,
HVACMode,
)
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.modbus.const import (
CONF_CLIMATES,
CONF_DATA_TYPE,
@ -66,15 +71,17 @@ from homeassistant.components.modbus.const import (
DataType,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_TEMPERATURE,
CONF_ADDRESS,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.setup import async_setup_component
from .conftest import TEST_ENTITY_NAME, ReadResult
@ -152,13 +159,13 @@ ENTITY_ID = f"{CLIMATE_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
CONF_HVAC_MODE_REGISTER: {
CONF_ADDRESS: 11,
CONF_HVAC_MODE_VALUES: {
"state_off": 0,
"state_heat": 1,
"state_cool": 2,
"state_heat_cool": 3,
"state_dry": 4,
"state_fan_only": 5,
"state_auto": 6,
CONF_HVAC_MODE_OFF: 0,
CONF_HVAC_MODE_HEAT: 1,
CONF_HVAC_MODE_COOL: 2,
CONF_HVAC_MODE_HEAT_COOL: 3,
CONF_HVAC_MODE_DRY: 4,
CONF_HVAC_MODE_FAN_ONLY: 5,
CONF_HVAC_MODE_AUTO: 6,
},
},
}
@ -176,13 +183,13 @@ ENTITY_ID = f"{CLIMATE_DOMAIN}.{TEST_ENTITY_NAME}".replace(" ", "_")
CONF_ADDRESS: 11,
CONF_WRITE_REGISTERS: True,
CONF_HVAC_MODE_VALUES: {
"state_off": 0,
"state_heat": 1,
"state_cool": 2,
"state_heat_cool": 3,
"state_dry": 4,
"state_fan_only": 5,
"state_auto": 6,
CONF_HVAC_MODE_OFF: 0,
CONF_HVAC_MODE_HEAT: 1,
CONF_HVAC_MODE_COOL: 2,
CONF_HVAC_MODE_HEAT_COOL: 3,
CONF_HVAC_MODE_DRY: 4,
CONF_HVAC_MODE_FAN_ONLY: 5,
CONF_HVAC_MODE_AUTO: 6,
},
},
}
@ -501,7 +508,10 @@ async def test_service_climate_update(
"""Run test for service homeassistant.update_entity."""
mock_modbus_ha.read_holding_registers.return_value = ReadResult(register_words)
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == result
@ -616,7 +626,10 @@ async def test_service_climate_fan_update(
"""Run test for service homeassistant.update_entity."""
mock_modbus_ha.read_holding_registers.return_value = ReadResult(register_words)
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).attributes[ATTR_FAN_MODE] == result
@ -756,7 +769,10 @@ async def test_service_climate_swing_update(
"""Run test for service homeassistant.update_entity."""
mock_modbus_ha.read_holding_registers.return_value = ReadResult(register_words)
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).attributes[ATTR_SWING_MODE] == result
@ -850,9 +866,9 @@ async def test_service_climate_set_temperature(
mock_modbus_ha.read_holding_registers.return_value = ReadResult(result)
await hass.services.async_call(
CLIMATE_DOMAIN,
"set_temperature",
SERVICE_SET_TEMPERATURE,
{
"entity_id": ENTITY_ID,
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_TEMPERATURE: temperature,
},
blocking=True,
@ -961,9 +977,9 @@ async def test_service_set_hvac_mode(
await hass.services.async_call(
CLIMATE_DOMAIN,
"set_hvac_mode",
SERVICE_SET_HVAC_MODE,
{
"entity_id": ENTITY_ID,
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_HVAC_MODE: hvac_mode,
},
blocking=True,
@ -1024,9 +1040,9 @@ async def test_service_set_fan_mode(
mock_modbus_ha.read_holding_registers.return_value = ReadResult(result)
await hass.services.async_call(
CLIMATE_DOMAIN,
"set_fan_mode",
SERVICE_SET_FAN_MODE,
{
"entity_id": ENTITY_ID,
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_FAN_MODE: fan_mode,
},
blocking=True,
@ -1087,9 +1103,9 @@ async def test_service_set_swing_mode(
mock_modbus_ha.read_holding_registers.return_value = ReadResult(result)
await hass.services.async_call(
CLIMATE_DOMAIN,
"set_swing_mode",
SERVICE_SET_SWING_MODE,
{
"entity_id": ENTITY_ID,
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_SWING_MODE: swing_mode,
},
blocking=True,
@ -1174,7 +1190,7 @@ async def test_no_discovery_info_climate(
assert await async_setup_component(
hass,
CLIMATE_DOMAIN,
{CLIMATE_DOMAIN: {"platform": MODBUS_DOMAIN}},
{CLIMATE_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert CLIMATE_DOMAIN in hass.config.components

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

@ -4,6 +4,7 @@ from pymodbus.exceptions import ModbusException
import pytest
from homeassistant.components.cover import DOMAIN as COVER_DOMAIN, CoverState
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.modbus.const import (
CALL_TYPE_COIL,
CALL_TYPE_REGISTER_HOLDING,
@ -18,14 +19,18 @@ from homeassistant.components.modbus.const import (
MODBUS_DOMAIN,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ADDRESS,
CONF_COVERS,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
SERVICE_CLOSE_COVER,
SERVICE_OPEN_COVER,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.setup import async_setup_component
from .conftest import TEST_ENTITY_NAME, ReadResult
@ -181,12 +186,18 @@ async def test_register_cover(hass: HomeAssistant, expected, mock_do_cycle) -> N
async def test_service_cover_update(hass: HomeAssistant, mock_modbus_ha) -> None:
"""Run test for service homeassistant.update_entity."""
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
"update_entity",
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == CoverState.CLOSED
mock_modbus_ha.read_holding_registers.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == CoverState.OPEN
@ -256,27 +267,27 @@ async def test_service_cover_move(hass: HomeAssistant, mock_modbus_ha) -> None:
mock_modbus_ha.read_holding_registers.return_value = ReadResult([0x01])
await hass.services.async_call(
"cover", "open_cover", {"entity_id": ENTITY_ID}, blocking=True
COVER_DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == CoverState.OPEN
mock_modbus_ha.read_holding_registers.return_value = ReadResult([0x00])
await hass.services.async_call(
"cover", "close_cover", {"entity_id": ENTITY_ID}, blocking=True
COVER_DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == CoverState.CLOSED
await mock_modbus_ha.reset()
mock_modbus_ha.read_holding_registers.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"cover", "close_cover", {"entity_id": ENTITY_ID}, blocking=True
COVER_DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
)
assert mock_modbus_ha.read_holding_registers.called
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
mock_modbus_ha.read_coils.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"cover", "close_cover", {"entity_id": ENTITY_ID2}, blocking=True
COVER_DOMAIN, SERVICE_CLOSE_COVER, {ATTR_ENTITY_ID: ENTITY_ID2}, blocking=True
)
assert hass.states.get(ENTITY_ID2).state == STATE_UNAVAILABLE
@ -289,7 +300,7 @@ async def test_no_discovery_info_cover(
assert await async_setup_component(
hass,
COVER_DOMAIN,
{COVER_DOMAIN: {"platform": MODBUS_DOMAIN}},
{COVER_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert COVER_DOMAIN in hass.config.components

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

@ -4,6 +4,7 @@ from pymodbus.exceptions import ModbusException
import pytest
from homeassistant.components.fan import DOMAIN as FAN_DOMAIN
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.modbus.const import (
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
@ -19,17 +20,21 @@ from homeassistant.components.modbus.const import (
MODBUS_DOMAIN,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ADDRESS,
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.setup import async_setup_component
from .conftest import TEST_ENTITY_NAME, ReadResult
@ -269,12 +274,12 @@ async def test_fan_service_turn(
assert hass.states.get(ENTITY_ID).state == STATE_OFF
await hass.services.async_call(
"fan", "turn_on", service_data={"entity_id": ENTITY_ID}
FAN_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_ON
await hass.services.async_call(
"fan", "turn_off", service_data={"entity_id": ENTITY_ID}
FAN_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
@ -282,26 +287,26 @@ async def test_fan_service_turn(
mock_modbus.read_holding_registers.return_value = ReadResult([0x01])
assert hass.states.get(ENTITY_ID2).state == STATE_OFF
await hass.services.async_call(
"fan", "turn_on", service_data={"entity_id": ENTITY_ID2}
FAN_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_ON
mock_modbus.read_holding_registers.return_value = ReadResult([0x00])
await hass.services.async_call(
"fan", "turn_off", service_data={"entity_id": ENTITY_ID2}
FAN_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_OFF
mock_modbus.write_register.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"fan", "turn_on", service_data={"entity_id": ENTITY_ID2}
FAN_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_UNAVAILABLE
mock_modbus.write_coil.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"fan", "turn_off", service_data={"entity_id": ENTITY_ID}
FAN_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
@ -325,12 +330,18 @@ async def test_fan_service_turn(
async def test_service_fan_update(hass: HomeAssistant, mock_modbus_ha) -> None:
"""Run test for service homeassistant.update_entity."""
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_modbus_ha.read_coils.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == STATE_ON
@ -343,7 +354,7 @@ async def test_no_discovery_info_fan(
assert await async_setup_component(
hass,
FAN_DOMAIN,
{FAN_DOMAIN: {"platform": MODBUS_DOMAIN}},
{FAN_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert FAN_DOMAIN in hass.config.components

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

@ -3,6 +3,7 @@
from pymodbus.exceptions import ModbusException
import pytest
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.components.modbus.const import (
CALL_TYPE_COIL,
@ -18,18 +19,22 @@ from homeassistant.components.modbus.const import (
MODBUS_DOMAIN,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ADDRESS,
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_LIGHTS,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.setup import async_setup_component
from .conftest import TEST_ENTITY_NAME, ReadResult
@ -269,12 +274,12 @@ async def test_light_service_turn(
assert hass.states.get(ENTITY_ID).state == STATE_OFF
await hass.services.async_call(
"light", "turn_on", service_data={"entity_id": ENTITY_ID}
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_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}
LIGHT_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
@ -282,20 +287,20 @@ async def test_light_service_turn(
mock_modbus.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}
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_ON
mock_modbus.read_holding_registers.return_value = ReadResult([0x00])
await hass.services.async_call(
"light", "turn_off", service_data={"entity_id": ENTITY_ID2}
LIGHT_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_OFF
mock_modbus.write_register.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"light", "turn_on", service_data={"entity_id": ENTITY_ID2}
LIGHT_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_UNAVAILABLE
@ -319,12 +324,18 @@ async def test_light_service_turn(
async def test_service_light_update(hass: HomeAssistant, mock_modbus_ha) -> None:
"""Run test for service homeassistant.update_entity."""
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_modbus_ha.read_coils.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == STATE_ON
@ -337,7 +348,7 @@ async def test_no_discovery_info_light(
assert await async_setup_component(
hass,
LIGHT_DOMAIN,
{LIGHT_DOMAIN: {"platform": MODBUS_DOMAIN}},
{LIGHT_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert LIGHT_DOMAIN in hass.config.components

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

@ -4,6 +4,7 @@ import struct
import pytest
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.modbus.const import (
CALL_TYPE_REGISTER_HOLDING,
CALL_TYPE_REGISTER_INPUT,
@ -32,11 +33,13 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ADDRESS,
CONF_COUNT,
CONF_DEVICE_CLASS,
CONF_NAME,
CONF_OFFSET,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SENSORS,
CONF_SLAVE,
@ -45,7 +48,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
@ -1395,12 +1398,18 @@ async def test_service_sensor_update(hass: HomeAssistant, mock_modbus_ha) -> Non
"""Run test for service homeassistant.update_entity."""
mock_modbus_ha.read_input_registers.return_value = ReadResult([27])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == "27"
mock_modbus_ha.read_input_registers.return_value = ReadResult([32])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == "32"
@ -1413,7 +1422,7 @@ async def test_no_discovery_info_sensor(
assert await async_setup_component(
hass,
SENSOR_DOMAIN,
{SENSOR_DOMAIN: {"platform": MODBUS_DOMAIN}},
{SENSOR_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert SENSOR_DOMAIN in hass.config.components

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

@ -6,6 +6,7 @@ from unittest import mock
from pymodbus.exceptions import ModbusException
import pytest
from homeassistant.components.homeassistant import SERVICE_UPDATE_ENTITY
from homeassistant.components.modbus.const import (
CALL_TYPE_COIL,
CALL_TYPE_DISCRETE,
@ -21,20 +22,24 @@ from homeassistant.components.modbus.const import (
)
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ADDRESS,
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_DELAY,
CONF_DEVICE_CLASS,
CONF_NAME,
CONF_PLATFORM,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
CONF_SWITCHES,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -75,7 +80,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_ADDRESS: 1235,
@ -93,7 +98,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_DEVICE_ADDRESS: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_ADDRESS: 1235,
@ -111,7 +116,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
CONF_ADDRESS: 1235,
@ -130,7 +135,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
CONF_ADDRESS: 1235,
@ -148,7 +153,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_SCAN_INTERVAL: 0,
CONF_VERIFY: None,
}
@ -162,7 +167,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_DEVICE_ADDRESS: 10,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_ADDRESS: 1235,
@ -180,7 +185,7 @@ ENTITY_ID3 = f"{ENTITY_ID}_3"
CONF_DEVICE_ADDRESS: 10,
CONF_COMMAND_OFF: 0x00,
CONF_COMMAND_ON: 0x01,
CONF_DEVICE_CLASS: "switch",
CONF_DEVICE_CLASS: SWITCH_DOMAIN,
CONF_VERIFY: {
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_ADDRESS: 1235,
@ -339,12 +344,12 @@ async def test_switch_service_turn(
assert hass.states.get(ENTITY_ID).state == STATE_OFF
await hass.services.async_call(
"switch", "turn_on", service_data={"entity_id": ENTITY_ID}
SWITCH_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_ON
await hass.services.async_call(
"switch", "turn_off", service_data={"entity_id": ENTITY_ID}
SWITCH_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
@ -352,45 +357,45 @@ async def test_switch_service_turn(
mock_modbus.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}
SWITCH_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_ON
mock_modbus.read_holding_registers.return_value = ReadResult([0x00])
await hass.services.async_call(
"switch", "turn_off", service_data={"entity_id": ENTITY_ID2}
SWITCH_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_OFF
mock_modbus.read_holding_registers.return_value = ReadResult([0x03])
assert hass.states.get(ENTITY_ID3).state == STATE_OFF
await hass.services.async_call(
"switch", "turn_on", service_data={"entity_id": ENTITY_ID3}
SWITCH_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID3}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID3).state == STATE_ON
mock_modbus.read_holding_registers.return_value = ReadResult([0x00])
await hass.services.async_call(
"switch", "turn_off", service_data={"entity_id": ENTITY_ID3}
SWITCH_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID3}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID3).state == STATE_OFF
mock_modbus.write_register.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"switch", "turn_on", service_data={"entity_id": ENTITY_ID2}
SWITCH_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID2}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID2).state == STATE_UNAVAILABLE
mock_modbus.write_coil.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"switch", "turn_off", service_data={"entity_id": ENTITY_ID}
SWITCH_DOMAIN, SERVICE_TURN_OFF, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
mock_modbus.write_register.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"switch", "turn_on", service_data={"entity_id": ENTITY_ID3}
SWITCH_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID3}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID3).state == STATE_UNAVAILABLE
@ -434,12 +439,18 @@ async def test_switch_service_turn(
async def test_service_switch_update(hass: HomeAssistant, mock_modbus_ha) -> None:
"""Run test for service homeassistant.update_entity."""
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_modbus_ha.read_coils.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
HOMEASSISTANT_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: ENTITY_ID},
blocking=True,
)
assert hass.states.get(ENTITY_ID).state == STATE_ON
@ -467,7 +478,7 @@ async def test_delay_switch(hass: HomeAssistant, mock_modbus) -> None:
mock_modbus.read_holding_registers.return_value = ReadResult([0x01])
now = dt_util.utcnow()
await hass.services.async_call(
"switch", "turn_on", service_data={"entity_id": ENTITY_ID}
SWITCH_DOMAIN, SERVICE_TURN_ON, service_data={ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
@ -486,7 +497,7 @@ async def test_no_discovery_info_switch(
assert await async_setup_component(
hass,
SWITCH_DOMAIN,
{SWITCH_DOMAIN: {"platform": MODBUS_DOMAIN}},
{SWITCH_DOMAIN: {CONF_PLATFORM: MODBUS_DOMAIN}},
)
await hass.async_block_till_done()
assert SWITCH_DOMAIN in hass.config.components