Add typed helpers and improve type hints in util/json (#88534)

* Add type hints to load_json

* Adjust ios

* Adjust nest

* Add use of load_json_array

* Add tests

* Adjust test patch

* Add test_load_json_os_error
This commit is contained in:
epenet 2023-02-22 10:09:28 +01:00 committed by GitHub
parent 906d397736
commit aa20c902db
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 137 additions and 27 deletions

View File

@ -1,7 +1,6 @@
"""Native Home Assistant iOS app component.""" """Native Home Assistant iOS app component."""
import datetime import datetime
from http import HTTPStatus from http import HTTPStatus
from typing import TYPE_CHECKING
import voluptuous as vol import voluptuous as vol
@ -14,7 +13,7 @@ from homeassistant.helpers import config_validation as cv, discovery
from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.json import save_json from homeassistant.helpers.json import save_json
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from homeassistant.util.json import load_json from homeassistant.util.json import load_json_object
from .const import ( from .const import (
CONF_ACTION_BACKGROUND_COLOR, CONF_ACTION_BACKGROUND_COLOR,
@ -252,22 +251,19 @@ def device_name_for_push_id(hass, push_id):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the iOS component.""" """Set up the iOS component."""
conf = config.get(DOMAIN) conf: ConfigType | None = config.get(DOMAIN)
ios_config = await hass.async_add_executor_job( ios_config = await hass.async_add_executor_job(
load_json, hass.config.path(CONFIGURATION_FILE) load_json_object, hass.config.path(CONFIGURATION_FILE)
) )
if TYPE_CHECKING:
assert isinstance(ios_config, dict)
if ios_config == {}: if ios_config == {}:
ios_config[ATTR_DEVICES] = {} ios_config[ATTR_DEVICES] = {}
ios_config[CONF_USER] = conf or {} if CONF_PUSH not in (conf_user := conf or {}):
conf_user[CONF_PUSH] = {}
if CONF_PUSH not in ios_config[CONF_USER]: ios_config[CONF_USER] = conf_user
ios_config[CONF_USER][CONF_PUSH] = {}
hass.data[DOMAIN] = ios_config hass.data[DOMAIN] = ios_config

View File

@ -33,7 +33,7 @@ from homeassistant.data_entry_flow import FlowResult
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_entry_oauth2_flow from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.util import get_random_string from homeassistant.util import get_random_string
from homeassistant.util.json import load_json from homeassistant.util.json import JsonObjectType, load_json_object
from . import api from . import api
from .const import ( from .const import (
@ -570,7 +570,7 @@ class NestFlowHandler(
return await self.async_step_link() return await self.async_step_link()
flow = self.hass.data[DATA_FLOW_IMPL][DOMAIN] flow = self.hass.data[DATA_FLOW_IMPL][DOMAIN]
tokens = await self.hass.async_add_executor_job(load_json, config_path) tokens = await self.hass.async_add_executor_job(load_json_object, config_path)
return self._entry_from_tokens( return self._entry_from_tokens(
"Nest (import from configuration.yaml)", flow, tokens "Nest (import from configuration.yaml)", flow, tokens
@ -578,7 +578,7 @@ class NestFlowHandler(
@callback @callback
def _entry_from_tokens( def _entry_from_tokens(
self, title: str, flow: dict[str, Any], tokens: list[Any] | dict[Any, Any] self, title: str, flow: dict[str, Any], tokens: JsonObjectType
) -> FlowResult: ) -> FlowResult:
"""Create an entry from tokens.""" """Create an entry from tokens."""
return self.async_create_entry( return self.async_create_entry(

View File

@ -15,7 +15,7 @@ from homeassistant.core import HomeAssistant, ServiceCall, callback
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.json import save_json from homeassistant.helpers.json import save_json
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from homeassistant.util.json import load_json from homeassistant.util.json import JsonArrayType, load_json_array
from .const import ( from .const import (
DOMAIN, DOMAIN,
@ -174,10 +174,10 @@ class NoMatchingShoppingListItem(Exception):
class ShoppingData: class ShoppingData:
"""Class to hold shopping list data.""" """Class to hold shopping list data."""
def __init__(self, hass): def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the shopping list.""" """Initialize the shopping list."""
self.hass = hass self.hass = hass
self.items = [] self.items: JsonArrayType = []
async def async_add(self, name, context=None): async def async_add(self, name, context=None):
"""Add a shopping list item.""" """Add a shopping list item."""
@ -277,16 +277,16 @@ class ShoppingData:
context=context, context=context,
) )
async def async_load(self): async def async_load(self) -> None:
"""Load items.""" """Load items."""
def load(): def load() -> JsonArrayType:
"""Load the items synchronously.""" """Load the items synchronously."""
return load_json(self.hass.config.path(PERSISTENCE), default=[]) return load_json_array(self.hass.config.path(PERSISTENCE))
self.items = await self.hass.async_add_executor_job(load) self.items = await self.hass.async_add_executor_job(load)
def save(self): def save(self) -> None:
"""Save the items.""" """Save the items."""
save_json(self.hass.config.path(PERSISTENCE), self.items) save_json(self.hass.config.path(PERSISTENCE), self.items)

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
import json import json
import logging import logging
from os import PathLike
from typing import Any from typing import Any
import orjson import orjson
@ -12,6 +13,7 @@ from homeassistant.exceptions import HomeAssistantError
from .file import WriteError # pylint: disable=unused-import # noqa: F401 from .file import WriteError # pylint: disable=unused-import # noqa: F401
_SENTINEL = object()
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
JsonValueType = ( JsonValueType = (
@ -54,8 +56,10 @@ def json_loads_object(__obj: bytes | bytearray | memoryview | str) -> JsonObject
raise ValueError(f"Expected JSON to be parsed as a dict got {type(value)}") raise ValueError(f"Expected JSON to be parsed as a dict got {type(value)}")
def load_json(filename: str, default: list | dict | None = None) -> list | dict: def load_json(
"""Load JSON data from a file and return as dict or list. filename: str | PathLike, default: JsonValueType = _SENTINEL # type: ignore[assignment]
) -> JsonValueType:
"""Load JSON data from a file.
Defaults to returning empty dict if file is not found. Defaults to returning empty dict if file is not found.
""" """
@ -71,7 +75,45 @@ def load_json(filename: str, default: list | dict | None = None) -> list | dict:
except OSError as error: except OSError as error:
_LOGGER.exception("JSON file reading failed: %s", filename) _LOGGER.exception("JSON file reading failed: %s", filename)
raise HomeAssistantError(error) from error raise HomeAssistantError(error) from error
return {} if default is None else default return {} if default is _SENTINEL else default
def load_json_array(
filename: str | PathLike, default: JsonArrayType = _SENTINEL # type: ignore[assignment]
) -> JsonArrayType:
"""Load JSON data from a file and return as list.
Defaults to returning empty list if file is not found.
"""
if default is _SENTINEL:
default = []
value: JsonValueType = load_json(filename, default=default)
# Avoid isinstance overhead as we are not interested in list subclasses
if type(value) is list: # pylint: disable=unidiomatic-typecheck
return value
_LOGGER.exception(
"Expected JSON to be parsed as a list got %s in: %s", {type(value)}, filename
)
raise HomeAssistantError(f"Expected JSON to be parsed as a list got {type(value)}")
def load_json_object(
filename: str | PathLike, default: JsonObjectType = _SENTINEL # type: ignore[assignment]
) -> JsonObjectType:
"""Load JSON data from a file and return as dict.
Defaults to returning empty dict if file is not found.
"""
if default is _SENTINEL:
default = {}
value: JsonValueType = load_json(filename, default=default)
# Avoid isinstance overhead as we are not interested in dict subclasses
if type(value) is dict: # pylint: disable=unidiomatic-typecheck
return value
_LOGGER.exception(
"Expected JSON to be parsed as a dict got %s in: %s", {type(value)}, filename
)
raise HomeAssistantError(f"Expected JSON to be parsed as a dict got {type(value)}")
def save_json( def save_json(

View File

@ -13,7 +13,7 @@ from tests.common import mock_component, mock_coro
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def mock_load_json(): def mock_load_json():
"""Mock load_json.""" """Mock load_json."""
with patch("homeassistant.components.ios.load_json", return_value={}): with patch("homeassistant.components.ios.load_json_object", return_value={}):
yield yield

View File

@ -228,7 +228,7 @@ async def test_step_import(hass: HomeAssistant) -> None:
async def test_step_import_with_token_cache(hass: HomeAssistant) -> None: async def test_step_import_with_token_cache(hass: HomeAssistant) -> None:
"""Test that we import existing token cache.""" """Test that we import existing token cache."""
with patch("os.path.isfile", return_value=True), patch( with patch("os.path.isfile", return_value=True), patch(
"homeassistant.components.nest.config_flow.load_json", "homeassistant.components.nest.config_flow.load_json_object",
return_value={"access_token": "yo"}, return_value={"access_token": "yo"},
), patch( ), patch(
"homeassistant.components.nest.async_setup_legacy_entry", return_value=True "homeassistant.components.nest.async_setup_legacy_entry", return_value=True

View File

@ -4,7 +4,13 @@ from pathlib import Path
import pytest import pytest
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.util.json import json_loads_array, json_loads_object, load_json from homeassistant.util.json import (
json_loads_array,
json_loads_object,
load_json,
load_json_array,
load_json_object,
)
# 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"}
@ -17,8 +23,74 @@ def test_load_bad_data(tmp_path: Path) -> None:
fname = tmp_path / "test5.json" 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) as err:
load_json(fname) load_json(fname)
assert isinstance(err.value.__cause__, ValueError)
def test_load_json_os_error() -> None:
"""Test trying to load JSON data from a directory."""
fname = "/"
with pytest.raises(HomeAssistantError) as err:
load_json(fname)
assert isinstance(err.value.__cause__, OSError)
def test_load_json_file_not_found_error() -> None:
"""Test trying to load object data from inexistent JSON file."""
fname = "invalid_file.json"
assert load_json(fname) == {}
assert load_json(fname, default="") == ""
assert load_json_object(fname) == {}
assert load_json_object(fname, default={"Hi": "Peter"}) == {"Hi": "Peter"}
assert load_json_array(fname) == []
assert load_json_array(fname, default=["Hi"]) == ["Hi"]
def test_load_json_value_data(tmp_path: Path) -> None:
"""Test trying to load object data from JSON file."""
fname = tmp_path / "test5.json"
with open(fname, "w", encoding="utf8") as handle:
handle.write('"two"')
assert load_json(fname) == "two"
with pytest.raises(
HomeAssistantError, match="Expected JSON to be parsed as a dict"
):
load_json_object(fname)
with pytest.raises(
HomeAssistantError, match="Expected JSON to be parsed as a list"
):
load_json_array(fname)
def test_load_json_object_data(tmp_path: Path) -> None:
"""Test trying to load object data from JSON file."""
fname = tmp_path / "test5.json"
with open(fname, "w", encoding="utf8") as handle:
handle.write('{"a": 1, "B": "two"}')
assert load_json(fname) == {"a": 1, "B": "two"}
assert load_json_object(fname) == {"a": 1, "B": "two"}
with pytest.raises(
HomeAssistantError, match="Expected JSON to be parsed as a list"
):
load_json_array(fname)
def test_load_json_array_data(tmp_path: Path) -> None:
"""Test trying to load array data from JSON file."""
fname = tmp_path / "test5.json"
with open(fname, "w", encoding="utf8") as handle:
handle.write('[{"a": 1, "B": "two"}]')
assert load_json(fname) == [{"a": 1, "B": "two"}]
assert load_json_array(fname) == [{"a": 1, "B": "two"}]
with pytest.raises(
HomeAssistantError, match="Expected JSON to be parsed as a dict"
):
load_json_object(fname)
def test_json_loads_array() -> None: def test_json_loads_array() -> None: