Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add todo component (#100019)

pull/102615/head^2
Allen Porter 2023-10-23 13:53:00 -07:00 committed by GitHub
parent fa1df7e334
commit 5d430f53cd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 1908 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -1303,6 +1303,8 @@ build.json @home-assistant/supervisor
/homeassistant/components/time_date/ @fabaff
/tests/components/time_date/ @fabaff
/homeassistant/components/tmb/ @alemuro
/homeassistant/components/todo/ @home-assistant/core
/tests/components/todo/ @home-assistant/core
/homeassistant/components/todoist/ @boralyl
/tests/components/todoist/ @boralyl
/homeassistant/components/tolo/ @MatthiasLohr

121
homeassistant/components/shopping_list/__init__.py
View File

@ -1,21 +1,22 @@
"""Support to manage a shopping list."""
from collections.abc import Callable
from http import HTTPStatus
import logging
from typing import Any
from typing import Any, cast
import uuid
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import frontend, http, websocket_api
from homeassistant.components import http, websocket_api
from homeassistant.components.http.data_validator import RequestDataValidator
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_NAME
from homeassistant.const import ATTR_NAME, Platform
from homeassistant.core import HomeAssistant, ServiceCall, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.json import save_json
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.json import JsonArrayType, load_json_array
from homeassistant.util.json import JsonValueType, load_json_array
from .const import (
ATTR_REVERSE,
@ -32,6 +33,8 @@ from .const import (
SERVICE_SORT,
)
PLATFORMS = [Platform.TODO]
ATTR_COMPLETE = "complete"
_LOGGER = logging.getLogger(__name__)
@ -169,10 +172,6 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
hass.http.register_view(UpdateShoppingListItemView)
hass.http.register_view(ClearCompletedItemsView)
frontend.async_register_built_in_panel(
hass, "shopping-list", "shopping_list", "mdi:cart"
)
websocket_api.async_register_command(hass, websocket_handle_items)
websocket_api.async_register_command(hass, websocket_handle_add)
websocket_api.async_register_command(hass, websocket_handle_remove)
@ -180,6 +179,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
websocket_api.async_register_command(hass, websocket_handle_clear)
websocket_api.async_register_command(hass, websocket_handle_reorder)
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
return True
@ -193,13 +194,15 @@ class ShoppingData:
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the shopping list."""
self.hass = hass
self.items: JsonArrayType = []
self.items: list[dict[str, JsonValueType]] = []
self._listeners: list[Callable[[], None]] = []
async def async_add(self, name, context=None):
async def async_add(self, name, complete=False, context=None):
"""Add a shopping list item."""
item = {"name": name, "id": uuid.uuid4().hex, "complete": False}
item = {"name": name, "id": uuid.uuid4().hex, "complete": complete}
self.items.append(item)
await self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "add", "item": item},
@ -207,21 +210,43 @@ class ShoppingData:
)
return item
async def async_remove(self, item_id, context=None):
async def async_remove(
self, item_id: str, context=None
) -> dict[str, JsonValueType] | None:
"""Remove a shopping list item."""
item = next((itm for itm in self.items if itm["id"] == item_id), None)
if item is None:
raise NoMatchingShoppingListItem
self.items.remove(item)
await self.hass.async_add_executor_job(self.save)
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "remove", "item": item},
context=context,
removed = await self.async_remove_items(
item_ids=set({item_id}), context=context
)
return item
return next(iter(removed), None)
async def async_remove_items(
self, item_ids: set[str], context=None
) -> list[dict[str, JsonValueType]]:
"""Remove a shopping list item."""
items_dict: dict[str, dict[str, JsonValueType]] = {}
for itm in self.items:
item_id = cast(str, itm["id"])
items_dict[item_id] = itm
removed = []
for item_id in item_ids:
_LOGGER.debug(
"Removing %s",
)
if not (item := items_dict.pop(item_id, None)):
raise NoMatchingShoppingListItem(
"Item '{item_id}' not found in shopping list"
)
removed.append(item)
self.items = list(items_dict.values())
await self.hass.async_add_executor_job(self.save)
self._async_notify()
for item in removed:
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "remove", "item": item},
context=context,
)
return removed
async def async_update(self, item_id, info, context=None):
"""Update a shopping list item."""
@ -233,6 +258,7 @@ class ShoppingData:
info = ITEM_UPDATE_SCHEMA(info)
item.update(info)
await self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "update", "item": item},
@ -244,6 +270,7 @@ class ShoppingData:
"""Clear completed items."""
self.items = [itm for itm in self.items if not itm["complete"]]
await self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "clear"},
@ -255,6 +282,7 @@ class ShoppingData:
for item in self.items:
item.update(info)
await self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "update_list"},
@ -287,16 +315,36 @@ class ShoppingData:
new_items.append(all_items_mapping[key])
self.items = new_items
self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "reorder"},
context=context,
)
async def async_move_item(self, uid: str, pos: int) -> None:
"""Re-order a shopping list item."""
found_item: dict[str, Any] | None = None
for idx, itm in enumerate(self.items):
if cast(str, itm["id"]) == uid:
found_item = itm
self.items.pop(idx)
break
if not found_item:
raise NoMatchingShoppingListItem(f"Item '{uid}' not found in shopping list")
self.items.insert(pos, found_item)
await self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "reorder"},
)
async def async_sort(self, reverse=False, context=None):
"""Sort items by name."""
self.items = sorted(self.items, key=lambda item: item["name"], reverse=reverse)
self.hass.async_add_executor_job(self.save)
self._async_notify()
self.hass.bus.async_fire(
EVENT_SHOPPING_LIST_UPDATED,
{"action": "sorted"},
@ -306,9 +354,12 @@ class ShoppingData:
async def async_load(self) -> None:
"""Load items."""
def load() -> JsonArrayType:
def load() -> list[dict[str, JsonValueType]]:
"""Load the items synchronously."""
return load_json_array(self.hass.config.path(PERSISTENCE))
return cast(
list[dict[str, JsonValueType]],
load_json_array(self.hass.config.path(PERSISTENCE)),
)
self.items = await self.hass.async_add_executor_job(load)
@ -316,6 +367,20 @@ class ShoppingData:
"""Save the items."""
save_json(self.hass.config.path(PERSISTENCE), self.items)
def async_add_listener(self, cb: Callable[[], None]) -> Callable[[], None]:
"""Add a listener to notify when data is updated."""
def unsub():
self._listeners.remove(cb)
self._listeners.append(cb)
return unsub
def _async_notify(self) -> None:
"""Notify all listeners that data has been updated."""
for listener in self._listeners:
listener()
class ShoppingListView(http.HomeAssistantView):
"""View to retrieve shopping list content."""
@ -397,7 +462,9 @@ async def websocket_handle_add(
msg: dict[str, Any],
) -> None:
"""Handle adding item to shopping_list."""
item = await hass.data[DOMAIN].async_add(msg["name"], connection.context(msg))
item = await hass.data[DOMAIN].async_add(
msg["name"], context=connection.context(msg)
)
connection.send_message(websocket_api.result_message(msg["id"], item))

7
homeassistant/components/shopping_list/strings.json
View File

@ -74,5 +74,12 @@
}
}
}
},
"entity": {
"todo": {
"shopping_list": {
"name": "[%key:component::shopping_list::title%]"
}
}
}
}

106
homeassistant/components/shopping_list/todo.py Normal file
View File

@ -0,0 +1,106 @@
"""A shopping list todo platform."""
from typing import Any, cast
from homeassistant.components.todo import (
TodoItem,
TodoItemStatus,
TodoListEntity,
TodoListEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import NoMatchingShoppingListItem, ShoppingData
from .const import DOMAIN
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the shopping_list todo platform."""
shopping_data = hass.data[DOMAIN]
entity = ShoppingTodoListEntity(shopping_data, unique_id=config_entry.entry_id)
async_add_entities([entity], True)
class ShoppingTodoListEntity(TodoListEntity):
"""A To-do List representation of the Shopping List."""
_attr_has_entity_name = True
_attr_translation_key = "shopping_list"
_attr_should_poll = False
_attr_supported_features = (
TodoListEntityFeature.CREATE_TODO_ITEM
| TodoListEntityFeature.DELETE_TODO_ITEM
| TodoListEntityFeature.UPDATE_TODO_ITEM
| TodoListEntityFeature.MOVE_TODO_ITEM
)
def __init__(self, data: ShoppingData, unique_id: str) -> None:
"""Initialize ShoppingTodoListEntity."""
self._attr_unique_id = unique_id
self._data = data
async def async_create_todo_item(self, item: TodoItem) -> None:
"""Add an item to the To-do list."""
await self._data.async_add(
item.summary, complete=(item.status == TodoItemStatus.COMPLETED)
)
async def async_update_todo_item(self, item: TodoItem) -> None:
"""Update an item to the To-do list."""
data: dict[str, Any] = {}
if item.summary:
data["name"] = item.summary
if item.status:
data["complete"] = item.status == TodoItemStatus.COMPLETED
try:
await self._data.async_update(item.uid, data)
except NoMatchingShoppingListItem as err:
raise HomeAssistantError(
f"Shopping list item '{item.uid}' was not found"
) from err
async def async_delete_todo_items(self, uids: list[str]) -> None:
"""Add an item to the To-do list."""
await self._data.async_remove_items(set(uids))
async def async_move_todo_item(self, uid: str, pos: int) -> None:
"""Re-order an item to the To-do list."""
try:
await self._data.async_move_item(uid, pos)
except NoMatchingShoppingListItem as err:
raise HomeAssistantError(
f"Shopping list item '{uid}' could not be re-ordered"
) from err
async def async_added_to_hass(self) -> None:
"""Entity has been added to hass."""
# Shopping list integration doesn't currently support config entry unload
# so this code may not be used in practice, however it is here in case
# this changes in the future.
self.async_on_remove(self._data.async_add_listener(self.async_write_ha_state))
@property
def todo_items(self) -> list[TodoItem]:
"""Get items in the To-do list."""
results = []
for item in self._data.items:
if cast(bool, item["complete"]):
status = TodoItemStatus.COMPLETED
else:
status = TodoItemStatus.NEEDS_ACTION
results.append(
TodoItem(
summary=cast(str, item["name"]),
uid=cast(str, item["id"]),
status=status,
)
)
return results

262
homeassistant/components/todo/__init__.py Normal file
View File

@ -0,0 +1,262 @@
"""The todo integration."""
import dataclasses
import datetime
import logging
from typing import Any
import voluptuous as vol
from homeassistant.components import frontend, websocket_api
from homeassistant.components.websocket_api import ERR_NOT_FOUND, ERR_NOT_SUPPORTED
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.config_validation import ( # noqa: F401
PLATFORM_SCHEMA,
PLATFORM_SCHEMA_BASE,
)
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, TodoItemStatus, TodoListEntityFeature
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = datetime.timedelta(seconds=60)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Todo entities."""
component = hass.data[DOMAIN] = EntityComponent[TodoListEntity](
_LOGGER, DOMAIN, hass, SCAN_INTERVAL
)
frontend.async_register_built_in_panel(hass, "todo", "todo", "mdi:clipboard-list")
websocket_api.async_register_command(hass, websocket_handle_todo_item_list)
websocket_api.async_register_command(hass, websocket_handle_todo_item_move)
component.async_register_entity_service(
"create_item",
{
vol.Required("summary"): vol.All(cv.string, vol.Length(min=1)),
vol.Optional("status", default=TodoItemStatus.NEEDS_ACTION): vol.In(
{TodoItemStatus.NEEDS_ACTION, TodoItemStatus.COMPLETED}
),
},
_async_create_todo_item,
required_features=[TodoListEntityFeature.CREATE_TODO_ITEM],
)
component.async_register_entity_service(
"update_item",
vol.All(
cv.make_entity_service_schema(
{
vol.Optional("uid"): cv.string,
vol.Optional("summary"): vol.All(cv.string, vol.Length(min=1)),
vol.Optional("status"): vol.In(
{TodoItemStatus.NEEDS_ACTION, TodoItemStatus.COMPLETED}
),
}
),
cv.has_at_least_one_key("uid", "summary"),
),
_async_update_todo_item,
required_features=[TodoListEntityFeature.UPDATE_TODO_ITEM],
)
component.async_register_entity_service(
"delete_item",
vol.All(
cv.make_entity_service_schema(
{
vol.Optional("uid"): vol.All(cv.ensure_list, [cv.string]),
vol.Optional("summary"): vol.All(cv.ensure_list, [cv.string]),
}
),
cv.has_at_least_one_key("uid", "summary"),
),
_async_delete_todo_items,
required_features=[TodoListEntityFeature.DELETE_TODO_ITEM],
)
await component.async_setup(config)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
component: EntityComponent[TodoListEntity] = hass.data[DOMAIN]
return await component.async_setup_entry(entry)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
component: EntityComponent[TodoListEntity] = hass.data[DOMAIN]
return await component.async_unload_entry(entry)
@dataclasses.dataclass
class TodoItem:
"""A To-do item in a To-do list."""
summary: str | None = None
"""The summary that represents the item."""
uid: str | None = None
"""A unique identifier for the To-do item."""
status: TodoItemStatus | None = None
"""A status or confirmation of the To-do item."""
@classmethod
def from_dict(cls, obj: dict[str, Any]) -> "TodoItem":
"""Create a To-do Item from a dictionary parsed by schema validators."""
return cls(
summary=obj.get("summary"), status=obj.get("status"), uid=obj.get("uid")
)
class TodoListEntity(Entity):
"""An entity that represents a To-do list."""
_attr_todo_items: list[TodoItem] | None = None
@property
def state(self) -> int | None:
"""Return the entity state as the count of incomplete items."""
items = self.todo_items
if items is None:
return None
return sum([item.status == TodoItemStatus.NEEDS_ACTION for item in items])
@property
def todo_items(self) -> list[TodoItem] | None:
"""Return the To-do items in the To-do list."""
return self._attr_todo_items
async def async_create_todo_item(self, item: TodoItem) -> None:
"""Add an item to the To-do list."""
raise NotImplementedError()
async def async_update_todo_item(self, item: TodoItem) -> None:
"""Update an item in the To-do list."""
raise NotImplementedError()
async def async_delete_todo_items(self, uids: list[str]) -> None:
"""Delete an item in the To-do list."""
raise NotImplementedError()
async def async_move_todo_item(self, uid: str, pos: int) -> None:
"""Move an item in the To-do list."""
raise NotImplementedError()
@websocket_api.websocket_command(
{
vol.Required("type"): "todo/item/list",
vol.Required("entity_id"): cv.entity_id,
}
)
@websocket_api.async_response
async def websocket_handle_todo_item_list(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle the list of To-do items in a To-do- list."""
component: EntityComponent[TodoListEntity] = hass.data[DOMAIN]
if (
not (entity_id := msg[CONF_ENTITY_ID])
or not (entity := component.get_entity(entity_id))
or not isinstance(entity, TodoListEntity)
):
connection.send_error(msg["id"], ERR_NOT_FOUND, "Entity not found")
return
items: list[TodoItem] = entity.todo_items or []
connection.send_message(
websocket_api.result_message(
msg["id"], {"items": [dataclasses.asdict(item) for item in items]}
)
)
@websocket_api.websocket_command(
{
vol.Required("type"): "todo/item/move",
vol.Required("entity_id"): cv.entity_id,
vol.Required("uid"): cv.string,
vol.Optional("pos", default=0): cv.positive_int,
}
)
@websocket_api.async_response
async def websocket_handle_todo_item_move(
hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict[str, Any]
) -> None:
"""Handle move of a To-do item within a To-do list."""
component: EntityComponent[TodoListEntity] = hass.data[DOMAIN]
if not (entity := component.get_entity(msg["entity_id"])):
connection.send_error(msg["id"], ERR_NOT_FOUND, "Entity not found")
return
if (
not entity.supported_features
or not entity.supported_features & TodoListEntityFeature.MOVE_TODO_ITEM
):
connection.send_message(
websocket_api.error_message(
msg["id"],
ERR_NOT_SUPPORTED,
"To-do list does not support To-do item reordering",
)
)
return
try:
await entity.async_move_todo_item(uid=msg["uid"], pos=msg["pos"])
except HomeAssistantError as ex:
connection.send_error(msg["id"], "failed", str(ex))
else:
connection.send_result(msg["id"])
def _find_by_summary(summary: str, items: list[TodoItem] | None) -> TodoItem | None:
"""Find a To-do List item by summary name."""
for item in items or ():
if item.summary == summary:
return item
return None
async def _async_create_todo_item(entity: TodoListEntity, call: ServiceCall) -> None:
"""Add an item to the To-do list."""
await entity.async_create_todo_item(item=TodoItem.from_dict(call.data))
async def _async_update_todo_item(entity: TodoListEntity, call: ServiceCall) -> None:
"""Update an item in the To-do list."""
item = TodoItem.from_dict(call.data)
if not item.uid:
found = _find_by_summary(call.data["summary"], entity.todo_items)
if not found:
raise ValueError(f"Unable to find To-do item with summary '{item.summary}'")
item.uid = found.uid
await entity.async_update_todo_item(item=item)
async def _async_delete_todo_items(entity: TodoListEntity, call: ServiceCall) -> None:
"""Delete an item in the To-do list."""
uids = call.data.get("uid", [])
if not uids:
summaries = call.data.get("summary", [])
for summary in summaries:
item = _find_by_summary(summary, entity.todo_items)
if not item:
raise ValueError(f"Unable to find To-do item with summary '{summary}")
uids.append(item.uid)
await entity.async_delete_todo_items(uids=uids)

24
homeassistant/components/todo/const.py Normal file
View File

@ -0,0 +1,24 @@
"""Constants for the To-do integration."""
from enum import IntFlag, StrEnum
DOMAIN = "todo"
class TodoListEntityFeature(IntFlag):
"""Supported features of the To-do List entity."""
CREATE_TODO_ITEM = 1
DELETE_TODO_ITEM = 2
UPDATE_TODO_ITEM = 4
MOVE_TODO_ITEM = 8
class TodoItemStatus(StrEnum):
"""Status or confirmation of a To-do List Item.
This is a subset of the statuses supported in rfc5545.
"""
NEEDS_ACTION = "needs_action"
COMPLETED = "completed"

9
homeassistant/components/todo/manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"domain": "todo",
"name": "To-do",
"codeowners": ["@home-assistant/core"],
"dependencies": ["http"],
"documentation": "https://www.home-assistant.io/integrations/todo",
"integration_type": "entity",
"quality_scale": "internal"
}

55
homeassistant/components/todo/services.yaml Normal file
View File

@ -0,0 +1,55 @@
create_item:
target:
entity:
domain: todo
supported_features:
- todo.TodoListEntityFeature.CREATE_TODO_ITEM
fields:
summary:
required: true
example: "Submit Income Tax Return"
selector:
text:
status:
example: "needs_action"
selector:
select:
translation_key: status
options:
- needs_action
- completed
update_item:
target:
entity:
domain: todo
supported_features:
- todo.TodoListEntityFeature.UPDATE_TODO_ITEM
fields:
uid:
selector:
text:
summary:
example: "Submit Income Tax Return"
selector:
text:
status:
example: "needs_action"
selector:
select:
translation_key: status
options:
- needs_action
- completed
delete_item:
target:
entity:
domain: todo
supported_features:
- todo.TodoListEntityFeature.DELETE_TODO_ITEM
fields:
uid:
selector:
object:
summary:
selector:
object:

64
homeassistant/components/todo/strings.json Normal file
View File

@ -0,0 +1,64 @@
{
"title": "To-do List",
"entity_component": {
"_": {
"name": "[%key:component::todo::title%]"
}
},
"services": {
"create_item": {
"name": "Create To-do List Item",
"description": "Add a new To-do List Item.",
"fields": {
"summary": {
"name": "Summary",
"description": "The short summary that represents the To-do item."
},
"status": {
"name": "Status",
"description": "A status or confirmation of the To-do item."
}
}
},
"update_item": {
"name": "Update To-do List Item",
"description": "Update an existing To-do List Item based on either its Unique Id or Summary.",
"fields": {
"uid": {
"name": "To-do Item Unique Id",
"description": "Unique Identifier for the To-do List Item."
},
"summary": {
"name": "Summary",
"description": "The short summary that represents the To-do item."
},
"status": {
"name": "Status",
"description": "A status or confirmation of the To-do item."
}
}
},
"delete_item": {
"name": "Delete a To-do List Item",
"description": "Delete an existing To-do List Item either by its Unique Id or Summary.",
"fields": {
"uid": {
"name": "To-do Item Unique Ids",
"description": "Unique Identifiers for the To-do List Items."
},
"summary": {
"name": "Summary",
"description": "The short summary that represents the To-do item."
}
}
}
},
"selector": {
"status": {
"options": {
"needs_action": "Needs Action",
"completed": "Completed"
}
}
}
}

1
homeassistant/const.py
View File

@ -55,6 +55,7 @@ class Platform(StrEnum):
SWITCH = "switch"
TEXT = "text"
TIME = "time"
TODO = "todo"
TTS = "tts"
VACUUM = "vacuum"
UPDATE = "update"

2
homeassistant/helpers/selector.py
View File

@ -99,6 +99,7 @@ def _entity_features() -> dict[str, type[IntFlag]]:
from homeassistant.components.media_player import MediaPlayerEntityFeature
from homeassistant.components.remote import RemoteEntityFeature
from homeassistant.components.siren import SirenEntityFeature
from homeassistant.components.todo import TodoListEntityFeature
from homeassistant.components.update import UpdateEntityFeature
from homeassistant.components.vacuum import VacuumEntityFeature
from homeassistant.components.water_heater import WaterHeaterEntityFeature
@ -118,6 +119,7 @@ def _entity_features() -> dict[str, type[IntFlag]]:
"MediaPlayerEntityFeature": MediaPlayerEntityFeature,
"RemoteEntityFeature": RemoteEntityFeature,
"SirenEntityFeature": SirenEntityFeature,
"TodoListEntityFeature": TodoListEntityFeature,
"UpdateEntityFeature": UpdateEntityFeature,
"VacuumEntityFeature": VacuumEntityFeature,
"WaterHeaterEntityFeature": WaterHeaterEntityFeature,

48
pylint/plugins/hass_enforce_type_hints.py
View File

@ -2428,6 +2428,54 @@ _INHERITANCE_MATCH: dict[str, list[ClassTypeHintMatch]] = {
],
),
],
"todo": [
ClassTypeHintMatch(
base_class="Entity",
matches=_ENTITY_MATCH,
),
ClassTypeHintMatch(
base_class="RestoreEntity",
matches=_RESTORE_ENTITY_MATCH,
),
ClassTypeHintMatch(
base_class="TodoListEntity",
matches=[
TypeHintMatch(
function_name="todo_items",
return_type=["list[TodoItem]", None],
),
TypeHintMatch(
function_name="async_create_todo_item",
arg_types={
1: "TodoItem",
},
return_type="None",
),
TypeHintMatch(
function_name="async_update_todo_item",
arg_types={
1: "TodoItem",
},
return_type="None",
),
TypeHintMatch(
function_name="async_delete_todo_items",
arg_types={
1: "list[str]",
},
return_type="None",
),
TypeHintMatch(
function_name="async_move_todo_item",
arg_types={
1: "str",
2: "int",
},
return_type="None",
),
],
),
],
"tts": [
ClassTypeHintMatch(
base_class="Provider",

14
tests/components/shopping_list/conftest.py
View File

@ -4,6 +4,7 @@ from unittest.mock import patch
import pytest
from homeassistant.components.shopping_list import intent as sl_intent
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -18,12 +19,17 @@ def mock_shopping_list_io():
@pytest.fixture
async def sl_setup(hass):
def mock_config_entry() -> MockConfigEntry:
"""Config Entry fixture."""
return MockConfigEntry(domain="shopping_list")
@pytest.fixture
async def sl_setup(hass: HomeAssistant, mock_config_entry: MockConfigEntry):
"""Set up the shopping list."""
entry = MockConfigEntry(domain="shopping_list")
entry.add_to_hass(hass)
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await sl_intent.async_setup_intents(hass)

493
tests/components/shopping_list/test_todo.py Normal file
View File

@ -0,0 +1,493 @@
"""Test shopping list todo platform."""
from collections.abc import Awaitable, Callable
from typing import Any
import pytest
from homeassistant.components.todo import DOMAIN as TODO_DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from tests.typing import WebSocketGenerator
TEST_ENTITY = "todo.shopping_list"
@pytest.fixture
def ws_req_id() -> Callable[[], int]:
"""Fixture for incremental websocket requests."""
id = 0
def next() -> int:
nonlocal id
id += 1
return id
return next
@pytest.fixture
async def ws_get_items(
hass_ws_client: WebSocketGenerator, ws_req_id: Callable[[], int]
) -> Callable[[], Awaitable[dict[str, str]]]:
"""Fixture to fetch items from the todo websocket."""
async def get() -> list[dict[str, str]]:
# Fetch items using To-do platform
client = await hass_ws_client()
id = ws_req_id()
await client.send_json(
{
"id": id,
"type": "todo/item/list",
"entity_id": TEST_ENTITY,
}
)
resp = await client.receive_json()
assert resp.get("id") == id
assert resp.get("success")
return resp.get("result", {}).get("items", [])
return get
@pytest.fixture
async def ws_move_item(
hass_ws_client: WebSocketGenerator,
ws_req_id: Callable[[], int],
) -> Callable[[str, int | None], Awaitable[None]]:
"""Fixture to move an item in the todo list."""
async def move(uid: str, pos: int | None) -> dict[str, Any]:
# Fetch items using To-do platform
client = await hass_ws_client()
id = ws_req_id()
data = {
"id": id,
"type": "todo/item/move",
"entity_id": TEST_ENTITY,
"uid": uid,
}
if pos is not None:
data["pos"] = pos
await client.send_json(data)
resp = await client.receive_json()
assert resp.get("id") == id
return resp
return move
async def test_get_items(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test creating a shopping list item with the WS API and verifying with To-do API."""
client = await hass_ws_client(hass)
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "0"
# Native shopping list websocket
await client.send_json(
{"id": ws_req_id(), "type": "shopping_list/items/add", "name": "soda"}
)
msg = await client.receive_json()
assert msg["success"] is True
data = msg["result"]
assert data["name"] == "soda"
assert data["complete"] is False
# Fetch items using To-do platform
items = await ws_get_items()
assert len(items) == 1
assert items[0]["summary"] == "soda"
assert items[0]["status"] == "needs_action"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "1"
async def test_create_item(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test creating shopping_list item and listing it."""
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{
"summary": "soda",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
# Fetch items using To-do platform
items = await ws_get_items()
assert len(items) == 1
assert items[0]["summary"] == "soda"
assert items[0]["status"] == "needs_action"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "1"
# Add a completed item
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{"summary": "paper", "status": "completed"},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 2
assert items[0]["summary"] == "soda"
assert items[0]["status"] == "needs_action"
assert items[1]["summary"] == "paper"
assert items[1]["status"] == "completed"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "1"
async def test_delete_item(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test deleting a todo item."""
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{"summary": "soda", "status": "needs_action"},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 1
assert items[0]["summary"] == "soda"
assert items[0]["status"] == "needs_action"
assert "uid" in items[0]
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "1"
await hass.services.async_call(
TODO_DOMAIN,
"delete_item",
{
"uid": [items[0]["uid"]],
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 0
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "0"
async def test_bulk_delete(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test deleting a todo item."""
for _i in range(0, 5):
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{
"summary": "soda",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 5
uids = [item["uid"] for item in items]
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "5"
await hass.services.async_call(
TODO_DOMAIN,
"delete_item",
{
"uid": uids,
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 0
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "0"
async def test_update_item(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test updating a todo item."""
# Create new item
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{
"summary": "soda",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
# Fetch item
items = await ws_get_items()
assert len(items) == 1
item = items[0]
assert item["summary"] == "soda"
assert item["status"] == "needs_action"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "1"
# Mark item completed
await hass.services.async_call(
TODO_DOMAIN,
"update_item",
{
**item,
"status": "completed",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
# Verify item is marked as completed
items = await ws_get_items()
assert len(items) == 1
item = items[0]
assert item["summary"] == "soda"
assert item["status"] == "completed"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "0"
async def test_partial_update_item(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test updating a todo item with partial information."""
# Create new item
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{
"summary": "soda",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
# Fetch item
items = await ws_get_items()
assert len(items) == 1
item = items[0]
assert item["summary"] == "soda"
assert item["status"] == "needs_action"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "1"
# Mark item completed without changing the summary
await hass.services.async_call(
TODO_DOMAIN,
"update_item",
{
"uid": item["uid"],
"status": "completed",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
# Verify item is marked as completed
items = await ws_get_items()
assert len(items) == 1
item = items[0]
assert item["summary"] == "soda"
assert item["status"] == "completed"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "0"
# Change the summary without changing the status
await hass.services.async_call(
TODO_DOMAIN,
"update_item",
{
"uid": item["uid"],
"summary": "other summary",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
# Verify item is changed and still marked as completed
items = await ws_get_items()
assert len(items) == 1
item = items[0]
assert item["summary"] == "other summary"
assert item["status"] == "completed"
state = hass.states.get(TEST_ENTITY)
assert state
assert state.state == "0"
async def test_update_invalid_item(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
) -> None:
"""Test updating a todo item that does not exist."""
with pytest.raises(HomeAssistantError, match="was not found"):
await hass.services.async_call(
TODO_DOMAIN,
"update_item",
{
"uid": "invalid-uid",
"summary": "Example task",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
@pytest.mark.parametrize(
("src_idx", "dst_idx", "expected_items"),
[
# Move any item to the front of the list
(0, 0, ["item 1", "item 2", "item 3", "item 4"]),
(1, 0, ["item 2", "item 1", "item 3", "item 4"]),
(2, 0, ["item 3", "item 1", "item 2", "item 4"]),
(3, 0, ["item 4", "item 1", "item 2", "item 3"]),
# Move items right
(0, 1, ["item 2", "item 1", "item 3", "item 4"]),
(0, 2, ["item 2", "item 3", "item 1", "item 4"]),
(0, 3, ["item 2", "item 3", "item 4", "item 1"]),
(1, 2, ["item 1", "item 3", "item 2", "item 4"]),
(1, 3, ["item 1", "item 3", "item 4", "item 2"]),
# Move items left
(2, 1, ["item 1", "item 3", "item 2", "item 4"]),
(3, 1, ["item 1", "item 4", "item 2", "item 3"]),
(3, 2, ["item 1", "item 2", "item 4", "item 3"]),
# No-ops
(0, 0, ["item 1", "item 2", "item 3", "item 4"]),
(1, 1, ["item 1", "item 2", "item 3", "item 4"]),
(2, 2, ["item 1", "item 2", "item 3", "item 4"]),
(3, 3, ["item 1", "item 2", "item 3", "item 4"]),
(3, 4, ["item 1", "item 2", "item 3", "item 4"]),
],
)
async def test_move_item(
hass: HomeAssistant,
sl_setup: None,
ws_req_id: Callable[[], int],
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
ws_move_item: Callable[[str, int | None], Awaitable[dict[str, Any]]],
src_idx: int,
dst_idx: int | None,
expected_items: list[str],
) -> None:
"""Test moving a todo item within the list."""
for i in range(1, 5):
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{
"summary": f"item {i}",
},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 4
uids = [item["uid"] for item in items]
summaries = [item["summary"] for item in items]
assert summaries == ["item 1", "item 2", "item 3", "item 4"]
resp = await ws_move_item(uids[src_idx], dst_idx)
assert resp.get("success")
items = await ws_get_items()
assert len(items) == 4
summaries = [item["summary"] for item in items]
assert summaries == expected_items
async def test_move_invalid_item(
hass: HomeAssistant,
sl_setup: None,
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
ws_move_item: Callable[[str, int | None], Awaitable[dict[str, Any]]],
) -> None:
"""Test moving an item that does not exist."""
await hass.services.async_call(
TODO_DOMAIN,
"create_item",
{"summary": "soda"},
target={"entity_id": TEST_ENTITY},
blocking=True,
)
items = await ws_get_items()
assert len(items) == 1
item = items[0]
assert item["summary"] == "soda"
resp = await ws_move_item("unknown", 0)
assert not resp.get("success")
assert resp.get("error", {}).get("code") == "failed"
assert "could not be re-ordered" in resp.get("error", {}).get("message")

1
tests/components/todo/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the To-do integration."""

730
tests/components/todo/test_init.py Normal file
View File

@ -0,0 +1,730 @@
"""Tests for the todo integration."""
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock
import pytest
import voluptuous as vol
from homeassistant.components.todo import (
DOMAIN,
TodoItem,
TodoItemStatus,
TodoListEntity,
TodoListEntityFeature,
)
from homeassistant.config_entries import ConfigEntry, ConfigEntryState, ConfigFlow
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from tests.common import (
MockConfigEntry,
MockModule,
MockPlatform,
mock_config_flow,
mock_integration,
mock_platform,
)
from tests.typing import WebSocketGenerator
TEST_DOMAIN = "test"
class MockFlow(ConfigFlow):
"""Test flow."""
@pytest.fixture(autouse=True)
def config_flow_fixture(hass: HomeAssistant) -> Generator[None, None, None]:
"""Mock config flow."""
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
with mock_config_flow(TEST_DOMAIN, MockFlow):
yield
@pytest.fixture(autouse=True)
def mock_setup_integration(hass: HomeAssistant) -> None:
"""Fixture to set up a mock integration."""
async def async_setup_entry_init(
hass: HomeAssistant, config_entry: ConfigEntry
) -> bool:
"""Set up test config entry."""
await hass.config_entries.async_forward_entry_setup(config_entry, DOMAIN)
return True
async def async_unload_entry_init(
hass: HomeAssistant,
config_entry: ConfigEntry,
) -> bool:
await hass.config_entries.async_unload_platforms(config_entry, [Platform.TODO])
return True
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
mock_integration(
hass,
MockModule(
TEST_DOMAIN,
async_setup_entry=async_setup_entry_init,
async_unload_entry=async_unload_entry_init,
),
)
async def create_mock_platform(
hass: HomeAssistant,
entities: list[TodoListEntity],
) -> MockConfigEntry:
"""Create a todo platform with the specified entities."""
async def async_setup_entry_platform(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up test event platform via config entry."""
async_add_entities(entities)
mock_platform(
hass,
f"{TEST_DOMAIN}.{DOMAIN}",
MockPlatform(async_setup_entry=async_setup_entry_platform),
)
config_entry = MockConfigEntry(domain=TEST_DOMAIN)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
@pytest.fixture(name="test_entity")
def mock_test_entity() -> TodoListEntity:
"""Fixture that creates a test TodoList entity with mock service calls."""
entity1 = TodoListEntity()
entity1.entity_id = "todo.entity1"
entity1._attr_supported_features = (
TodoListEntityFeature.CREATE_TODO_ITEM
| TodoListEntityFeature.UPDATE_TODO_ITEM
| TodoListEntityFeature.DELETE_TODO_ITEM
| TodoListEntityFeature.MOVE_TODO_ITEM
)
entity1._attr_todo_items = [
TodoItem(summary="Item #1", uid="1", status=TodoItemStatus.NEEDS_ACTION),
TodoItem(summary="Item #2", uid="2", status=TodoItemStatus.COMPLETED),
]
entity1.async_create_todo_item = AsyncMock()
entity1.async_update_todo_item = AsyncMock()
entity1.async_delete_todo_items = AsyncMock()
entity1.async_move_todo_item = AsyncMock()
return entity1
async def test_unload_entry(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test unloading a config entry with a todo entity."""
config_entry = await create_mock_platform(hass, [test_entity])
assert config_entry.state == ConfigEntryState.LOADED
state = hass.states.get("todo.entity1")
assert state
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.NOT_LOADED
state = hass.states.get("todo.entity1")
assert not state
async def test_list_todo_items(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
test_entity: TodoListEntity,
) -> None:
"""Test listing items in a To-do list."""
await create_mock_platform(hass, [test_entity])
state = hass.states.get("todo.entity1")
assert state
assert state.state == "1"
assert state.attributes == {"supported_features": 15}
client = await hass_ws_client(hass)
await client.send_json(
{"id": 1, "type": "todo/item/list", "entity_id": "todo.entity1"}
)
resp = await client.receive_json()
assert resp.get("id") == 1
assert resp.get("success")
assert resp.get("result") == {
"items": [
{"summary": "Item #1", "uid": "1", "status": "needs_action"},
{"summary": "Item #2", "uid": "2", "status": "completed"},
]
}
async def test_unsupported_websocket(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a To-do list that does not support features."""
entity1 = TodoListEntity()
entity1.entity_id = "todo.entity1"
await create_mock_platform(hass, [entity1])
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 1,
"type": "todo/item/list",
"entity_id": "todo.unknown",
}
)
resp = await client.receive_json()
assert resp.get("id") == 1
assert resp.get("error", {}).get("code") == "not_found"
@pytest.mark.parametrize(
("item_data", "expected_status"),
[
({}, TodoItemStatus.NEEDS_ACTION),
({"status": "needs_action"}, TodoItemStatus.NEEDS_ACTION),
({"status": "completed"}, TodoItemStatus.COMPLETED),
],
)
async def test_create_item_service(
hass: HomeAssistant,
item_data: dict[str, Any],
expected_status: TodoItemStatus,
test_entity: TodoListEntity,
) -> None:
"""Test creating an item in a To-do list."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"create_item",
{"summary": "New item", **item_data},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_create_todo_item.call_args
assert args
item = args.kwargs.get("item")
assert item
assert item.uid is None
assert item.summary == "New item"
assert item.status == expected_status
async def test_create_item_service_raises(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test creating an item in a To-do list that raises an error."""
await create_mock_platform(hass, [test_entity])
test_entity.async_create_todo_item.side_effect = HomeAssistantError("Ooops")
with pytest.raises(HomeAssistantError, match="Ooops"):
await hass.services.async_call(
DOMAIN,
"create_item",
{"summary": "New item", "status": "needs_action"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
@pytest.mark.parametrize(
("item_data", "expected_error"),
[
({}, "required key not provided"),
({"status": "needs_action"}, "required key not provided"),
(
{"summary": "", "status": "needs_action"},
"length of value must be at least 1",
),
],
)
async def test_create_item_service_invalid_input(
hass: HomeAssistant,
test_entity: TodoListEntity,
item_data: dict[str, Any],
expected_error: str,
) -> None:
"""Test invalid input to the create item service."""
await create_mock_platform(hass, [test_entity])
with pytest.raises(vol.Invalid, match=expected_error):
await hass.services.async_call(
DOMAIN,
"create_item",
item_data,
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_update_todo_item_service_by_id(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test updating an item in a To-do list."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"update_item",
{"uid": "item-1", "summary": "Updated item", "status": "completed"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_update_todo_item.call_args
assert args
item = args.kwargs.get("item")
assert item
assert item.uid == "item-1"
assert item.summary == "Updated item"
assert item.status == TodoItemStatus.COMPLETED
async def test_update_todo_item_service_by_id_status_only(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test updating an item in a To-do list."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"update_item",
{"uid": "item-1", "status": "completed"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_update_todo_item.call_args
assert args
item = args.kwargs.get("item")
assert item
assert item.uid == "item-1"
assert item.summary is None
assert item.status == TodoItemStatus.COMPLETED
async def test_update_todo_item_service_by_id_summary_only(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test updating an item in a To-do list."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"update_item",
{"uid": "item-1", "summary": "Updated item"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_update_todo_item.call_args
assert args
item = args.kwargs.get("item")
assert item
assert item.uid == "item-1"
assert item.summary == "Updated item"
assert item.status is None
async def test_update_todo_item_service_raises(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test updating an item in a To-do list that raises an error."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"update_item",
{"uid": "item-1", "summary": "Updated item", "status": "completed"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
test_entity.async_update_todo_item.side_effect = HomeAssistantError("Ooops")
with pytest.raises(HomeAssistantError, match="Ooops"):
await hass.services.async_call(
DOMAIN,
"update_item",
{"uid": "item-1", "summary": "Updated item", "status": "completed"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_update_todo_item_service_by_summary(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test updating an item in a To-do list by summary."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"update_item",
{"summary": "Item #1", "status": "completed"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_update_todo_item.call_args
assert args
item = args.kwargs.get("item")
assert item
assert item.uid == "1"
assert item.summary == "Item #1"
assert item.status == TodoItemStatus.COMPLETED
async def test_update_todo_item_service_by_summary_not_found(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test updating an item in a To-do list by summary which is not found."""
await create_mock_platform(hass, [test_entity])
with pytest.raises(ValueError, match="Unable to find"):
await hass.services.async_call(
DOMAIN,
"update_item",
{"summary": "Item #7", "status": "completed"},
target={"entity_id": "todo.entity1"},
blocking=True,
)
@pytest.mark.parametrize(
("item_data", "expected_error"),
[
({}, "must contain at least one of"),
({"status": "needs_action"}, "must contain at least one of"),
(
{"summary": "", "status": "needs_action"},
"length of value must be at least 1",
),
],
)
async def test_update_item_service_invalid_input(
hass: HomeAssistant,
test_entity: TodoListEntity,
item_data: dict[str, Any],
expected_error: str,
) -> None:
"""Test invalid input to the update item service."""
await create_mock_platform(hass, [test_entity])
with pytest.raises(vol.Invalid, match=expected_error):
await hass.services.async_call(
DOMAIN,
"update_item",
item_data,
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_delete_todo_item_service_by_id(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test deleting an item in a To-do list."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"delete_item",
{"uid": ["item-1", "item-2"]},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_delete_todo_items.call_args
assert args
assert args.kwargs.get("uids") == ["item-1", "item-2"]
async def test_delete_todo_item_service_raises(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test deleting an item in a To-do list that raises an error."""
await create_mock_platform(hass, [test_entity])
test_entity.async_delete_todo_items.side_effect = HomeAssistantError("Ooops")
with pytest.raises(HomeAssistantError, match="Ooops"):
await hass.services.async_call(
DOMAIN,
"delete_item",
{"uid": ["item-1", "item-2"]},
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_delete_todo_item_service_invalid_input(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test invalid input to the delete item service."""
await create_mock_platform(hass, [test_entity])
with pytest.raises(vol.Invalid, match="must contain at least one of"):
await hass.services.async_call(
DOMAIN,
"delete_item",
{},
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_delete_todo_item_service_by_summary(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test deleting an item in a To-do list by summary."""
await create_mock_platform(hass, [test_entity])
await hass.services.async_call(
DOMAIN,
"delete_item",
{"summary": ["Item #1"]},
target={"entity_id": "todo.entity1"},
blocking=True,
)
args = test_entity.async_delete_todo_items.call_args
assert args
assert args.kwargs.get("uids") == ["1"]
async def test_delete_todo_item_service_by_summary_not_found(
hass: HomeAssistant,
test_entity: TodoListEntity,
) -> None:
"""Test deleting an item in a To-do list by summary which is not found."""
await create_mock_platform(hass, [test_entity])
with pytest.raises(ValueError, match="Unable to find"):
await hass.services.async_call(
DOMAIN,
"delete_item",
{"summary": ["Item #7"]},
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_move_todo_item_service_by_id(
hass: HomeAssistant,
test_entity: TodoListEntity,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test moving an item in a To-do list."""
await create_mock_platform(hass, [test_entity])
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "todo/item/move",
"entity_id": "todo.entity1",
"uid": "item-1",
"pos": "1",
}
)
resp = await client.receive_json()
assert resp.get("id") == 1
assert resp.get("success")
args = test_entity.async_move_todo_item.call_args
assert args
assert args.kwargs.get("uid") == "item-1"
assert args.kwargs.get("pos") == 1
async def test_move_todo_item_service_raises(
hass: HomeAssistant,
test_entity: TodoListEntity,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test moving an item in a To-do list that raises an error."""
await create_mock_platform(hass, [test_entity])
test_entity.async_move_todo_item.side_effect = HomeAssistantError("Ooops")
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "todo/item/move",
"entity_id": "todo.entity1",
"uid": "item-1",
"pos": "1",
}
)
resp = await client.receive_json()
assert resp.get("id") == 1
assert resp.get("error", {}).get("code") == "failed"
assert resp.get("error", {}).get("message") == "Ooops"
@pytest.mark.parametrize(
("item_data", "expected_status", "expected_error"),
[
(
{"entity_id": "todo.unknown", "uid": "item-1"},
"not_found",
"Entity not found",
),
({"entity_id": "todo.entity1"}, "invalid_format", "required key not provided"),
(
{"entity_id": "todo.entity1", "pos": "2"},
"invalid_format",
"required key not provided",
),
(
{"entity_id": "todo.entity1", "uid": "item-1", "pos": "-2"},
"invalid_format",
"value must be at least 0",
),
],
)
async def test_move_todo_item_service_invalid_input(
hass: HomeAssistant,
test_entity: TodoListEntity,
hass_ws_client: WebSocketGenerator,
item_data: dict[str, Any],
expected_status: str,
expected_error: str,
) -> None:
"""Test invalid input for the move item service."""
await create_mock_platform(hass, [test_entity])
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "todo/item/move",
**item_data,
}
)
resp = await client.receive_json()
assert resp.get("id") == 1
assert resp.get("error", {}).get("code") == expected_status
assert expected_error in resp.get("error", {}).get("message")
@pytest.mark.parametrize(
("service_name", "payload"),
[
(
"create_item",
{
"summary": "New item",
},
),
(
"delete_item",
{
"uid": ["1"],
},
),
(
"update_item",
{
"uid": "1",
"summary": "Updated item",
},
),
],
)
async def test_unsupported_service(
hass: HomeAssistant,
service_name: str,
payload: dict[str, Any],
) -> None:
"""Test a To-do list that does not support features."""
entity1 = TodoListEntity()
entity1.entity_id = "todo.entity1"
await create_mock_platform(hass, [entity1])
with pytest.raises(
HomeAssistantError,
match="does not support this service",
):
await hass.services.async_call(
DOMAIN,
service_name,
payload,
target={"entity_id": "todo.entity1"},
blocking=True,
)
async def test_move_item_unsupported(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test invalid input for the move item service."""
entity1 = TodoListEntity()
entity1.entity_id = "todo.entity1"
await create_mock_platform(hass, [entity1])
client = await hass_ws_client()
await client.send_json(
{
"id": 1,
"type": "todo/item/move",
"entity_id": "todo.entity1",
"uid": "item-1",
"pos": "1",
}
)
resp = await client.receive_json()
assert resp.get("id") == 1
assert resp.get("error", {}).get("code") == "not_supported"