Add Google tasks integration, with initial read-only To-do list (#102629)

* Add Google Tasks integration

* Update tests and unique id

* Revert devcontainer change

* Increase test coverage

* Apply suggestions from code review

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Remove ternary

* Fix JSON

---------

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

CODEOWNERS

@@ -477,6 +477,8 @@ build.json @home-assistant/supervisor
 /tests/components/google_mail/ @tkdrob
 /homeassistant/components/google_sheets/ @tkdrob
 /tests/components/google_sheets/ @tkdrob
+/homeassistant/components/google_tasks/ @allenporter
+/tests/components/google_tasks/ @allenporter
 /homeassistant/components/google_travel_time/ @eifinger
 /tests/components/google_travel_time/ @eifinger
 /homeassistant/components/govee_ble/ @bdraco @PierreAronnax

homeassistant/brands/google.json

@@ -11,6 +11,7 @@
     "google_maps",
     "google_pubsub",
     "google_sheets",
+    "google_tasks",
     "google_translate",
     "google_travel_time",
     "google_wifi",

homeassistant/components/google_tasks/__init__.py

@@ -0,0 +1,46 @@
+"""The Google Tasks integration."""
+from __future__ import annotations
+
+from aiohttp import ClientError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from . import api
+from .const import DOMAIN
+
+PLATFORMS: list[Platform] = [Platform.TODO]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Google Tasks from a config entry."""
+    hass.data.setdefault(DOMAIN, {})
+
+    implementation = (
+        await config_entry_oauth2_flow.async_get_config_entry_implementation(
+            hass, entry
+        )
+    )
+    session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
+    auth = api.AsyncConfigEntryAuth(hass, session)
+    try:
+        await auth.async_get_access_token()
+    except ClientError as err:
+        raise ConfigEntryNotReady from err
+
+    hass.data[DOMAIN][entry.entry_id] = auth
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/google_tasks/api.py

@@ -0,0 +1,53 @@
+"""API for Google Tasks bound to Home Assistant OAuth."""
+
+from typing import Any
+
+from google.oauth2.credentials import Credentials
+from googleapiclient.discovery import Resource, build
+from googleapiclient.http import HttpRequest
+
+from homeassistant.const import CONF_ACCESS_TOKEN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+
+MAX_TASK_RESULTS = 100
+
+
+class AsyncConfigEntryAuth:
+    """Provide Google Tasks authentication tied to an OAuth2 based config entry."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        oauth2_session: config_entry_oauth2_flow.OAuth2Session,
+    ) -> None:
+        """Initialize Google Tasks Auth."""
+        self._hass = hass
+        self._oauth_session = oauth2_session
+
+    async def async_get_access_token(self) -> str:
+        """Return a valid access token."""
+        if not self._oauth_session.valid_token:
+            await self._oauth_session.async_ensure_token_valid()
+        return self._oauth_session.token[CONF_ACCESS_TOKEN]
+
+    async def _get_service(self) -> Resource:
+        """Get current resource."""
+        token = await self.async_get_access_token()
+        return build("tasks", "v1", credentials=Credentials(token=token))
+
+    async def list_task_lists(self) -> list[dict[str, Any]]:
+        """Get all TaskList resources."""
+        service = await self._get_service()
+        cmd: HttpRequest = service.tasklists().list()
+        result = await self._hass.async_add_executor_job(cmd.execute)
+        return result["items"]
+
+    async def list_tasks(self, task_list_id: str) -> list[dict[str, Any]]:
+        """Get all Task resources for the task list."""
+        service = await self._get_service()
+        cmd: HttpRequest = service.tasks().list(
+            tasklist=task_list_id, maxResults=MAX_TASK_RESULTS
+        )
+        result = await self._hass.async_add_executor_job(cmd.execute)
+        return result["items"]

homeassistant/components/google_tasks/application_credentials.py

@@ -0,0 +1,23 @@
+"""Application credentials platform for the Google Tasks integration."""
+
+from homeassistant.components.application_credentials import AuthorizationServer
+from homeassistant.core import HomeAssistant
+
+from .const import OAUTH2_AUTHORIZE, OAUTH2_TOKEN
+
+
+async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
+    """Return authorization server."""
+    return AuthorizationServer(
+        authorize_url=OAUTH2_AUTHORIZE,
+        token_url=OAUTH2_TOKEN,
+    )
+
+
+async def async_get_description_placeholders(hass: HomeAssistant) -> dict[str, str]:
+    """Return description placeholders for the credentials dialog."""
+    return {
+        "oauth_consent_url": "https://console.cloud.google.com/apis/credentials/consent",
+        "more_info_url": "https://www.home-assistant.io/integrations/google_tasks/",
+        "oauth_creds_url": "https://console.cloud.google.com/apis/credentials",
+    }

