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

Replace prepare_service_call with a simpler fixture in modbus (#53975) 

* Convert prepare_service_call to a fixture.
pull/54474/head
jan iversen 2021-08-11 12:42:28 +02:00 committed by GitHub
parent 4d93184197
commit 4ef9269790
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 253 additions and 218 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,4 +1,5 @@
"""The tests for the Modbus sensor component."""
from dataclasses import dataclass
from datetime import timedelta
import logging
from unittest import mock
@ -24,6 +25,16 @@ TEST_MODBUS_NAME = "modbusTest"
_LOGGER = logging.getLogger(__name__)
@dataclass
class ReadResult:
"""Storage class for register read results."""
def __init__(self, register_words):
"""Init."""
self.registers = register_words
self.bits = register_words
@pytest.fixture
def mock_pymodbus():
"""Mock pymodbus."""
@ -59,9 +70,15 @@ async def mock_modbus(hass, caplog, request, do_config):
}
]
}
mock_pb = mock.MagicMock()
with mock.patch(
"homeassistant.components.modbus.modbus.ModbusTcpClient", autospec=True
) as mock_pb:
"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
else:
@ -77,14 +94,11 @@ async def mock_test_state(hass, request):
return request.param
# dataclass
class ReadResult:
"""Storage class for register read results."""
def __init__(self, register_words):
"""Init."""
self.registers = register_words
self.bits = register_words
@pytest.fixture
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(
@ -191,21 +205,3 @@ async def base_test(
# Check state
entity_id = f"{entity_domain}.{device_name}"
return hass.states.get(entity_id).state
async def prepare_service_update(hass, config):
"""Run test for service write_coil."""
config_modbus = {
DOMAIN: {
CONF_NAME: DEFAULT_HUB,
CONF_TYPE: "tcp",
CONF_HOST: "modbusTest",
CONF_PORT: 5001,
**config,
},
}
assert await async_setup_component(hass, DOMAIN, config_modbus)
await hass.async_block_till_done()
assert await async_setup_component(hass, "homeassistant", {})
await hass.async_block_till_done()

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

@ -20,7 +20,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
SENSOR_NAME = "test_binary_sensor"
ENTITY_ID = f"{SENSOR_DOMAIN}.{SENSOR_NAME}"
@ -102,33 +102,34 @@ async def test_all_binary_sensor(hass, do_type, regs, expected):
assert state == expected
async def test_service_binary_sensor_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_BINARY_SENSORS: [
{
CONF_NAME: SENSOR_NAME,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
}
]
},
],
)
async def test_service_binary_sensor_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_BINARY_SENSORS: [
{
CONF_NAME: SENSOR_NAME,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
}
]
}
mock_pymodbus.read_coils.return_value = ReadResult([0x00])
await prepare_service_update(
hass,
config,
)
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_OFF
mock_pymodbus.read_coils.return_value = ReadResult([0x01])
mock_modbus.read_coils.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
await hass.async_block_till_done()
assert hass.states.get(ENTITY_ID).state == STATE_ON

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

