Improve remember the milk storage (#138618)

This commit is contained in:
Martin Hjelmare 2025-02-16 21:17:26 +01:00 committed by GitHub
parent ccd0e27e84
commit 0b7ec96448
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 148 additions and 76 deletions

View File

@ -2,7 +2,7 @@
import json import json
import logging import logging
import os from pathlib import Path
from rtmapi import Rtm from rtmapi import Rtm
import voluptuous as vol import voluptuous as vol
@ -160,56 +160,64 @@ class RememberTheMilkConfiguration:
This class stores the authentication token it get from the backend. This class stores the authentication token it get from the backend.
""" """
def __init__(self, hass): def __init__(self, hass: HomeAssistant) -> None:
"""Create new instance of configuration.""" """Create new instance of configuration."""
self._config_file_path = hass.config.path(CONFIG_FILE_NAME) self._config_file_path = hass.config.path(CONFIG_FILE_NAME)
if not os.path.isfile(self._config_file_path): self._config = {}
self._config = {} _LOGGER.debug("Loading configuration from file: %s", self._config_file_path)
return
try: try:
_LOGGER.debug("Loading configuration from file: %s", self._config_file_path) self._config = json.loads(
with open(self._config_file_path, encoding="utf8") as config_file: Path(self._config_file_path).read_text(encoding="utf8")
self._config = json.load(config_file) )
except ValueError: except FileNotFoundError:
_LOGGER.error( _LOGGER.debug("Missing configuration file: %s", self._config_file_path)
"Failed to load configuration file, creating a new one: %s", except OSError:
_LOGGER.debug(
"Failed to read from configuration file, %s, using empty configuration",
self._config_file_path,
)
except ValueError:
_LOGGER.error(
"Failed to parse configuration file, %s, using empty configuration",
self._config_file_path, self._config_file_path,
) )
self._config = {}
def save_config(self): def _save_config(self) -> None:
"""Write the configuration to a file.""" """Write the configuration to a file."""
with open(self._config_file_path, "w", encoding="utf8") as config_file: Path(self._config_file_path).write_text(
json.dump(self._config, config_file) json.dumps(self._config), encoding="utf8"
)
def get_token(self, profile_name): def get_token(self, profile_name: str) -> str | None:
"""Get the server token for a profile.""" """Get the server token for a profile."""
if profile_name in self._config: if profile_name in self._config:
return self._config[profile_name][CONF_TOKEN] return self._config[profile_name][CONF_TOKEN]
return None return None
def set_token(self, profile_name, token): def set_token(self, profile_name: str, token: str) -> None:
"""Store a new server token for a profile.""" """Store a new server token for a profile."""
self._initialize_profile(profile_name) self._initialize_profile(profile_name)
self._config[profile_name][CONF_TOKEN] = token self._config[profile_name][CONF_TOKEN] = token
self.save_config() self._save_config()
def delete_token(self, profile_name): def delete_token(self, profile_name: str) -> None:
"""Delete a token for a profile. """Delete a token for a profile.
Usually called when the token has expired. Usually called when the token has expired.
""" """
self._config.pop(profile_name, None) self._config.pop(profile_name, None)
self.save_config() self._save_config()
def _initialize_profile(self, profile_name): def _initialize_profile(self, profile_name: str) -> None:
"""Initialize the data structures for a profile.""" """Initialize the data structures for a profile."""
if profile_name not in self._config: if profile_name not in self._config:
self._config[profile_name] = {} self._config[profile_name] = {}
if CONF_ID_MAP not in self._config[profile_name]: if CONF_ID_MAP not in self._config[profile_name]:
self._config[profile_name][CONF_ID_MAP] = {} self._config[profile_name][CONF_ID_MAP] = {}
def get_rtm_id(self, profile_name, hass_id): def get_rtm_id(
self, profile_name: str, hass_id: str
) -> tuple[str, str, str] | None:
"""Get the RTM ids for a Home Assistant task ID. """Get the RTM ids for a Home Assistant task ID.
The id of a RTM tasks consists of the tuple: The id of a RTM tasks consists of the tuple:
@ -221,7 +229,14 @@ class RememberTheMilkConfiguration:
return None return None
return ids[CONF_LIST_ID], ids[CONF_TIMESERIES_ID], ids[CONF_TASK_ID] return ids[CONF_LIST_ID], ids[CONF_TIMESERIES_ID], ids[CONF_TASK_ID]
def set_rtm_id(self, profile_name, hass_id, list_id, time_series_id, rtm_task_id): def set_rtm_id(
self,
profile_name: str,
hass_id: str,
list_id: str,
time_series_id: str,
rtm_task_id: str,
) -> None:
"""Add/Update the RTM task ID for a Home Assistant task IS.""" """Add/Update the RTM task ID for a Home Assistant task IS."""
self._initialize_profile(profile_name) self._initialize_profile(profile_name)
id_tuple = { id_tuple = {
@ -230,11 +245,11 @@ class RememberTheMilkConfiguration:
CONF_TASK_ID: rtm_task_id, CONF_TASK_ID: rtm_task_id,
} }
self._config[profile_name][CONF_ID_MAP][hass_id] = id_tuple self._config[profile_name][CONF_ID_MAP][hass_id] = id_tuple
self.save_config() self._save_config()
def delete_rtm_id(self, profile_name, hass_id): def delete_rtm_id(self, profile_name: str, hass_id: str) -> None:
"""Delete a key mapping.""" """Delete a key mapping."""
self._initialize_profile(profile_name) self._initialize_profile(profile_name)
if hass_id in self._config[profile_name][CONF_ID_MAP]: if hass_id in self._config[profile_name][CONF_ID_MAP]:
del self._config[profile_name][CONF_ID_MAP][hass_id] del self._config[profile_name][CONF_ID_MAP][hass_id]
self.save_config() self._save_config()

View File

@ -8,7 +8,7 @@ JSON_STRING = json.dumps(
{ {
"myprofile": { "myprofile": {
"token": "mytoken", "token": "mytoken",
"id_map": {"1234": {"list_id": "0", "timeseries_id": "1", "task_id": "2"}}, "id_map": {"123": {"list_id": "1", "timeseries_id": "2", "task_id": "3"}},
} }
} }
) )