homeassistant/components/google_tasks/config_flow.py

@@ -0,0 +1,30 @@
+"""Config flow for Google Tasks."""
+import logging
+from typing import Any
+
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from .const import DOMAIN, OAUTH2_SCOPES
+
+
+class OAuth2FlowHandler(
+    config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
+):
+    """Config flow to handle Google Tasks OAuth2 authentication."""
+
+    DOMAIN = DOMAIN
+
+    @property
+    def logger(self) -> logging.Logger:
+        """Return logger."""
+        return logging.getLogger(__name__)
+
+    @property
+    def extra_authorize_data(self) -> dict[str, Any]:
+        """Extra data that needs to be appended to the authorize url."""
+        return {
+            "scope": " ".join(OAUTH2_SCOPES),
+            # Add params to ensure we get back a refresh token
+            "access_type": "offline",
+            "prompt": "consent",
+        }

homeassistant/components/google_tasks/const.py

@@ -0,0 +1,16 @@
+"""Constants for the Google Tasks integration."""
+
+from enum import StrEnum
+
+DOMAIN = "google_tasks"
+
+OAUTH2_AUTHORIZE = "https://accounts.google.com/o/oauth2/v2/auth"
+OAUTH2_TOKEN = "https://oauth2.googleapis.com/token"
+OAUTH2_SCOPES = ["https://www.googleapis.com/auth/tasks"]
+
+
+class TaskStatus(StrEnum):
+    """Status of a Google Task."""
+
+    NEEDS_ACTION = "needsAction"
+    COMPLETED = "completed"

homeassistant/components/google_tasks/coordinator.py

@@ -0,0 +1,38 @@
+"""Coordinator for fetching data from Google Tasks API."""
+
+import asyncio
+import datetime
+import logging
+from typing import Any, Final
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .api import AsyncConfigEntryAuth
+
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_INTERVAL: Final = datetime.timedelta(minutes=30)
+TIMEOUT = 10
+
+
+class TaskUpdateCoordinator(DataUpdateCoordinator):
+    """Coordinator for fetching Google Tasks for a Task List form the API."""
+
+    def __init__(
+        self, hass: HomeAssistant, api: AsyncConfigEntryAuth, task_list_id: str
+    ) -> None:
+        """Initialize TaskUpdateCoordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=f"Google Tasks {task_list_id}",
+            update_interval=UPDATE_INTERVAL,
+        )
+        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)

homeassistant/components/google_tasks/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "google_tasks",
+  "name": "Google Tasks",
+  "codeowners": ["@allenporter"],
+  "config_flow": true,
+  "dependencies": ["application_credentials"],
+  "documentation": "https://www.home-assistant.io/integrations/google_tasks",
+  "iot_class": "cloud_polling",
+  "requirements": ["google-api-python-client==2.71.0"]
+}

homeassistant/components/google_tasks/strings.json

@@ -0,0 +1,24 @@
+{
+  "application_credentials": {
+    "description": "Follow the [instructions]({more_info_url}) for [OAuth consent screen]({oauth_consent_url}) to give Home Assistant access to your Google Tasks. You also need to create Application Credentials linked to your account:\n1. Go to [Credentials]({oauth_creds_url}) and click **Create Credentials**.\n1. From the drop-down list select **OAuth client ID**.\n1. Select **Web application** for the Application Type.\n\n"
+  },
+  "config": {
+    "step": {
+      "pick_implementation": {
+        "title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
+      }
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
+      "missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
+      "authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
+      "no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
+      "user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]"
+    },
+    "create_entry": {
+      "default": "[%key:common::config_flow::create_entry::authenticated%]"
+    }
+  }
+}

homeassistant/components/google_tasks/todo.py

@@ -0,0 +1,75 @@
+"""Google Tasks todo platform."""
+from __future__ import annotations
+
+from datetime import timedelta
+
+from homeassistant.components.todo import TodoItem, TodoItemStatus, TodoListEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .api import AsyncConfigEntryAuth
+from .const import DOMAIN
+from .coordinator import TaskUpdateCoordinator
+
+SCAN_INTERVAL = timedelta(minutes=15)
+
+TODO_STATUS_MAP = {
+    "needsAction": TodoItemStatus.NEEDS_ACTION,
+    "completed": TodoItemStatus.COMPLETED,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up the Google Tasks todo platform."""
+    api: AsyncConfigEntryAuth = hass.data[DOMAIN][entry.entry_id]
+    task_lists = await api.list_task_lists()
+    async_add_entities(
+        (
+            GoogleTaskTodoListEntity(
+                TaskUpdateCoordinator(hass, api, task_list["id"]),
+                task_list["title"],
+                entry.entry_id,
+                task_list["id"],
+            )
+            for task_list in task_lists
+        ),
+        True,
+    )
+
+
+class GoogleTaskTodoListEntity(CoordinatorEntity, TodoListEntity):
+    """A To-do List representation of the Shopping List."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: TaskUpdateCoordinator,
+        name: str,
+        config_entry_id: str,
+        task_list_id: str,
+    ) -> None:
+        """Initialize LocalTodoListEntity."""
+        super().__init__(coordinator)
+        self._attr_name = name.capitalize()
+        self._attr_unique_id = f"{config_entry_id}-{task_list_id}"
+
+    @property
+    def todo_items(self) -> list[TodoItem] | None:
+        """Get the current set of To-do items."""
+        if self.coordinator.data is None:
+            return None
+        return [
+            TodoItem(
+                summary=item["title"],
+                uid=item["id"],
+                status=TODO_STATUS_MAP.get(
+                    item.get("status"), TodoItemStatus.NEEDS_ACTION
+                ),
+            )
+            for item in self.coordinator.data
+        ]

homeassistant/generated/application_credentials.py

@@ -11,6 +11,7 @@ APPLICATION_CREDENTIALS = [
     "google_assistant_sdk",
     "google_mail",
     "google_sheets",
+    "google_tasks",
     "home_connect",
     "lametric",
     "lyric",

homeassistant/generated/config_flows.py

@@ -181,6 +181,7 @@ FLOWS = {
         "google_generative_ai_conversation",
         "google_mail",
         "google_sheets",
+        "google_tasks",
         "google_translate",
         "google_travel_time",
         "govee_ble",

homeassistant/generated/integrations.json

@@ -2156,6 +2156,12 @@
           "iot_class": "cloud_polling",
           "name": "Google Sheets"
         },
+        "google_tasks": {
+          "integration_type": "hub",
+          "config_flow": true,
+          "iot_class": "cloud_polling",
+          "name": "Google Tasks"
+        },
         "google_translate": {
           "integration_type": "hub",
           "config_flow": true,

requirements_all.txt

@@ -897,6 +897,7 @@ goalzero==0.2.2
 goodwe==0.2.31
 
 # homeassistant.components.google_mail
+# homeassistant.components.google_tasks
 google-api-python-client==2.71.0
 
 # homeassistant.components.google_pubsub

requirements_test_all.txt

@@ -716,6 +716,7 @@ goalzero==0.2.2
 goodwe==0.2.31
 
 # homeassistant.components.google_mail
+# homeassistant.components.google_tasks
 google-api-python-client==2.71.0
 
 # homeassistant.components.google_pubsub

tests/components/google_tasks/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Google Tasks integration."""