@ -22,7 +22,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
CLIMATE_NAME = "test_climate"
ENTITY_ID = f"{CLIMATE_DOMAIN}.{CLIMATE_NAME}"
@ -94,25 +94,24 @@ async def test_temperature_climate(hass, regs, expected):
assert state == expected
async def test_service_climate_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_SCAN_INTERVAL: 0,
}
]
},
],
)
async def test_service_climate_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_SCAN_INTERVAL: 0,
}
]
}
mock_pymodbus.read_input_registers.return_value = ReadResult([0x00])
await prepare_service_update(
hass,
config,
)
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
@ -120,34 +119,75 @@ async def test_service_climate_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"data_type, temperature, result",
"temperature, result, do_config",
[
(DATA_TYPE_INT16, 35, [0x00]),
(DATA_TYPE_INT32, 36, [0x00, 0x00]),
(DATA_TYPE_FLOAT32, 37.5, [0x00, 0x00]),
(DATA_TYPE_FLOAT64, "39", [0x00, 0x00, 0x00, 0x00]),
(
35,
[0x00],
{
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_DATA_TYPE: DATA_TYPE_INT16,
}
]
},
),
(
36,
[0x00, 0x00],
{
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_DATA_TYPE: DATA_TYPE_INT32,
}
]
},
),
(
37.5,
[0x00, 0x00],
{
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_DATA_TYPE: DATA_TYPE_FLOAT32,
}
]
},
),
(
"39",
[0x00, 0x00, 0x00, 0x00],
{
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_DATA_TYPE: DATA_TYPE_FLOAT64,
}
]
},
),
],
)
async def test_service_climate_set_temperature(
hass, data_type, temperature, result, mock_pymodbus
hass, temperature, result, mock_modbus, mock_ha
):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_CLIMATES: [
{
CONF_NAME: CLIMATE_NAME,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
CONF_DATA_TYPE: data_type,
}
]
}
mock_pymodbus.read_holding_registers.return_value = ReadResult(result)
await prepare_service_update(
hass,
config,
)
"""Test set_temperature."""
mock_modbus.read_holding_registers.return_value = ReadResult(result)
await hass.services.async_call(
CLIMATE_DOMAIN,
"set_temperature",

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

@ -29,7 +29,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
COVER_NAME = "test_cover"
ENTITY_ID = f"{COVER_DOMAIN}.{COVER_NAME}"
@ -158,28 +158,27 @@ async def test_register_cover(hass, regs, expected):
assert state == expected
async def test_service_cover_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_COVERS: [
{
CONF_NAME: COVER_NAME,
CONF_ADDRESS: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
}
]
},
],
)
async def test_service_cover_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_COVERS: [
{
CONF_NAME: COVER_NAME,
CONF_ADDRESS: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
}
]
}
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x00])
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_CLOSED
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x01])
mock_modbus.read_holding_registers.return_value = ReadResult([0x01])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
@ -223,51 +222,52 @@ async def test_restore_state_cover(hass, mock_test_state, mock_modbus):
assert hass.states.get(ENTITY_ID).state == test_state
async def test_service_cover_move(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_COVERS: [
{
CONF_NAME: COVER_NAME,
CONF_ADDRESS: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_SCAN_INTERVAL: 0,
},
{
CONF_NAME: f"{COVER_NAME}2",
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
CONF_SCAN_INTERVAL: 0,
},
]
},
],
)
async def test_service_cover_move(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
ENTITY_ID2 = f"{ENTITY_ID}2"
config = {
CONF_COVERS: [
{
CONF_NAME: COVER_NAME,
CONF_ADDRESS: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_SCAN_INTERVAL: 0,
},
{
CONF_NAME: f"{COVER_NAME}2",
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
CONF_SCAN_INTERVAL: 0,
},
]
}
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x01])
await prepare_service_update(
hass,
config,
)
mock_modbus.read_holding_registers.return_value = ReadResult([0x01])
await hass.services.async_call(
"cover", "open_cover", {"entity_id": ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == STATE_OPEN
mock_pymodbus.read_holding_registers.return_value = ReadResult([0x00])
mock_modbus.read_holding_registers.return_value = ReadResult([0x00])
await hass.services.async_call(
"cover", "close_cover", {"entity_id": ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == STATE_CLOSED
mock_pymodbus.reset()
mock_pymodbus.read_holding_registers.side_effect = ModbusException("fail write_")
mock_modbus.reset()
mock_modbus.read_holding_registers.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"cover", "close_cover", {"entity_id": ENTITY_ID}, blocking=True
)
assert mock_pymodbus.read_holding_registers.called
assert mock_modbus.read_holding_registers.called
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
mock_pymodbus.read_coils.side_effect = ModbusException("fail write_")
mock_modbus.read_coils.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"cover", "close_cover", {"entity_id": ENTITY_ID2}, blocking=True
)

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

@ -33,7 +33,7 @@ from homeassistant.const import (
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
FAN_NAME = "test_fan"
ENTITY_ID = f"{FAN_DOMAIN}.{FAN_NAME}"
@ -277,30 +277,29 @@ async def test_fan_service_turn(hass, caplog, mock_pymodbus):
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
async def test_service_fan_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_FANS: [
{
CONF_NAME: FAN_NAME,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},
}
]
},
],
)
async def test_service_fan_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_FANS: [
{
CONF_NAME: FAN_NAME,
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
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

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

@ -33,7 +33,7 @@ from homeassistant.const import (
from homeassistant.core import State
from homeassistant.setup import async_setup_component
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
LIGHT_NAME = "test_light"
ENTITY_ID = f"{LIGHT_DOMAIN}.{LIGHT_NAME}"
@ -277,30 +277,29 @@ async def test_light_service_turn(hass, caplog, mock_pymodbus):
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
async def test_service_light_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_LIGHTS: [
{
CONF_NAME: LIGHT_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."""
config = {
CONF_LIGHTS: [
{
CONF_NAME: LIGHT_NAME,
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
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

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

@ -35,7 +35,7 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
SENSOR_NAME = "test_sensor"
ENTITY_ID = f"{SENSOR_DOMAIN}.{SENSOR_NAME}"
@ -599,27 +599,28 @@ async def test_restore_state_sensor(hass, mock_test_state, mock_modbus):
assert hass.states.get(ENTITY_ID).state == mock_test_state[0].state
async def test_service_sensor_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_SENSORS: [
{
CONF_NAME: SENSOR_NAME,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
}
]
},
],
)
async def test_service_sensor_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_SENSORS: [
{
CONF_NAME: SENSOR_NAME,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
}
]
}
mock_pymodbus.read_input_registers.return_value = ReadResult([27])
await prepare_service_update(
hass,
config,
)
mock_modbus.read_input_registers.return_value = ReadResult([27])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)
assert hass.states.get(ENTITY_ID).state == "27"
mock_pymodbus.read_input_registers.return_value = ReadResult([32])
mock_modbus.read_input_registers.return_value = ReadResult([32])
await hass.services.async_call(
"homeassistant", "update_entity", {"entity_id": ENTITY_ID}, blocking=True
)

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

@ -39,7 +39,7 @@ from homeassistant.core import State
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from .conftest import ReadResult, base_test, prepare_service_update
from .conftest import ReadResult, base_test
from tests.common import async_fire_time_changed
@ -291,33 +291,32 @@ async def test_switch_service_turn(hass, caplog, mock_pymodbus):
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
async def test_service_switch_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_SWITCHES: [
{
CONF_NAME: SWITCH_NAME,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},
}
]
},
],
)
async def test_service_switch_update(hass, mock_modbus, mock_ha):
"""Run test for service homeassistant.update_entity."""
config = {
CONF_SWITCHES: [
{
CONF_NAME: SWITCH_NAME,
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
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
async def test_delay_switch(hass, mock_pymodbus):