Use freezegun in kraken tests (#99035)

This commit is contained in:
Erik Montnemery 2023-08-25 16:03:12 +02:00 committed by GitHub
parent cb8842b1fb
commit e6728f2f19
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -2,6 +2,7 @@
from datetime import timedelta from datetime import timedelta
from unittest.mock import patch from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
from pykrakenapi.pykrakenapi import KrakenAPIError from pykrakenapi.pykrakenapi import KrakenAPIError
from homeassistant.components.kraken.const import ( from homeassistant.components.kraken.const import (
@ -13,7 +14,6 @@ from homeassistant.components.kraken.const import (
from homeassistant.const import CONF_SCAN_INTERVAL, EVENT_HOMEASSISTANT_START from homeassistant.const import CONF_SCAN_INTERVAL, EVENT_HOMEASSISTANT_START
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
import homeassistant.util.dt as dt_util
from .const import ( from .const import (
MISSING_PAIR_TICKER_INFORMATION_RESPONSE, MISSING_PAIR_TICKER_INFORMATION_RESPONSE,
@ -25,11 +25,9 @@ from .const import (
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import MockConfigEntry, async_fire_time_changed
async def test_sensor(hass: HomeAssistant) -> None: async def test_sensor(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
"""Test that sensor has a value.""" """Test that sensor has a value."""
utcnow = dt_util.utcnow() with patch(
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"pykrakenapi.KrakenAPI.get_tradable_asset_pairs", "pykrakenapi.KrakenAPI.get_tradable_asset_pairs",
return_value=TRADEABLE_ASSET_PAIR_RESPONSE, return_value=TRADEABLE_ASSET_PAIR_RESPONSE,
), patch( ), patch(
@ -230,11 +228,11 @@ async def test_sensor(hass: HomeAssistant) -> None:
assert xbt_usd_opening_price_today.state == "0.0003513" assert xbt_usd_opening_price_today.state == "0.0003513"
async def test_sensors_available_after_restart(hass: HomeAssistant) -> None: async def test_sensors_available_after_restart(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test that all sensors are added again after a restart.""" """Test that all sensors are added again after a restart."""
utcnow = dt_util.utcnow() with patch(
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"pykrakenapi.KrakenAPI.get_tradable_asset_pairs", "pykrakenapi.KrakenAPI.get_tradable_asset_pairs",
return_value=TRADEABLE_ASSET_PAIR_RESPONSE, return_value=TRADEABLE_ASSET_PAIR_RESPONSE,
), patch( ), patch(
@ -271,11 +269,11 @@ async def test_sensors_available_after_restart(hass: HomeAssistant) -> None:
assert sensor.state == "0.0003494" assert sensor.state == "0.0003494"
async def test_sensors_added_after_config_update(hass: HomeAssistant) -> None: async def test_sensors_added_after_config_update(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test that sensors are added when another tracked asset pair is added.""" """Test that sensors are added when another tracked asset pair is added."""
utcnow = dt_util.utcnow() with patch(
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"pykrakenapi.KrakenAPI.get_tradable_asset_pairs", "pykrakenapi.KrakenAPI.get_tradable_asset_pairs",
return_value=TRADEABLE_ASSET_PAIR_RESPONSE, return_value=TRADEABLE_ASSET_PAIR_RESPONSE,
), patch( ), patch(
@ -309,19 +307,18 @@ async def test_sensors_added_after_config_update(hass: HomeAssistant) -> None:
CONF_TRACKED_ASSET_PAIRS: [DEFAULT_TRACKED_ASSET_PAIR, "ADA/XBT"], CONF_TRACKED_ASSET_PAIRS: [DEFAULT_TRACKED_ASSET_PAIR, "ADA/XBT"],
}, },
) )
async_fire_time_changed( freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL * 2))
hass, utcnow + timedelta(seconds=DEFAULT_SCAN_INTERVAL * 2) async_fire_time_changed(hass)
)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get("sensor.ada_xbt_ask") assert hass.states.get("sensor.ada_xbt_ask")
async def test_missing_pair_marks_sensor_unavailable(hass: HomeAssistant) -> None: async def test_missing_pair_marks_sensor_unavailable(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test that a missing tradable asset pair marks the sensor unavailable.""" """Test that a missing tradable asset pair marks the sensor unavailable."""
utcnow = dt_util.utcnow() with patch(
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"pykrakenapi.KrakenAPI.get_tradable_asset_pairs", "pykrakenapi.KrakenAPI.get_tradable_asset_pairs",
return_value=TRADEABLE_ASSET_PAIR_RESPONSE, return_value=TRADEABLE_ASSET_PAIR_RESPONSE,
) as tradeable_asset_pairs_mock, patch( ) as tradeable_asset_pairs_mock, patch(
@ -353,16 +350,14 @@ async def test_missing_pair_marks_sensor_unavailable(hass: HomeAssistant) -> Non
ticket_information_mock.side_effect = KrakenAPIError( ticket_information_mock.side_effect = KrakenAPIError(
"EQuery:Unknown asset pair" "EQuery:Unknown asset pair"
) )
async_fire_time_changed( freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL * 2))
hass, utcnow + timedelta(seconds=DEFAULT_SCAN_INTERVAL * 2) async_fire_time_changed(hass)
)
await hass.async_block_till_done() await hass.async_block_till_done()
ticket_information_mock.side_effect = None ticket_information_mock.side_effect = None
ticket_information_mock.return_value = MISSING_PAIR_TICKER_INFORMATION_RESPONSE ticket_information_mock.return_value = MISSING_PAIR_TICKER_INFORMATION_RESPONSE
async_fire_time_changed( freezer.tick(timedelta(seconds=DEFAULT_SCAN_INTERVAL * 2))
hass, utcnow + timedelta(seconds=DEFAULT_SCAN_INTERVAL * 2) async_fire_time_changed(hass)
)
await hass.async_block_till_done() await hass.async_block_till_done()
sensor = hass.states.get("sensor.xbt_usd_ask") sensor = hass.states.get("sensor.xbt_usd_ask")