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

Use freezegun in shelly tests (#99042)

Browse source
This commit is contained in:
Erik Montnemery 2023-08-25 16:06:43 +02:00 • committed by GitHub
parent 452caee41a
commit b0952bc54a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 80 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
tests/components/shelly/__init__.py
View file

@ -7,6 +7,7 @@ from datetime import timedelta
from typing import Any
from unittest.mock import Mock
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.shelly.const import (
@ -20,7 +21,6 @@ from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, format_mac
from homeassistant.helpers.entity_registry import async_get
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
@ -78,17 +78,21 @@ def inject_rpc_device_event(
mock_rpc_device.mock_event()
async def mock_rest_update(hass: HomeAssistant, seconds=REST_SENSORS_UPDATE_INTERVAL):
async def mock_rest_update(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
seconds=REST_SENSORS_UPDATE_INTERVAL,
):
"""Move time to create REST sensors update event."""
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=seconds))
freezer.tick(timedelta(seconds=seconds))
async_fire_time_changed(hass)
await hass.async_block_till_done()
async def mock_polling_rpc_update(hass: HomeAssistant):
async def mock_polling_rpc_update(hass: HomeAssistant, freezer: FrozenDateTimeFactory):
"""Move time to create polling RPC sensors update event."""
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=RPC_SENSORS_POLLING_INTERVAL)
)
freezer.tick(timedelta(seconds=RPC_SENSORS_POLLING_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()

12
tests/components/shelly/test_binary_sensor.py
View file

@ -1,4 +1,6 @@
"""Tests for Shelly binary sensor platform."""
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.shelly.const import SLEEP_PERIOD_MULTIPLIER
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNKNOWN
@ -54,7 +56,7 @@ async def test_block_binary_sensor_extra_state_attr(
async def test_block_rest_binary_sensor(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block REST binary sensor."""
entity_id = register_entity(hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud")
@ -64,13 +66,13 @@ async def test_block_rest_binary_sensor(
assert hass.states.get(entity_id).state == STATE_OFF
monkeypatch.setitem(mock_block_device.status["cloud"], "connected", True)
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
assert hass.states.get(entity_id).state == STATE_ON
async def test_block_rest_binary_sensor_connected_battery_devices(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block REST binary sensor for connected battery devices."""
entity_id = register_entity(hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud")
@ -84,11 +86,11 @@ async def test_block_rest_binary_sensor_connected_battery_devices(
monkeypatch.setitem(mock_block_device.status["cloud"], "connected", True)
# Verify no update on fast intervals
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
assert hass.states.get(entity_id).state == STATE_OFF
# Verify update on slow intervals
await mock_rest_update(hass, seconds=SLEEP_PERIOD_MULTIPLIER * 3600)
await mock_rest_update(hass, freezer, seconds=SLEEP_PERIOD_MULTIPLIER * 3600)
assert hass.states.get(entity_id).state == STATE_ON

106
tests/components/shelly/test_coordinator.py
View file

@ -3,6 +3,7 @@ from datetime import timedelta
from unittest.mock import AsyncMock, patch
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
@ -26,7 +27,6 @@ from homeassistant.helpers.device_registry import (
async_get as async_get_dev_reg,
)
import homeassistant.helpers.issue_registry as ir
from homeassistant.util import dt as dt_util
from . import (
MOCK_MAC,
@ -46,7 +46,7 @@ DEVICE_BLOCK_ID = 4
async def test_block_reload_on_cfg_change(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block reload on config change."""
await init_integration(hass, 1)
@ -66,16 +66,15 @@ async def test_block_reload_on_cfg_change(
assert hass.states.get("switch.test_name_channel_1") is not None
# Wait for debouncer
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=ENTRY_RELOAD_COOLDOWN)
)
freezer.tick(timedelta(seconds=ENTRY_RELOAD_COOLDOWN))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_channel_1") is None
async def test_block_no_reload_on_bulb_changes(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block no reload on bulb mode/effect change."""
await init_integration(hass, 1, model="SHBLB-1")
@ -96,9 +95,8 @@ async def test_block_no_reload_on_bulb_changes(
assert hass.states.get("switch.test_name_channel_1") is not None
# Wait for debouncer
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=ENTRY_RELOAD_COOLDOWN)
)
freezer.tick(timedelta(seconds=ENTRY_RELOAD_COOLDOWN))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_channel_1") is not None
@ -112,16 +110,15 @@ async def test_block_no_reload_on_bulb_changes(
assert hass.states.get("switch.test_name_channel_1") is not None
# Wait for debouncer
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=ENTRY_RELOAD_COOLDOWN)
)
freezer.tick(timedelta(seconds=ENTRY_RELOAD_COOLDOWN))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_channel_1") is not None
async def test_block_polling_auth_error(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block device polling authentication error."""
monkeypatch.setattr(
@ -134,9 +131,8 @@ async def test_block_polling_auth_error(
assert entry.state == ConfigEntryState.LOADED
# Move time to generate polling
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UPDATE_PERIOD_MULTIPLIER * 15)
)
freezer.tick(timedelta(seconds=UPDATE_PERIOD_MULTIPLIER * 15))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
@ -154,7 +150,7 @@ async def test_block_polling_auth_error(
async def test_block_rest_update_auth_error(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block REST update authentication error."""
register_entity(hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud")
@ -170,7 +166,7 @@ async def test_block_rest_update_auth_error(
assert entry.state == ConfigEntryState.LOADED
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
assert entry.state == ConfigEntryState.LOADED
@ -187,7 +183,7 @@ async def test_block_rest_update_auth_error(
async def test_block_polling_connection_error(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block device polling connection error."""
monkeypatch.setattr(
@ -200,16 +196,15 @@ async def test_block_polling_connection_error(
assert hass.states.get("switch.test_name_channel_1").state == STATE_ON
# Move time to generate polling
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UPDATE_PERIOD_MULTIPLIER * 15)
)
freezer.tick(timedelta(seconds=UPDATE_PERIOD_MULTIPLIER * 15))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_channel_1").state == STATE_UNAVAILABLE
async def test_block_rest_update_connection_error(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block REST update connection error."""
entity_id = register_entity(hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud")
@ -217,7 +212,7 @@ async def test_block_rest_update_connection_error(
monkeypatch.setitem(mock_block_device.status, "uptime", 1)
await init_integration(hass, 1)
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
assert hass.states.get(entity_id).state == STATE_ON
monkeypatch.setattr(
@ -225,13 +220,13 @@ async def test_block_rest_update_connection_error(
"update_shelly",
AsyncMock(side_effect=DeviceConnectionError),
)
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_block_sleeping_device_no_periodic_updates(
hass: HomeAssistant, mock_block_device
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device
) -> None:
"""Test block sleeping device no periodic updates."""
entity_id = f"{SENSOR_DOMAIN}.test_name_temperature"
@ -244,9 +239,8 @@ async def test_block_sleeping_device_no_periodic_updates(
assert hass.states.get(entity_id).state == "22.1"
# Move time to generate polling
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=SLEEP_PERIOD_MULTIPLIER * 1000)
)
freezer.tick(timedelta(seconds=UPDATE_PERIOD_MULTIPLIER * 1000))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
@ -322,7 +316,7 @@ async def test_block_button_click_event(
async def test_rpc_reload_on_cfg_change(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC reload on config change."""
await init_integration(hass, 2)
@ -356,16 +350,15 @@ async def test_rpc_reload_on_cfg_change(
assert hass.states.get("switch.test_name_test_switch_0") is not None
# Wait for debouncer
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=ENTRY_RELOAD_COOLDOWN)
)
freezer.tick(timedelta(seconds=ENTRY_RELOAD_COOLDOWN))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_test_switch_0") is None
async def test_rpc_reload_with_invalid_auth(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC when InvalidAuthError is raising during config entry reload."""
with patch(
@ -398,9 +391,8 @@ async def test_rpc_reload_with_invalid_auth(
await hass.async_block_till_done()
# Move time to generate reconnect
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=RPC_RECONNECT_INTERVAL)
)
freezer.tick(timedelta(seconds=RPC_RECONNECT_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
@ -455,7 +447,7 @@ async def test_rpc_click_event(
async def test_rpc_update_entry_sleep_period(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC update entry sleep period."""
entry = await init_integration(hass, 2, sleep_period=600)
@ -475,16 +467,15 @@ async def test_rpc_update_entry_sleep_period(
# Move time to generate sleep period update
monkeypatch.setitem(mock_rpc_device.status["sys"], "wakeup_period", 3600)
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=600 * SLEEP_PERIOD_MULTIPLIER)
)
freezer.tick(timedelta(seconds=600 * SLEEP_PERIOD_MULTIPLIER))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert entry.data["sleep_period"] == 3600
async def test_rpc_sleeping_device_no_periodic_updates(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC sleeping device no periodic updates."""
entity_id = f"{SENSOR_DOMAIN}.test_name_temperature"
@ -504,16 +495,15 @@ async def test_rpc_sleeping_device_no_periodic_updates(
assert hass.states.get(entity_id).state == "22.9"
# Move time to generate polling
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=SLEEP_PERIOD_MULTIPLIER * 1000)
)
freezer.tick(timedelta(seconds=SLEEP_PERIOD_MULTIPLIER * 1000))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_rpc_reconnect_auth_error(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC reconnect authentication error."""
entry = await init_integration(hass, 2)
@ -530,9 +520,8 @@ async def test_rpc_reconnect_auth_error(
assert entry.state == ConfigEntryState.LOADED
# Move time to generate reconnect
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=RPC_RECONNECT_INTERVAL)
)
freezer.tick(timedelta(seconds=RPC_RECONNECT_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
@ -550,7 +539,7 @@ async def test_rpc_reconnect_auth_error(
async def test_rpc_polling_auth_error(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC polling authentication error."""
register_entity(hass, SENSOR_DOMAIN, "test_name_rssi", "wifi-rssi")
@ -566,7 +555,7 @@ async def test_rpc_polling_auth_error(
assert entry.state == ConfigEntryState.LOADED
await mock_polling_rpc_update(hass)
await mock_polling_rpc_update(hass, freezer)
assert entry.state == ConfigEntryState.LOADED
@ -583,7 +572,7 @@ async def test_rpc_polling_auth_error(
async def test_rpc_reconnect_error(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC reconnect error."""
await init_integration(hass, 2)
@ -600,16 +589,15 @@ async def test_rpc_reconnect_error(
)
# Move time to generate reconnect
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=RPC_RECONNECT_INTERVAL)
)
freezer.tick(timedelta(seconds=RPC_RECONNECT_INTERVAL))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_test_switch_0").state == STATE_UNAVAILABLE
async def test_rpc_polling_connection_error(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC polling connection error."""
entity_id = register_entity(hass, SENSOR_DOMAIN, "test_name_rssi", "wifi-rssi")
@ -625,13 +613,13 @@ async def test_rpc_polling_connection_error(
assert hass.states.get(entity_id).state == "-63"
await mock_polling_rpc_update(hass)
await mock_polling_rpc_update(hass, freezer)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_rpc_polling_disconnected(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC polling device disconnected."""
entity_id = register_entity(hass, SENSOR_DOMAIN, "test_name_rssi", "wifi-rssi")
@ -641,6 +629,6 @@ async def test_rpc_polling_disconnected(
assert hass.states.get(entity_id).state == "-63"
await mock_polling_rpc_update(hass)
await mock_polling_rpc_update(hass, freezer)
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE

10
tests/components/shelly/test_sensor.py
View file

@ -1,4 +1,6 @@
"""Tests for Shelly sensor platform."""
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.const import (
@ -89,7 +91,7 @@ async def test_power_factory_without_unit_migration(
async def test_block_rest_sensor(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block REST sensor."""
entity_id = register_entity(hass, SENSOR_DOMAIN, "test_name_rssi", "rssi")
@ -98,7 +100,7 @@ async def test_block_rest_sensor(
assert hass.states.get(entity_id).state == "-64"
monkeypatch.setitem(mock_block_device.status["wifi_sta"], "rssi", -71)
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
assert hass.states.get(entity_id).state == "-71"
@ -304,7 +306,7 @@ async def test_rpc_sensor_error(
async def test_rpc_polling_sensor(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC polling sensor."""
entity_id = register_entity(hass, SENSOR_DOMAIN, "test_name_rssi", "wifi-rssi")
@ -313,7 +315,7 @@ async def test_rpc_polling_sensor(
assert hass.states.get(entity_id).state == "-63"
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "wifi", "rssi", "-70")
await mock_polling_rpc_update(hass)
await mock_polling_rpc_update(hass, freezer)
assert hass.states.get(entity_id).state == "-70"

17
tests/components/shelly/test_update.py
View file

@ -2,6 +2,7 @@
from unittest.mock import AsyncMock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.shelly.const import DOMAIN
@ -37,7 +38,7 @@ from tests.common import mock_restore_cache
async def test_block_update(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block device update entity."""
entity_registry = async_get(hass)
@ -75,7 +76,7 @@ async def test_block_update(
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "2")
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_OFF
@ -85,7 +86,7 @@ async def test_block_update(
async def test_block_beta_update(
hass: HomeAssistant, mock_block_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_block_device, monkeypatch
) -> None:
"""Test block device beta update entity."""
entity_registry = async_get(hass)
@ -108,7 +109,7 @@ async def test_block_beta_update(
assert state.attributes[ATTR_IN_PROGRESS] is False
monkeypatch.setitem(mock_block_device.status["update"], "beta_version", "2b")
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
@ -131,7 +132,7 @@ async def test_block_beta_update(
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "2b")
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF
@ -389,7 +390,7 @@ async def test_rpc_restored_sleeping_update_no_last_state(
async def test_rpc_beta_update(
hass: HomeAssistant, mock_rpc_device, monkeypatch
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_rpc_device, monkeypatch
) -> None:
"""Test RPC device beta update entity."""
entity_registry = async_get(hass)
@ -425,7 +426,7 @@ async def test_rpc_beta_update(
"beta": {"version": "2b"},
},
)
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
@ -448,7 +449,7 @@ async def test_rpc_beta_update(
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "2b")
await mock_rest_update(hass)
await mock_rest_update(hass, freezer)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF