Use freezer in KNX tests (#123044)

use freezer in tests
pull/121786/head^2
Matthias Alphart 2024-08-02 08:48:41 +02:00 committed by GitHub
parent ed6d6575d7
commit 8ec8aef02e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 44 additions and 26 deletions

View File

@ -2,6 +2,8 @@
from datetime import timedelta from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.knx.const import CONF_STATE_ADDRESS, CONF_SYNC_STATE from homeassistant.components.knx.const import CONF_STATE_ADDRESS, CONF_SYNC_STATE
from homeassistant.components.knx.schema import BinarySensorSchema from homeassistant.components.knx.schema import BinarySensorSchema
from homeassistant.const import ( from homeassistant.const import (
@ -13,7 +15,6 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant, State from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from .conftest import KNXTestKit from .conftest import KNXTestKit
@ -141,9 +142,12 @@ async def test_binary_sensor_ignore_internal_state(
assert len(events) == 6 assert len(events) == 6
async def test_binary_sensor_counter(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_binary_sensor_counter(
hass: HomeAssistant,
knx: KNXTestKit,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test KNX binary_sensor with context timeout.""" """Test KNX binary_sensor with context timeout."""
async_fire_time_changed(hass, dt_util.utcnow())
context_timeout = 1 context_timeout = 1
await knx.setup_integration( await knx.setup_integration(
@ -167,7 +171,8 @@ async def test_binary_sensor_counter(hass: HomeAssistant, knx: KNXTestKit) -> No
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state is STATE_OFF assert state.state is STATE_OFF
assert state.attributes.get("counter") == 0 assert state.attributes.get("counter") == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=context_timeout)) freezer.tick(timedelta(seconds=context_timeout))
async_fire_time_changed(hass)
await knx.xknx.task_registry.block_till_done() await knx.xknx.task_registry.block_till_done()
# state changed twice after context timeout - once to ON with counter 1 and once to counter 0 # state changed twice after context timeout - once to ON with counter 1 and once to counter 0
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
@ -188,7 +193,8 @@ async def test_binary_sensor_counter(hass: HomeAssistant, knx: KNXTestKit) -> No
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state is STATE_ON assert state.state is STATE_ON
assert state.attributes.get("counter") == 0 assert state.attributes.get("counter") == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=context_timeout)) freezer.tick(timedelta(seconds=context_timeout))
async_fire_time_changed(hass)
await knx.xknx.task_registry.block_till_done() await knx.xknx.task_registry.block_till_done()
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state is STATE_ON assert state.state is STATE_ON
@ -202,10 +208,12 @@ async def test_binary_sensor_counter(hass: HomeAssistant, knx: KNXTestKit) -> No
assert event.get("old_state").attributes.get("counter") == 2 assert event.get("old_state").attributes.get("counter") == 2
async def test_binary_sensor_reset(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_binary_sensor_reset(
hass: HomeAssistant,
knx: KNXTestKit,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test KNX binary_sensor with reset_after function.""" """Test KNX binary_sensor with reset_after function."""
async_fire_time_changed(hass, dt_util.utcnow())
await knx.setup_integration( await knx.setup_integration(
{ {
BinarySensorSchema.PLATFORM: [ BinarySensorSchema.PLATFORM: [
@ -223,7 +231,8 @@ async def test_binary_sensor_reset(hass: HomeAssistant, knx: KNXTestKit) -> None
await knx.receive_write("2/2/2", True) await knx.receive_write("2/2/2", True)
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")
assert state.state is STATE_ON assert state.state is STATE_ON
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1)) freezer.tick(timedelta(seconds=1))
async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
# state reset after after timeout # state reset after after timeout
state = hass.states.get("binary_sensor.test") state = hass.states.get("binary_sensor.test")

View File

@ -3,20 +3,22 @@
from datetime import timedelta from datetime import timedelta
import logging import logging
from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
from homeassistant.components.knx.const import CONF_PAYLOAD_LENGTH, DOMAIN, KNX_ADDRESS from homeassistant.components.knx.const import CONF_PAYLOAD_LENGTH, DOMAIN, KNX_ADDRESS
from homeassistant.components.knx.schema import ButtonSchema from homeassistant.components.knx.schema import ButtonSchema
from homeassistant.const import CONF_NAME, CONF_PAYLOAD, CONF_TYPE from homeassistant.const import CONF_NAME, CONF_PAYLOAD, CONF_TYPE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from .conftest import KNXTestKit from .conftest import KNXTestKit
from tests.common import async_capture_events, async_fire_time_changed from tests.common import async_capture_events, async_fire_time_changed
async def test_button_simple(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_button_simple(
hass: HomeAssistant, knx: KNXTestKit, freezer: FrozenDateTimeFactory
) -> None:
"""Test KNX button with default payload.""" """Test KNX button with default payload."""
await knx.setup_integration( await knx.setup_integration(
{ {
@ -38,7 +40,8 @@ async def test_button_simple(hass: HomeAssistant, knx: KNXTestKit) -> None:
# received telegrams on button GA are ignored by the entity # received telegrams on button GA are ignored by the entity
old_state = hass.states.get("button.test") old_state = hass.states.get("button.test")
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=3)) freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await knx.receive_write("1/2/3", False) await knx.receive_write("1/2/3", False)
await knx.receive_write("1/2/3", True) await knx.receive_write("1/2/3", True)
new_state = hass.states.get("button.test") new_state = hass.states.get("button.test")

View File

@ -3,6 +3,7 @@
from datetime import timedelta from datetime import timedelta
from freezegun import freeze_time from freezegun import freeze_time
from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
from homeassistant.components.knx import CONF_KNX_EXPOSE, DOMAIN, KNX_ADDRESS from homeassistant.components.knx import CONF_KNX_EXPOSE, DOMAIN, KNX_ADDRESS
@ -14,11 +15,10 @@ from homeassistant.const import (
CONF_VALUE_TEMPLATE, CONF_VALUE_TEMPLATE,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from .conftest import KNXTestKit from .conftest import KNXTestKit
from tests.common import async_fire_time_changed_exact from tests.common import async_fire_time_changed
async def test_binary_expose(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_binary_expose(hass: HomeAssistant, knx: KNXTestKit) -> None:
@ -206,7 +206,9 @@ async def test_expose_string(hass: HomeAssistant, knx: KNXTestKit) -> None:
) )
async def test_expose_cooldown(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_expose_cooldown(
hass: HomeAssistant, knx: KNXTestKit, freezer: FrozenDateTimeFactory
) -> None:
"""Test an expose with cooldown.""" """Test an expose with cooldown."""
cooldown_time = 2 cooldown_time = 2
entity_id = "fake.entity" entity_id = "fake.entity"
@ -234,9 +236,8 @@ async def test_expose_cooldown(hass: HomeAssistant, knx: KNXTestKit) -> None:
await hass.async_block_till_done() await hass.async_block_till_done()
await knx.assert_no_telegram() await knx.assert_no_telegram()
# Wait for cooldown to pass # Wait for cooldown to pass
async_fire_time_changed_exact( freezer.tick(timedelta(seconds=cooldown_time))
hass, dt_util.utcnow() + timedelta(seconds=cooldown_time) async_fire_time_changed(hass)
)
await hass.async_block_till_done() await hass.async_block_till_done()
await knx.assert_write("1/1/8", (3,)) await knx.assert_write("1/1/8", (3,))

View File

@ -2,6 +2,7 @@
from unittest.mock import patch from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
from xknx.core import XknxConnectionState, XknxConnectionType from xknx.core import XknxConnectionState, XknxConnectionType
from xknx.telegram import IndividualAddress from xknx.telegram import IndividualAddress
@ -10,7 +11,6 @@ from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .conftest import KNXTestKit from .conftest import KNXTestKit
@ -19,7 +19,10 @@ from tests.typing import WebSocketGenerator
async def test_diagnostic_entities( async def test_diagnostic_entities(
hass: HomeAssistant, knx: KNXTestKit, entity_registry: er.EntityRegistry hass: HomeAssistant,
knx: KNXTestKit,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
) -> None: ) -> None:
"""Test diagnostic entities.""" """Test diagnostic entities."""
await knx.setup_integration({}) await knx.setup_integration({})
@ -50,7 +53,8 @@ async def test_diagnostic_entities(
knx.xknx.connection_manager.cemi_count_outgoing_error = 2 knx.xknx.connection_manager.cemi_count_outgoing_error = 2
events = async_capture_events(hass, "state_changed") events = async_capture_events(hass, "state_changed")
async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL) freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(events) == 3 # 5 polled sensors - 2 disabled assert len(events) == 3 # 5 polled sensors - 2 disabled

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from datetime import timedelta from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory
from xknx.core import XknxConnectionState from xknx.core import XknxConnectionState
from xknx.devices.light import Light as XknxLight from xknx.devices.light import Light as XknxLight
@ -19,7 +20,6 @@ from homeassistant.components.light import (
) )
from homeassistant.const import CONF_NAME, STATE_OFF, STATE_ON from homeassistant.const import CONF_NAME, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from . import KnxEntityGenerator from . import KnxEntityGenerator
from .conftest import KNXTestKit from .conftest import KNXTestKit
@ -643,7 +643,9 @@ async def test_light_rgb_individual(hass: HomeAssistant, knx: KNXTestKit) -> Non
await knx.assert_write(test_blue, (45,)) await knx.assert_write(test_blue, (45,))
async def test_light_rgbw_individual(hass: HomeAssistant, knx: KNXTestKit) -> None: async def test_light_rgbw_individual(
hass: HomeAssistant, knx: KNXTestKit, freezer: FrozenDateTimeFactory
) -> None:
"""Test KNX light with rgbw color in individual GAs.""" """Test KNX light with rgbw color in individual GAs."""
test_red = "1/1/3" test_red = "1/1/3"
test_red_state = "1/1/4" test_red_state = "1/1/4"
@ -763,9 +765,8 @@ async def test_light_rgbw_individual(hass: HomeAssistant, knx: KNXTestKit) -> No
await knx.receive_write(test_green, (0,)) await knx.receive_write(test_green, (0,))
# # individual color debounce takes 0.2 seconds if not all 4 addresses received # # individual color debounce takes 0.2 seconds if not all 4 addresses received
knx.assert_state("light.test", STATE_ON) knx.assert_state("light.test", STATE_ON)
async_fire_time_changed( freezer.tick(timedelta(seconds=XknxLight.DEBOUNCE_TIMEOUT))
hass, dt_util.utcnow() + timedelta(seconds=XknxLight.DEBOUNCE_TIMEOUT) async_fire_time_changed(hass)
)
await knx.xknx.task_registry.block_till_done() await knx.xknx.task_registry.block_till_done()
knx.assert_state("light.test", STATE_OFF) knx.assert_state("light.test", STATE_OFF)
# turn ON from KNX # turn ON from KNX