tests/components/google_tasks/conftest.py

@@ -0,0 +1,91 @@
+"""Test fixtures for Google Tasks."""
+
+
+from collections.abc import Awaitable, Callable
+import time
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.application_credentials import (
+    ClientCredential,
+    async_import_client_credential,
+)
+from homeassistant.components.google_tasks.const import DOMAIN, OAUTH2_SCOPES
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+
+CLIENT_ID = "1234"
+CLIENT_SECRET = "5678"
+FAKE_ACCESS_TOKEN = "some-access-token"
+FAKE_REFRESH_TOKEN = "some-refresh-token"
+FAKE_AUTH_IMPL = "conftest-imported-cred"
+
+
+@pytest.fixture
+def platforms() -> list[Platform]:
+    """Fixture to specify platforms to test."""
+    return []
+
+
+@pytest.fixture(name="expires_at")
+def mock_expires_at() -> int:
+    """Fixture to set the oauth token expiration time."""
+    return time.time() + 3600
+
+
+@pytest.fixture(name="token_entry")
+def mock_token_entry(expires_at: int) -> dict[str, Any]:
+    """Fixture for OAuth 'token' data for a ConfigEntry."""
+    return {
+        "access_token": FAKE_ACCESS_TOKEN,
+        "refresh_token": FAKE_REFRESH_TOKEN,
+        "scope": " ".join(OAUTH2_SCOPES),
+        "token_type": "Bearer",
+        "expires_at": expires_at,
+    }
+
+
+@pytest.fixture(name="config_entry")
+def mock_config_entry(token_entry: dict[str, Any]) -> MockConfigEntry:
+    """Fixture for a config entry."""
+    return MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            "auth_implementation": DOMAIN,
+            "token": token_entry,
+        },
+    )
+
+
+@pytest.fixture
+async def setup_credentials(hass: HomeAssistant) -> None:
+    """Fixture to setup credentials."""
+    assert await async_setup_component(hass, "application_credentials", {})
+    await async_import_client_credential(
+        hass,
+        DOMAIN,
+        ClientCredential(CLIENT_ID, CLIENT_SECRET),
+    )
+
+
+@pytest.fixture(name="integration_setup")
+async def mock_integration_setup(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    platforms: list[str],
+) -> Callable[[], Awaitable[bool]]:
+    """Fixture to set up the integration."""
+    config_entry.add_to_hass(hass)
+
+    async def run() -> bool:
+        with patch(f"homeassistant.components.{DOMAIN}.PLATFORMS", platforms):
+            result = await hass.config_entries.async_setup(config_entry.entry_id)
+            await hass.async_block_till_done()
+        return result
+
+    return run

