jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reorder MQTT binary sensor tests (#41277)

Browse Source
This commit is contained in:
Erik Montnemery 2020-10-05 19:42:28 +02:00 committed by GitHub
parent 7eaa304fc3
commit 13dad34019
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 82 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
tests/components/mqtt/test_binary_sensor.py
View File

@ -161,6 +161,88 @@ async def expires_helper(hass, mqtt_mock, caplog):
assert state.state == STATE_UNAVAILABLE
async def test_expiration_on_discovery_and_discovery_update_of_binary_sensor(
hass, mqtt_mock, legacy_patchable_time, caplog
):
"""Test that binary_sensor with expire_after set behaves correctly on discovery and discovery update."""
entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
await async_start(hass, "homeassistant", entry)
config = {
"name": "Test",
"state_topic": "test-topic",
"expire_after": 4,
"force_update": True,
}
config_msg = json.dumps(config)
# Set time and publish config message to create binary_sensor via discovery with 4 s expiry
now = datetime(2017, 1, 1, 1, tzinfo=dt_util.UTC)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
async_fire_mqtt_message(
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor is not available
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
# Publish state message
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_mqtt_message(hass, "test-topic", "ON")
await hass.async_block_till_done()
# Test that binary_sensor has correct state
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON
# Advance +3 seconds
now = now + timedelta(seconds=3)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# binary_sensor is not yet expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON
# Resend config message to update discovery
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
async_fire_mqtt_message(
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor has not expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON
# Add +2 seconds
now = now + timedelta(seconds=2)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Test that binary_sensor has expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
# Resend config message to update discovery
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_mqtt_message(
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor is still expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
async def test_setting_sensor_value_via_mqtt_message(hass, mqtt_mock):
"""Test the setting of the value via MQTT."""
assert await async_setup_component(
@ -666,88 +748,6 @@ async def test_discovery_update_unchanged_binary_sensor(hass, mqtt_mock, caplog)
)
async def test_expiration_on_discovery_and_discovery_update_of_binary_sensor(
hass, mqtt_mock, legacy_patchable_time, caplog
):
"""Test that binary_sensor with expire_after set behaves correctly on discovery and discovery update."""
entry = hass.config_entries.async_entries(mqtt.DOMAIN)[0]
await async_start(hass, "homeassistant", entry)
config = {
"name": "Test",
"state_topic": "test-topic",
"expire_after": 4,
"force_update": True,
}
config_msg = json.dumps(config)
# Set time and publish config message to create binary_sensor via discovery with 4 s expiry
now = datetime(2017, 1, 1, 1, tzinfo=dt_util.UTC)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
async_fire_mqtt_message(
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor is not available
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
# Publish state message
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_mqtt_message(hass, "test-topic", "ON")
await hass.async_block_till_done()
# Test that binary_sensor has correct state
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON
# Advance +3 seconds
now = now + timedelta(seconds=3)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# binary_sensor is not yet expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON
# Resend config message to update discovery
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
async_fire_mqtt_message(
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor has not expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON
# Add +2 seconds
now = now + timedelta(seconds=2)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Test that binary_sensor has expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
# Resend config message to update discovery
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now):
async_fire_mqtt_message(
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor is still expired
state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE
@pytest.mark.no_fail_on_log_exception
async def test_discovery_broken(hass, mqtt_mock, caplog):
"""Test handling of bad discovery message."""