Add Google Tasks create and update for todo platform (#102754)

* Add Google Tasks create and update for todo platform

* Update comments

* Update comments

homeassistant/components/google_tasks/api.py

@@ -51,3 +51,31 @@ class AsyncConfigEntryAuth:
         )
         result = await self._hass.async_add_executor_job(cmd.execute)
         return result["items"]
+
+    async def insert(
+        self,
+        task_list_id: str,
+        task: dict[str, Any],
+    ) -> None:
+        """Create a new Task resource on the task list."""
+        service = await self._get_service()
+        cmd: HttpRequest = service.tasks().insert(
+            tasklist=task_list_id,
+            body=task,
+        )
+        await self._hass.async_add_executor_job(cmd.execute)
+
+    async def patch(
+        self,
+        task_list_id: str,
+        task_id: str,
+        task: dict[str, Any],
+    ) -> None:
+        """Update a task resource."""
+        service = await self._get_service()
+        cmd: HttpRequest = service.tasks().patch(
+            tasklist=task_list_id,
+            task=task_id,
+            body=task,
+        )
+        await self._hass.async_add_executor_job(cmd.execute)

homeassistant/components/google_tasks/coordinator.py

@@ -29,10 +29,10 @@ class TaskUpdateCoordinator(DataUpdateCoordinator):
             name=f"Google Tasks {task_list_id}",
             update_interval=UPDATE_INTERVAL,
         )
-        self._api = api
+        self.api = api
         self._task_list_id = task_list_id
 
     async def _async_update_data(self) -> list[dict[str, Any]]:
         """Fetch tasks from API endpoint."""
         async with asyncio.timeout(TIMEOUT):
-            return await self._api.list_tasks(self._task_list_id)
+            return await self.api.list_tasks(self._task_list_id)

homeassistant/components/google_tasks/todo.py

@@ -2,8 +2,14 @@
 from __future__ import annotations
 
 from datetime import timedelta
+from typing import cast
 
-from homeassistant.components.todo import TodoItem, TodoItemStatus, TodoListEntity
+from homeassistant.components.todo import (
+    TodoItem,
+    TodoItemStatus,
+    TodoListEntity,
+    TodoListEntityFeature,
+)
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
@@ -19,6 +25,17 @@ TODO_STATUS_MAP = {
     "needsAction": TodoItemStatus.NEEDS_ACTION,
     "completed": TodoItemStatus.COMPLETED,
 }
+TODO_STATUS_MAP_INV = {v: k for k, v in TODO_STATUS_MAP.items()}
+
+
+def _convert_todo_item(item: TodoItem) -> dict[str, str]:
+    """Convert TodoItem dataclass items to dictionary of attributes the tasks API."""
+    result: dict[str, str] = {}
+    if item.summary is not None:
+        result["title"] = item.summary
+    if item.status is not None:
+        result["status"] = TODO_STATUS_MAP_INV[item.status]
+    return result
 
 
 async def async_setup_entry(
@@ -45,6 +62,9 @@ class GoogleTaskTodoListEntity(CoordinatorEntity, TodoListEntity):
     """A To-do List representation of the Shopping List."""
 
     _attr_has_entity_name = True
+    _attr_supported_features = (
+        TodoListEntityFeature.CREATE_TODO_ITEM | TodoListEntityFeature.UPDATE_TODO_ITEM
+    )
 
     def __init__(
         self,
@@ -57,6 +77,7 @@ class GoogleTaskTodoListEntity(CoordinatorEntity, TodoListEntity):
         super().__init__(coordinator)
         self._attr_name = name.capitalize()
         self._attr_unique_id = f"{config_entry_id}-{task_list_id}"
+        self._task_list_id = task_list_id
 
     @property
     def todo_items(self) -> list[TodoItem] | None:
@@ -73,3 +94,21 @@ class GoogleTaskTodoListEntity(CoordinatorEntity, TodoListEntity):
             )
             for item in self.coordinator.data
         ]
