starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-27 18:14:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add Remember the Milk todo platform (#180057)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Martin Hjelmare
2026-08-26 17:40:28 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 38ac696fa4
commit 9feefc0ebe
13 changed files with 2168 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/remember_the_milk/__init__.py
+55 -14
View File
@@ -1,19 +1,19 @@
"""The Remember The Milk integration."""
from copy import deepcopy
from dataclasses import dataclass
from typing import Any
from aiortm import AioRTMClient, AioRTMError, Auth, AuthError
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_API_KEY,
CONF_ID,
CONF_NAME,
CONF_TOKEN,
CONF_USERNAME,
Platform,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@@ -24,10 +24,17 @@ from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType
from .const import CONF_SHARED_SECRET, DOMAIN, LOGGER
from .const import CONF_LIST_ID, CONF_SHARED_SECRET, DOMAIN, LOGGER, SUBENTRY_TYPE_LIST
from .coordinator import (
RememberTheMilkConfigEntry,
RememberTheMilkData,
RtmTodoCoordinator,
)
from .entity import RememberTheMilkEntity
from .storage import RememberTheMilkConfiguration
PLATFORMS = [Platform.TODO]
RTM_SCHEMA = vol.Schema(
{
vol.Required(CONF_NAME): cv.string,
@@ -52,15 +59,6 @@ SERVICE_SCHEMA_COMPLETE_TASK = vol.Schema({vol.Required(CONF_ID): cv.string})
DATA_COMPONENT = "component"
DATA_STORAGE = "storage"
type RememberTheMilkConfigEntry = ConfigEntry[RememberTheMilkData]
@dataclass
class RememberTheMilkData:
"""Runtime data for a Remember The Milk config entry."""
entity_id: str
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Remember the milk component."""
@@ -174,7 +172,14 @@ async def async_setup_entry(
token_valid=token_valid,
)
await component.async_add_entities([entity])
entry.runtime_data = RememberTheMilkData(entity_id=entity.entity_id)
coordinator = RtmTodoCoordinator(hass, entry, client)
entry.runtime_data = RememberTheMilkData(
entity_id=entity.entity_id,
client=client,
coordinator=coordinator,
)
# The services are registered here for now because they need the account name.
# The services will be deprecated when a todo platform is added.
@@ -195,9 +200,45 @@ async def async_setup_entry(
if not token_valid:
raise ConfigEntryAuthFailed("Invalid token")
await coordinator.async_config_entry_first_refresh()
# Keep the coordinator polling even when there are no todo entities so that
# lists created later in RTM are discovered and synced to subentries.
entry.async_on_unload(coordinator.async_add_listener(lambda: None))
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def _async_update_listener(
hass: HomeAssistant, entry: RememberTheMilkConfigEntry
) -> None:
"""Delete removed lists on the server and reload when subentries change."""
data = entry.runtime_data
# Coordinator-driven syncs mutate subentries one at a time and schedule a
# single reload themselves; skip here to avoid one reload per mutation and
# to avoid deleting server lists from an incomplete mid-sync subentry set.
if data.coordinator.syncing_subentries:
return
current_list_ids = {
subentry.data[CONF_LIST_ID]
for subentry in entry.subentries.values()
if subentry.subentry_type == SUBENTRY_TYPE_LIST
}
removed_list_ids = set(data.coordinator.data or {}) - current_list_ids
if removed_list_ids:
try:
timeline_response = await data.client.rtm.timelines.create()
for list_id in removed_list_ids:
await data.client.rtm.lists.delete(
timeline=timeline_response.timeline,
list_id=list_id,
)
except AioRTMError as err:
LOGGER.warning("Failed to delete list on Remember The Milk: %s", err)
hass.config_entries.async_schedule_reload(entry.entry_id)
async def async_unload_entry(
hass: HomeAssistant, entry: RememberTheMilkConfigEntry
) -> bool:
@@ -206,4 +247,4 @@ async def async_unload_entry(
DATA_COMPONENT
]
await component.async_remove_entity(entry.runtime_data.entity_id)
return True
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
homeassistant/components/remember_the_milk/config_flow.py
+102 -3
View File
@@ -4,10 +4,18 @@ import asyncio
from collections.abc import Mapping
from typing import Any, override
from aiortm import AioRTMError, Auth, AuthError
from aiortm import AioRTMClient, AioRTMError, Auth, AuthError
import voluptuous as vol
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_API_KEY, CONF_NAME, CONF_TOKEN, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
@@ -16,7 +24,8 @@ from homeassistant.helpers.selector import (
TextSelectorType,
)
from .const import CONF_SHARED_SECRET, DOMAIN, LOGGER
from .const import CONF_LIST_ID, CONF_SHARED_SECRET, DOMAIN, LOGGER, SUBENTRY_TYPE_LIST
from .coordinator import RememberTheMilkData
TOKEN_TIMEOUT_SEC = 30
@@ -37,6 +46,14 @@ class RTMConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1
@classmethod
@override
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {SUBENTRY_TYPE_LIST: ListSubentryFlowHandler}
def __init__(self) -> None:
"""Initialize the config flow."""
self._auth: Auth | None = None
@@ -205,3 +222,85 @@ class RTMConfigFlow(ConfigFlow, domain=DOMAIN):
import_info[CONF_API_KEY],
import_info[CONF_SHARED_SECRET],
)
class ListSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for adding and reconfiguring RTM lists."""
@property
def _client(self) -> AioRTMClient:
"""Return the RTM client from the parent entry."""
data: RememberTheMilkData = self._get_entry().runtime_data
return data.client
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Create a new RTM list."""
if self._get_entry().state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
errors: dict[str, str] = {}
if user_input is not None:
name: str = user_input[CONF_NAME]
try:
timeline_response = await self._client.rtm.timelines.create()
list_response = await self._client.rtm.lists.add(
timeline=timeline_response.timeline,
name=name,
)
except AioRTMError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001 pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
new_list_id = list_response.list.id
return self.async_create_entry(
title=name,
data={CONF_LIST_ID: new_list_id},
unique_id=str(new_list_id),
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_NAME): TextSelector()}),
errors=errors,
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Rename the RTM list on the server and update the sub-entry title."""
if self._get_entry().state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
subentry = self._get_reconfigure_subentry()
errors: dict[str, str] = {}
if user_input is not None:
name: str = user_input[CONF_NAME]
try:
timeline_response = await self._client.rtm.timelines.create()
await self._client.rtm.lists.set_name(
timeline=timeline_response.timeline,
list_id=subentry.data[CONF_LIST_ID],
name=name,
)
except AioRTMError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001 pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return self.async_update_and_abort(
self._get_entry(),
subentry,
title=name,
data=subentry.data,
)
return self.async_show_form(
step_id="reconfigure",
data_schema=self.add_suggested_values_to_schema(
vol.Schema({vol.Required(CONF_NAME): TextSelector()}),
{CONF_NAME: subentry.title},
),
errors=errors,
)
homeassistant/components/remember_the_milk/const.py
+2
View File
@@ -2,6 +2,8 @@
import logging
CONF_LIST_ID = "list_id"
CONF_SHARED_SECRET = "shared_secret"
DOMAIN = "remember_the_milk"
LOGGER = logging.getLogger(__package__)
SUBENTRY_TYPE_LIST = "list"
homeassistant/components/remember_the_milk/coordinator.py
+175
View File
@@ -0,0 +1,175 @@
"""DataUpdateCoordinator for the Remember The Milk integration."""
import asyncio
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from types import MappingProxyType
from typing import override
from aiortm import AioRTMClient, AioRTMError, AuthError
from homeassistant.components.todo import TodoItem, TodoItemStatus
from homeassistant.config_entries import ConfigEntry, ConfigSubentry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_LIST_ID, DOMAIN, LOGGER, SUBENTRY_TYPE_LIST
UPDATE_INTERVAL = timedelta(minutes=5)
@dataclass(kw_only=True, frozen=True)
class RtmList:
"""An RTM list with its name and current tasks."""
name: str
tasks: dict[str, RtmTask]
@dataclass(kw_only=True, frozen=True)
class RtmTask:
"""An RTM task with its HA representation and note metadata."""
uid: str
todo_item: TodoItem
note_id: int | None
@dataclass(kw_only=True, frozen=True)
class RememberTheMilkData:
"""Runtime data for a Remember The Milk config entry."""
entity_id: str
client: AioRTMClient
coordinator: RtmTodoCoordinator
type RememberTheMilkConfigEntry = ConfigEntry[RememberTheMilkData]
class RtmTodoCoordinator(DataUpdateCoordinator[dict[int, RtmList]]):
"""Coordinator for updating task data from RTM."""
config_entry: RememberTheMilkConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: RememberTheMilkConfigEntry,
client: AioRTMClient,
) -> None:
"""Initialize the RTM coordinator."""
super().__init__(
hass,
LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=UPDATE_INTERVAL,
)
self.client = client
self.syncing_subentries = False
@override
async def _async_update_data(self) -> dict[int, RtmList]:
"""Fetch lists and tasks from the RTM API and sync subentries."""
try:
lists_response, tasks_response = await asyncio.gather(
self.client.rtm.lists.get_list(),
self.client.rtm.tasks.get_list(),
)
except AuthError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="invalid_auth",
) from err
except AioRTMError as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="api_error",
) from err
result: dict[int, RtmList] = {
lst.id: RtmList(name=lst.name, tasks={})
for lst in lists_response.lists
if not (lst.smart or lst.archived or lst.locked or lst.deleted)
}
for task_list in tasks_response.tasks.task_list:
if task_list.id not in result:
continue
for taskseries in task_list.taskseries:
for task in taskseries.task:
if task.deleted is not None:
continue
uid = f"{task_list.id}_{taskseries.id}_{task.id}"
status = (
TodoItemStatus.COMPLETED
if task.completed is not None
else TodoItemStatus.NEEDS_ACTION
)
due: date | datetime | None = None
if task.due is not None:
due = task.due if task.has_due_time else task.due.date()
description: str | None = None
note_id: int | None = None
if taskseries.notes:
first_note = taskseries.notes[0]
description = first_note.body or None
note_id = first_note.id
result[task_list.id].tasks[uid] = RtmTask(
uid=uid,
todo_item=TodoItem(
uid=uid,
summary=taskseries.name,
status=status,
due=due,
description=description,
),
note_id=note_id,
)
# Schedule after return so self.data is set before the sync runs.
# The update listener fired by subentry mutations reads coordinator.data,
# and eager task start means it can run mid-callback synchronously.
self.config_entry.async_create_task(
self.hass, self._async_sync_subentries(result), eager_start=False
)
return result
async def _async_sync_subentries(self, lists: dict[int, RtmList]) -> None:
"""Add, update, or remove list subentries to match the fetched lists."""
entry = self.config_entry
existing = {
subentry.data[CONF_LIST_ID]: subentry
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_LIST)
}
self.syncing_subentries = True
changed = False
try:
for list_id, rtm_list in lists.items():
subentry = existing.get(list_id)
if subentry is None:
self.hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType({CONF_LIST_ID: list_id}),
subentry_type=SUBENTRY_TYPE_LIST,
title=rtm_list.name,
unique_id=str(list_id),
),
)
changed = True
elif subentry.title != rtm_list.name:
self.hass.config_entries.async_update_subentry(
entry, subentry, title=rtm_list.name
)
changed = True
for list_id, subentry in existing.items():
if list_id not in lists:
self.hass.config_entries.async_remove_subentry(
entry, subentry.subentry_id
)
changed = True
finally:
self.syncing_subentries = False
if changed:
self.hass.config_entries.async_schedule_reload(entry.entry_id)
homeassistant/components/remember_the_milk/manifest.json
+1 -1
View File
@@ -5,7 +5,7 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/remember_the_milk",
"integration_type": "service",
"iot_class": "cloud_push",
"iot_class": "cloud_polling",
"loggers": ["aiortm"],
"quality_scale": "legacy",
"requirements": ["aiortm==0.20.0"]
homeassistant/components/remember_the_milk/strings.json
+45
View File
@@ -35,6 +35,51 @@
}
}
},
"config_subentries": {
"list": {
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"entry_not_loaded": "The Remember The Milk account must be loaded before adding or renaming a list.",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "List",
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"initiate_flow": {
"user": "Add list"
},
"step": {
"reconfigure": {
"data": {
"name": "[%key:common::config_flow::data::name%]"
},
"data_description": {
"name": "The new name for this To-do list. The list will also be renamed in your Remember The Milk account."
},
"description": "Rename the To-do list."
},
"user": {
"data": {
"name": "[%key:common::config_flow::data::name%]"
},
"data_description": {
"name": "A name for this To-do list. A new list with this name will also be created in your Remember The Milk account."
},
"description": "Enter a name for the new Remember The Milk list."
}
}
}
},
"exceptions": {
"api_error": {
"message": "Error communicating with the Remember The Milk API."
},
"invalid_auth": {
"message": "Invalid authentication for the Remember The Milk API."
}
},
"issues": {
"deprecated_yaml_import_issue_cannot_connect": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, a connection error occurred. Please restart Home Assistant to try again, or remove the {domain} configuration from your YAML and set the integration up via the UI.",
homeassistant/components/remember_the_milk/todo.py
+252
View File
@@ -0,0 +1,252 @@
"""A todo platform for Remember The Milk."""
import asyncio
from collections.abc import Awaitable, Callable, Coroutine
from datetime import datetime
from functools import wraps
from typing import TYPE_CHECKING, Any, cast, override
from aiortm import AioRTMError, AuthError
from homeassistant.components.todo import (
TodoItem,
TodoItemStatus,
TodoListEntity,
TodoListEntityFeature,
)
from homeassistant.config_entries import ConfigSubentry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_LIST_ID, DOMAIN, SUBENTRY_TYPE_LIST
from .coordinator import RememberTheMilkConfigEntry, RtmTodoCoordinator
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: RememberTheMilkConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the RTM todo platform."""
coordinator = entry.runtime_data.coordinator
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_LIST):
async_add_entities(
[RtmTodoListEntity(coordinator, subentry)],
config_subentry_id=subentry.subentry_id,
)
def handle_api_errors[**_P](
func: Callable[_P, Awaitable[None]],
) -> Callable[_P, Coroutine[Any, Any, None]]:
"""Catch aiortm errors and re-raise as HomeAssistantError."""
@wraps(func)
async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
try:
await func(*args, **kwargs)
except AuthError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="invalid_auth",
) from err
except AioRTMError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="api_error",
) from err
return wrapper
class RtmTodoListEntity(CoordinatorEntity[RtmTodoCoordinator], TodoListEntity):
"""A Remember The Milk TodoListEntity."""
_attr_has_entity_name = True
_attr_name = None
_attr_supported_features = (
TodoListEntityFeature.CREATE_TODO_ITEM
| TodoListEntityFeature.UPDATE_TODO_ITEM
| TodoListEntityFeature.DELETE_TODO_ITEM
| TodoListEntityFeature.SET_DUE_DATE_ON_ITEM
| TodoListEntityFeature.SET_DUE_DATETIME_ON_ITEM
| TodoListEntityFeature.SET_DESCRIPTION_ON_ITEM
)
def __init__(
self,
coordinator: RtmTodoCoordinator,
subentry: ConfigSubentry,
) -> None:
"""Initialize the RtmTodoListEntity."""
super().__init__(coordinator=coordinator)
self._list_id: int = subentry.data[CONF_LIST_ID]
self._attr_unique_id = subentry.subentry_id
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, subentry.subentry_id)},
name=subentry.title,
manufacturer="Remember The Milk",
entry_type=dr.DeviceEntryType.SERVICE,
)
@property
@override
def todo_items(self) -> list[TodoItem]:
"""Return the To-do items in the To-do list."""
rtm_list = self.coordinator.data.get(self._list_id)
return [
rtm_task.todo_item
for rtm_task in (rtm_list.tasks.values() if rtm_list else [])
]
@handle_api_errors
@override
async def async_create_todo_item(self, item: TodoItem) -> None:
"""Create a To-do item."""
client = self.coordinator.client
timeline_response = await client.rtm.timelines.create()
timeline = timeline_response.timeline
if TYPE_CHECKING:
assert item.summary is not None
result = await client.rtm.tasks.add(
timeline=timeline,
name=item.summary,
list_id=self._list_id,
parse=True,
)
taskseries = result.task_list.taskseries[0]
task = taskseries.task[0]
if item.due is not None:
await client.rtm.tasks.set_due_date(
timeline=timeline,
list_id=result.task_list.id,
taskseries_id=taskseries.id,
task_id=task.id,
due=(
item.due if isinstance(item.due, datetime) else item.due.isoformat()
),
has_due_time=isinstance(item.due, datetime),
)
if item.description:
await client.rtm.tasks.notes.add(
timeline=timeline,
list_id=result.task_list.id,
taskseries_id=taskseries.id,
task_id=task.id,
title="",
text=item.description,
)
await self.coordinator.async_refresh()
@handle_api_errors
@override
async def async_update_todo_item(self, item: TodoItem) -> None:
"""Update a To-do item."""
uid = cast(str, item.uid)
list_id, taskseries_id, task_id = _parse_uid(uid)
rtm_list = self.coordinator.data.get(self._list_id)
existing = rtm_list.tasks.get(uid) if rtm_list else None
client = self.coordinator.client
timeline_response = await client.rtm.timelines.create()
timeline = timeline_response.timeline
if item.summary is not None:
await client.rtm.tasks.set_name(
timeline=timeline,
list_id=list_id,
taskseries_id=taskseries_id,
task_id=task_id,
name=item.summary,
)
if item.status is not None:
if item.status == TodoItemStatus.COMPLETED:
await client.rtm.tasks.complete(
timeline=timeline,
list_id=list_id,
taskseries_id=taskseries_id,
task_id=task_id,
)
else:
await client.rtm.tasks.uncomplete( # codespell:ignore uncomplete
timeline=timeline,
list_id=list_id,
taskseries_id=taskseries_id,
task_id=task_id,
)
await client.rtm.tasks.set_due_date(
timeline=timeline,
list_id=list_id,
taskseries_id=taskseries_id,
task_id=task_id,
due=(
item.due
if isinstance(item.due, (datetime, type(None)))
else item.due.isoformat()
),
has_due_time=(
isinstance(item.due, datetime) if item.due is not None else None
),
)
note_id = existing.note_id if existing else None
new_description = item.description or None
if new_description and note_id is not None:
await client.rtm.tasks.notes.edit(
timeline=timeline,
note_id=note_id,
title="",
text=new_description,
)
elif new_description:
await client.rtm.tasks.notes.add(
timeline=timeline,
list_id=list_id,
taskseries_id=taskseries_id,
task_id=task_id,
title="",
text=new_description,
)
elif (
note_id is not None
and existing is not None
and existing.todo_item.description is not None
):
await client.rtm.tasks.notes.delete(
timeline=timeline,
note_id=note_id,
)
await self.coordinator.async_refresh()
@handle_api_errors
@override
async def async_delete_todo_items(self, uids: list[str]) -> None:
"""Delete To-do items."""
client = self.coordinator.client
timeline_response = await client.rtm.timelines.create()
timeline = timeline_response.timeline
await asyncio.gather(
*[
client.rtm.tasks.delete(
timeline=timeline,
list_id=list_id,
taskseries_id=taskseries_id,
task_id=task_id,
)
for uid in uids
for list_id, taskseries_id, task_id in (_parse_uid(uid),)
]
)
await self.coordinator.async_refresh()
def _parse_uid(uid: str) -> tuple[int, int, int]:
"""Split a task UID into (list_id, taskseries_id, task_id)."""
parts = uid.split("_", 2)
return int(parts[0]), int(parts[1]), int(parts[2])
homeassistant/generated/integrations.json
+1 -1
View File
@@ -6058,7 +6058,7 @@
"name": "Remember The Milk",
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_push"
"iot_class": "cloud_polling"
},
"remote_calendar": {
"integration_type": "service",
tests/components/remember_the_milk/conftest.py
+70 -9
View File
@@ -1,6 +1,6 @@
"""Provide common pytest fixtures."""
from collections.abc import AsyncGenerator, Generator
from collections.abc import AsyncGenerator, Callable, Generator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -20,7 +20,7 @@ def ignore_missing_translations(request: pytest.FixtureRequest) -> list[str]:
The services are only registered when the integration is set up, so only
ignore them for the test modules that load the integration.
"""
if request.module.__name__.endswith((".test_entity", ".test_init")):
if request.module.__name__.endswith((".test_entity", ".test_init", ".test_todo")):
return [
f"component.{DOMAIN}.services.{PROFILE}_create_task.",
f"component.{DOMAIN}.services.{PROFILE}_complete_task.",
@@ -53,19 +53,39 @@ def client_fixture() -> Generator[MagicMock]:
timelines = MagicMock()
timelines.timeline = 1234
client.rtm.timelines.create = AsyncMock(return_value=timelines)
response = MagicMock()
response.task_list.id = 1
response.task_list.taskseries = []
task_modified_response = MagicMock()
task_modified_response.task_list.id = 1
task_modified_response.task_list.taskseries = []
task_series = MagicMock()
task_series.id = 2
task_series.task = []
task = MagicMock()
task.id = 3
task_series.task.append(task)
response.task_list.taskseries.append(task_series)
client.rtm.tasks.add = AsyncMock(return_value=response)
client.rtm.tasks.complete = AsyncMock(return_value=response)
client.rtm.tasks.set_name = AsyncMock(return_value=response)
task_modified_response.task_list.taskseries.append(task_series)
client.rtm.tasks.add = AsyncMock(return_value=task_modified_response)
client.rtm.tasks.complete = AsyncMock(return_value=task_modified_response)
client.rtm.tasks.uncomplete = AsyncMock( # codespell:ignore uncomplete
return_value=task_modified_response
)
client.rtm.tasks.delete = AsyncMock(return_value=task_modified_response)
client.rtm.tasks.set_name = AsyncMock(return_value=task_modified_response)
client.rtm.tasks.set_due_date = AsyncMock(return_value=task_modified_response)
tasks_response = MagicMock()
tasks_response.tasks.task_list = []
client.rtm.tasks.get_list = AsyncMock(return_value=tasks_response)
note_response = MagicMock()
client.rtm.tasks.notes.add = AsyncMock(return_value=note_response)
client.rtm.tasks.notes.edit = AsyncMock(return_value=note_response)
client.rtm.tasks.notes.delete = AsyncMock(return_value=note_response)
lists_response = MagicMock()
lists_response.lists = []
client.rtm.lists.get_list = AsyncMock(return_value=lists_response)
list_add_response = MagicMock()
list_add_response.list.id = 42
client.rtm.lists.add = AsyncMock(return_value=list_add_response)
client.rtm.lists.set_name = AsyncMock(return_value=MagicMock())
client.rtm.lists.delete = AsyncMock(return_value=MagicMock())
yield client
@@ -94,6 +114,47 @@ def config_entry(hass: HomeAssistant) -> MockConfigEntry:
return entry
@pytest.fixture
def make_rtm_list_mock() -> Callable[..., MagicMock]:
"""Return a factory that creates a single RTM list mock with optional flags."""
def factory(
list_id: int,
name: str,
*,
smart: bool = False,
archived: bool = False,
locked: bool = False,
deleted: bool = False,
) -> MagicMock:
lst = MagicMock()
lst.id = list_id
lst.name = name
lst.smart = smart
lst.archived = archived
lst.locked = locked
lst.deleted = deleted
return lst
return factory
@pytest.fixture
def rtm_list_mock(
client: MagicMock, make_rtm_list_mock: Callable[..., MagicMock]
) -> Callable[[int, str], MagicMock]:
"""Return a helper that configures get_list to return a single standard list."""
def factory(list_id: int, name: str) -> MagicMock:
lst = make_rtm_list_mock(list_id, name)
response = MagicMock()
response.lists = [lst]
client.rtm.lists.get_list.return_value = response
return lst
return factory
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
tests/components/remember_the_milk/snapshots/test_todo.ambr
+31
View File
@@ -0,0 +1,31 @@
# serializer version: 1
# name: test_device_entry
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': <DeviceEntryType.SERVICE: 'service'>,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'remember_the_milk',
'test-subentry-id',
),
}),
'labels': set({
}),
'manufacturer': 'Remember The Milk',
'model': None,
'model_id': None,
'name': 'My Shopping List',
'name_by_user': None,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---
tests/components/remember_the_milk/test_config_flow.py
+301 -27
View File
@@ -1,10 +1,10 @@
"""Test the Remember The Milk config flow."""
import asyncio
from collections.abc import Awaitable
from collections.abc import Awaitable, Callable
from copy import deepcopy
from typing import Any
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
from aiortm import AioRTMError, AuthError
import pytest
@@ -12,7 +12,17 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.remember_the_milk.config_flow import TOKEN_TIMEOUT_SEC
from homeassistant.components.remember_the_milk.const import DOMAIN
from homeassistant.components.remember_the_milk.const import (
CONF_LIST_ID,
DOMAIN,
SUBENTRY_TYPE_LIST,
)
from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
SOURCE_USER,
ConfigSubentryDataWithId,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@@ -20,7 +30,21 @@ from .const import CREATE_ENTRY_DATA, PROFILE, TOKEN_RESPONSE
from tests.common import MockConfigEntry
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
SUBENTRY_ID = "test-subentry-id"
LIST_ID = 99
NEW_LIST_ID = 42
@pytest.fixture
def ignore_missing_translations(request: pytest.FixtureRequest) -> list[str]:
"""Ignore per-account service translations for subentry tests that do real setup."""
for marker in request.node.iter_markers("usefixtures"):
if "storage" in marker.args:
return [
f"component.{DOMAIN}.services.{PROFILE}_create_task.",
f"component.{DOMAIN}.services.{PROFILE}_complete_task.",
]
return []
def get_suggested_value(data_schema: vol.Schema, key: str) -> Any:
@@ -31,10 +55,32 @@ def get_suggested_value(data_schema: vol.Schema, key: str) -> Any:
return None
@pytest.fixture
def config_entry_with_subentry(hass: HomeAssistant) -> MockConfigEntry:
"""Return a loaded mock config entry with one list subentry."""
entry = MockConfigEntry(
data=CREATE_ENTRY_DATA,
domain=DOMAIN,
subentries_data=[
ConfigSubentryDataWithId(
data={CONF_LIST_ID: LIST_ID},
subentry_type=SUBENTRY_TYPE_LIST,
title="Shopping",
unique_id=str(LIST_ID),
subentry_id=SUBENTRY_ID,
)
],
)
entry.add_to_hass(hass)
return entry
@pytest.mark.usefixtures("client")
async def test_successful_flow(
hass: HomeAssistant, client: AsyncMock, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
) -> None:
"""Test successful flow."""
"""Test successful flow creates subentries for existing lists."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@@ -56,6 +102,7 @@ async def test_successful_flow(
assert result["data"] == CREATE_ENTRY_DATA
assert result["result"].unique_id == TOKEN_RESPONSE["user"]["id"]
assert len(mock_setup_entry.mock_calls) == 1
assert len(result["result"].subentries) == 0
@pytest.mark.parametrize(
@@ -66,11 +113,11 @@ async def test_successful_flow(
(Exception, "unknown"),
],
)
@pytest.mark.usefixtures("client")
async def test_form_errors(
hass: HomeAssistant,
client: AsyncMock,
mock_setup_entry: AsyncMock,
exception: Exception,
exception: type[Exception],
error: str,
) -> None:
"""Test form errors when getting the authentication URL."""
@@ -108,6 +155,7 @@ async def test_form_errors(
assert result["data"] == CREATE_ENTRY_DATA
assert result["result"].unique_id == TOKEN_RESPONSE["user"]["id"]
assert len(mock_setup_entry.mock_calls) == 1
assert len(result["result"].subentries) == 0
async def mock_get_token(*args: Any) -> None:
@@ -115,6 +163,7 @@ async def mock_get_token(*args: Any) -> None:
await asyncio.Future()
@pytest.mark.usefixtures("client", "mock_setup_entry")
@pytest.mark.parametrize(
("side_effect", "reason", "timeout"),
[
@@ -126,8 +175,7 @@ async def mock_get_token(*args: Any) -> None:
)
async def test_token_abort_reasons(
hass: HomeAssistant,
client: AsyncMock,
side_effect: Exception | Awaitable[None],
side_effect: type[Exception] | Awaitable[None],
reason: str,
timeout: int,
) -> None:
@@ -160,9 +208,8 @@ async def test_token_abort_reasons(
assert result["reason"] == reason
async def test_abort_if_already_configured(
hass: HomeAssistant, client: AsyncMock, config_entry: MockConfigEntry
) -> None:
@pytest.mark.usefixtures("client", "config_entry")
async def test_abort_if_already_configured(hass: HomeAssistant) -> None:
"""Test abort if the same username is already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@@ -184,6 +231,7 @@ async def test_abort_if_already_configured(
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("client", "mock_setup_entry")
@pytest.mark.parametrize(
"source", [config_entries.SOURCE_IMPORT, config_entries.SOURCE_USER]
)
@@ -200,7 +248,6 @@ async def test_abort_if_already_configured(
)
async def test_reauth(
hass: HomeAssistant,
client: AsyncMock,
source: str,
reauth_unique_id: str,
abort_reason: str,
@@ -248,9 +295,8 @@ async def test_reauth(
assert mock_entry.unique_id == "test-user-id"
async def test_reauth_change_credentials(
hass: HomeAssistant, client: AsyncMock
) -> None:
@pytest.mark.usefixtures("client", "mock_setup_entry")
async def test_reauth_change_credentials(hass: HomeAssistant) -> None:
"""Test reauth flow where the user changes the stored credentials."""
mock_entry = MockConfigEntry(
domain=DOMAIN, unique_id=TOKEN_RESPONSE["user"]["id"], data=CREATE_ENTRY_DATA
@@ -288,6 +334,7 @@ async def test_reauth_change_credentials(
assert mock_entry.unique_id == TOKEN_RESPONSE["user"]["id"]
@pytest.mark.usefixtures("client", "mock_setup_entry")
@pytest.mark.parametrize(
("exception", "error"),
[
@@ -298,8 +345,6 @@ async def test_reauth_change_credentials(
)
async def test_reauth_form_errors(
hass: HomeAssistant,
client: AsyncMock,
mock_setup_entry: AsyncMock,
exception: type[Exception],
error: str,
) -> None:
@@ -352,6 +397,7 @@ async def test_reauth_form_errors(
assert result["reason"] == "reauth_successful"
@pytest.mark.usefixtures("client")
@pytest.mark.parametrize(
("side_effect", "reason", "timeout"),
[
@@ -363,7 +409,6 @@ async def test_reauth_form_errors(
)
async def test_reauth_token_abort(
hass: HomeAssistant,
client: AsyncMock,
side_effect: type[Exception | Awaitable[None]],
reason: str,
timeout: int,
@@ -402,8 +447,10 @@ async def test_reauth_token_abort(
assert result["reason"] == reason
@pytest.mark.usefixtures("client")
async def test_import_flow(
hass: HomeAssistant, client: AsyncMock, mock_setup_entry: AsyncMock
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
) -> None:
"""Test import flow with a valid stored token."""
result = await hass.config_entries.flow.async_init(
@@ -426,8 +473,10 @@ async def test_import_flow(
}
assert result["result"].unique_id == TOKEN_RESPONSE["user"]["id"]
assert len(mock_setup_entry.mock_calls) == 1
assert len(result["result"].subentries) == 0
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("token", "side_effect", "reason"),
[
@@ -462,9 +511,8 @@ async def test_import_flow_abort(
assert result["reason"] == reason
async def test_import_flow_username_mismatch(
hass: HomeAssistant, client: AsyncMock
) -> None:
@pytest.mark.usefixtures("client")
async def test_import_flow_username_mismatch(hass: HomeAssistant) -> None:
"""Test import flow aborts when the token username doesn't match the name."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
@@ -480,9 +528,8 @@ async def test_import_flow_username_mismatch(
assert result["reason"] == "invalid_auth"
async def test_import_flow_already_configured(
hass: HomeAssistant, client: AsyncMock, config_entry: MockConfigEntry
) -> None:
@pytest.mark.usefixtures("client", "config_entry")
async def test_import_flow_already_configured(hass: HomeAssistant) -> None:
"""Test import flow aborts when the account name is already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
@@ -496,3 +543,230 @@ async def test_import_flow_already_configured(
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("storage")
async def test_create_new_list(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test creating a new RTM list creates a subentry."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, SUBENTRY_TYPE_LIST),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
# After adding the list on the server, reflect it in get_list so the coordinator
# sync keeps the newly created subentry rather than removing it on the next refresh.
list_add_return = client.rtm.lists.add.return_value
async def _add_list(*args: object, **kwargs: object) -> MagicMock:
rtm_list_mock(NEW_LIST_ID, "Work Tasks")
return list_add_return
client.rtm.lists.add.side_effect = _add_list
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_NAME: "Work Tasks"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Work Tasks"
assert result["data"] == {CONF_LIST_ID: NEW_LIST_ID}
assert result["unique_id"] == str(NEW_LIST_ID)
client.rtm.lists.add.assert_called_once_with(
timeline=1234,
name="Work Tasks",
)
assert len(config_entry.subentries) == 1
@pytest.mark.usefixtures("storage")
@pytest.mark.parametrize(
("exception", "error"),
[
(AioRTMError, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_create_error(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
exception: type[Exception],
error: str,
) -> None:
"""Test that an API error in the create step shows an error and can recover."""
client.rtm.lists.add = AsyncMock(side_effect=exception("server error"))
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, SUBENTRY_TYPE_LIST),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_NAME: "Failing List"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
# Clear the error and verify the flow recovers to success.
list_add_response = MagicMock()
list_add_response.list.id = NEW_LIST_ID
async def _add_list(*args: object, **kwargs: object) -> MagicMock:
rtm_list_mock(NEW_LIST_ID, "Failing List")
return list_add_response
client.rtm.lists.add.side_effect = _add_list
result = await hass.config_entries.subentries.async_configure(
result["flow_id"], user_input={CONF_NAME: "Failing List"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Failing List"
assert result["data"] == {CONF_LIST_ID: NEW_LIST_ID}
assert result["unique_id"] == str(NEW_LIST_ID)
async def test_entry_not_loaded(
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> None:
"""Test abort when the parent config entry is not loaded."""
result = await hass.config_entries.subentries.async_init(
(config_entry.entry_id, SUBENTRY_TYPE_LIST),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
async def test_reconfigure_entry_not_loaded(
hass: HomeAssistant,
config_entry_with_subentry: MockConfigEntry,
) -> None:
"""Test abort in reconfigure when the parent config entry is not loaded."""
subentry = next(iter(config_entry_with_subentry.subentries.values()))
result = await hass.config_entries.subentries.async_init(
(config_entry_with_subentry.entry_id, SUBENTRY_TYPE_LIST),
context={"source": SOURCE_RECONFIGURE, "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
@pytest.mark.usefixtures("storage")
async def test_reconfigure_subentry(
hass: HomeAssistant,
client: MagicMock,
config_entry_with_subentry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test renaming a subentry via reconfigure."""
lst = rtm_list_mock(LIST_ID, "Shopping")
# Reflect the rename in get_list so the coordinator sync stays idempotent after reload.
def _set_name(*args: object, name: str, **kwargs: object) -> MagicMock:
lst.name = name
return MagicMock()
client.rtm.lists.set_name.side_effect = _set_name
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
subentry = next(iter(config_entry_with_subentry.subentries.values()))
assert subentry.title == "Shopping"
result = await hass.config_entries.subentries.async_init(
(config_entry_with_subentry.entry_id, SUBENTRY_TYPE_LIST),
context={"source": SOURCE_RECONFIGURE, "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_NAME: "Grocery Shopping"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
subentry = next(iter(config_entry_with_subentry.subentries.values()))
assert subentry.title == "Grocery Shopping"
assert subentry.data[CONF_LIST_ID] == LIST_ID
client.rtm.lists.set_name.assert_called_once_with(
timeline=1234, list_id=LIST_ID, name="Grocery Shopping"
)
@pytest.mark.usefixtures("storage")
@pytest.mark.parametrize(
("exception", "error"),
[
(AioRTMError, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_reconfigure_error(
hass: HomeAssistant,
client: MagicMock,
config_entry_with_subentry: MockConfigEntry,
exception: type[Exception],
error: str,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that an API error in the reconfigure step shows an error and can recover."""
lst = rtm_list_mock(LIST_ID, "Shopping")
client.rtm.lists.set_name = AsyncMock(side_effect=exception("server error"))
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
subentry = next(iter(config_entry_with_subentry.subentries.values()))
result = await hass.config_entries.subentries.async_init(
(config_entry_with_subentry.entry_id, SUBENTRY_TYPE_LIST),
context={"source": SOURCE_RECONFIGURE, "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reconfigure"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_NAME: "Grocery Shopping"},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
# Clear the error and verify the flow recovers to success.
def _set_name(*args: object, name: str, **kwargs: object) -> MagicMock:
lst.name = name
return MagicMock()
client.rtm.lists.set_name.side_effect = _set_name
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
user_input={CONF_NAME: "Grocery Shopping"},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
subentry = next(iter(config_entry_with_subentry.subentries.values()))
assert subentry.title == "Grocery Shopping"
tests/components/remember_the_milk/test_init.py
+498 -7
View File
@@ -1,19 +1,35 @@
"""Test the Remember The Milk integration."""
from unittest.mock import MagicMock
from collections.abc import Callable
from datetime import timedelta
from types import MappingProxyType
from unittest.mock import MagicMock, patch
from aiortm import AioRTMError, AuthError
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.remember_the_milk.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.components.remember_the_milk.const import (
CONF_LIST_ID,
DOMAIN,
SUBENTRY_TYPE_LIST,
)
from homeassistant.config_entries import (
ConfigEntryState,
ConfigSubentry,
ConfigSubentryDataWithId,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from .const import PROFILE
from .const import CREATE_ENTRY_DATA, PROFILE
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
LIST_ID = 42
NEW_LIST_ID = 100
SUBENTRY_ID = "test-subentry-id"
CONFIG = {
"name": "myprofile",
@@ -22,10 +38,29 @@ CONFIG = {
}
@pytest.mark.usefixtures("storage")
@pytest.fixture
def config_entry_with_subentry(hass: HomeAssistant) -> MockConfigEntry:
"""Return a mock config entry with one list subentry."""
entry = MockConfigEntry(
data=CREATE_ENTRY_DATA,
domain=DOMAIN,
subentries_data=[
ConfigSubentryDataWithId(
data={CONF_LIST_ID: LIST_ID},
subentry_type=SUBENTRY_TYPE_LIST,
title="Shopping",
unique_id=str(LIST_ID),
subentry_id=SUBENTRY_ID,
)
],
)
entry.add_to_hass(hass)
return entry
@pytest.mark.usefixtures("client", "storage")
async def test_load_unload_config_entry(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test loading and unloading a config entry."""
@@ -92,6 +127,36 @@ async def test_import_creates_deprecation_issue(
)
@pytest.mark.parametrize(
("side_effect", "expected_state"),
[
pytest.param(
AuthError("Invalid token!"),
ConfigEntryState.SETUP_ERROR,
id="auth_error",
),
pytest.param(
AioRTMError("Boom!"),
ConfigEntryState.SETUP_RETRY,
id="api_error",
),
],
)
@pytest.mark.usefixtures("storage")
async def test_coordinator_update_errors(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test config entry state when the first coordinator refresh fails."""
client.rtm.tasks.get_list.side_effect = side_effect
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is expected_state
@pytest.mark.parametrize("ignore_missing_translations", [[]])
@pytest.mark.usefixtures("client")
async def test_import_without_token_creates_issue(
@@ -113,3 +178,429 @@ async def test_import_without_token_creates_issue(
assert issue_registry.async_get_issue(
DOMAIN, "deprecated_yaml_import_issue_invalid_auth"
)
@pytest.mark.usefixtures("storage")
async def test_remove_subentry_deletes_list(
hass: HomeAssistant,
client: MagicMock,
config_entry_with_subentry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that removing a list sub-entry deletes the list on the RTM server."""
rtm_list_mock(LIST_ID, "Shopping")
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
hass.config_entries.async_remove_subentry(config_entry_with_subentry, SUBENTRY_ID)
await hass.async_block_till_done()
client.rtm.timelines.create.assert_called_once()
client.rtm.lists.delete.assert_called_once_with(
timeline=client.rtm.timelines.create.return_value.timeline,
list_id=LIST_ID,
)
@pytest.mark.usefixtures("storage")
async def test_rename_subentry_does_not_delete_list(
hass: HomeAssistant,
client: MagicMock,
config_entry_with_subentry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that renaming a sub-entry (no list removed) does not trigger deletion."""
rtm_list_mock(LIST_ID, "Shopping")
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
subentry = next(iter(config_entry_with_subentry.subentries.values()))
hass.config_entries.async_update_subentry(
config_entry_with_subentry, subentry, title="Grocery Shopping"
)
await hass.async_block_till_done()
client.rtm.lists.delete.assert_not_called()
@pytest.mark.parametrize(
"side_effect",
[
pytest.param(AuthError("Invalid token!"), id="auth_error"),
pytest.param(AioRTMError("Boom!"), id="api_error"),
],
)
@pytest.mark.usefixtures("storage")
async def test_remove_subentry_delete_list_error(
hass: HomeAssistant,
client: MagicMock,
config_entry_with_subentry: MockConfigEntry,
side_effect: Exception,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that a server error when deleting a list is logged and reload still runs."""
rtm_list_mock(LIST_ID, "Shopping")
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
client.rtm.lists.delete.side_effect = side_effect
hass.config_entries.async_remove_subentry(config_entry_with_subentry, SUBENTRY_ID)
await hass.async_block_till_done()
client.rtm.lists.delete.assert_called_once()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
@pytest.mark.usefixtures("storage")
async def test_coordinator_creates_subentry_for_new_list(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that a list on the server creates a subentry and todo entity during first refresh."""
rtm_list_mock(LIST_ID, "Shopping")
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 1
subentry = next(iter(config_entry.subentries.values()))
assert subentry.data[CONF_LIST_ID] == LIST_ID
assert subentry.title == "Shopping"
assert subentry.unique_id == str(LIST_ID)
assert hass.states.get("todo.shopping") is not None
client.rtm.lists.delete.assert_not_called()
@pytest.mark.usefixtures("storage")
async def test_coordinator_removes_subentry_when_list_gone(
hass: HomeAssistant,
client: MagicMock,
config_entry_with_subentry: MockConfigEntry,
) -> None:
"""Test that a list gone from the server removes the subentry without a server delete."""
# rtm.lists.get_list returns empty by default — list 42 has disappeared from the server.
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
assert len(config_entry_with_subentry.subentries) == 0
client.rtm.lists.delete.assert_not_called()
@pytest.mark.usefixtures("client", "storage")
async def test_coordinator_updates_subentry_title_on_rename(
hass: HomeAssistant,
config_entry_with_subentry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that a server-side list rename updates the subentry title."""
rtm_list_mock(
LIST_ID, "Grocery Shopping"
) # Different from the subentry title "Shopping"
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
subentry = config_entry_with_subentry.subentries.get(SUBENTRY_ID)
assert subentry is not None
assert subentry.title == "Grocery Shopping"
@pytest.mark.usefixtures("storage")
async def test_coordinator_ignores_filtered_lists(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
make_rtm_list_mock: Callable[..., MagicMock],
) -> None:
"""Test that smart, archived, locked, and deleted lists are not synced as subentries."""
lists_response = MagicMock()
lists_response.lists = [
make_rtm_list_mock(1, "Normal"),
make_rtm_list_mock(2, "Smart", smart=True),
make_rtm_list_mock(3, "Archived", archived=True),
make_rtm_list_mock(4, "Locked", locked=True),
make_rtm_list_mock(5, "Deleted", deleted=True),
]
client.rtm.lists.get_list.return_value = lists_response
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 1
subentry = next(iter(config_entry.subentries.values()))
assert subentry.title == "Normal"
assert subentry.data[CONF_LIST_ID] == 1
@pytest.mark.usefixtures("storage")
async def test_coordinator_skips_tasks_for_filtered_list(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
) -> None:
"""Test that tasks for a list absent from the lists result (e.g. filtered) are ignored."""
rtm_list_mock(LIST_ID, "Shopping")
# A task_list whose id is not in the coordinator result (simulates a filtered list).
task_list = MagicMock()
task_list.id = 999
tasks_response = MagicMock()
tasks_response.tasks.task_list = [task_list]
client.rtm.tasks.get_list.return_value = tasks_response
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 1
@pytest.mark.parametrize(
("side_effect", "expected_state", "ignore_missing_translations"),
[
pytest.param(
AuthError("Invalid token!"),
ConfigEntryState.SETUP_ERROR,
[
f"component.{DOMAIN}.services.{PROFILE}_create_task.",
f"component.{DOMAIN}.services.{PROFILE}_complete_task.",
],
id="auth_error",
),
pytest.param(
AioRTMError("Boom!"),
ConfigEntryState.SETUP_RETRY,
[
f"component.{DOMAIN}.services.{PROFILE}_create_task.",
f"component.{DOMAIN}.services.{PROFILE}_complete_task.",
],
id="api_error",
),
],
)
@pytest.mark.usefixtures("storage")
async def test_coordinator_lists_fetch_errors(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test config entry state when the list fetch in the coordinator fails."""
client.rtm.lists.get_list.side_effect = side_effect
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is expected_state
@pytest.mark.usefixtures("storage")
async def test_coordinator_does_not_delete_server_removed_list(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that a list removed from the server on a subsequent poll is not deleted.
The update listener computes lists to delete by comparing the current subentries
against the coordinator's last-known server data. If it runs before coordinator.data
is updated with the fresh (shorter) list, a server-side removal looks like a
user-initiated deletion and the list would be permanently deleted on the server.
"""
rtm_list_mock(LIST_ID, "Shopping")
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 1
# List disappears from the server (e.g. archived) on the next poll.
lists_response = MagicMock()
lists_response.lists = []
client.rtm.lists.get_list.return_value = lists_response
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(config_entry.subentries) == 0
client.rtm.lists.delete.assert_not_called()
@pytest.mark.usefixtures("storage")
async def test_coordinator_sync_multiple_new_lists_no_deletion(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
make_rtm_list_mock: Callable[..., MagicMock],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that discovering multiple new lists in one poll doesn't delete any of them.
Each async_add_subentry call fires the update listener eagerly. Without the
syncing_subentries guard, the listener would see an incomplete subentry set
mid-sync and wrongly delete the not-yet-added list from the server.
"""
lists_response = MagicMock()
lists_response.lists = [make_rtm_list_mock(LIST_ID, "Shopping")]
client.rtm.lists.get_list.return_value = lists_response
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 1
# Two new lists appear simultaneously on the server on the next poll.
lists_response = MagicMock()
lists_response.lists = [
make_rtm_list_mock(LIST_ID, "Shopping"),
make_rtm_list_mock(LIST_ID + 1, "Work"),
make_rtm_list_mock(LIST_ID + 2, "Personal"),
]
client.rtm.lists.get_list.return_value = lists_response
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(config_entry.subentries) == 3
client.rtm.lists.delete.assert_not_called()
@pytest.mark.usefixtures("storage")
async def test_coordinator_sync_multiple_lists_single_reload(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
make_rtm_list_mock: Callable[..., MagicMock],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that discovering multiple new lists in one poll schedules a single reload."""
# Empty list response during setup so no subentries are added and no reload fires.
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 0
# Three new lists appear simultaneously on the next poll.
lists_response = MagicMock()
lists_response.lists = [
make_rtm_list_mock(LIST_ID, "Shopping"),
make_rtm_list_mock(LIST_ID + 1, "Work"),
make_rtm_list_mock(LIST_ID + 2, "Personal"),
]
client.rtm.lists.get_list.return_value = lists_response
with patch.object(hass.config_entries, "async_schedule_reload") as mock_reload:
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
mock_reload.assert_called_once_with(config_entry.entry_id)
@pytest.mark.usefixtures("storage")
async def test_coordinator_sync_no_changes_no_reload(
hass: HomeAssistant,
config_entry_with_subentry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that a coordinator poll with unchanged subentries schedules no reload."""
rtm_list_mock(LIST_ID, "Shopping")
await hass.config_entries.async_setup(config_entry_with_subentry.entry_id)
await hass.async_block_till_done()
assert config_entry_with_subentry.state is ConfigEntryState.LOADED
assert len(config_entry_with_subentry.subentries) == 1
with patch.object(hass.config_entries, "async_schedule_reload") as mock_reload:
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
mock_reload.assert_not_called()
@pytest.mark.usefixtures("client", "storage")
async def test_coordinator_polls_when_no_entities(
hass: HomeAssistant,
config_entry: MockConfigEntry,
rtm_list_mock: Callable[[int, str], MagicMock],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test that the coordinator keeps polling even when there are no todo entities.
When there are no eligible RTM lists there are no subentries and therefore no
CoordinatorEntity listeners. Without a permanent listener the coordinator stops
scheduling refreshes after the first one, so lists created later in RTM are
never discovered.
"""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert len(config_entry.subentries) == 0
# A new list appears on the server.
rtm_list_mock(LIST_ID, "Shopping")
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(config_entry.subentries) == 1
subentry = next(iter(config_entry.subentries.values()))
assert subentry.data[CONF_LIST_ID] == LIST_ID
@pytest.mark.usefixtures("client", "storage")
async def test_update_listener_registered_before_forward(
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> None:
"""Test that a subentry added during platform forwarding triggers a reload.
The update listener must be registered before async_forward_entry_setups so
that a subentry appearing in the reconciliation window (after todo.async_setup_entry
enumerated subentries but before the listener is installed) still fires a reload and
gets picked up.
"""
original_forward = hass.config_entries.async_forward_entry_setups
async def forward_and_add(entry, platforms):
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType({CONF_LIST_ID: NEW_LIST_ID}),
subentry_type=SUBENTRY_TYPE_LIST,
title="Late List",
unique_id=str(NEW_LIST_ID),
),
)
await original_forward(entry, platforms)
with (
patch.object(
hass.config_entries,
"async_forward_entry_setups",
side_effect=forward_and_add,
),
patch.object(hass.config_entries, "async_schedule_reload") as mock_reload,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
mock_reload.assert_called_with(config_entry.entry_id)
tests/components/remember_the_milk/test_todo.py
+635
View File
@@ -0,0 +1,635 @@
"""Test the Remember The Milk todo platform."""
from collections.abc import Callable
from datetime import UTC, datetime
from unittest.mock import MagicMock
from aiortm import AioRTMError, AuthError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.remember_the_milk.const import (
CONF_LIST_ID,
DOMAIN,
SUBENTRY_TYPE_LIST,
)
from homeassistant.components.todo import (
ATTR_DESCRIPTION,
ATTR_DUE_DATE,
ATTR_ITEM,
ATTR_RENAME,
ATTR_STATUS,
DOMAIN as TODO_DOMAIN,
TodoItemStatus,
TodoServices,
)
from homeassistant.config_entries import ConfigSubentryDataWithId
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from .const import CREATE_ENTRY_DATA
from tests.common import MockConfigEntry
SUBENTRY_ID = "test-subentry-id"
LIST_ID = 99
ENTITY_ID = "todo.my_shopping_list"
def _make_task_list(
list_id: int,
taskseries_id: int,
task_id: int,
name: str,
completed: datetime | None = None,
deleted: datetime | None = None,
due: datetime | None = None,
has_due_time: bool = False,
notes: list | None = None,
) -> MagicMock:
"""Build a minimal mock RTM task list response for one task."""
task_list = MagicMock()
task_list.id = list_id
taskseries = MagicMock()
taskseries.id = taskseries_id
taskseries.name = name
taskseries.notes = notes or []
task = MagicMock()
task.id = task_id
task.completed = completed
task.deleted = deleted
task.due = due
task.has_due_time = has_due_time
taskseries.task = [task]
task_list.taskseries = [taskseries]
return task_list
def _set_tasks_response(client: MagicMock, *task_lists: MagicMock) -> None:
"""Configure the client to return the given task lists from tasks.get_list."""
tasks_response = MagicMock()
tasks_response.tasks.task_list = list(task_lists)
client.rtm.tasks.get_list.return_value = tasks_response
@pytest.fixture(autouse=True)
def _lists_response(rtm_list_mock: Callable[[int, str], MagicMock]) -> None:
"""Return list 99 for all todo tests so the list subentry is kept during coordinator sync."""
rtm_list_mock(LIST_ID, "My Shopping List")
@pytest.fixture
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Return a mock config entry with one list subentry."""
entry = MockConfigEntry(
data=CREATE_ENTRY_DATA,
domain=DOMAIN,
subentries_data=[
ConfigSubentryDataWithId(
data={CONF_LIST_ID: LIST_ID},
subentry_type=SUBENTRY_TYPE_LIST,
title="My Shopping List",
unique_id=str(LIST_ID),
subentry_id=SUBENTRY_ID,
)
],
)
entry.add_to_hass(hass)
return entry
@pytest.mark.usefixtures("storage")
async def test_entity_state(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test that the entity state reflects incomplete task count."""
_set_tasks_response(
client,
_make_task_list(LIST_ID, 10, 1, "Buy milk"),
_make_task_list(
LIST_ID, 20, 2, "Eggs", completed=datetime(2024, 1, 1, tzinfo=UTC)
),
_make_task_list(
LIST_ID, 30, 3, "Bread", deleted=datetime(2024, 1, 1, tzinfo=UTC)
),
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state is not None
# 1 active ("Buy milk"), 1 completed ("Eggs"), 1 deleted ("Bread" — excluded)
assert state.state == "1"
@pytest.mark.usefixtures("client", "storage")
async def test_device_entry(
hass: HomeAssistant,
config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test that a device entry is created for the todo list entity."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
device_entry = device_registry.async_get_device_by_identifier(
(DOMAIN, SUBENTRY_ID), config_entry.entry_id
)
assert device_entry is not None
assert device_entry == snapshot
@pytest.mark.usefixtures("storage")
async def test_create_todo_item(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test creating a todo item calls the RTM tasks.add API."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.ADD_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: "Buy butter"},
blocking=True,
)
client.rtm.tasks.add.assert_called_once_with(
timeline=1234,
name="Buy butter",
list_id=LIST_ID,
parse=True,
)
client.rtm.tasks.get_list.assert_called()
@pytest.mark.usefixtures("storage")
async def test_create_todo_item_with_due_date(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test creating a todo item with a due date."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.ADD_ITEM,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_ITEM: "Buy butter",
ATTR_DUE_DATE: "2024-03-15",
},
blocking=True,
)
client.rtm.tasks.add.assert_called_once()
client.rtm.tasks.set_due_date.assert_called_once_with(
timeline=1234,
list_id=1, # from mock response task_list.id
taskseries_id=2, # from mock response taskseries.id
task_id=3, # from mock response task.id
due="2024-03-15",
has_due_time=False,
)
@pytest.mark.usefixtures("storage")
async def test_create_todo_item_with_description(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test creating a todo item with a description adds a note."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.ADD_ITEM,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_ITEM: "Buy butter",
ATTR_DESCRIPTION: "Full fat please",
},
blocking=True,
)
client.rtm.tasks.add.assert_called_once()
client.rtm.tasks.notes.add.assert_called_once_with(
timeline=1234,
list_id=1,
taskseries_id=2,
task_id=3,
title="",
text="Full fat please",
)
@pytest.mark.parametrize(
"new_name",
[
pytest.param("Buy whole milk", id="different_name"),
pytest.param("Buy milk", id="same_name"),
],
)
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_rename(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
new_name: str,
) -> None:
"""Test renaming a todo item always calls tasks.set_name, even when the name is unchanged."""
_set_tasks_response(client, _make_task_list(LIST_ID, 10, 1, "Buy milk"))
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: uid, ATTR_RENAME: new_name},
blocking=True,
)
client.rtm.tasks.set_name.assert_called_once_with(
timeline=1234,
list_id=LIST_ID,
taskseries_id=10,
task_id=1,
name=new_name,
)
@pytest.mark.parametrize(
("initial_completed", "target_status", "expected_api", "not_expected_api"),
[
pytest.param(
None,
TodoItemStatus.COMPLETED,
"complete",
"uncomplete", # codespell:ignore uncomplete
id="incomplete_to_complete",
),
pytest.param(
datetime(2024, 1, 1, tzinfo=UTC),
TodoItemStatus.NEEDS_ACTION,
"uncomplete", # codespell:ignore uncomplete
"complete",
id="complete_to_incomplete",
),
pytest.param(
datetime(2024, 1, 1, tzinfo=UTC),
TodoItemStatus.COMPLETED,
"complete",
"uncomplete", # codespell:ignore uncomplete
id="already_complete",
),
pytest.param(
None,
TodoItemStatus.NEEDS_ACTION,
"uncomplete", # codespell:ignore uncomplete
"complete",
id="already_incomplete",
),
],
)
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_status(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
initial_completed: datetime | None,
target_status: TodoItemStatus,
expected_api: str,
not_expected_api: str,
) -> None:
"""Test updating item status always calls the correct RTM API, even when unchanged."""
_set_tasks_response(
client, _make_task_list(LIST_ID, 10, 1, "Buy milk", completed=initial_completed)
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: uid, ATTR_STATUS: target_status},
blocking=True,
)
getattr(client.rtm.tasks, expected_api).assert_called_once_with(
timeline=1234,
list_id=LIST_ID,
taskseries_id=10,
task_id=1,
)
getattr(client.rtm.tasks, not_expected_api).assert_not_called()
@pytest.mark.usefixtures("storage")
async def test_delete_todo_items(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test deleting todo items calls tasks.delete for each uid."""
_set_tasks_response(
client,
_make_task_list(LIST_ID, 10, 1, "Buy milk"),
_make_task_list(LIST_ID, 20, 2, "Buy eggs"),
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid1 = f"{LIST_ID}_10_1"
uid2 = f"{LIST_ID}_20_2"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.REMOVE_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: [uid1, uid2]},
blocking=True,
)
assert client.rtm.tasks.delete.call_count == 2
client.rtm.tasks.delete.assert_any_call(
timeline=1234, list_id=LIST_ID, taskseries_id=10, task_id=1
)
client.rtm.tasks.delete.assert_any_call(
timeline=1234, list_id=LIST_ID, taskseries_id=20, task_id=2
)
@pytest.mark.parametrize(
("initial_due", "new_due_str", "expected_due", "expected_has_due_time"),
[
pytest.param(None, "2024-03-15", "2024-03-15", False, id="no_due_to_due"),
pytest.param(
datetime(2024, 3, 15, tzinfo=UTC),
"2024-03-15",
"2024-03-15",
False,
id="same_due",
),
],
)
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_set_due_date(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
initial_due: datetime | None,
new_due_str: str,
expected_due: str,
expected_has_due_time: bool,
) -> None:
"""Test updating due date always calls tasks.set_due_date, even when unchanged."""
_set_tasks_response(
client, _make_task_list(LIST_ID, 10, 1, "Buy milk", due=initial_due)
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: uid, ATTR_DUE_DATE: new_due_str},
blocking=True,
)
client.rtm.tasks.set_due_date.assert_called_once_with(
timeline=1234,
list_id=LIST_ID,
taskseries_id=10,
task_id=1,
due=expected_due,
has_due_time=expected_has_due_time,
)
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_add_description(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test updating a todo item to add a description adds a note."""
_set_tasks_response(client, _make_task_list(LIST_ID, 10, 1, "Buy milk"))
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_ITEM: uid,
ATTR_DESCRIPTION: "Organic if possible",
},
blocking=True,
)
client.rtm.tasks.notes.add.assert_called_once_with(
timeline=1234,
list_id=LIST_ID,
taskseries_id=10,
task_id=1,
title="",
text="Organic if possible",
)
client.rtm.tasks.notes.edit.assert_not_called()
client.rtm.tasks.notes.delete.assert_not_called()
@pytest.mark.parametrize(
"new_description",
[
pytest.param("New description", id="different_description"),
pytest.param("Old description", id="same_description"),
],
)
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_edit_description(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
new_description: str,
) -> None:
"""Test updating a todo item description always edits the note, even when unchanged."""
note = MagicMock()
note.id = 55
note.body = "Old description"
_set_tasks_response(
client, _make_task_list(LIST_ID, 10, 1, "Buy milk", notes=[note])
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: uid, ATTR_DESCRIPTION: new_description},
blocking=True,
)
client.rtm.tasks.notes.edit.assert_called_once_with(
timeline=1234,
note_id=55,
title="",
text=new_description,
)
client.rtm.tasks.notes.add.assert_not_called()
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_delete_description(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test clearing a todo item description deletes the note."""
note = MagicMock()
note.id = 55
note.body = "Existing description"
_set_tasks_response(
client, _make_task_list(LIST_ID, 10, 1, "Buy milk", notes=[note])
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: uid, ATTR_DESCRIPTION: ""},
blocking=True,
)
client.rtm.tasks.notes.delete.assert_called_once_with(
timeline=1234,
note_id=55,
)
client.rtm.tasks.notes.add.assert_not_called()
client.rtm.tasks.notes.edit.assert_not_called()
@pytest.mark.usefixtures("storage")
async def test_update_todo_item_empty_note_preserved(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
) -> None:
"""Test that an unrelated update does not delete a note with an empty body.
RTM notes can have a title but an empty body. The coordinator represents those
as description=None (because body or None collapses an empty string), but the
note still has a note_id. An unrelated update (e.g. completing the task) must
not delete that note.
"""
note = MagicMock()
note.id = 55
note.body = ""
_set_tasks_response(
client, _make_task_list(LIST_ID, 10, 1, "Buy milk", notes=[note])
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
uid = f"{LIST_ID}_10_1"
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.UPDATE_ITEM,
{
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_ITEM: uid,
ATTR_STATUS: TodoItemStatus.COMPLETED,
},
blocking=True,
)
client.rtm.tasks.notes.delete.assert_not_called()
client.rtm.tasks.notes.edit.assert_not_called()
client.rtm.tasks.notes.add.assert_not_called()
@pytest.mark.usefixtures("client", "storage")
async def test_setup_skips_non_list_subentries(hass: HomeAssistant) -> None:
"""Test that async_setup_entry ignores subentries that are not list subentries."""
entry = MockConfigEntry(
data=CREATE_ENTRY_DATA,
domain=DOMAIN,
subentries_data=[
ConfigSubentryDataWithId(
data={CONF_LIST_ID: LIST_ID},
subentry_type=SUBENTRY_TYPE_LIST,
title="My Shopping List",
unique_id=str(LIST_ID),
subentry_id=SUBENTRY_ID,
),
ConfigSubentryDataWithId(
data={},
subentry_type="other",
title="Other Subentry",
unique_id=None,
subentry_id="extra-subentry-id",
),
],
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Only the list subentry should produce a todo entity.
assert hass.states.get(ENTITY_ID) is not None
assert hass.states.get("todo.other_subentry") is None
@pytest.mark.parametrize(
("side_effect", "translation_key"),
[
pytest.param(AuthError("Boom!"), "invalid_auth", id="auth_error"),
pytest.param(AioRTMError("Boom!"), "api_error", id="api_error"),
],
)
@pytest.mark.usefixtures("storage")
async def test_todo_item_api_errors(
hass: HomeAssistant,
client: MagicMock,
config_entry: MockConfigEntry,
side_effect: Exception,
translation_key: str,
) -> None:
"""Test that RTM API errors during a todo operation raise HomeAssistantError."""
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
client.rtm.timelines.create.side_effect = side_effect
with pytest.raises(HomeAssistantError) as exc_info:
await hass.services.async_call(
TODO_DOMAIN,
TodoServices.ADD_ITEM,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_ITEM: "Buy butter"},
blocking=True,
)
assert exc_info.value.translation_key == translation_key