tests/components/google_tasks/test_config_flow.py

@@ -0,0 +1,66 @@
+"""Test the Google Tasks config flow."""
+
+from unittest.mock import patch
+
+from homeassistant import config_entries
+from homeassistant.components.google_tasks.const import (
+    DOMAIN,
+    OAUTH2_AUTHORIZE,
+    OAUTH2_TOKEN,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+
+CLIENT_ID = "1234"
+CLIENT_SECRET = "5678"
+
+
+async def test_full_flow(
+    hass: HomeAssistant,
+    hass_client_no_auth,
+    aioclient_mock,
+    current_request_with_host,
+    setup_credentials,
+) -> None:
+    """Check full flow."""
+    result = await hass.config_entries.flow.async_init(
+        "google_tasks", context={"source": config_entries.SOURCE_USER}
+    )
+    state = config_entry_oauth2_flow._encode_jwt(
+        hass,
+        {
+            "flow_id": result["flow_id"],
+            "redirect_uri": "https://example.com/auth/external/callback",
+        },
+    )
+
+    assert result["url"] == (
+        f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
+        "&redirect_uri=https://example.com/auth/external/callback"
+        f"&state={state}"
+        "&scope=https://www.googleapis.com/auth/tasks"
+        "&access_type=offline&prompt=consent"
+    )
+
+    client = await hass_client_no_auth()
+    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
+    assert resp.status == 200
+    assert resp.headers["content-type"] == "text/html; charset=utf-8"
+
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "refresh_token": "mock-refresh-token",
+            "access_token": "mock-access-token",
+            "type": "Bearer",
+            "expires_in": 60,
+        },
+    )
+
+    with patch(
+        "homeassistant.components.google_tasks.async_setup_entry", return_value=True
+    ) as mock_setup:
+        await hass.config_entries.flow.async_configure(result["flow_id"])
+
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
+    assert len(mock_setup.mock_calls) == 1

tests/components/google_tasks/test_init.py

@@ -0,0 +1,99 @@
+"""Tests for Google Tasks."""
+from collections.abc import Awaitable, Callable
+import http
+import time
+
+import pytest
+
+from homeassistant.components.google_tasks import DOMAIN
+from homeassistant.components.google_tasks.const import OAUTH2_TOKEN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_setup(
+    hass: HomeAssistant,
+    integration_setup: Callable[[], Awaitable[bool]],
+    config_entry: MockConfigEntry,
+    setup_credentials: None,
+) -> None:
+    """Test successful setup and unload."""
+    assert config_entry.state is ConfigEntryState.NOT_LOADED
+
+    await integration_setup()
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert config_entry.state == ConfigEntryState.NOT_LOADED
+    assert not hass.services.async_services().get(DOMAIN)
+
+
+@pytest.mark.parametrize("expires_at", [time.time() - 3600], ids=["expired"])
+async def test_expired_token_refresh_success(
+    hass: HomeAssistant,
+    integration_setup: Callable[[], Awaitable[bool]],
+    aioclient_mock: AiohttpClientMocker,
+    config_entry: MockConfigEntry,
+    setup_credentials: None,
+) -> None:
+    """Test expired token is refreshed."""
+
+    aioclient_mock.clear_requests()
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        json={
+            "access_token": "updated-access-token",
+            "refresh_token": "updated-refresh-token",
+            "expires_at": time.time() + 3600,
+            "expires_in": 3600,
+        },
+    )
+
+    await integration_setup()
+
+    assert config_entry.state is ConfigEntryState.LOADED
+    assert config_entry.data["token"]["access_token"] == "updated-access-token"
+    assert config_entry.data["token"]["expires_in"] == 3600
+
+
+@pytest.mark.parametrize(
+    ("expires_at", "status", "expected_state"),
+    [
+        (
+            time.time() - 3600,
+            http.HTTPStatus.UNAUTHORIZED,
+            ConfigEntryState.SETUP_RETRY,  # Will trigger reauth in the future
+        ),
+        (
+            time.time() - 3600,
+            http.HTTPStatus.INTERNAL_SERVER_ERROR,
+            ConfigEntryState.SETUP_RETRY,
+        ),
+    ],
+    ids=["unauthorized", "internal_server_error"],
+)
+async def test_expired_token_refresh_failure(
+    hass: HomeAssistant,
+    integration_setup: Callable[[], Awaitable[bool]],
+    aioclient_mock: AiohttpClientMocker,
+    config_entry: MockConfigEntry,
+    setup_credentials: None,
+    status: http.HTTPStatus,
+    expected_state: ConfigEntryState,
+) -> None:
+    """Test failure while refreshing token with a transient error."""
+
+    aioclient_mock.clear_requests()
+    aioclient_mock.post(
+        OAUTH2_TOKEN,
+        status=status,
+    )
+
+    await integration_setup()
+
+    assert config_entry.state is expected_state

