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

Freeze time for MQTT sensor expire tests (#99496)

Browse Source
This commit is contained in:
Jan Bouwhuis 2023-09-02 11:10:57 +02:00 committed by GitHub
parent 1e46ecbb48
commit 3d1efaa4ad
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 90 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,6 +6,7 @@ from pathlib import Path
from typing import Any from typing import Any
from unittest.mock import patch from unittest.mock import patch
from freezegun import freeze_time
from freezegun.api import FrozenDateTimeFactory from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
@ -146,57 +147,64 @@ async def expires_helper(hass: HomeAssistant) -> None:
"""Run the basic expiry code.""" """Run the basic expiry code."""
realnow = dt_util.utcnow() realnow = dt_util.utcnow()
now = datetime(realnow.year + 1, 1, 1, 1, tzinfo=dt_util.UTC) now = datetime(realnow.year + 1, 1, 1, 1, tzinfo=dt_util.UTC)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): with freeze_time(now) as freezer:
freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
async_fire_mqtt_message(hass, "test-topic", "ON") async_fire_mqtt_message(hass, "test-topic", "ON")
await hass.async_block_till_done() await hass.async_block_till_done()
# Value was set correctly. # Value was set correctly.
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON assert state.state == STATE_ON
# Time jump +3s # Time jump +3s
now = now + timedelta(seconds=3) now += timedelta(seconds=3)
async_fire_time_changed(hass, now) freezer.move_to(now)
await hass.async_block_till_done() async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Value is not yet expired # Value is not yet expired
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON assert state.state == STATE_ON
# Next message resets timer # Next message resets timer
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): # Time jump 0.5s
now += timedelta(seconds=0.5)
freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
async_fire_mqtt_message(hass, "test-topic", "OFF") async_fire_mqtt_message(hass, "test-topic", "OFF")
await hass.async_block_till_done() await hass.async_block_till_done()
# Value was updated correctly. # Value was updated correctly.
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_OFF assert state.state == STATE_OFF
# Time jump +3s # Time jump +3s
now = now + timedelta(seconds=3) now += timedelta(seconds=3)
async_fire_time_changed(hass, now) freezer.move_to(now)
await hass.async_block_till_done() async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Value is not yet expired # Value is not yet expired
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_OFF assert state.state == STATE_OFF
# Time jump +2s # Time jump +2s
now = now + timedelta(seconds=2) now += timedelta(seconds=2)
async_fire_time_changed(hass, now) freezer.move_to(now)
await hass.async_block_till_done() async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Value is expired now # Value is expired now
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
async def test_expiration_on_discovery_and_discovery_update_of_binary_sensor( async def test_expiration_on_discovery_and_discovery_update_of_binary_sensor(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry: MqttMockHAClientGenerator, mqtt_mock_entry: MqttMockHAClientGenerator,
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
freezer: FrozenDateTimeFactory,
) -> None: ) -> None:
"""Test that binary_sensor with expire_after set behaves correctly on discovery and discovery update.""" """Test that binary_sensor with expire_after set behaves correctly on discovery and discovery update."""
await mqtt_mock_entry() await mqtt_mock_entry()
@ -212,31 +220,28 @@ async def test_expiration_on_discovery_and_discovery_update_of_binary_sensor(
# Set time and publish config message to create binary_sensor via discovery with 4 s expiry # Set time and publish config message to create binary_sensor via discovery with 4 s expiry
realnow = dt_util.utcnow() realnow = dt_util.utcnow()
now = datetime(realnow.year + 1, 1, 1, 1, tzinfo=dt_util.UTC) now = datetime(realnow.year + 1, 1, 1, 1, tzinfo=dt_util.UTC)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
async_fire_mqtt_message( async_fire_mqtt_message(hass, "homeassistant/binary_sensor/bla/config", config_msg)
hass, "homeassistant/binary_sensor/bla/config", config_msg await hass.async_block_till_done()
)
await hass.async_block_till_done()
# Test that binary_sensor is not available # Test that binary_sensor is not available
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
# Publish state message # Publish state message
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): async_fire_mqtt_message(hass, "test-topic", "ON")
async_fire_mqtt_message(hass, "test-topic", "ON") await hass.async_block_till_done()
await hass.async_block_till_done()
# Test that binary_sensor has correct state # Test that binary_sensor has correct state
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_ON assert state.state == STATE_ON
# Advance +3 seconds # Advance +3 seconds
now = now + timedelta(seconds=3) now += timedelta(seconds=3)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
await hass.async_block_till_done() await hass.async_block_till_done()
# binary_sensor is not yet expired # binary_sensor is not yet expired
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
@ -255,21 +260,18 @@ async def test_expiration_on_discovery_and_discovery_update_of_binary_sensor(
assert state.state == STATE_ON assert state.state == STATE_ON
# Add +2 seconds # Add +2 seconds
now = now + timedelta(seconds=2) now += timedelta(seconds=2)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
await hass.async_block_till_done() await hass.async_block_till_done()
# Test that binary_sensor has expired # Test that binary_sensor has expired
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
# Resend config message to update discovery # 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)
async_fire_mqtt_message( await hass.async_block_till_done()
hass, "homeassistant/binary_sensor/bla/config", config_msg
)
await hass.async_block_till_done()
# Test that binary_sensor is still expired # Test that binary_sensor is still expired
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")

66
tests/components/mqtt/test_sensor.py
View File

@ -6,6 +6,7 @@ from pathlib import Path
from typing import Any from typing import Any
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from freezegun import freeze_time
from freezegun.api import FrozenDateTimeFactory from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
@ -360,51 +361,56 @@ async def expires_helper(hass: HomeAssistant) -> None:
"""Run the basic expiry code.""" """Run the basic expiry code."""
realnow = dt_util.utcnow() realnow = dt_util.utcnow()
now = datetime(realnow.year + 1, 1, 1, 1, tzinfo=dt_util.UTC) now = datetime(realnow.year + 1, 1, 1, 1, tzinfo=dt_util.UTC)
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): with freeze_time(now) as freezer:
freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
async_fire_mqtt_message(hass, "test-topic", "100") async_fire_mqtt_message(hass, "test-topic", "100")
await hass.async_block_till_done() await hass.async_block_till_done()
# Value was set correctly. # Value was set correctly.
state = hass.states.get("sensor.test") state = hass.states.get("sensor.test")
assert state.state == "100" assert state.state == "100"
# Time jump +3s # Time jump +3s
now = now + timedelta(seconds=3) now += timedelta(seconds=3)
async_fire_time_changed(hass, now) freezer.move_to(now)
await hass.async_block_till_done() async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Value is not yet expired # Value is not yet expired
state = hass.states.get("sensor.test") state = hass.states.get("sensor.test")
assert state.state == "100" assert state.state == "100"
# Next message resets timer # Next message resets timer
with patch(("homeassistant.helpers.event.dt_util.utcnow"), return_value=now): now += timedelta(seconds=0.5)
freezer.move_to(now)
async_fire_time_changed(hass, now) async_fire_time_changed(hass, now)
async_fire_mqtt_message(hass, "test-topic", "101") async_fire_mqtt_message(hass, "test-topic", "101")
await hass.async_block_till_done() await hass.async_block_till_done()
# Value was updated correctly. # Value was updated correctly.
state = hass.states.get("sensor.test") state = hass.states.get("sensor.test")
assert state.state == "101" assert state.state == "101"
# Time jump +3s # Time jump +3s
now = now + timedelta(seconds=3) now += timedelta(seconds=3)
async_fire_time_changed(hass, now) freezer.move_to(now)
await hass.async_block_till_done() async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Value is not yet expired # Value is not yet expired
state = hass.states.get("sensor.test") state = hass.states.get("sensor.test")
assert state.state == "101" assert state.state == "101"
# Time jump +2s # Time jump +2s
now = now + timedelta(seconds=2) now += timedelta(seconds=2)
async_fire_time_changed(hass, now) freezer.move_to(now)
await hass.async_block_till_done() async_fire_time_changed(hass, now)
await hass.async_block_till_done()
# Value is expired now # Value is expired now
state = hass.states.get("sensor.test") state = hass.states.get("sensor.test")
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
@pytest.mark.parametrize( @pytest.mark.parametrize(