mirror of
https://github.com/home-assistant/core.git
synced 2026-09-13 12:10:22 +01:00
457 lines
14 KiB
Python
457 lines
14 KiB
Python
"""Test Home Assistant remote methods and classes."""
|
|
|
|
from collections.abc import Callable
|
|
import datetime
|
|
from functools import partial
|
|
import gc
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import time
|
|
import tracemalloc
|
|
from typing import Any, NamedTuple
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from homeassistant.core import Event, HomeAssistant, State
|
|
from homeassistant.helpers.json import (
|
|
ExtendedJSONEncoder,
|
|
JSONEncoder as DefaultHASSJSONEncoder,
|
|
cached_json_bytes,
|
|
cached_json_fragment,
|
|
cached_json_fragment_sorted,
|
|
find_paths_unserializable_data,
|
|
json_bytes,
|
|
json_bytes_sorted,
|
|
json_bytes_strip_null,
|
|
json_dumps,
|
|
json_dumps_sorted,
|
|
json_fragment,
|
|
save_json,
|
|
)
|
|
from homeassistant.util import dt as dt_util
|
|
from homeassistant.util.color import RGBColor
|
|
from homeassistant.util.json import SerializationError, load_json
|
|
|
|
from tests.common import json_round_trip
|
|
|
|
# Test data that can be saved as JSON
|
|
TEST_JSON_A = {"a": 1, "B": "two"}
|
|
TEST_JSON_B = {"a": "one", "B": 2}
|
|
|
|
|
|
@pytest.mark.parametrize("encoder", [DefaultHASSJSONEncoder, ExtendedJSONEncoder])
|
|
def test_json_encoder(hass: HomeAssistant, encoder: type[json.JSONEncoder]) -> None:
|
|
"""Test the JSON encoders."""
|
|
ha_json_enc = encoder()
|
|
state = State("test.test", "hello")
|
|
|
|
# Test serializing a datetime
|
|
now = dt_util.utcnow()
|
|
assert ha_json_enc.default(now) == now.isoformat()
|
|
|
|
# Test serializing a set()
|
|
data = {"milk", "beer"}
|
|
assert sorted(ha_json_enc.default(data)) == sorted(data)
|
|
|
|
# Test serializing an object which implements as_dict
|
|
default = ha_json_enc.default(state)
|
|
assert json_round_trip(default) == json_round_trip(state.as_dict())
|
|
|
|
|
|
def test_default_json_encoder(hass: HomeAssistant) -> None:
|
|
"""Test the default JSON encoder for date and time."""
|
|
ha_json_enc = DefaultHASSJSONEncoder()
|
|
|
|
today = datetime.date(2026, 8, 23)
|
|
assert ha_json_enc.default(today) == today.isoformat()
|
|
|
|
current_time = datetime.time(12, 0)
|
|
assert ha_json_enc.default(current_time) == current_time.isoformat()
|
|
|
|
|
|
def test_json_encoder_raises(hass: HomeAssistant) -> None:
|
|
"""Test the JSON encoder raises on unsupported types."""
|
|
ha_json_enc = DefaultHASSJSONEncoder()
|
|
|
|
# Default method raises TypeError if non HA object
|
|
with pytest.raises(TypeError):
|
|
ha_json_enc.default(1)
|
|
|
|
|
|
def test_extended_json_encoder(hass: HomeAssistant) -> None:
|
|
"""Test the extended JSON encoder."""
|
|
ha_json_enc = ExtendedJSONEncoder()
|
|
# Test serializing a timedelta
|
|
data = datetime.timedelta(
|
|
days=50,
|
|
seconds=27,
|
|
microseconds=10,
|
|
milliseconds=29000,
|
|
minutes=5,
|
|
hours=8,
|
|
weeks=2,
|
|
)
|
|
assert ha_json_enc.default(data) == {
|
|
"__type": str(type(data)),
|
|
"total_seconds": data.total_seconds(),
|
|
}
|
|
|
|
# Test serializing a time
|
|
o = datetime.time(7, 20)
|
|
assert ha_json_enc.default(o) == {"__type": str(type(o)), "isoformat": "07:20:00"}
|
|
|
|
# Test serializing a date
|
|
o = datetime.date(2021, 12, 24)
|
|
assert ha_json_enc.default(o) == {"__type": str(type(o)), "isoformat": "2021-12-24"}
|
|
|
|
# Default method falls back to repr(o)
|
|
o = object()
|
|
assert ha_json_enc.default(o) == {"__type": str(type(o)), "repr": repr(o)}
|
|
|
|
|
|
def test_json_dumps_sorted() -> None:
|
|
"""Test the json dumps sorted function."""
|
|
data = {"c": 3, "a": 1, "b": 2}
|
|
assert json_dumps_sorted(data) == json.dumps(
|
|
data, sort_keys=True, separators=(",", ":")
|
|
)
|
|
|
|
|
|
def test_json_bytes_sorted() -> None:
|
|
"""Test the json bytes sorted function."""
|
|
data = {"c": 3, "a": 1, "b": 2}
|
|
assert json_bytes_sorted(data) == json.dumps(
|
|
data, sort_keys=True, separators=(",", ":")
|
|
).encode("utf-8")
|
|
|
|
|
|
def test_json_dumps_float_subclass() -> None:
|
|
"""Test the json dumps a float subclass."""
|
|
|
|
class FloatSubclass(float):
|
|
"""A float subclass."""
|
|
|
|
assert json_dumps({"c": FloatSubclass(1.2)}) == '{"c":1.2}'
|
|
|
|
|
|
def test_json_dumps_tuple_subclass() -> None:
|
|
"""Test the json dumps a tuple subclass."""
|
|
|
|
tt = time.struct_time((1999, 3, 17, 32, 44, 55, 2, 76, 0))
|
|
|
|
assert json_dumps(tt) == "[1999,3,17,32,44,55,2,76,0]"
|
|
|
|
|
|
def test_json_dumps_named_tuple_subclass() -> None:
|
|
"""Test the json dumps a tuple subclass."""
|
|
|
|
class NamedTupleSubclass(NamedTuple):
|
|
"""A NamedTuple subclass."""
|
|
|
|
name: str
|
|
|
|
nts = NamedTupleSubclass("a")
|
|
|
|
assert json_dumps(nts) == '["a"]'
|
|
|
|
|
|
def test_json_dumps_rgb_color_subclass() -> None:
|
|
"""Test the json dumps of RGBColor."""
|
|
rgb = RGBColor(4, 2, 1)
|
|
|
|
assert json_dumps(rgb) == "[4,2,1]"
|
|
|
|
|
|
def test_json_dumps_date_time_subclasses() -> None:
|
|
"""Test the json dumps with date and time subclasses."""
|
|
|
|
class CustomDate(datetime.date):
|
|
"""Custom date subclass."""
|
|
|
|
class CustomTime(datetime.time):
|
|
"""Custom time subclass."""
|
|
|
|
class CustomDatetime(datetime.datetime):
|
|
"""Custom datetime subclass."""
|
|
|
|
d = CustomDate(2026, 8, 23)
|
|
t = CustomTime(12, 30, 45)
|
|
dt = CustomDatetime(2026, 8, 23, 12, 30, 45)
|
|
|
|
assert json_dumps({"date": d, "time": t, "datetime": dt}) == (
|
|
'{"date":"2026-08-23","time":"12:30:45","datetime":"2026-08-23T12:30:45"}'
|
|
)
|
|
|
|
|
|
def test_json_fragments() -> None:
|
|
"""Test the json dumps with a fragment."""
|
|
|
|
assert (
|
|
json_dumps(
|
|
[
|
|
json_fragment('{"inner":"fragment2"}'),
|
|
json_fragment('{"inner":"fragment2"}'),
|
|
]
|
|
)
|
|
== '[{"inner":"fragment2"},{"inner":"fragment2"}]'
|
|
)
|
|
|
|
class Fragment1:
|
|
@property
|
|
def json_fragment(self):
|
|
return json_fragment('{"inner":"fragment1"}')
|
|
|
|
class Fragment2:
|
|
@property
|
|
def json_fragment(self):
|
|
return json_fragment('{"inner":"fragment2"}')
|
|
|
|
assert (
|
|
json_dumps([Fragment1(), Fragment2()])
|
|
== '[{"inner":"fragment1"},{"inner":"fragment2"}]'
|
|
)
|
|
|
|
|
|
def test_cached_json_fragment() -> None:
|
|
"""Test cached_json_fragment serializes identically to a plain fragment."""
|
|
data = {"a": 1, "b": [1, 2, 3], "c": {"nested": True}, "d": None}
|
|
|
|
fragment = cached_json_fragment(data)
|
|
assert isinstance(fragment, json_fragment)
|
|
assert json_dumps([fragment]) == json_dumps([json_fragment(json_bytes(data))])
|
|
assert (
|
|
json_dumps([fragment]) == '[{"a":1,"b":[1,2,3],"c":{"nested":true},"d":null}]'
|
|
)
|
|
|
|
|
|
def test_cached_json_bytes() -> None:
|
|
"""Test cached_json_bytes serializes identically to json_bytes."""
|
|
data = {"a": 1, "b": [1, 2, 3], "c": {"nested": True}, "d": None}
|
|
|
|
assert cached_json_bytes(data) == json_bytes(data)
|
|
assert (
|
|
cached_json_bytes(data) == b'{"a":1,"b":[1,2,3],"c":{"nested":true},"d":null}'
|
|
)
|
|
|
|
|
|
def test_cached_json_fragment_sorted() -> None:
|
|
"""Test cached_json_fragment_sorted serializes with sorted keys."""
|
|
data = {"c": 3, "a": 1, "b": 2}
|
|
|
|
fragment = cached_json_fragment_sorted(data)
|
|
assert isinstance(fragment, json_fragment)
|
|
assert json_dumps([fragment]) == '[{"a":1,"b":2,"c":3}]'
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"cached_serializer",
|
|
[cached_json_bytes, cached_json_fragment, cached_json_fragment_sorted],
|
|
ids=["cached_json_bytes", "cached_json_fragment", "cached_json_fragment_sorted"],
|
|
)
|
|
def test_cached_json_helpers_trim_buffer(
|
|
cached_serializer: Callable[[Any], object],
|
|
) -> None:
|
|
"""Test the cached_json_* helpers cache right-sized bytes, not orjson's slack.
|
|
|
|
orjson.dumps returns bytes whose backing buffer is rounded up to a power of
|
|
two and not shrunk; the helpers copy them to a right-sized buffer. Without
|
|
that copy the cached value would retain the full over-allocated buffer
|
|
(several KiB even for a small payload), which is the memory regression this
|
|
guards against.
|
|
|
|
The waste is invisible to normal object inspection: sys.getsizeof() reports
|
|
the logical length, not the backing buffer, and orjson.Fragment exposes no way
|
|
to reach the bytes it wraps, so the retained allocation can only be observed
|
|
via tracemalloc.
|
|
"""
|
|
data = {f"key_{index}": "value" * 5 for index in range(40)}
|
|
serialized_size = len(json_bytes(data))
|
|
|
|
tracemalloc.start()
|
|
try:
|
|
# clear_traces resets the baseline to zero so pre-existing garbage from
|
|
# the test session is not counted; the transient over-allocated buffer is
|
|
# freed by refcounting before get_traced_memory, leaving only `cached`.
|
|
gc.collect()
|
|
tracemalloc.clear_traces()
|
|
cached = cached_serializer(data)
|
|
retained, _ = tracemalloc.get_traced_memory()
|
|
finally:
|
|
tracemalloc.stop()
|
|
|
|
assert cached is not None # keep alive until measured
|
|
# The cache holds ~the serialized size; without the copy it would hold
|
|
# orjson's oversized power-of-two buffer, which is far larger.
|
|
assert retained < serialized_size * 1.5
|
|
|
|
|
|
def test_json_bytes_strip_null() -> None:
|
|
"""Test stripping nul from strings."""
|
|
|
|
assert json_bytes_strip_null("\0") == b'""'
|
|
assert json_bytes_strip_null("silly\0stuff") == b'"silly"'
|
|
assert json_bytes_strip_null(["one", "two\0", "three"]) == b'["one","two","three"]'
|
|
assert (
|
|
json_bytes_strip_null({"k1": "one", "k2": "two\0", "k3": "three"})
|
|
== b'{"k1":"one","k2":"two","k3":"three"}'
|
|
)
|
|
assert (
|
|
json_bytes_strip_null([[{"k1": {"k2": ["silly\0stuff"]}}]])
|
|
== b'[[{"k1":{"k2":["silly"]}}]]'
|
|
)
|
|
|
|
|
|
def test_save_and_load(tmp_path: Path) -> None:
|
|
"""Test saving and loading back."""
|
|
fname = tmp_path / "test1.json"
|
|
save_json(fname, TEST_JSON_A)
|
|
data = load_json(fname)
|
|
assert data == TEST_JSON_A
|
|
|
|
|
|
def test_save_and_load_int_keys(tmp_path: Path) -> None:
|
|
"""Test saving and loading back stringifies the keys."""
|
|
fname = tmp_path / "test1.json"
|
|
save_json(fname, {1: "a", 2: "b"})
|
|
data = load_json(fname)
|
|
assert data == {"1": "a", "2": "b"}
|
|
|
|
|
|
def test_save_and_load_private(tmp_path: Path) -> None:
|
|
"""Test we can load private files and that they are protected."""
|
|
fname = tmp_path / "test2.json"
|
|
save_json(fname, TEST_JSON_A, private=True)
|
|
data = load_json(fname)
|
|
assert data == TEST_JSON_A
|
|
stats = os.stat(fname)
|
|
assert stats.st_mode & 0o77 == 0
|
|
|
|
|
|
@pytest.mark.parametrize("atomic_writes", [True, False])
|
|
def test_overwrite_and_reload(atomic_writes: bool, tmp_path: Path) -> None:
|
|
"""Test that we can overwrite an existing file and read back."""
|
|
fname = tmp_path / "test3.json"
|
|
save_json(fname, TEST_JSON_A, atomic_writes=atomic_writes)
|
|
save_json(fname, TEST_JSON_B, atomic_writes=atomic_writes)
|
|
data = load_json(fname)
|
|
assert data == TEST_JSON_B
|
|
|
|
|
|
def test_save_bad_data() -> None:
|
|
"""Test error from trying to save unserializable data."""
|
|
|
|
class CannotSerializeMe:
|
|
"""Cannot serialize this."""
|
|
|
|
with pytest.raises(SerializationError) as excinfo:
|
|
save_json("test4", {"hello": CannotSerializeMe()})
|
|
|
|
assert "Bad data at $.hello=" in str(excinfo.value)
|
|
|
|
|
|
def test_custom_encoder(tmp_path: Path) -> None:
|
|
"""Test serializing with a custom encoder."""
|
|
|
|
class MockJSONEncoder(json.JSONEncoder):
|
|
"""Mock JSON encoder."""
|
|
|
|
def default(self, o):
|
|
"""Mock JSON encode method."""
|
|
return "9"
|
|
|
|
fname = tmp_path / "test6.json"
|
|
save_json(fname, Mock(), encoder=MockJSONEncoder)
|
|
data = load_json(fname)
|
|
assert data == "9"
|
|
|
|
|
|
def test_saving_subclassed_datetime(tmp_path: Path) -> None:
|
|
"""Test saving subclassed datetime objects."""
|
|
|
|
class SubClassDateTime(datetime.datetime):
|
|
"""Subclass datetime."""
|
|
|
|
time = SubClassDateTime.fromtimestamp(0)
|
|
|
|
fname = tmp_path / "test6.json"
|
|
save_json(fname, {"time": time})
|
|
data = load_json(fname)
|
|
assert data == {"time": time.isoformat()}
|
|
|
|
|
|
def test_default_encoder_is_passed(tmp_path: Path) -> None:
|
|
"""Test we use orjson if they pass in the default encoder."""
|
|
fname = tmp_path / "test6.json"
|
|
with patch(
|
|
"homeassistant.helpers.json.orjson.dumps", return_value=b"{}"
|
|
) as mock_orjson_dumps:
|
|
save_json(fname, {"any": 1}, encoder=DefaultHASSJSONEncoder)
|
|
assert len(mock_orjson_dumps.mock_calls) == 1
|
|
# Patch json.dumps to make sure we are using the orjson path
|
|
with patch("homeassistant.helpers.json.json.dumps", side_effect=Exception):
|
|
save_json(fname, {"any": {1}}, encoder=DefaultHASSJSONEncoder)
|
|
data = load_json(fname)
|
|
assert data == {"any": [1]}
|
|
|
|
|
|
def test_find_unserializable_data() -> None:
|
|
"""Find unserializeable data."""
|
|
assert find_paths_unserializable_data(1) == {}
|
|
assert find_paths_unserializable_data([1, 2]) == {}
|
|
assert find_paths_unserializable_data({"something": "yo"}) == {}
|
|
|
|
assert find_paths_unserializable_data({"something": set()}) == {
|
|
"$.something": set()
|
|
}
|
|
assert find_paths_unserializable_data({"something": [1, set()]}) == {
|
|
"$.something[1]": set()
|
|
}
|
|
assert find_paths_unserializable_data([1, {"bla": set(), "blub": set()}]) == {
|
|
"$[1].bla": set(),
|
|
"$[1].blub": set(),
|
|
}
|
|
assert find_paths_unserializable_data({("A",): 1}) == {"$<key: ('A',)>": ("A",)}
|
|
assert math.isnan(
|
|
find_paths_unserializable_data(
|
|
math.nan, dump=partial(json.dumps, allow_nan=False)
|
|
)["$"]
|
|
)
|
|
|
|
# Test custom encoder + State support.
|
|
|
|
class MockJSONEncoder(json.JSONEncoder):
|
|
"""Mock JSON encoder."""
|
|
|
|
def default(self, o):
|
|
"""Mock JSON encode method."""
|
|
if isinstance(o, datetime.datetime):
|
|
return o.isoformat()
|
|
return super().default(o)
|
|
|
|
bad_data = object()
|
|
|
|
assert find_paths_unserializable_data(
|
|
[State("mock_domain.mock_entity", "on", {"bad": bad_data})],
|
|
dump=partial(json.dumps, cls=MockJSONEncoder),
|
|
) == {"$[0](State: mock_domain.mock_entity).attributes.bad": bad_data}
|
|
|
|
assert find_paths_unserializable_data(
|
|
[Event("bad_event", {"bad_attribute": bad_data})],
|
|
dump=partial(json.dumps, cls=MockJSONEncoder),
|
|
) == {"$[0](Event: bad_event).data.bad_attribute": bad_data}
|
|
|
|
class BadData:
|
|
def __init__(self) -> None:
|
|
self.bla = bad_data
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {"bla": self.bla}
|
|
|
|
assert find_paths_unserializable_data(
|
|
BadData(),
|
|
dump=partial(json.dumps, cls=MockJSONEncoder),
|
|
) == {"$(BadData).bla": bad_data}
|