View File

@ -1,6 +1,9 @@
"""Tests for the Remember The Milk component.""" """Tests for the Remember The Milk integration."""
from unittest.mock import Mock, mock_open, patch import json
from unittest.mock import mock_open, patch
import pytest
from homeassistant.components import remember_the_milk as rtm from homeassistant.components import remember_the_milk as rtm
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -8,63 +11,117 @@ from homeassistant.core import HomeAssistant
from .const import JSON_STRING, PROFILE, TOKEN from .const import JSON_STRING, PROFILE, TOKEN
def test_create_new(hass: HomeAssistant) -> None: def test_set_get_delete_token(hass: HomeAssistant) -> None:
"""Test creating a new config file.""" """Test set, get and delete token."""
with ( open_mock = mock_open()
patch("builtins.open", mock_open()), with patch("homeassistant.components.remember_the_milk.Path.open", open_mock):
patch("os.path.isfile", Mock(return_value=False)),
patch.object(rtm.RememberTheMilkConfiguration, "save_config"),
):
config = rtm.RememberTheMilkConfiguration(hass) config = rtm.RememberTheMilkConfiguration(hass)
assert open_mock.return_value.write.call_count == 0
assert config.get_token(PROFILE) is None
assert open_mock.return_value.write.call_count == 0
config.set_token(PROFILE, TOKEN) config.set_token(PROFILE, TOKEN)
assert config.get_token(PROFILE) == TOKEN assert open_mock.return_value.write.call_count == 1
assert open_mock.return_value.write.call_args[0][0] == json.dumps(
{
"myprofile": {
"id_map": {},
"token": "mytoken",
}
}
)
assert config.get_token(PROFILE) == TOKEN
assert open_mock.return_value.write.call_count == 1
config.delete_token(PROFILE)
assert open_mock.return_value.write.call_count == 2
assert open_mock.return_value.write.call_args[0][0] == json.dumps({})
assert config.get_token(PROFILE) is None
assert open_mock.return_value.write.call_count == 2
def test_load_config(hass: HomeAssistant) -> None: def test_config_load(hass: HomeAssistant) -> None:
"""Test loading an existing token from the file.""" """Test loading from the file."""
with ( with (
patch("builtins.open", mock_open(read_data=JSON_STRING)), patch(
patch("os.path.isfile", Mock(return_value=True)), "homeassistant.components.remember_the_milk.Path.open",
): mock_open(read_data=JSON_STRING),
config = rtm.RememberTheMilkConfiguration(hass) ),
assert config.get_token(PROFILE) == TOKEN
def test_invalid_data(hass: HomeAssistant) -> None:
"""Test starts with invalid data and should not raise an exception."""
with (
patch("builtins.open", mock_open(read_data="random characters")),
patch("os.path.isfile", Mock(return_value=True)),
):
config = rtm.RememberTheMilkConfiguration(hass)
assert config is not None
def test_id_map(hass: HomeAssistant) -> None:
"""Test the hass to rtm task is mapping."""
hass_id = "hass-id-1234"
list_id = "mylist"
timeseries_id = "my_timeseries"
rtm_id = "rtm-id-4567"
with (
patch("builtins.open", mock_open()),
patch("os.path.isfile", Mock(return_value=False)),
patch.object(rtm.RememberTheMilkConfiguration, "save_config"),
): ):
config = rtm.RememberTheMilkConfiguration(hass) config = rtm.RememberTheMilkConfiguration(hass)
rtm_id = config.get_rtm_id(PROFILE, "123")
assert rtm_id is not None
assert rtm_id == ("1", "2", "3")
@pytest.mark.parametrize(
"side_effect", [FileNotFoundError("Missing file"), OSError("IO error")]
)
def test_config_load_file_error(hass: HomeAssistant, side_effect: Exception) -> None:
"""Test loading with file error."""
config = rtm.RememberTheMilkConfiguration(hass)
with (
patch(
"homeassistant.components.remember_the_milk.Path.open",
side_effect=side_effect,
),
):
config = rtm.RememberTheMilkConfiguration(hass)
# The config should be empty and we should not have any errors
# when trying to access it.
rtm_id = config.get_rtm_id(PROFILE, "123")
assert rtm_id is None
def test_config_load_invalid_data(hass: HomeAssistant) -> None:
"""Test loading invalid data."""
config = rtm.RememberTheMilkConfiguration(hass)
with (
patch(
"homeassistant.components.remember_the_milk.Path.open",
mock_open(read_data="random characters"),
),
):
config = rtm.RememberTheMilkConfiguration(hass)
# The config should be empty and we should not have any errors
# when trying to access it.
rtm_id = config.get_rtm_id(PROFILE, "123")
assert rtm_id is None
def test_config_set_delete_id(hass: HomeAssistant) -> None:
"""Test setting and deleting an id from the config."""
hass_id = "123"
list_id = "1"
timeseries_id = "2"
rtm_id = "3"
open_mock = mock_open()
config = rtm.RememberTheMilkConfiguration(hass)
with patch("homeassistant.components.remember_the_milk.Path.open", open_mock):
config = rtm.RememberTheMilkConfiguration(hass)
assert open_mock.return_value.write.call_count == 0
assert config.get_rtm_id(PROFILE, hass_id) is None assert config.get_rtm_id(PROFILE, hass_id) is None
assert open_mock.return_value.write.call_count == 0
config.set_rtm_id(PROFILE, hass_id, list_id, timeseries_id, rtm_id) config.set_rtm_id(PROFILE, hass_id, list_id, timeseries_id, rtm_id)
assert (list_id, timeseries_id, rtm_id) == config.get_rtm_id(PROFILE, hass_id) assert (list_id, timeseries_id, rtm_id) == config.get_rtm_id(PROFILE, hass_id)
assert open_mock.return_value.write.call_count == 1
assert open_mock.return_value.write.call_args[0][0] == json.dumps(
{
"myprofile": {
"id_map": {
"123": {"list_id": "1", "timeseries_id": "2", "task_id": "3"}
}
}
}
)
config.delete_rtm_id(PROFILE, hass_id) config.delete_rtm_id(PROFILE, hass_id)
assert config.get_rtm_id(PROFILE, hass_id) is None assert config.get_rtm_id(PROFILE, hass_id) is None
assert open_mock.return_value.write.call_count == 2
assert open_mock.return_value.write.call_args[0][0] == json.dumps(
def test_load_key_map(hass: HomeAssistant) -> None: {
"""Test loading an existing key map from the file.""" "myprofile": {
with ( "id_map": {},
patch("builtins.open", mock_open(read_data=JSON_STRING)), }
patch("os.path.isfile", Mock(return_value=True)), }
): )
config = rtm.RememberTheMilkConfiguration(hass)
assert config.get_rtm_id(PROFILE, "1234") == ("0", "1", "2")