mirror of
https://github.com/home-assistant/core.git
synced 2025-07-25 14:17:45 +00:00
Refactor json tests to align with new code (#88247)
* Refactor json tests to align with new code * Use tmp_path
This commit is contained in:
parent
dc30210237
commit
3873484849
@ -1,25 +1,37 @@
|
|||||||
"""Test Home Assistant remote methods and classes."""
|
"""Test Home Assistant remote methods and classes."""
|
||||||
import datetime
|
import datetime
|
||||||
|
from functools import partial
|
||||||
import json
|
import json
|
||||||
|
import math
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
import time
|
import time
|
||||||
from typing import NamedTuple
|
from typing import NamedTuple
|
||||||
|
from unittest.mock import Mock, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from homeassistant.core import HomeAssistant, State
|
from homeassistant.core import Event, HomeAssistant, State
|
||||||
from homeassistant.helpers.json import (
|
from homeassistant.helpers.json import (
|
||||||
ExtendedJSONEncoder,
|
ExtendedJSONEncoder,
|
||||||
JSONEncoder,
|
JSONEncoder as DefaultHASSJSONEncoder,
|
||||||
|
find_paths_unserializable_data,
|
||||||
json_bytes_strip_null,
|
json_bytes_strip_null,
|
||||||
json_dumps,
|
json_dumps,
|
||||||
json_dumps_sorted,
|
json_dumps_sorted,
|
||||||
|
save_json,
|
||||||
)
|
)
|
||||||
from homeassistant.util import dt as dt_util
|
from homeassistant.util import dt as dt_util
|
||||||
from homeassistant.util.color import RGBColor
|
from homeassistant.util.color import RGBColor
|
||||||
|
from homeassistant.util.json import SerializationError, load_json
|
||||||
|
|
||||||
|
# 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", (JSONEncoder, ExtendedJSONEncoder))
|
@pytest.mark.parametrize("encoder", (DefaultHASSJSONEncoder, ExtendedJSONEncoder))
|
||||||
def test_json_encoder(hass, encoder):
|
def test_json_encoder(hass: HomeAssistant, encoder: type[json.JSONEncoder]) -> None:
|
||||||
"""Test the JSON encoders."""
|
"""Test the JSON encoders."""
|
||||||
ha_json_enc = encoder()
|
ha_json_enc = encoder()
|
||||||
state = State("test.test", "hello")
|
state = State("test.test", "hello")
|
||||||
@ -38,7 +50,7 @@ def test_json_encoder(hass, encoder):
|
|||||||
|
|
||||||
def test_json_encoder_raises(hass: HomeAssistant) -> None:
|
def test_json_encoder_raises(hass: HomeAssistant) -> None:
|
||||||
"""Test the JSON encoder raises on unsupported types."""
|
"""Test the JSON encoder raises on unsupported types."""
|
||||||
ha_json_enc = JSONEncoder()
|
ha_json_enc = DefaultHASSJSONEncoder()
|
||||||
|
|
||||||
# Default method raises TypeError if non HA object
|
# Default method raises TypeError if non HA object
|
||||||
with pytest.raises(TypeError):
|
with pytest.raises(TypeError):
|
||||||
@ -135,3 +147,143 @@ def test_json_bytes_strip_null() -> None:
|
|||||||
json_bytes_strip_null([[{"k1": {"k2": ["silly\0stuff"]}}]])
|
json_bytes_strip_null([[{"k1": {"k2": ["silly\0stuff"]}}]])
|
||||||
== b'[[{"k1":{"k2":["silly"]}}]]'
|
== 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 "Failed to serialize to JSON: test4. 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_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(
|
||||||
|
float("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):
|
||||||
|
self.bla = bad_data
|
||||||
|
|
||||||
|
def as_dict(self):
|
||||||
|
return {"bla": self.bla}
|
||||||
|
|
||||||
|
assert find_paths_unserializable_data(
|
||||||
|
BadData(),
|
||||||
|
dump=partial(json.dumps, cls=MockJSONEncoder),
|
||||||
|
) == {"$(BadData).bla": bad_data}
|
||||||
|
@ -1,200 +1,26 @@
|
|||||||
"""Test Home Assistant json utility functions."""
|
"""Test Home Assistant json utility functions."""
|
||||||
from datetime import datetime
|
from pathlib import Path
|
||||||
from functools import partial
|
|
||||||
from json import JSONEncoder, dumps
|
|
||||||
import math
|
|
||||||
import os
|
|
||||||
from tempfile import mkdtemp
|
|
||||||
from unittest.mock import Mock, patch
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from homeassistant.core import Event, State
|
|
||||||
from homeassistant.exceptions import HomeAssistantError
|
from homeassistant.exceptions import HomeAssistantError
|
||||||
from homeassistant.helpers.json import JSONEncoder as DefaultHASSJSONEncoder
|
from homeassistant.util.json import json_loads_array, json_loads_object, load_json
|
||||||
from homeassistant.util.json import (
|
|
||||||
SerializationError,
|
|
||||||
find_paths_unserializable_data,
|
|
||||||
json_loads_array,
|
|
||||||
json_loads_object,
|
|
||||||
load_json,
|
|
||||||
save_json,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Test data that can be saved as JSON
|
# Test data that can be saved as JSON
|
||||||
TEST_JSON_A = {"a": 1, "B": "two"}
|
TEST_JSON_A = {"a": 1, "B": "two"}
|
||||||
TEST_JSON_B = {"a": "one", "B": 2}
|
|
||||||
# Test data that cannot be loaded as JSON
|
# Test data that cannot be loaded as JSON
|
||||||
TEST_BAD_SERIALIED = "THIS IS NOT JSON\n"
|
TEST_BAD_SERIALIED = "THIS IS NOT JSON\n"
|
||||||
TMP_DIR = None
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
def test_load_bad_data(tmp_path: Path) -> None:
|
||||||
def setup_and_teardown():
|
|
||||||
"""Clean up after tests."""
|
|
||||||
global TMP_DIR
|
|
||||||
TMP_DIR = mkdtemp()
|
|
||||||
|
|
||||||
yield
|
|
||||||
|
|
||||||
for fname in os.listdir(TMP_DIR):
|
|
||||||
os.remove(os.path.join(TMP_DIR, fname))
|
|
||||||
os.rmdir(TMP_DIR)
|
|
||||||
|
|
||||||
|
|
||||||
def _path_for(leaf_name):
|
|
||||||
return os.path.join(TMP_DIR, f"{leaf_name}.json")
|
|
||||||
|
|
||||||
|
|
||||||
def test_save_and_load() -> None:
|
|
||||||
"""Test saving and loading back."""
|
|
||||||
fname = _path_for("test1")
|
|
||||||
save_json(fname, TEST_JSON_A)
|
|
||||||
data = load_json(fname)
|
|
||||||
assert data == TEST_JSON_A
|
|
||||||
|
|
||||||
|
|
||||||
def test_save_and_load_int_keys() -> None:
|
|
||||||
"""Test saving and loading back stringifies the keys."""
|
|
||||||
fname = _path_for("test1")
|
|
||||||
save_json(fname, {1: "a", 2: "b"})
|
|
||||||
data = load_json(fname)
|
|
||||||
assert data == {"1": "a", "2": "b"}
|
|
||||||
|
|
||||||
|
|
||||||
def test_save_and_load_private() -> None:
|
|
||||||
"""Test we can load private files and that they are protected."""
|
|
||||||
fname = _path_for("test2")
|
|
||||||
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):
|
|
||||||
"""Test that we can overwrite an existing file and read back."""
|
|
||||||
fname = _path_for("test3")
|
|
||||||
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 "Failed to serialize to JSON: test4. Bad data at $.hello=" in str(
|
|
||||||
excinfo.value
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_load_bad_data() -> None:
|
|
||||||
"""Test error from trying to load unserialisable data."""
|
"""Test error from trying to load unserialisable data."""
|
||||||
fname = _path_for("test5")
|
fname = tmp_path / "test5.json"
|
||||||
with open(fname, "w") as fh:
|
with open(fname, "w") as fh:
|
||||||
fh.write(TEST_BAD_SERIALIED)
|
fh.write(TEST_BAD_SERIALIED)
|
||||||
with pytest.raises(HomeAssistantError):
|
with pytest.raises(HomeAssistantError):
|
||||||
load_json(fname)
|
load_json(fname)
|
||||||
|
|
||||||
|
|
||||||
def test_custom_encoder() -> None:
|
|
||||||
"""Test serializing with a custom encoder."""
|
|
||||||
|
|
||||||
class MockJSONEncoder(JSONEncoder):
|
|
||||||
"""Mock JSON encoder."""
|
|
||||||
|
|
||||||
def default(self, o):
|
|
||||||
"""Mock JSON encode method."""
|
|
||||||
return "9"
|
|
||||||
|
|
||||||
fname = _path_for("test6")
|
|
||||||
save_json(fname, Mock(), encoder=MockJSONEncoder)
|
|
||||||
data = load_json(fname)
|
|
||||||
assert data == "9"
|
|
||||||
|
|
||||||
|
|
||||||
def test_default_encoder_is_passed() -> None:
|
|
||||||
"""Test we use orjson if they pass in the default encoder."""
|
|
||||||
fname = _path_for("test6")
|
|
||||||
with patch(
|
|
||||||
"homeassistant.util.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.util.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(
|
|
||||||
float("nan"), dump=partial(dumps, allow_nan=False)
|
|
||||||
)["$"]
|
|
||||||
)
|
|
||||||
|
|
||||||
# Test custom encoder + State support.
|
|
||||||
|
|
||||||
class MockJSONEncoder(JSONEncoder):
|
|
||||||
"""Mock JSON encoder."""
|
|
||||||
|
|
||||||
def default(self, o):
|
|
||||||
"""Mock JSON encode method."""
|
|
||||||
if isinstance(o, 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(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(dumps, cls=MockJSONEncoder),
|
|
||||||
) == {"$[0](Event: bad_event).data.bad_attribute": bad_data}
|
|
||||||
|
|
||||||
class BadData:
|
|
||||||
def __init__(self):
|
|
||||||
self.bla = bad_data
|
|
||||||
|
|
||||||
def as_dict(self):
|
|
||||||
return {"bla": self.bla}
|
|
||||||
|
|
||||||
assert find_paths_unserializable_data(
|
|
||||||
BadData(),
|
|
||||||
dump=partial(dumps, cls=MockJSONEncoder),
|
|
||||||
) == {"$(BadData).bla": bad_data}
|
|
||||||
|
|
||||||
|
|
||||||
def test_json_loads_array() -> None:
|
def test_json_loads_array() -> None:
|
||||||
"""Test json_loads_array validates result."""
|
"""Test json_loads_array validates result."""
|
||||||
assert json_loads_array('[{"c":1.2}]') == [{"c": 1.2}]
|
assert json_loads_array('[{"c":1.2}]') == [{"c": 1.2}]
|
||||||
@ -233,6 +59,9 @@ async def test_deprecated_test_find_unserializable_data(
|
|||||||
caplog: pytest.LogCaptureFixture,
|
caplog: pytest.LogCaptureFixture,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Test deprecated test_find_unserializable_data logs a warning."""
|
"""Test deprecated test_find_unserializable_data logs a warning."""
|
||||||
|
# pylint: disable-next=hass-deprecated-import,import-outside-toplevel
|
||||||
|
from homeassistant.util.json import find_paths_unserializable_data
|
||||||
|
|
||||||
find_paths_unserializable_data(1)
|
find_paths_unserializable_data(1)
|
||||||
assert (
|
assert (
|
||||||
"uses find_paths_unserializable_data from homeassistant.util.json"
|
"uses find_paths_unserializable_data from homeassistant.util.json"
|
||||||
@ -241,9 +70,14 @@ async def test_deprecated_test_find_unserializable_data(
|
|||||||
assert "should be updated to use homeassistant.helpers.json module" in caplog.text
|
assert "should be updated to use homeassistant.helpers.json module" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
async def test_deprecated_save_json(caplog: pytest.LogCaptureFixture) -> None:
|
async def test_deprecated_save_json(
|
||||||
|
caplog: pytest.LogCaptureFixture, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
"""Test deprecated save_json logs a warning."""
|
"""Test deprecated save_json logs a warning."""
|
||||||
fname = _path_for("test1")
|
# pylint: disable-next=hass-deprecated-import,import-outside-toplevel
|
||||||
|
from homeassistant.util.json import save_json
|
||||||
|
|
||||||
|
fname = tmp_path / "test1.json"
|
||||||
save_json(fname, TEST_JSON_A)
|
save_json(fname, TEST_JSON_A)
|
||||||
assert "uses save_json from homeassistant.util.json" in caplog.text
|
assert "uses save_json from homeassistant.util.json" in caplog.text
|
||||||
assert "should be updated to use homeassistant.helpers.json module" in caplog.text
|
assert "should be updated to use homeassistant.helpers.json module" in caplog.text
|
||||||
|
Loading…
x
Reference in New Issue
Block a user