jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 19:27:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move remember the milk config storage to own module (#138999)

Browse Source
This commit is contained in:
Martin Hjelmare 2025-02-23 16:32:55 +01:00 committed by GitHub
parent 1cd82ab8ee
commit 0b961d98f5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 148 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
homeassistant/components/remember_the_milk/__init__.py
View File

@ -1,33 +1,25 @@
"""Support to interact with Remember The Milk."""
import json
import logging
from pathlib import Path
from rtmapi import Rtm
import voluptuous as vol
from homeassistant.components import configurator
from homeassistant.const import CONF_API_KEY, CONF_ID, CONF_NAME, CONF_TOKEN
from homeassistant.const import CONF_API_KEY, CONF_ID, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from .const import LOGGER
from .entity import RememberTheMilkEntity
from .storage import RememberTheMilkConfiguration
# httplib2 is a transitive dependency from RtmAPI. If this dependency is not
# set explicitly, the library does not work.
_LOGGER = logging.getLogger(__name__)
DOMAIN = "remember_the_milk"
DEFAULT_NAME = DOMAIN
CONF_SHARED_SECRET = "shared_secret"
CONF_ID_MAP = "id_map"
CONF_LIST_ID = "list_id"
CONF_TIMESERIES_ID = "timeseries_id"
CONF_TASK_ID = "task_id"
RTM_SCHEMA = vol.Schema(
{
@ -41,7 +33,6 @@ CONFIG_SCHEMA = vol.Schema(
{DOMAIN: vol.All(cv.ensure_list, [RTM_SCHEMA])}, extra=vol.ALLOW_EXTRA
)
CONFIG_FILE_NAME = ".remember_the_milk.conf"
SERVICE_CREATE_TASK = "create_task"
SERVICE_COMPLETE_TASK = "complete_task"
@ -54,17 +45,17 @@ SERVICE_SCHEMA_COMPLETE_TASK = vol.Schema({vol.Required(CONF_ID): cv.string})
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Remember the milk component."""
component = EntityComponent[RememberTheMilkEntity](_LOGGER, DOMAIN, hass)
component = EntityComponent[RememberTheMilkEntity](LOGGER, DOMAIN, hass)
stored_rtm_config = RememberTheMilkConfiguration(hass)
for rtm_config in config[DOMAIN]:
account_name = rtm_config[CONF_NAME]
_LOGGER.debug("Adding Remember the milk account %s", account_name)
LOGGER.debug("Adding Remember the milk account %s", account_name)
api_key = rtm_config[CONF_API_KEY]
shared_secret = rtm_config[CONF_SHARED_SECRET]
token = stored_rtm_config.get_token(account_name)
if token:
_LOGGER.debug("found token for account %s", account_name)
LOGGER.debug("found token for account %s", account_name)
_create_instance(
hass,
account_name,
@ -79,7 +70,7 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
hass, account_name, api_key, shared_secret, stored_rtm_config, component
)
_LOGGER.debug("Finished adding all Remember the milk accounts")
LOGGER.debug("Finished adding all Remember the milk accounts")
return True
@ -110,21 +101,21 @@ def _register_new_account(
request_id = None
api = Rtm(api_key, shared_secret, "write", None)
url, frob = api.authenticate_desktop()
_LOGGER.debug("Sent authentication request to server")
LOGGER.debug("Sent authentication request to server")
def register_account_callback(fields: list[dict[str, str]]) -> None:
"""Call for register the configurator."""
api.retrieve_token(frob)
token = api.token
if api.token is None:
_LOGGER.error("Failed to register, please try again")
LOGGER.error("Failed to register, please try again")
configurator.notify_errors(
hass, request_id, "Failed to register, please try again."
)
return
stored_rtm_config.set_token(account_name, token)
_LOGGER.debug("Retrieved new token from server")
LOGGER.debug("Retrieved new token from server")
_create_instance(
hass,
@ -152,104 +143,3 @@ def _register_new_account(
link_url=url,
submit_caption="login completed",
)
class RememberTheMilkConfiguration:
"""Internal configuration data for RememberTheMilk class.
This class stores the authentication token it get from the backend.
"""
def __init__(self, hass: HomeAssistant) -> None:
"""Create new instance of configuration."""
self._config_file_path = hass.config.path(CONFIG_FILE_NAME)
self._config = {}
_LOGGER.debug("Loading configuration from file: %s", self._config_file_path)
try:
self._config = json.loads(
Path(self._config_file_path).read_text(encoding="utf8")
)
except FileNotFoundError:
_LOGGER.debug("Missing configuration file: %s", self._config_file_path)
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,
)
def _save_config(self) -> None:
"""Write the configuration to a file."""
Path(self._config_file_path).write_text(
json.dumps(self._config), encoding="utf8"
)
def get_token(self, profile_name: str) -> str | None:
"""Get the server token for a profile."""
if profile_name in self._config:
return self._config[profile_name][CONF_TOKEN]
return None
def set_token(self, profile_name: str, token: str) -> None:
"""Store a new server token for a profile."""
self._initialize_profile(profile_name)
self._config[profile_name][CONF_TOKEN] = token
self._save_config()
def delete_token(self, profile_name: str) -> None:
"""Delete a token for a profile.
Usually called when the token has expired.
"""
self._config.pop(profile_name, None)
self._save_config()
def _initialize_profile(self, profile_name: str) -> None:
"""Initialize the data structures for a profile."""
if profile_name not in self._config:
self._config[profile_name] = {}
if CONF_ID_MAP not in self._config[profile_name]:
self._config[profile_name][CONF_ID_MAP] = {}
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.
The id of a RTM tasks consists of the tuple:
list id, timeseries id and the task id.
"""
self._initialize_profile(profile_name)
ids = self._config[profile_name][CONF_ID_MAP].get(hass_id)
if ids is None:
return None
return ids[CONF_LIST_ID], ids[CONF_TIMESERIES_ID], ids[CONF_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."""
self._initialize_profile(profile_name)
id_tuple = {
CONF_LIST_ID: list_id,
CONF_TIMESERIES_ID: time_series_id,
CONF_TASK_ID: rtm_task_id,
}
self._config[profile_name][CONF_ID_MAP][hass_id] = id_tuple
self._save_config()
def delete_rtm_id(self, profile_name: str, hass_id: str) -> None:
"""Delete a key mapping."""
self._initialize_profile(profile_name)
if hass_id in self._config[profile_name][CONF_ID_MAP]:
del self._config[profile_name][CONF_ID_MAP][hass_id]
self._save_config()

5
homeassistant/components/remember_the_milk/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants for the Remember The Milk integration."""
import logging
LOGGER = logging.getLogger(__package__)

22
homeassistant/components/remember_the_milk/entity.py
View File

@ -1,14 +1,12 @@
"""Support to interact with Remember The Milk."""
import logging
from rtmapi import Rtm, RtmRequestFailedException
from homeassistant.const import CONF_ID, CONF_NAME, STATE_OK
from homeassistant.core import ServiceCall
from homeassistant.helpers.entity import Entity
_LOGGER = logging.getLogger(__name__)
from .const import LOGGER
class RememberTheMilkEntity(Entity):
@ -24,7 +22,7 @@ class RememberTheMilkEntity(Entity):
self._rtm_api = Rtm(api_key, shared_secret, "delete", token)
self._token_valid = None
self._check_token()
_LOGGER.debug("Instance created for account %s", self._name)
LOGGER.debug("Instance created for account %s", self._name)
def _check_token(self):
"""Check if the API token is still valid.
@ -34,7 +32,7 @@ class RememberTheMilkEntity(Entity):
"""
valid = self._rtm_api.token_valid()
if not valid:
_LOGGER.error(
LOGGER.error(
"Token for account %s is invalid. You need to register again!",
self.name,
)
@ -64,7 +62,7 @@ class RememberTheMilkEntity(Entity):
result = self._rtm_api.rtm.tasks.add(
timeline=timeline, name=task_name, parse="1"
)
_LOGGER.debug(
LOGGER.debug(
"Created new task '%s' in account %s", task_name, self.name
)
if hass_id is not None:
@ -83,14 +81,14 @@ class RememberTheMilkEntity(Entity):
task_id=rtm_id[2],
timeline=timeline,
)
_LOGGER.debug(
LOGGER.debug(
"Updated task with id '%s' in account %s to name %s",
hass_id,
self.name,
task_name,
)
except RtmRequestFailedException as rtm_exception:
_LOGGER.error(
LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,
@ -101,7 +99,7 @@ class RememberTheMilkEntity(Entity):
hass_id = call.data[CONF_ID]
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
if rtm_id is None:
_LOGGER.error(
LOGGER.error(
(
"Could not find task with ID %s in account %s. "
"So task could not be closed"
@ -120,11 +118,9 @@ class RememberTheMilkEntity(Entity):
timeline=timeline,
)
self._rtm_config.delete_rtm_id(self._name, hass_id)
_LOGGER.debug(
"Completed task with id %s in account %s", hass_id, self._name
)
LOGGER.debug("Completed task with id %s in account %s", hass_id, self._name)
except RtmRequestFailedException as rtm_exception:
_LOGGER.error(
LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,

115
homeassistant/components/remember_the_milk/storage.py Normal file
View File

@ -0,0 +1,115 @@
"""Store RTM configuration in Home Assistant storage."""
from __future__ import annotations
import json
from pathlib import Path
from homeassistant.const import CONF_TOKEN
from homeassistant.core import HomeAssistant
from .const import LOGGER
CONFIG_FILE_NAME = ".remember_the_milk.conf"
CONF_ID_MAP = "id_map"
CONF_LIST_ID = "list_id"
CONF_TASK_ID = "task_id"
CONF_TIMESERIES_ID = "timeseries_id"
class RememberTheMilkConfiguration:
"""Internal configuration data for Remember The Milk."""
def __init__(self, hass: HomeAssistant) -> None:
"""Create new instance of configuration."""
self._config_file_path = hass.config.path(CONFIG_FILE_NAME)
self._config = {}
LOGGER.debug("Loading configuration from file: %s", self._config_file_path)
try:
self._config = json.loads(
Path(self._config_file_path).read_text(encoding="utf8")
)
except FileNotFoundError:
LOGGER.debug("Missing configuration file: %s", self._config_file_path)
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,
)
def _save_config(self) -> None:
"""Write the configuration to a file."""
Path(self._config_file_path).write_text(
json.dumps(self._config), encoding="utf8"
)
def get_token(self, profile_name: str) -> str | None:
"""Get the server token for a profile."""
if profile_name in self._config:
return self._config[profile_name][CONF_TOKEN]
return None
def set_token(self, profile_name: str, token: str) -> None:
"""Store a new server token for a profile."""
self._initialize_profile(profile_name)
self._config[profile_name][CONF_TOKEN] = token
self._save_config()
def delete_token(self, profile_name: str) -> None:
"""Delete a token for a profile.
Usually called when the token has expired.
"""
self._config.pop(profile_name, None)
self._save_config()
def _initialize_profile(self, profile_name: str) -> None:
"""Initialize the data structures for a profile."""
if profile_name not in self._config:
self._config[profile_name] = {}
if CONF_ID_MAP not in self._config[profile_name]:
self._config[profile_name][CONF_ID_MAP] = {}
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.
The id of a RTM tasks consists of the tuple:
list id, timeseries id and the task id.
"""
self._initialize_profile(profile_name)
ids = self._config[profile_name][CONF_ID_MAP].get(hass_id)
if ids is None:
return None
return ids[CONF_LIST_ID], ids[CONF_TIMESERIES_ID], ids[CONF_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."""
self._initialize_profile(profile_name)
id_tuple = {
CONF_LIST_ID: list_id,
CONF_TIMESERIES_ID: time_series_id,
CONF_TASK_ID: rtm_task_id,
}
self._config[profile_name][CONF_ID_MAP][hass_id] = id_tuple
self._save_config()
def delete_rtm_id(self, profile_name: str, hass_id: str) -> None:
"""Delete a key mapping."""
self._initialize_profile(profile_name)
if hass_id in self._config[profile_name][CONF_ID_MAP]:
del self._config[profile_name][CONF_ID_MAP][hass_id]
self._save_config()

14
tests/components/remember_the_milk/test_init.py → tests/components/remember_the_milk/test_storage.py
View File

@ -14,7 +14,9 @@ from .const import JSON_STRING, PROFILE, TOKEN
def test_set_get_delete_token(hass: HomeAssistant) -> None:
"""Test set, get and delete token."""
open_mock = mock_open()
with patch("homeassistant.components.remember_the_milk.Path.open", open_mock):
with patch(
"homeassistant.components.remember_the_milk.storage.Path.open", open_mock
):
config = rtm.RememberTheMilkConfiguration(hass)
assert open_mock.return_value.write.call_count == 0
assert config.get_token(PROFILE) is None
@ -42,7 +44,7 @@ def test_config_load(hass: HomeAssistant) -> None:
"""Test loading from the file."""
with (
patch(
"homeassistant.components.remember_the_milk.Path.open",
"homeassistant.components.remember_the_milk.storage.Path.open",
mock_open(read_data=JSON_STRING),
),
):
@ -61,7 +63,7 @@ def test_config_load_file_error(hass: HomeAssistant, side_effect: Exception) ->
config = rtm.RememberTheMilkConfiguration(hass)
with (
patch(
"homeassistant.components.remember_the_milk.Path.open",
"homeassistant.components.remember_the_milk.storage.Path.open",
side_effect=side_effect,
),
):
@ -78,7 +80,7 @@ def test_config_load_invalid_data(hass: HomeAssistant) -> None:
config = rtm.RememberTheMilkConfiguration(hass)
with (
patch(
"homeassistant.components.remember_the_milk.Path.open",
"homeassistant.components.remember_the_milk.storage.Path.open",
mock_open(read_data="random characters"),
),
):
@ -98,7 +100,9 @@ def test_config_set_delete_id(hass: HomeAssistant) -> None:
rtm_id = "3"
open_mock = mock_open()
config = rtm.RememberTheMilkConfiguration(hass)
with patch("homeassistant.components.remember_the_milk.Path.open", open_mock):
with patch(
"homeassistant.components.remember_the_milk.storage.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