+
+    async def async_create_todo_item(self, item: TodoItem) -> None:
+        """Add an item to the To-do list."""
+        await self.coordinator.api.insert(
+            self._task_list_id,
+            task=_convert_todo_item(item),
+        )
+        await self.coordinator.async_refresh()
+
+    async def async_update_todo_item(self, item: TodoItem) -> None:
+        """Update a To-do item."""
+        uid: str = cast(str, item.uid)
+        await self.coordinator.api.patch(
+            self._task_list_id,
+            uid,
+            task=_convert_todo_item(item),
+        )
+        await self.coordinator.async_refresh()

tests/components/google_tasks/snapshots/test_todo.ambr

@@ -0,0 +1,37 @@
+# serializer version: 1
+# name: test_create_todo_list_item[api_responses0]
+  tuple(
+    'https://tasks.googleapis.com/tasks/v1/lists/task-list-id-1/tasks?alt=json',
+    'POST',
+  )
+# ---
+# name: test_create_todo_list_item[api_responses0].1
+  '{"title": "Soda", "status": "needsAction"}'
+# ---
+# name: test_partial_update_status[api_responses0]
+  tuple(
+    'https://tasks.googleapis.com/tasks/v1/lists/task-list-id-1/tasks/some-task-id?alt=json',
+    'PATCH',
+  )
+# ---
+# name: test_partial_update_status[api_responses0].1
+  '{"status": "needsAction"}'
+# ---
+# name: test_partial_update_title[api_responses0]
+  tuple(
+    'https://tasks.googleapis.com/tasks/v1/lists/task-list-id-1/tasks/some-task-id?alt=json',
+    'PATCH',
+  )
+# ---
+# name: test_partial_update_title[api_responses0].1
+  '{"title": "Soda"}'
+# ---
+# name: test_update_todo_list_item[api_responses0]
+  tuple(
+    'https://tasks.googleapis.com/tasks/v1/lists/task-list-id-1/tasks/some-task-id?alt=json',
+    'PATCH',
+  )
+# ---
+# name: test_update_todo_list_item[api_responses0].1
+  '{"title": "Soda", "status": "completed"}'
+# ---

tests/components/google_tasks/test_todo.py

@@ -3,11 +3,14 @@
 
 from collections.abc import Awaitable, Callable
 import json
-from unittest.mock import patch
+from typing import Any
+from unittest.mock import Mock, patch
 
 from httplib2 import Response
 import pytest
+from syrupy.assertion import SnapshotAssertion
 
+from homeassistant.components.todo import DOMAIN as TODO_DOMAIN
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 
@@ -22,6 +25,10 @@ LIST_TASK_LIST_RESPONSE = {
         },
     ]
 }
+EMPTY_RESPONSE = {}
+LIST_TASKS_RESPONSE = {
+    "items": [],
+}
 
 
 @pytest.fixture
@@ -76,14 +83,14 @@ def mock_api_responses() -> list[dict | list]:
 
 
 @pytest.fixture(autouse=True)
-def mock_http_response(api_responses: list[dict | list]) -> None:
+def mock_http_response(api_responses: list[dict | list]) -> Mock:
     """Fixture to fake out http2lib responses."""
     responses = [
         (Response({}), bytes(json.dumps(api_response), encoding="utf-8"))
         for api_response in api_responses
     ]
