starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Clean up Evohome's storage tests (#175660)

Browse Source
This commit is contained in:
David Bonnes
2026-07-08 22:41:20 +02:00
committed by GitHub
parent fe7346052b
commit 944233376c
1 changed files with 39 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/evohome/test_storage.py
+39 -52
View File
@@ -1,8 +1,9 @@
"""The tests for evohome storage load & save."""
from datetime import datetime, timedelta
from typing import Any, Final, NotRequired, TypedDict
from typing import Any, Final, TypedDict
from evohomeasync.auth import SZ_SESSION_ID, SZ_SESSION_ID_EXPIRES
from evohomeasync2.auth import (
SZ_ACCESS_TOKEN,
SZ_ACCESS_TOKEN_EXPIRES,
@@ -11,6 +12,7 @@ from evohomeasync2.auth import (
import pytest
from homeassistant.components.evohome.const import DOMAIN, STORAGE_KEY, STORAGE_VER
from homeassistant.components.evohome.storage import _TokenStoreT
from homeassistant.const import CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
@@ -19,26 +21,10 @@ from .conftest import setup_evohome
from .const import ACCESS_TOKEN, REFRESH_TOKEN, SESSION_ID, USERNAME
class _SessionDataT(TypedDict):
session_id: str
session_id_expires: NotRequired[str] # 2024-07-27T23:57:30+01:00
class _TokenStoreT(TypedDict):
username: str
refresh_token: str
access_token: str
access_token_expires: str # 2024-07-27T23:57:30+01:00
user_data: NotRequired[_SessionDataT]
class _EmptyStoreT(TypedDict):
pass
SZ_USER_DATA: Final = "user_data"
def dt_pair(dt_dtm: datetime) -> tuple[datetime, str]:
"""Return a datetime without milliseconds and its string representation."""
dt_str = dt_dtm.isoformat(timespec="seconds") # e.g. 2024-07-28T00:57:29+01:00
@@ -46,12 +32,10 @@ def dt_pair(dt_dtm: datetime) -> tuple[datetime, str]:
ACCESS_TOKEN_EXP_DTM, ACCESS_TOKEN_EXP_STR = dt_pair(dt_util.now() + timedelta(hours=1))
USERNAME_DIFF: Final = f"not_{USERNAME}"
USERNAME_SAME: Final = USERNAME
_, SESSION_ID_EXP_STR = dt_pair(dt_util.now() + timedelta(minutes=15))
_TEST_STORAGE_BASE: Final[_TokenStoreT] = {
CONF_USERNAME: USERNAME_SAME,
CONF_USERNAME: USERNAME,
SZ_REFRESH_TOKEN: REFRESH_TOKEN,
SZ_ACCESS_TOKEN: ACCESS_TOKEN,
SZ_ACCESS_TOKEN_EXPIRES: ACCESS_TOKEN_EXP_STR,
@@ -59,8 +43,11 @@ _TEST_STORAGE_BASE: Final[_TokenStoreT] = {
TEST_STORAGE_DATA: Final[dict[str, _TokenStoreT]] = {
"sans_session_id": _TEST_STORAGE_BASE,
"null_session_id": _TEST_STORAGE_BASE | {SZ_USER_DATA: None}, # type: ignore[dict-item]
"with_session_id": _TEST_STORAGE_BASE | {SZ_USER_DATA: {"session_id": SESSION_ID}},
"with_session_id": _TEST_STORAGE_BASE
| {
SZ_SESSION_ID: SESSION_ID,
SZ_SESSION_ID_EXPIRES: SESSION_ID_EXP_STR,
}, # pyright: ignore[reportAssignmentType]
}
TEST_STORAGE_NULL: Final[dict[str, _EmptyStoreT | None]] = {
@@ -68,7 +55,7 @@ TEST_STORAGE_NULL: Final[dict[str, _EmptyStoreT | None]] = {
"store_was_reset": {},
}
DOMAIN_STORAGE_BASE: Final = {
DOMAIN_STORAGE_ROOT: Final = {
"version": STORAGE_VER,
"minor_version": 1,
"key": STORAGE_KEY,
@@ -86,21 +73,20 @@ async def test_auth_tokens_null(
) -> None:
"""Test credentials manager when cache is empty."""
hass_storage[DOMAIN] = DOMAIN_STORAGE_BASE | {"data": TEST_STORAGE_NULL[idx]}
hass_storage[DOMAIN] = DOMAIN_STORAGE_ROOT | {"data": TEST_STORAGE_NULL[idx]}
async for _ in setup_evohome(hass, config, install=install):
pass
# Confirm the expected tokens were cached to storage...
data: _TokenStoreT = hass_storage[DOMAIN]["data"]
assert data[CONF_USERNAME] == USERNAME_SAME
# Confirm the expected tokens were cached to storage...
assert data[CONF_USERNAME] == USERNAME
assert data[SZ_REFRESH_TOKEN] == f"new_{REFRESH_TOKEN}"
assert data[SZ_ACCESS_TOKEN] == f"new_{ACCESS_TOKEN}"
assert (
dt_util.parse_datetime(data[SZ_ACCESS_TOKEN_EXPIRES], raise_on_error=True)
> dt_util.now()
)
assert (expires := data.get(SZ_ACCESS_TOKEN_EXPIRES)) is not None
assert dt_util.parse_datetime(expires, raise_on_error=True) > dt_util.now()
@pytest.mark.parametrize("install", ["minimal"])
@@ -114,18 +100,20 @@ async def test_auth_tokens_same(
) -> None:
"""Test credentials manager when cache contains valid data for this user."""
hass_storage[DOMAIN] = DOMAIN_STORAGE_BASE | {"data": TEST_STORAGE_DATA[idx]}
hass_storage[DOMAIN] = DOMAIN_STORAGE_ROOT | {"data": TEST_STORAGE_DATA[idx]}
async for _ in setup_evohome(hass, config, install=install):
pass
# Confirm the expected tokens were cached to storage...
data: _TokenStoreT = hass_storage[DOMAIN]["data"]
assert data[CONF_USERNAME] == USERNAME_SAME
# Confirm the expected tokens were cached to storage...
assert data[CONF_USERNAME] == USERNAME
assert data[SZ_REFRESH_TOKEN] == REFRESH_TOKEN
assert data[SZ_ACCESS_TOKEN] == ACCESS_TOKEN
assert dt_util.parse_datetime(data[SZ_ACCESS_TOKEN_EXPIRES]) == ACCESS_TOKEN_EXP_DTM
assert (expires := data[SZ_ACCESS_TOKEN_EXPIRES]) is not None
assert dt_util.parse_datetime(expires, raise_on_error=True) == ACCESS_TOKEN_EXP_DTM
@pytest.mark.parametrize("install", ["minimal"])
@@ -139,27 +127,26 @@ async def test_auth_tokens_past(
) -> None:
"""Test credentials manager when cache contains expired data for this user."""
_dt_dtm, dt_str = dt_pair(dt_util.now() - timedelta(hours=1))
# Make this access token have expired in the past...
_, dt_str = dt_pair(dt_util.now() - timedelta(hours=1))
# make this access token have expired in the past...
test_data = TEST_STORAGE_DATA[idx].copy() # shallow copy is OK here
test_data[SZ_ACCESS_TOKEN_EXPIRES] = dt_str
hass_storage[DOMAIN] = DOMAIN_STORAGE_BASE | {"data": test_data}
hass_storage[DOMAIN] = DOMAIN_STORAGE_ROOT | {"data": test_data}
async for _ in setup_evohome(hass, config, install=install):
pass
# Confirm the expected tokens were cached to storage...
data: _TokenStoreT = hass_storage[DOMAIN]["data"]
assert data[CONF_USERNAME] == USERNAME_SAME
# Confirm the expected tokens were cached to storage...
assert data[CONF_USERNAME] == USERNAME
assert data[SZ_REFRESH_TOKEN] == f"new_{REFRESH_TOKEN}"
assert data[SZ_ACCESS_TOKEN] == f"new_{ACCESS_TOKEN}"
assert (
dt_util.parse_datetime(data[SZ_ACCESS_TOKEN_EXPIRES], raise_on_error=True)
> dt_util.now()
)
assert (expires := data[SZ_ACCESS_TOKEN_EXPIRES]) is not None
assert dt_util.parse_datetime(expires, raise_on_error=True) > dt_util.now()
@pytest.mark.parametrize("install", ["minimal"])
@@ -173,19 +160,19 @@ async def test_auth_tokens_diff(
) -> None:
"""Test credentials manager when cache contains data for a different user."""
hass_storage[DOMAIN] = DOMAIN_STORAGE_BASE | {"data": TEST_STORAGE_DATA[idx]}
config["username"] = USERNAME_DIFF
# Make this access token be for a different user...
hass_storage[DOMAIN] = DOMAIN_STORAGE_ROOT | {"data": TEST_STORAGE_DATA[idx]}
config[CONF_USERNAME] = f"new_{USERNAME}"
async for _ in setup_evohome(hass, config, install=install):
pass
# Confirm the expected tokens were cached to storage...
data: _TokenStoreT = hass_storage[DOMAIN]["data"]
assert data[CONF_USERNAME] == USERNAME_DIFF
# Confirm the expected tokens were cached to storage...
assert data[CONF_USERNAME] == f"new_{USERNAME}"
assert data[SZ_REFRESH_TOKEN] == f"new_{REFRESH_TOKEN}"
assert data[SZ_ACCESS_TOKEN] == f"new_{ACCESS_TOKEN}"
assert (
dt_util.parse_datetime(data[SZ_ACCESS_TOKEN_EXPIRES], raise_on_error=True)
> dt_util.now()
)
assert (expires := data[SZ_ACCESS_TOKEN_EXPIRES]) is not None
assert dt_util.parse_datetime(expires, raise_on_error=True) > dt_util.now()