UGA-Innovation-Factory
/
FactoryAssistant
mirror of https://github.com/home-assistant/core.git
3
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add config flow to remember_the_milk

This commit is contained in:
Martin Hjelmare 2023-11-05 17:43:06 +01:00
parent 6a4f5188b1
commit 0b854635a2
15 changed files with 926 additions and 304 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,33 +1,28 @@
"""Support to interact with Remember The Milk."""
"""The Remember The Milk integration."""
import json
import logging
import os
from __future__ import annotations
from rtmapi import Rtm
from aiortm import AioRTMClient, Auth
import voluptuous as vol
from homeassistant.components import configurator
from homeassistant.const import CONF_API_KEY, CONF_ID, CONF_NAME, CONF_TOKEN
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_ID,
CONF_NAME,
CONF_TOKEN,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from .const import CONF_SHARED_SECRET, DOMAIN, LOGGER
from .entity import RememberTheMilkEntity
# 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"
from .storage import RememberTheMilkConfiguration
RTM_SCHEMA = vol.Schema(
{
@ -41,7 +36,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"
@ -51,190 +45,104 @@ SERVICE_SCHEMA_CREATE_TASK = vol.Schema(
SERVICE_SCHEMA_COMPLETE_TASK = vol.Schema({vol.Required(CONF_ID): cv.string})
DATA_COMPONENT = "component"
DATA_ENTITY_ID = "entity_id"
DATA_STORAGE = "storage"
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Remember the milk component."""
component = EntityComponent[RememberTheMilkEntity](_LOGGER, DOMAIN, hass)
stored_rtm_config = RememberTheMilkConfiguration(hass)
hass.data[DOMAIN] = {}
hass.data[DOMAIN][DATA_COMPONENT] = EntityComponent[RememberTheMilkEntity](
LOGGER, DOMAIN, hass
)
storage = hass.data[DOMAIN][DATA_STORAGE] = RememberTheMilkConfiguration(hass)
await hass.async_add_executor_job(storage.setup)
if DOMAIN not in config:
return True
for rtm_config in config[DOMAIN]:
account_name = rtm_config[CONF_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)
_create_instance(
hass,
account_name,
api_key,
shared_secret,
token,
stored_rtm_config,
component,
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=rtm_config,
)
else:
_register_new_account(
hass, account_name, api_key, shared_secret, stored_rtm_config, component
)
_LOGGER.debug("Finished adding all Remember the milk accounts")
)
return True
def _create_instance(
hass, account_name, api_key, shared_secret, token, stored_rtm_config, component
):
entity = RememberTheMilkEntity(
account_name, api_key, shared_secret, token, stored_rtm_config
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Remember The Milk from a config entry."""
component: EntityComponent[RememberTheMilkEntity] = hass.data[DOMAIN][
DATA_COMPONENT
]
storage: RememberTheMilkConfiguration = hass.data[DOMAIN][DATA_STORAGE]
rtm_config = entry.data
account_name: str = rtm_config[CONF_USERNAME]
LOGGER.debug("Adding Remember the milk account %s", account_name)
api_key: str = rtm_config[CONF_API_KEY]
shared_secret: str = rtm_config[CONF_SHARED_SECRET]
token: str | None = rtm_config[CONF_TOKEN] # None if imported from YAML
client = AioRTMClient(
Auth(
client_session=async_get_clientsession(hass),
api_key=api_key,
shared_secret=shared_secret,
auth_token=token,
permission="delete",
)
)
component.add_entities([entity])
hass.services.register(
token_valid = True
if not await client.rtm.api.check_token():
token_valid = False
if entry.source == SOURCE_IMPORT:
raise ConfigEntryAuthFailed("Missing token")
if (known_entity_ids := hass.data[DOMAIN].get(DATA_ENTITY_ID)) and (
entity_id := known_entity_ids.get(account_name)
):
await component.async_remove_entity(entity_id)
# The entity will be deprecated when a todo platform is added.
entity = RememberTheMilkEntity(
name=account_name,
client=client,
config_entry_id=entry.entry_id,
storage=storage,
token_valid=token_valid,
)
await component.async_add_entities([entity])
known_entity_ids = hass.data[DOMAIN].setdefault(DATA_ENTITY_ID, {})
known_entity_ids[account_name] = entity.entity_id
# The services are registered here for now because they need the account name.
# The services will be deprecated when a todo platform is added.
hass.services.async_register(
DOMAIN,
f"{account_name}_create_task",
entity.create_task,
schema=SERVICE_SCHEMA_CREATE_TASK,
)
hass.services.register(
hass.services.async_register(
DOMAIN,
f"{account_name}_complete_task",
entity.complete_task,
schema=SERVICE_SCHEMA_COMPLETE_TASK,
)
if not token_valid:
raise ConfigEntryAuthFailed("Invalid token")
def _register_new_account(
hass, account_name, api_key, shared_secret, stored_rtm_config, component
):
request_id = None
api = Rtm(api_key, shared_secret, "write", None)
url, frob = api.authenticate_desktop()
_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")
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")
_create_instance(
hass,
account_name,
api_key,
shared_secret,
token,
stored_rtm_config,
component,
)
configurator.request_done(hass, request_id)
request_id = configurator.request_config(
hass,
f"{DOMAIN} - {account_name}",
callback=register_account_callback,
description=(
"You need to log in to Remember The Milk to"
"connect your account. \n\n"
"Step 1: Click on the link 'Remember The Milk login'\n\n"
"Step 2: Click on 'login completed'"
),
link_name="Remember The Milk login",
link_url=url,
submit_caption="login completed",
)
return True
class RememberTheMilkConfiguration:
"""Internal configuration data for RememberTheMilk class.
This class stores the authentication token it get from the backend.
"""
def __init__(self, hass):
"""Create new instance of configuration."""
self._config_file_path = hass.config.path(CONFIG_FILE_NAME)
if not os.path.isfile(self._config_file_path):
self._config = {}
return
try:
_LOGGER.debug("Loading configuration from file: %s", self._config_file_path)
with open(self._config_file_path, encoding="utf8") as config_file:
self._config = json.load(config_file)
except ValueError:
_LOGGER.error(
"Failed to load configuration file, creating a new one: %s",
self._config_file_path,
)
self._config = {}
def save_config(self):
"""Write the configuration to a file."""
with open(self._config_file_path, "w", encoding="utf8") as config_file:
json.dump(self._config, config_file)
def get_token(self, profile_name):
"""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, token):
"""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):
"""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):
"""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, hass_id):
"""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, hass_id, list_id, time_series_id, rtm_task_id):
"""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, hass_id):
"""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()
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
component: EntityComponent[RememberTheMilkEntity] = hass.data[DOMAIN][
DATA_COMPONENT
]
entity_id = hass.data[DOMAIN][DATA_ENTITY_ID].pop(entry.data[CONF_USERNAME])
await component.async_remove_entity(entity_id)
return True

157
homeassistant/components/remember_the_milk/config_flow.py Normal file
View File

@ -0,0 +1,157 @@
"""Config flow for Remember The Milk integration."""
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from typing import Any
from aiortm import Auth, AuthError, ResponseError
import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_IMPORT,
SOURCE_REAUTH,
ConfigFlow,
ConfigFlowResult,
)
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_TOKEN, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_SHARED_SECRET, DOMAIN, LOGGER
TOKEN_TIMEOUT_SEC = 30
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_API_KEY): str,
vol.Required(CONF_SHARED_SECRET): str,
}
)
class RTMConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Remember The Milk."""
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._auth: Auth | None = None
self._url: str | None = None
self._frob: str | None = None
self._auth_data: dict[str, str] | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
self._auth_data = user_input
auth = self._auth = Auth(
client_session=async_get_clientsession(self.hass),
api_key=user_input[CONF_API_KEY],
shared_secret=user_input[CONF_SHARED_SECRET],
permission="delete",
)
try:
self._url, self._frob = await auth.authenticate_desktop()
except AuthError:
errors["base"] = "invalid_auth"
except ResponseError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001 pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return await self.async_step_auth()
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def async_step_auth(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Authorize the application."""
assert self._url is not None
if user_input is not None:
return await self._get_token()
return self.async_show_form(
step_id="auth", description_placeholders={"url": self._url}
)
async def _get_token(self) -> ConfigFlowResult:
"""Get token and create config entry."""
assert self._auth is not None
assert self._frob is not None
assert self._auth_data is not None
try:
async with asyncio.timeout(TOKEN_TIMEOUT_SEC):
token = await self._auth.get_token(self._frob)
except TimeoutError:
return self.async_abort(reason="timeout_token")
except AuthError:
return self.async_abort(reason="invalid_auth")
except ResponseError:
return self.async_abort(reason="cannot_connect")
except Exception: # noqa: BLE001 pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
await self.async_set_unique_id(token["user"]["id"])
data = {
**self._auth_data,
CONF_TOKEN: token["token"],
CONF_USERNAME: token["user"]["username"],
}
if self.source == SOURCE_REAUTH:
reauth_entry = self._get_reauth_entry()
if reauth_entry.source == SOURCE_IMPORT and reauth_entry.unique_id is None:
# Imported entries do not have a token nor unique id.
# Update unique id to match the new token.
# This case can be removed when the import step is removed.
self.hass.config_entries.async_update_entry(
reauth_entry, data=data, unique_id=token["user"]["id"]
)
else:
self._abort_if_unique_id_mismatch()
return self.async_update_reload_and_abort(
self._get_reauth_entry(),
data_updates=data,
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=token["user"]["fullname"],
data=data,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that informs the user that reauth is required."""
if user_input is None:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({}),
)
return await self.async_step_user()
async def async_step_import(self, import_info: dict[str, Any]) -> ConfigFlowResult:
"""Import a config entry.
The token will be retrieved after config entry setup in a reauth flow.
"""
name = import_info.pop(CONF_NAME)
return self.async_create_entry(
title=name,
data=import_info | {CONF_USERNAME: name, CONF_TOKEN: None},
)

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

@ -0,0 +1,11 @@
"""Constants for the Remember The Milk integration."""
import logging
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"
DOMAIN = "remember_the_milk"
LOGGER = logging.getLogger(__package__)

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

@ -2,49 +2,37 @@
import logging
from rtmapi import Rtm, RtmRequestFailedException
from aiortm import AioRTMClient, AioRTMError, AuthError
from homeassistant.const import CONF_ID, CONF_NAME, STATE_OK
from homeassistant.core import ServiceCall
from homeassistant.core import ServiceCall, callback
from homeassistant.helpers.entity import Entity
from .storage import RememberTheMilkConfiguration
_LOGGER = logging.getLogger(__name__)
class RememberTheMilkEntity(Entity):
"""Representation of an interface to Remember The Milk."""
def __init__(self, name, api_key, shared_secret, token, rtm_config):
def __init__(
self,
*,
name: str,
client: AioRTMClient,
config_entry_id: str,
storage: RememberTheMilkConfiguration,
token_valid: bool,
) -> None:
"""Create new instance of Remember The Milk component."""
self._name = name
self._api_key = api_key
self._shared_secret = shared_secret
self._token = token
self._rtm_config = rtm_config
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)
self._rtm_config = storage
self._client = client
self._config_entry_id = config_entry_id
self._token_valid = token_valid
def _check_token(self):
"""Check if the API token is still valid.
If it is not valid any more, delete it from the configuration. This
will trigger a new authentication process.
"""
valid = self._rtm_api.token_valid()
if not valid:
_LOGGER.error(
"Token for account %s is invalid. You need to register again!",
self.name,
)
self._rtm_config.delete_token(self._name)
self._token_valid = False
else:
self._token_valid = True
return self._token_valid
def create_task(self, call: ServiceCall) -> None:
async def create_task(self, call: ServiceCall) -> None:
"""Create a new task on Remember The Milk.
You can use the smart syntax to define the attributes of a new task,
@ -52,30 +40,37 @@ class RememberTheMilkEntity(Entity):
due date to today.
"""
try:
task_name = call.data[CONF_NAME]
hass_id = call.data.get(CONF_ID)
rtm_id = None
task_name: str = call.data[CONF_NAME]
hass_id: str | None = call.data.get(CONF_ID)
rtm_id: tuple[int, int, int] | None = None
if hass_id is not None:
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
result = self._rtm_api.rtm.timelines.create()
timeline = result.timeline.value
rtm_id = await self.hass.async_add_executor_job(
self._rtm_config.get_rtm_id, self._name, hass_id
)
timeline_response = await self._client.rtm.timelines.create()
timeline = timeline_response.timeline
if hass_id is None or rtm_id is None:
result = self._rtm_api.rtm.tasks.add(
timeline=timeline, name=task_name, parse="1"
add_response = await self._client.rtm.tasks.add(
timeline=timeline, name=task_name, parse=True
)
_LOGGER.debug(
"Created new task '%s' in account %s", task_name, self.name
)
self._rtm_config.set_rtm_id(
if hass_id is None:
return
task_list = add_response.task_list
taskseries = task_list.taskseries[0]
await self.hass.async_add_executor_job(
self._rtm_config.set_rtm_id,
self._name,
hass_id,
result.list.id,
result.list.taskseries.id,
result.list.taskseries.task.id,
task_list.id,
taskseries.id,
taskseries.task[0].id,
)
else:
self._rtm_api.rtm.tasks.setName(
await self._client.rtm.tasks.set_name(
name=task_name,
list_id=rtm_id[0],
taskseries_id=rtm_id[1],
@ -88,17 +83,26 @@ class RememberTheMilkEntity(Entity):
self.name,
task_name,
)
except RtmRequestFailedException as rtm_exception:
except AuthError as err:
_LOGGER.error(
"Invalid authentication when creating task for account %s: %s",
self._name,
err,
)
self._handle_token(False)
except AioRTMError as err:
_LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,
err,
)
def complete_task(self, call: ServiceCall) -> None:
async def complete_task(self, call: ServiceCall) -> None:
"""Complete a task that was previously created by this component."""
hass_id = call.data[CONF_ID]
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
rtm_id = await self.hass.async_add_executor_job(
self._rtm_config.get_rtm_id, self._name, hass_id
)
if rtm_id is None:
_LOGGER.error(
(
@ -110,23 +114,34 @@ class RememberTheMilkEntity(Entity):
)
return
try:
result = self._rtm_api.rtm.timelines.create()
timeline = result.timeline.value
self._rtm_api.rtm.tasks.complete(
result = await self._client.rtm.timelines.create()
timeline = result.timeline
await self._client.rtm.tasks.complete(
list_id=rtm_id[0],
taskseries_id=rtm_id[1],
task_id=rtm_id[2],
timeline=timeline,
)
self._rtm_config.delete_rtm_id(self._name, hass_id)
await self.hass.async_add_executor_job(
self._rtm_config.delete_rtm_id, self._name, hass_id
)
_LOGGER.debug(
"Completed task with id %s in account %s", hass_id, self._name
)
except RtmRequestFailedException as rtm_exception:
except AuthError as err:
_LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
"Invalid authentication when completing task with id %s for account %s: %s",
hass_id,
self._name,
rtm_exception,
err,
)
self._handle_token(False)
except AioRTMError as err:
_LOGGER.error(
"Error completing task with id %s for account %s: %s",
hass_id,
self._name,
err,
)
@property
@ -140,3 +155,11 @@ class RememberTheMilkEntity(Entity):
if not self._token_valid:
return "API token invalid"
return STATE_OK
@callback
def _handle_token(self, token_valid: bool) -> None:
self._token_valid = token_valid
self.async_write_ha_state()
self.hass.async_create_task(
self.hass.config_entries.async_reload(self._config_entry_id)
)

6
homeassistant/components/remember_the_milk/manifest.json
View File

@ -2,10 +2,10 @@
"domain": "remember_the_milk",
"name": "Remember The Milk",
"codeowners": [],
"dependencies": ["configurator"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/remember_the_milk",
"iot_class": "cloud_push",
"loggers": ["rtmapi"],
"loggers": ["aiortm"],
"quality_scale": "legacy",
"requirements": ["RtmAPI==0.7.2", "httplib2==0.20.4"]
"requirements": ["aiortm==0.9.45"]
}

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

@ -0,0 +1,97 @@
"""Provide storage for Remember The Milk integration."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from homeassistant.core import HomeAssistant
from .const import CONF_ID_MAP, CONF_LIST_ID, CONF_TASK_ID, CONF_TIMESERIES_ID, LOGGER
CONFIG_FILE_NAME = ".remember_the_milk.conf"
class RememberTheMilkConfiguration:
"""Internal configuration data for Remember The Milk.
Store the map between Home Assistant task id and RTM task id.
"""
def __init__(self, hass: HomeAssistant) -> None:
"""Create new instance of configuration."""
self._config_file_path = hass.config.path(CONFIG_FILE_NAME)
self._config: dict[str, Any] = {}
def setup(self) -> None:
"""Set up the configuration."""
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 _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[int, int, int] | 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: int,
time_series_id: int,
rtm_task_id: int,
) -> None:
"""Add/Update the RTM task ID for a Home Assistant task ID."""
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, hass_id) -> 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()

31
homeassistant/components/remember_the_milk/strings.json
View File

@ -24,5 +24,36 @@
}
}
}
},
"config": {
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"shared_secret": "Shared secret"
}
},
"auth": {
"description": "Follow the link to authorize Home Assistant to access your Remember The Milk account. When done, click on the button below to continue.\n\n[Authorize]({url})"
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The Remember The Milk integration needs to re-authenticate your account"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"timeout_token": "Timeout getting access token",
"unique_id_mismatch": "The login details correspond to a different account. Please re-authenticate to the previously configured account.",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@ -512,6 +512,7 @@ FLOWS = {
"rdw",
"recollect_waste",
"refoss",
"remember_the_milk",
"renault",
"renson",
"reolink",

2
homeassistant/generated/integrations.json
View File

@ -5226,7 +5226,7 @@
"remember_the_milk": {
"name": "Remember The Milk",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "cloud_push"
},
"renault": {

9
requirements_all.txt generated
View File

@ -111,9 +111,6 @@ RachioPy==1.1.0
# homeassistant.components.python_script
RestrictedPython==8.0
# homeassistant.components.remember_the_milk
RtmAPI==0.7.2
# homeassistant.components.recorder
# homeassistant.components.sql
SQLAlchemy==2.0.38
@ -358,6 +355,9 @@ aiorecollect==2023.09.0
# homeassistant.components.ridwell
aioridwell==2024.01.0
# homeassistant.components.remember_the_milk
aiortm==0.9.45
# homeassistant.components.ruckus_unleashed
aioruckus==0.42
@ -1157,9 +1157,6 @@ homematicip==1.1.7
# homeassistant.components.horizon
horimote==0.4.1
# homeassistant.components.remember_the_milk
httplib2==0.20.4
# homeassistant.components.huawei_lte
huawei-lte-api==1.10.0

9
requirements_test_all.txt generated
View File

@ -105,9 +105,6 @@ RachioPy==1.1.0
# homeassistant.components.python_script
RestrictedPython==8.0
# homeassistant.components.remember_the_milk
RtmAPI==0.7.2
# homeassistant.components.recorder
# homeassistant.components.sql
SQLAlchemy==2.0.38
@ -340,6 +337,9 @@ aiorecollect==2023.09.0
# homeassistant.components.ridwell
aioridwell==2024.01.0
# homeassistant.components.remember_the_milk
aiortm==0.9.45
# homeassistant.components.ruckus_unleashed
aioruckus==0.42
@ -983,9 +983,6 @@ home-assistant-intents==2025.2.5
# homeassistant.components.homematicip_cloud
homematicip==1.1.7
# homeassistant.components.remember_the_milk
httplib2==0.20.4
# homeassistant.components.huawei_lte
huawei-lte-api==1.10.0

16
tests/components/remember_the_milk/conftest.py Normal file
View File

@ -0,0 +1,16 @@
"""Common fixtures for the Remember The Milk tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.remember_the_milk.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry

14
tests/components/remember_the_milk/const.py
View File

@ -3,12 +3,20 @@
import json
PROFILE = "myprofile"
TOKEN = "mytoken"
# The legacy configuration file format:
# {
# "myprofile": {
# "token": "mytoken",
# "id_map": {"123": {"list_id": 1, "timeseries_id": 2, "task_id": 3}},
# }
# }
# The new configuration file format:
JSON_STRING = json.dumps(
{
"myprofile": {
"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}},
}
}
)

351
tests/components/remember_the_milk/test_config_flow.py Normal file
View File

@ -0,0 +1,351 @@
"""Test the Remember The Milk config flow."""
import asyncio
from collections.abc import Awaitable
from copy import deepcopy
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant import config_entries
from homeassistant.components.remember_the_milk.config_flow import (
TOKEN_TIMEOUT_SEC,
AuthError,
ResponseError,
)
from homeassistant.components.remember_the_milk.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
TOKEN_DATA = {
"token": "test-token",
"user": {
"fullname": "test-fullname",
"id": "test-user-id",
"username": "test-username",
},
}
CREATE_ENTRY_DATA = {
"api_key": "test-api-key",
"shared_secret": "test-secret",
"token": "test-token",
"username": "test-username",
}
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
async def test_successful_flow(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test successful flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert not result["errors"]
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
return_value=("https://test-url.com", "test-frob"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.get_token",
return_value=TOKEN_DATA,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-fullname"
assert result["data"] == CREATE_ENTRY_DATA
assert result["result"].unique_id == "test-user-id"
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "error"),
[
(AuthError, "invalid_auth"),
(ResponseError, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_form_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test form errors when getting the authentication URL."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
side_effect=exception,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {"base": error}
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
return_value=("https://test-url.com", "test-frob"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.get_token",
return_value=TOKEN_DATA,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-fullname"
assert result["data"] == CREATE_ENTRY_DATA
assert result["result"].unique_id == "test-user-id"
assert len(mock_setup_entry.mock_calls) == 1
async def mock_get_token(*args: Any) -> None:
"""Handle get token."""
await asyncio.Future()
@pytest.mark.parametrize(
("side_effect", "reason", "timeout"),
[
(AuthError, "invalid_auth", TOKEN_TIMEOUT_SEC),
(ResponseError, "cannot_connect", TOKEN_TIMEOUT_SEC),
(Exception, "unknown", TOKEN_TIMEOUT_SEC),
(mock_get_token, "timeout_token", 0),
],
)
async def test_token_abort_reasons(
hass: HomeAssistant,
side_effect: Exception | Awaitable[None],
reason: str,
timeout: int,
) -> None:
"""Test abort result when getting token."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
return_value=("https://test-url.com", "test-frob"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
with (
patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.get_token",
side_effect=side_effect,
),
patch(
"homeassistant.components.remember_the_milk.config_flow.TOKEN_TIMEOUT_SEC",
timeout,
),
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == reason
async def test_abort_if_already_configured(hass: HomeAssistant) -> None:
"""Test abort if the same username is already configured."""
mock_entry = MockConfigEntry(domain=DOMAIN, unique_id="test-user-id")
mock_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert not result["errors"]
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
return_value=("https://test-url.com", "test-frob"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.get_token",
return_value=TOKEN_DATA,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.parametrize(
"source", [config_entries.SOURCE_IMPORT, config_entries.SOURCE_USER]
)
@pytest.mark.parametrize(
("reauth_unique_id", "abort_reason", "abort_entry_data"),
[
(
"test-user-id",
"reauth_successful",
CREATE_ENTRY_DATA | {"token": "new-test-token"},
),
("other-user-id", "unique_id_mismatch", CREATE_ENTRY_DATA),
],
)
async def test_reauth(
hass: HomeAssistant,
source: str,
reauth_unique_id: str,
abort_reason: str,
abort_entry_data: dict[str, str],
) -> None:
"""Test reauth flow."""
mock_entry = MockConfigEntry(
domain=DOMAIN, unique_id="test-user-id", data=CREATE_ENTRY_DATA, source=source
)
mock_entry.add_to_hass(hass)
result = await mock_entry.start_reauth_flow(hass)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
return_value=("https://test-url.com", "test-frob"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
reauth_data: dict[str, Any] = deepcopy(TOKEN_DATA) | {"token": "new-test-token"}
reauth_data["user"]["id"] = reauth_unique_id
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.get_token",
return_value=reauth_data,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == abort_reason
assert mock_entry.data == abort_entry_data
assert mock_entry.unique_id == "test-user-id"
async def test_import_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test import flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"api_key": "test-api-key",
"shared_secret": "test-secret",
"name": "test-name",
},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test-name"
assert result["data"] == {
"api_key": "test-api-key",
"shared_secret": "test-secret",
"token": None,
"username": "test-name",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_reauth_imported_entry(hass: HomeAssistant) -> None:
"""Test reauth flow for an imported entry."""
mock_entry = MockConfigEntry(
domain=DOMAIN,
data={
"api_key": "test-api-key",
"shared_secret": "test-secret",
"token": None,
"username": "test-name",
},
source=config_entries.SOURCE_IMPORT,
)
mock_entry.add_to_hass(hass)
result = await mock_entry.start_reauth_flow(hass)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.authenticate_desktop",
return_value=("https://test-url.com", "test-frob"),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"api_key": "test-api-key",
"shared_secret": "test-secret",
},
)
with patch(
"homeassistant.components.remember_the_milk.config_flow.Auth.get_token",
return_value=TOKEN_DATA,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert mock_entry.data == CREATE_ENTRY_DATA
assert mock_entry.unique_id == "test-user-id"

115
tests/components/remember_the_milk/test_init.py
View File

@ -1,70 +1,95 @@
"""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.core import HomeAssistant
from .const import JSON_STRING, PROFILE, TOKEN
from .const import JSON_STRING, PROFILE
def test_create_new(hass: HomeAssistant) -> None:
"""Test creating a new config file."""
def test_config_load(hass: HomeAssistant) -> None:
"""Test loading from the file."""
config = rtm.RememberTheMilkConfiguration(hass)
with (
patch("builtins.open", mock_open()),
patch("os.path.isfile", Mock(return_value=False)),
patch.object(rtm.RememberTheMilkConfiguration, "save_config"),
patch(
"homeassistant.components.remember_the_milk.storage.Path.open",
mock_open(read_data=JSON_STRING),
),
):
config = rtm.RememberTheMilkConfiguration(hass)
config.set_token(PROFILE, TOKEN)
assert config.get_token(PROFILE) == TOKEN
config.setup()
rtm_id = config.get_rtm_id(PROFILE, "123")
assert rtm_id is not None
assert rtm_id == (1, 2, 3)
def test_load_config(hass: HomeAssistant) -> None:
"""Test loading an existing token from the file."""
@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("builtins.open", mock_open(read_data=JSON_STRING)),
patch("os.path.isfile", Mock(return_value=True)),
patch(
"homeassistant.components.remember_the_milk.storage.Path.open",
side_effect=side_effect,
),
):
config = rtm.RememberTheMilkConfiguration(hass)
assert config.get_token(PROFILE) == TOKEN
config.setup()
# 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_invalid_data(hass: HomeAssistant) -> None:
"""Test starts with invalid data and should not raise an exception."""
def test_config_load_invalid_data(hass: HomeAssistant) -> None:
"""Test loading invalid data."""
config = rtm.RememberTheMilkConfiguration(hass)
with (
patch("builtins.open", mock_open(read_data="random characters")),
patch("os.path.isfile", Mock(return_value=True)),
patch(
"homeassistant.components.remember_the_milk.storage.Path.open",
mock_open(read_data="random characters"),
),
):
config = rtm.RememberTheMilkConfiguration(hass)
assert config is not None
config.setup()
# 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_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"),
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.storage.Path.open", open_mock
):
config = rtm.RememberTheMilkConfiguration(hass)
config.setup()
assert open_mock.return_value.write.call_count == 0
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)
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_STRING
config.delete_rtm_id(PROFILE, hass_id)
assert config.get_rtm_id(PROFILE, hass_id) is None
def test_load_key_map(hass: HomeAssistant) -> None:
"""Test loading an existing key map from the file."""
with (
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")
assert open_mock.return_value.write.call_count == 2
assert open_mock.return_value.write.call_args[0][0] == json.dumps(
{
"myprofile": {
"id_map": {},
}
}
)