evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add fixture to handle mock restore state (#52198)

Browse source
This commit is contained in:
jan iversen 2021-06-29 07:21:04 +02:00 • committed by GitHub
parent 1676bf220f
commit 74e1600a84
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 259 additions and 269 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
tests/components/modbus/conftest.py
View file

@ -18,7 +18,7 @@ from homeassistant.const import (
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed
from tests.common import async_fire_time_changed, mock_restore_cache
TEST_MODBUS_NAME = "modbusTest"
_LOGGER = logging.getLogger(__name__)
@ -61,6 +61,13 @@ async def mock_modbus(hass, do_config):
yield mock_pb
@pytest.fixture
async def mock_test_state(hass, request):
"""Mock restore cache."""
mock_restore_cache(hass, request.param)
return request.param
# dataclass
class ReadResult:
"""Storage class for register read results."""

53
tests/components/modbus/test_binary_sensor.py
View file

@ -19,9 +19,10 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_config_test, base_test, prepare_service_update
from .conftest import ReadResult, base_test, prepare_service_update
from tests.common import mock_restore_cache
sensor_name = "test_binary_sensor"
entity_id = f"{SENSOR_DOMAIN}.{sensor_name}"
@pytest.mark.parametrize(
@ -30,7 +31,7 @@ from tests.common import mock_restore_cache
{
CONF_BINARY_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
}
]
@ -38,7 +39,7 @@ from tests.common import mock_restore_cache
{
CONF_BINARY_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_SLAVE: 10,
CONF_INPUT_TYPE: CALL_TYPE_DISCRETE,
@ -85,7 +86,6 @@ async def test_config_binary_sensor(hass, mock_modbus):
)
async def test_all_binary_sensor(hass, do_type, regs, expected):
"""Run test for given config."""
sensor_name = "modbus_test_binary_sensor"
state = await base_test(
hass,
{CONF_NAME: sensor_name, CONF_ADDRESS: 1234, CONF_INPUT_TYPE: do_type},
@ -104,11 +104,10 @@ async def test_all_binary_sensor(hass, do_type, regs, expected):
async def test_service_binary_sensor_update(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "binary_sensor.test"
config = {
CONF_BINARY_SENSORS: [
{
CONF_NAME: "test",
CONF_NAME: sensor_name,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
}
@ -132,24 +131,24 @@ async def test_service_binary_sensor_update(hass, mock_pymodbus):
assert hass.states.get(entity_id).state == STATE_ON
async def test_restore_state_binary_sensor(hass):
@pytest.mark.parametrize(
"mock_test_state",
[(State(entity_id, STATE_ON),)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_BINARY_SENSORS: [
{
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
}
]
},
],
)
async def test_restore_state_binary_sensor(hass, mock_test_state, mock_modbus):
"""Run test for binary sensor restore state."""
sensor_name = "test_binary_sensor"
test_value = STATE_ON
config_sensor = {CONF_NAME: sensor_name, CONF_ADDRESS: 17}
mock_restore_cache(
hass,
(State(f"{SENSOR_DOMAIN}.{sensor_name}", test_value),),
)
await base_config_test(
hass,
config_sensor,
sensor_name,
SENSOR_DOMAIN,
CONF_BINARY_SENSORS,
None,
method_discovery=True,
)
entity_id = f"{SENSOR_DOMAIN}.{sensor_name}"
assert hass.states.get(entity_id).state == test_value
assert hass.states.get(entity_id).state == mock_test_state[0].state

63
tests/components/modbus/test_climate.py
View file

@ -14,9 +14,10 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_config_test, base_test, prepare_service_update
from .conftest import ReadResult, base_test, prepare_service_update
from tests.common import mock_restore_cache
climate_name = "test_climate"
entity_id = f"{CLIMATE_DOMAIN}.{climate_name}"
@pytest.mark.parametrize(
@ -25,7 +26,7 @@ from tests.common import mock_restore_cache
{
CONF_CLIMATES: [
{
CONF_NAME: "test_climate",
CONF_NAME: climate_name,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
@ -35,7 +36,7 @@ from tests.common import mock_restore_cache
{
CONF_CLIMATES: [
{
CONF_NAME: "test_climate",
CONF_NAME: climate_name,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
@ -88,11 +89,10 @@ async def test_temperature_climate(hass, regs, expected):
async def test_service_climate_update(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "climate.test"
config = {
CONF_CLIMATES: [
{
CONF_NAME: "test",
CONF_NAME: climate_name,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
CONF_SLAVE: 10,
@ -110,32 +110,31 @@ async def test_service_climate_update(hass, mock_pymodbus):
assert hass.states.get(entity_id).state == "auto"
async def test_restore_state_climate(hass):
"""Run test for sensor restore state."""
test_value = State(entity_id, 35)
test_value.attributes = {ATTR_TEMPERATURE: 37}
climate_name = "test_climate"
test_temp = 37
entity_id = f"{CLIMATE_DOMAIN}.{climate_name}"
test_value = State(entity_id, 35)
test_value.attributes = {ATTR_TEMPERATURE: test_temp}
config_sensor = {
CONF_NAME: climate_name,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
}
mock_restore_cache(
hass,
(test_value,),
)
await base_config_test(
hass,
config_sensor,
climate_name,
CLIMATE_DOMAIN,
CONF_CLIMATES,
None,
method_discovery=True,
)
@pytest.mark.parametrize(
"mock_test_state",
[(test_value,)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_CLIMATES: [
{
CONF_NAME: climate_name,
CONF_TARGET_TEMP: 117,
CONF_ADDRESS: 117,
}
],
},
],
)
async def test_restore_state_climate(hass, mock_test_state, mock_modbus):
"""Run test for sensor restore state."""
state = hass.states.get(entity_id)
assert state.state == HVAC_MODE_AUTO
assert state.attributes[ATTR_TEMPERATURE] == test_temp
assert state.attributes[ATTR_TEMPERATURE] == 37

82
tests/components/modbus/test_cover.py
View file

@ -29,9 +29,10 @@ from homeassistant.const import (
)
from homeassistant.core import State
from .conftest import ReadResult, base_config_test, base_test, prepare_service_update
from .conftest import ReadResult, base_test, prepare_service_update
from tests.common import mock_restore_cache
cover_name = "test_cover"
entity_id = f"{COVER_DOMAIN}.{cover_name}"
@pytest.mark.parametrize(
@ -40,7 +41,7 @@ from tests.common import mock_restore_cache
{
CONF_COVERS: [
{
CONF_NAME: "test_cover",
CONF_NAME: cover_name,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
}
@ -49,7 +50,7 @@ from tests.common import mock_restore_cache
{
CONF_COVERS: [
{
CONF_NAME: "test_cover",
CONF_NAME: cover_name,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_SLAVE: 10,
@ -91,7 +92,6 @@ async def test_config_cover(hass, mock_modbus):
)
async def test_coil_cover(hass, regs, expected):
"""Run test for given config."""
cover_name = "modbus_test_cover"
state = await base_test(
hass,
{
@ -139,7 +139,6 @@ async def test_coil_cover(hass, regs, expected):
)
async def test_register_cover(hass, regs, expected):
"""Run test for given config."""
cover_name = "modbus_test_cover"
state = await base_test(
hass,
{
@ -162,11 +161,10 @@ async def test_register_cover(hass, regs, expected):
async def test_service_cover_update(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "cover.test"
config = {
CONF_COVERS: [
{
CONF_NAME: "test",
CONF_NAME: cover_name,
CONF_ADDRESS: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
}
@ -189,54 +187,54 @@ async def test_service_cover_update(hass, mock_pymodbus):
@pytest.mark.parametrize(
"state", [STATE_CLOSED, STATE_CLOSING, STATE_OPENING, STATE_OPEN]
"mock_test_state",
[
(State(entity_id, STATE_CLOSED),),
(State(entity_id, STATE_CLOSING),),
(State(entity_id, STATE_OPENING),),
(State(entity_id, STATE_OPEN),),
],
indirect=True,
)
async def test_restore_state_cover(hass, state):
@pytest.mark.parametrize(
"do_config",
[
{
CONF_COVERS: [
{
CONF_NAME: cover_name,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
CONF_STATE_OPEN: 1,
CONF_STATE_CLOSED: 0,
CONF_STATE_OPENING: 2,
CONF_STATE_CLOSING: 3,
CONF_STATUS_REGISTER: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
}
]
},
],
)
async def test_restore_state_cover(hass, mock_test_state, mock_modbus):
"""Run test for cover restore state."""
entity_id = "cover.test"
cover_name = "test"
config = {
CONF_NAME: cover_name,
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
CONF_STATE_OPEN: 1,
CONF_STATE_CLOSED: 0,
CONF_STATE_OPENING: 2,
CONF_STATE_CLOSING: 3,
CONF_STATUS_REGISTER: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
}
mock_restore_cache(
hass,
(State(f"{entity_id}", state),),
)
await base_config_test(
hass,
config,
cover_name,
COVER_DOMAIN,
CONF_COVERS,
None,
method_discovery=True,
)
assert hass.states.get(entity_id).state == state
test_state = mock_test_state[0].state
assert hass.states.get(entity_id).state == test_state
async def test_service_cover_move(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "cover.test"
entity_id2 = "cover.test2"
entity_id2 = f"{entity_id}2"
config = {
CONF_COVERS: [
{
CONF_NAME: "test",
CONF_NAME: cover_name,
CONF_ADDRESS: 1234,
CONF_STATUS_REGISTER_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
{
CONF_NAME: "test2",
CONF_NAME: f"{cover_name}2",
CONF_INPUT_TYPE: CALL_TYPE_COIL,
CONF_ADDRESS: 1234,
},

82
tests/components/modbus/test_fan.py
View file

@ -32,9 +32,10 @@ from homeassistant.const import (
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 .conftest import ReadResult, base_test, prepare_service_update
from tests.common import mock_restore_cache
fan_name = "test_fan"
entity_id = f"{FAN_DOMAIN}.{fan_name}"
@pytest.mark.parametrize(
@ -43,7 +44,7 @@ from tests.common import mock_restore_cache
{
CONF_FANS: [
{
CONF_NAME: "test_fan",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
}
]
@ -51,7 +52,7 @@ from tests.common import mock_restore_cache
{
CONF_FANS: [
{
CONF_NAME: "test_fan",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
}
@ -60,7 +61,7 @@ from tests.common import mock_restore_cache
{
CONF_FANS: [
{
CONF_NAME: "test_fan",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -77,7 +78,7 @@ from tests.common import mock_restore_cache
{
CONF_FANS: [
{
CONF_NAME: "test_fan",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -94,7 +95,7 @@ from tests.common import mock_restore_cache
{
CONF_FANS: [
{
CONF_NAME: "test_fan",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -111,7 +112,7 @@ from tests.common import mock_restore_cache
{
CONF_FANS: [
{
CONF_NAME: "test_fan",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -160,7 +161,6 @@ async def test_config_fan(hass, mock_modbus):
)
async def test_all_fan(hass, call_type, regs, verify, expected):
"""Run test for given config."""
fan_name = "modbus_test_fan"
state = await base_test(
hass,
{
@ -182,34 +182,33 @@ async def test_all_fan(hass, call_type, regs, verify, expected):
assert state == expected
async def test_restore_state_fan(hass):
@pytest.mark.parametrize(
"mock_test_state",
[(State(entity_id, STATE_ON),)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_FANS: [
{
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
}
]
},
],
)
async def test_restore_state_fan(hass, mock_test_state, mock_modbus):
"""Run test for fan restore state."""
fan_name = "test_fan"
entity_id = f"{FAN_DOMAIN}.{fan_name}"
test_value = STATE_ON
config_fan = {CONF_NAME: fan_name, CONF_ADDRESS: 17}
mock_restore_cache(
hass,
(State(f"{entity_id}", test_value),),
)
await base_config_test(
hass,
config_fan,
fan_name,
FAN_DOMAIN,
CONF_FANS,
None,
method_discovery=True,
)
assert hass.states.get(entity_id).state == test_value
assert hass.states.get(entity_id).state == STATE_ON
async def test_fan_service_turn(hass, caplog, mock_pymodbus):
"""Run test for service turn_on/turn_off."""
entity_id1 = f"{FAN_DOMAIN}.fan1"
entity_id2 = f"{FAN_DOMAIN}.fan2"
entity_id2 = f"{FAN_DOMAIN}.{fan_name}2"
config = {
MODBUS_DOMAIN: {
CONF_TYPE: "tcp",
@ -217,12 +216,12 @@ async def test_fan_service_turn(hass, caplog, mock_pymodbus):
CONF_PORT: 5501,
CONF_FANS: [
{
CONF_NAME: "fan1",
CONF_NAME: fan_name,
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
{
CONF_NAME: "fan2",
CONF_NAME: f"{fan_name}2",
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_VERIFY: {},
@ -234,17 +233,17 @@ async def test_fan_service_turn(hass, caplog, mock_pymodbus):
await hass.async_block_till_done()
assert MODBUS_DOMAIN in hass.config.components
assert hass.states.get(entity_id1).state == STATE_OFF
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
"fan", "turn_on", service_data={"entity_id": entity_id1}
"fan", "turn_on", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_ON
assert hass.states.get(entity_id).state == STATE_ON
await hass.services.async_call(
"fan", "turn_off", service_data={"entity_id": entity_id1}
"fan", "turn_off", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_OFF
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
@ -268,20 +267,19 @@ async def test_fan_service_turn(hass, caplog, mock_pymodbus):
assert hass.states.get(entity_id2).state == STATE_UNAVAILABLE
mock_pymodbus.write_coil.side_effect = ModbusException("fail write_")
await hass.services.async_call(
"fan", "turn_off", service_data={"entity_id": entity_id1}
"fan", "turn_off", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_UNAVAILABLE
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_service_fan_update(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "fan.test"
config = {
CONF_FANS: [
{
CONF_NAME: "test",
CONF_NAME: fan_name,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},

82
tests/components/modbus/test_light.py
View file

@ -32,9 +32,10 @@ from homeassistant.const import (
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 .conftest import ReadResult, base_test, prepare_service_update
from tests.common import mock_restore_cache
light_name = "test_light"
entity_id = f"{LIGHT_DOMAIN}.{light_name}"
@pytest.mark.parametrize(
@ -43,7 +44,7 @@ from tests.common import mock_restore_cache
{
CONF_LIGHTS: [
{
CONF_NAME: "test_light",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
}
]
@ -51,7 +52,7 @@ from tests.common import mock_restore_cache
{
CONF_LIGHTS: [
{
CONF_NAME: "test_light",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
}
@ -60,7 +61,7 @@ from tests.common import mock_restore_cache
{
CONF_LIGHTS: [
{
CONF_NAME: "test_light",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -77,7 +78,7 @@ from tests.common import mock_restore_cache
{
CONF_LIGHTS: [
{
CONF_NAME: "test_light",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -94,7 +95,7 @@ from tests.common import mock_restore_cache
{
CONF_LIGHTS: [
{
CONF_NAME: "test_light",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -111,7 +112,7 @@ from tests.common import mock_restore_cache
{
CONF_LIGHTS: [
{
CONF_NAME: "test_light",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -160,7 +161,6 @@ async def test_config_light(hass, mock_modbus):
)
async def test_all_light(hass, call_type, regs, verify, expected):
"""Run test for given config."""
light_name = "modbus_test_light"
state = await base_test(
hass,
{
@ -182,34 +182,33 @@ async def test_all_light(hass, call_type, regs, verify, expected):
assert state == expected
async def test_restore_state_light(hass):
@pytest.mark.parametrize(
"mock_test_state",
[(State(entity_id, STATE_ON),)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_LIGHTS: [
{
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
}
]
},
],
)
async def test_restore_state_light(hass, mock_test_state, mock_modbus):
"""Run test for sensor restore state."""
light_name = "test_light"
entity_id = f"{LIGHT_DOMAIN}.{light_name}"
test_value = STATE_ON
config_light = {CONF_NAME: light_name, CONF_ADDRESS: 17}
mock_restore_cache(
hass,
(State(f"{entity_id}", test_value),),
)
await base_config_test(
hass,
config_light,
light_name,
LIGHT_DOMAIN,
CONF_LIGHTS,
None,
method_discovery=True,
)
assert hass.states.get(entity_id).state == test_value
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."""
entity_id1 = f"{LIGHT_DOMAIN}.light1"
entity_id2 = f"{LIGHT_DOMAIN}.light2"
entity_id2 = f"{entity_id}2"
config = {
MODBUS_DOMAIN: {
CONF_TYPE: "tcp",
@ -217,12 +216,12 @@ async def test_light_service_turn(hass, caplog, mock_pymodbus):
CONF_PORT: 5501,
CONF_LIGHTS: [
{
CONF_NAME: "light1",
CONF_NAME: light_name,
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
{
CONF_NAME: "light2",
CONF_NAME: f"{light_name}2",
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_VERIFY: {},
@ -234,17 +233,17 @@ async def test_light_service_turn(hass, caplog, mock_pymodbus):
await hass.async_block_till_done()
assert MODBUS_DOMAIN in hass.config.components
assert hass.states.get(entity_id1).state == STATE_OFF
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
"light", "turn_on", service_data={"entity_id": entity_id1}
"light", "turn_on", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_ON
assert hass.states.get(entity_id).state == STATE_ON
await hass.services.async_call(
"light", "turn_off", service_data={"entity_id": entity_id1}
"light", "turn_off", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_OFF
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
@ -268,20 +267,19 @@ async def test_light_service_turn(hass, caplog, mock_pymodbus):
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_id1}
"light", "turn_off", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_UNAVAILABLE
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_service_light_update(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "light.test"
config = {
CONF_LIGHTS: [
{
CONF_NAME: "test",
CONF_NAME: light_name,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},

69
tests/components/modbus/test_sensor.py
View file

@ -38,7 +38,8 @@ from homeassistant.core import State
from .conftest import ReadResult, base_config_test, base_test, prepare_service_update
from tests.common import mock_restore_cache
sensor_name = "test_sensor"
entity_id = f"{SENSOR_DOMAIN}.{sensor_name}"
@pytest.mark.parametrize(
@ -47,7 +48,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
}
]
@ -55,7 +56,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_SLAVE: 10,
CONF_COUNT: 1,
@ -71,7 +72,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_SLAVE: 10,
CONF_COUNT: 1,
@ -87,7 +88,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_COUNT: 1,
CONF_SWAP: CONF_SWAP_NONE,
@ -97,7 +98,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_COUNT: 1,
CONF_SWAP: CONF_SWAP_BYTE,
@ -107,7 +108,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_COUNT: 2,
CONF_SWAP: CONF_SWAP_WORD,
@ -117,7 +118,7 @@ from tests.common import mock_restore_cache
{
CONF_SENSORS: [
{
CONF_NAME: "test_sensor",
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
CONF_COUNT: 2,
CONF_SWAP: CONF_SWAP_WORD_BYTE,
@ -210,7 +211,6 @@ async def test_config_wrong_struct_sensor(
):
"""Run test for sensor with wrong struct."""
sensor_name = "test_sensor"
config_sensor = {
CONF_NAME: sensor_name,
**do_config,
@ -505,7 +505,6 @@ async def test_config_wrong_struct_sensor(
async def test_all_sensor(hass, cfg, regs, expected):
"""Run test for sensor."""
sensor_name = "modbus_test_sensor"
state = await base_test(
hass,
{CONF_NAME: sensor_name, CONF_ADDRESS: 1234, **cfg},
@ -560,7 +559,6 @@ async def test_all_sensor(hass, cfg, regs, expected):
async def test_struct_sensor(hass, cfg, regs, expected):
"""Run test for sensor struct."""
sensor_name = "modbus_test_sensor"
state = await base_test(
hass,
{CONF_NAME: sensor_name, CONF_ADDRESS: 1234, **cfg},
@ -576,27 +574,27 @@ async def test_struct_sensor(hass, cfg, regs, expected):
assert state == expected
async def test_restore_state_sensor(hass):
@pytest.mark.parametrize(
"mock_test_state",
[(State(entity_id, "117"),)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_SENSORS: [
{
CONF_NAME: sensor_name,
CONF_ADDRESS: 51,
}
]
},
],
)
async def test_restore_state_sensor(hass, mock_test_state, mock_modbus):
"""Run test for sensor restore state."""
sensor_name = "test_sensor"
test_value = "117"
config_sensor = {CONF_NAME: sensor_name, CONF_ADDRESS: 17}
mock_restore_cache(
hass,
(State(f"{SENSOR_DOMAIN}.{sensor_name}", test_value),),
)
await base_config_test(
hass,
config_sensor,
sensor_name,
SENSOR_DOMAIN,
CONF_SENSORS,
None,
method_discovery=True,
)
entity_id = f"{SENSOR_DOMAIN}.{sensor_name}"
assert hass.states.get(entity_id).state == test_value
assert hass.states.get(entity_id).state == mock_test_state[0].state
@pytest.mark.parametrize(
@ -604,11 +602,11 @@ async def test_restore_state_sensor(hass):
[
(
CONF_SWAP_WORD,
"Error in sensor modbus_test_sensor swap(word) not possible due to the registers count: 1, needed: 2",
f"Error in sensor {sensor_name} swap(word) not possible due to the registers count: 1, needed: 2",
),
(
CONF_SWAP_WORD_BYTE,
"Error in sensor modbus_test_sensor swap(word_byte) not possible due to the registers count: 1, needed: 2",
f"Error in sensor {sensor_name} swap(word_byte) not possible due to the registers count: 1, needed: 2",
),
],
)
@ -616,7 +614,6 @@ async def test_swap_sensor_wrong_config(
hass, caplog, swap_type, error_message, mock_pymodbus
):
"""Run test for sensor swap."""
sensor_name = "modbus_test_sensor"
config = {
CONF_NAME: sensor_name,
CONF_ADDRESS: 1234,
@ -642,12 +639,10 @@ async def test_swap_sensor_wrong_config(
async def test_service_sensor_update(hass, mock_pymodbus):
"""Run test for service homeassistant.update_entity."""
entity_id = "sensor.test"
config = {
CONF_SENSORS: [
{
CONF_NAME: "test",
CONF_NAME: sensor_name,
CONF_ADDRESS: 1234,
CONF_INPUT_TYPE: CALL_TYPE_REGISTER_INPUT,
}

88
tests/components/modbus/test_switch.py
View file

@ -39,9 +39,12 @@ 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 .conftest import ReadResult, base_test, prepare_service_update
from tests.common import async_fire_time_changed, mock_restore_cache
from tests.common import async_fire_time_changed
switch_name = "test_switch"
entity_id = f"{SWITCH_DOMAIN}.{switch_name}"
@pytest.mark.parametrize(
@ -50,7 +53,7 @@ from tests.common import async_fire_time_changed, mock_restore_cache
{
CONF_SWITCHES: [
{
CONF_NAME: "test_switch",
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
}
]
@ -58,7 +61,7 @@ from tests.common import async_fire_time_changed, mock_restore_cache
{
CONF_SWITCHES: [
{
CONF_NAME: "test_switch",
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
}
@ -67,7 +70,7 @@ from tests.common import async_fire_time_changed, mock_restore_cache
{
CONF_SWITCHES: [
{
CONF_NAME: "test_switch",
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -85,7 +88,7 @@ from tests.common import async_fire_time_changed, mock_restore_cache
{
CONF_SWITCHES: [
{
CONF_NAME: "test_switch",
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -104,7 +107,7 @@ from tests.common import async_fire_time_changed, mock_restore_cache
{
CONF_SWITCHES: [
{
CONF_NAME: "test_switch",
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -122,7 +125,7 @@ from tests.common import async_fire_time_changed, mock_restore_cache
{
CONF_SWITCHES: [
{
CONF_NAME: "test_switch",
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_SLAVE: 1,
CONF_COMMAND_OFF: 0x00,
@ -173,7 +176,6 @@ async def test_config_switch(hass, mock_modbus):
)
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,
{
@ -195,34 +197,33 @@ async def test_all_switch(hass, call_type, regs, verify, expected):
assert state == expected
async def test_restore_state_switch(hass):
@pytest.mark.parametrize(
"mock_test_state",
[(State(entity_id, STATE_ON),)],
indirect=True,
)
@pytest.mark.parametrize(
"do_config",
[
{
CONF_SWITCHES: [
{
CONF_NAME: switch_name,
CONF_ADDRESS: 1234,
}
]
},
],
)
async def test_restore_state_switch(hass, mock_test_state, mock_modbus):
"""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
assert hass.states.get(entity_id).state == mock_test_state[0].state
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"
entity_id2 = f"{SWITCH_DOMAIN}.{switch_name}2"
config = {
MODBUS_DOMAIN: {
CONF_TYPE: "tcp",
@ -230,12 +231,12 @@ async def test_switch_service_turn(hass, caplog, mock_pymodbus):
CONF_PORT: 5501,
CONF_SWITCHES: [
{
CONF_NAME: "switch1",
CONF_NAME: switch_name,
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
},
{
CONF_NAME: "switch2",
CONF_NAME: f"{switch_name}2",
CONF_ADDRESS: 17,
CONF_WRITE_TYPE: CALL_TYPE_REGISTER_HOLDING,
CONF_VERIFY: {},
@ -247,17 +248,17 @@ async def test_switch_service_turn(hass, caplog, mock_pymodbus):
await hass.async_block_till_done()
assert MODBUS_DOMAIN in hass.config.components
assert hass.states.get(entity_id1).state == STATE_OFF
assert hass.states.get(entity_id).state == STATE_OFF
await hass.services.async_call(
"switch", "turn_on", service_data={"entity_id": entity_id1}
"switch", "turn_on", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_ON
assert hass.states.get(entity_id).state == STATE_ON
await hass.services.async_call(
"switch", "turn_off", service_data={"entity_id": entity_id1}
"switch", "turn_off", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_OFF
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
@ -281,20 +282,19 @@ async def test_switch_service_turn(hass, caplog, mock_pymodbus):
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}
"switch", "turn_off", service_data={"entity_id": entity_id}
)
await hass.async_block_till_done()
assert hass.states.get(entity_id1).state == STATE_UNAVAILABLE
assert hass.states.get(entity_id).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_NAME: switch_name,
CONF_ADDRESS: 1234,
CONF_WRITE_TYPE: CALL_TYPE_COIL,
CONF_VERIFY: {},
@ -319,10 +319,6 @@ async def test_service_switch_update(hass, mock_pymodbus):
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: [
{