tests/components/google_tasks/test_todo.py

@@ -0,0 +1,165 @@
+"""Tests for Google Tasks todo platform."""
+
+
+from collections.abc import Awaitable, Callable
+import json
+from unittest.mock import patch
+
+from httplib2 import Response
+import pytest
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from tests.typing import WebSocketGenerator
+
+ENTITY_ID = "todo.my_tasks"
+LIST_TASK_LIST_RESPONSE = {
+    "items": [
+        {
+            "id": "task-list-id-1",
+            "title": "My tasks",
+        },
+    ]
+}
+
+
+@pytest.fixture
+def platforms() -> list[str]:
+    """Fixture to specify platforms to test."""
+    return [Platform.TODO]
+
+
+@pytest.fixture
+def ws_req_id() -> Callable[[], int]:
+    """Fixture for incremental websocket requests."""
+
+    id = 0
+
+    def next_id() -> int:
+        nonlocal id
+        id += 1
+        return id
+
+    return next_id
+
+
+@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": ENTITY_ID,
+            }
+        )
+        resp = await client.receive_json()
+        assert resp.get("id") == id
+        assert resp.get("success")
+        return resp.get("result", {}).get("items", [])
+
+    return get
+
+
+@pytest.fixture(name="api_responses")
+def mock_api_responses() -> list[dict | list]:
+    """Fixture for API responses to return during test."""
+    return []
+
+
+@pytest.fixture(autouse=True)
+def mock_http_response(api_responses: list[dict | list]) -> None:
+    """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
+
+
+@pytest.mark.parametrize(
+    "api_responses",
+    [
+        [
+            LIST_TASK_LIST_RESPONSE,
+            {
+                "items": [
+                    {"id": "task-1", "title": "Task 1", "status": "needsAction"},
+                    {"id": "task-2", "title": "Task 2", "status": "completed"},
+                ],
+            },
+        ]
+    ],
+)
+async def test_get_items(
+    hass: HomeAssistant,
+    setup_credentials: None,
+    integration_setup: Callable[[], Awaitable[bool]],
+    hass_ws_client: WebSocketGenerator,
+    ws_get_items: Callable[[], Awaitable[dict[str, str]]],
+) -> None:
+    """Test getting todo list items."""
+
+    assert await integration_setup()
+
+    await hass_ws_client(hass)
+
+    items = await ws_get_items()
+    assert items == [
+        {
+            "uid": "task-1",
+            "summary": "Task 1",
+            "status": "needs_action",
+        },
+        {
+            "uid": "task-2",
+            "summary": "Task 2",
+            "status": "completed",
+        },
+    ]
+
+    # State reflect that one task needs action
+    state = hass.states.get("todo.my_tasks")
+    assert state
+    assert state.state == "1"
+
+
+@pytest.mark.parametrize(
+    "api_responses",
+    [
+        [
+            LIST_TASK_LIST_RESPONSE,
+            {
+                "items": [],
+            },
+        ]
+    ],
+)
+async def test_empty_todo_list(
+    hass: HomeAssistant,
+    setup_credentials: None,
+    integration_setup: Callable[[], Awaitable[bool]],
+    hass_ws_client: WebSocketGenerator,
+    ws_get_items: Callable[[], Awaitable[dict[str, str]]],
+) -> None:
+    """Test getting todo list items."""
+
+    assert await integration_setup()
+
+    await hass_ws_client(hass)
+
+    items = await ws_get_items()
+    assert items == []
+
+    state = hass.states.get("todo.my_tasks")
+    assert state
+    assert state.state == "0"