forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_coordinator.py
51 lines (43 loc) · 1.75 KB
/
test_coordinator.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
"""Tests for the Tankerkoening integration."""
from __future__ import annotations
from datetime import timedelta
from unittest.mock import AsyncMock
from aiotankerkoenig.exceptions import TankerkoenigRateLimitError
import pytest
from homeassistant.components.tankerkoenig.const import DEFAULT_SCAN_INTERVAL
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.usefixtures("setup_integration")
async def test_rate_limit(
hass: HomeAssistant,
config_entry: MockConfigEntry,
tankerkoenig: AsyncMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test detection of API rate limit."""
assert config_entry.state == ConfigEntryState.LOADED
state = hass.states.get("binary_sensor.station_somewhere_street_1_status")
assert state
assert state.state == "on"
tankerkoenig.prices.side_effect = TankerkoenigRateLimitError
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(minutes=DEFAULT_SCAN_INTERVAL)
)
await hass.async_block_till_done()
assert (
"API rate limit reached, consider to increase polling interval" in caplog.text
)
state = hass.states.get("binary_sensor.station_somewhere_street_1_status")
assert state
assert state.state == STATE_UNAVAILABLE
tankerkoenig.prices.side_effect = None
async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(minutes=DEFAULT_SCAN_INTERVAL * 2)
)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.station_somewhere_street_1_status")
assert state
assert state.state == "on"