-    with patch("httplib2.Http.request", side_effect=responses):
-        yield
+    with patch("httplib2.Http.request", side_effect=responses) as mock_response:
+        yield mock_response
 
 
 @pytest.mark.parametrize(
@@ -138,9 +145,7 @@ async def test_get_items(
     [
         [
             LIST_TASK_LIST_RESPONSE,
-            {
-                "items": [],
-            },
+            LIST_TASKS_RESPONSE,
         ]
     ],
 )
@@ -163,3 +168,163 @@ async def test_empty_todo_list(
     state = hass.states.get("todo.my_tasks")
     assert state
     assert state.state == "0"
+
+
+@pytest.mark.parametrize(
+    "api_responses",
+    [
+        [
+            LIST_TASK_LIST_RESPONSE,
+            LIST_TASKS_RESPONSE,
+            EMPTY_RESPONSE,  # create
+            LIST_TASKS_RESPONSE,  # refresh after create
+        ]
+    ],
+)
+async def test_create_todo_list_item(
+    hass: HomeAssistant,
+    setup_credentials: None,
+    integration_setup: Callable[[], Awaitable[bool]],
+    mock_http_response: Mock,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test for creating a To-do Item."""
+
+    assert await integration_setup()
+
+    state = hass.states.get("todo.my_tasks")
+    assert state
+    assert state.state == "0"
+
+    await hass.services.async_call(
+        TODO_DOMAIN,
+        "create_item",
+        {"summary": "Soda"},
+        target={"entity_id": "todo.my_tasks"},
+        blocking=True,
+    )
+    assert len(mock_http_response.call_args_list) == 4
+    call = mock_http_response.call_args_list[2]
+    assert call
+    assert call.args == snapshot
+    assert call.kwargs.get("body") == snapshot
+
+
+@pytest.mark.parametrize(
+    "api_responses",
+    [
+        [
+            LIST_TASK_LIST_RESPONSE,
+            LIST_TASKS_RESPONSE,
+            EMPTY_RESPONSE,  # update
+            LIST_TASKS_RESPONSE,  # refresh after update
+        ]
+    ],
+)
+async def test_update_todo_list_item(
+    hass: HomeAssistant,
+    setup_credentials: None,
+    integration_setup: Callable[[], Awaitable[bool]],
+    mock_http_response: Any,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test for updating a To-do Item."""
+
+    assert await integration_setup()
+
+    state = hass.states.get("todo.my_tasks")
+    assert state
+    assert state.state == "0"
+
+    await hass.services.async_call(
+        TODO_DOMAIN,
+        "update_item",
+        {"uid": "some-task-id", "summary": "Soda", "status": "completed"},
+        target={"entity_id": "todo.my_tasks"},
+        blocking=True,
+    )
+    assert len(mock_http_response.call_args_list) == 4
+    call = mock_http_response.call_args_list[2]
+    assert call
+    assert call.args == snapshot
+    assert call.kwargs.get("body") == snapshot
+
+
+@pytest.mark.parametrize(
+    "api_responses",
+    [
+        [
+            LIST_TASK_LIST_RESPONSE,
+            LIST_TASKS_RESPONSE,
+            EMPTY_RESPONSE,  # update
+            LIST_TASKS_RESPONSE,  # refresh after update
+        ]
+    ],
+)
+async def test_partial_update_title(
+    hass: HomeAssistant,
+    setup_credentials: None,
+    integration_setup: Callable[[], Awaitable[bool]],
+    mock_http_response: Any,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test for partial update with title only."""
+
+    assert await integration_setup()
+
+    state = hass.states.get("todo.my_tasks")
+    assert state
+    assert state.state == "0"
+
+    await hass.services.async_call(
+        TODO_DOMAIN,
+        "update_item",
+        {"uid": "some-task-id", "summary": "Soda"},
+        target={"entity_id": "todo.my_tasks"},
+        blocking=True,
+    )
+    assert len(mock_http_response.call_args_list) == 4
+    call = mock_http_response.call_args_list[2]
+    assert call
+    assert call.args == snapshot
+    assert call.kwargs.get("body") == snapshot
+
+
+@pytest.mark.parametrize(
+    "api_responses",
+    [
+        [
+            LIST_TASK_LIST_RESPONSE,
+            LIST_TASKS_RESPONSE,
+            EMPTY_RESPONSE,  # update
+            LIST_TASKS_RESPONSE,  # refresh after update
+        ]
+    ],
+)
+async def test_partial_update_status(
+    hass: HomeAssistant,
+    setup_credentials: None,
+    integration_setup: Callable[[], Awaitable[bool]],
+    mock_http_response: Any,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test for partial update with status only."""
+
+    assert await integration_setup()
+
+    state = hass.states.get("todo.my_tasks")
+    assert state
+    assert state.state == "0"
+
+    await hass.services.async_call(
+        TODO_DOMAIN,
+        "update_item",
+        {"uid": "some-task-id", "status": "needs_action"},
+        target={"entity_id": "todo.my_tasks"},
+        blocking=True,
+    )
+    assert len(mock_http_response.call_args_list) == 4
+    call = mock_http_response.call_args_list[2]
+    assert call
+    assert call.args == snapshot
+    assert call.kwargs.get("body") == snapshot