717 lines
19 KiB
Python
717 lines
19 KiB
Python
"""Tests for Google Tasks todo platform."""
|
|
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from http import HTTPStatus
|
|
import json
|
|
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
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
|
|
from tests.typing import WebSocketGenerator
|
|
|
|
ENTITY_ID = "todo.my_tasks"
|
|
LIST_TASK_LIST_RESPONSE = {
|
|
"items": [
|
|
{
|
|
"id": "task-list-id-1",
|
|
"title": "My tasks",
|
|
},
|
|
]
|
|
}
|
|
EMPTY_RESPONSE = {}
|
|
LIST_TASKS_RESPONSE = {
|
|
"items": [],
|
|
}
|
|
ERROR_RESPONSE = {
|
|
"error": {
|
|
"code": 400,
|
|
"message": "Invalid task ID",
|
|
"errors": [
|
|
{"message": "Invalid task ID", "domain": "global", "reason": "invalid"}
|
|
],
|
|
}
|
|
}
|
|
CONTENT_ID = "Content-ID"
|
|
BOUNDARY = "batch_00972cc8-75bd-11ee-9692-0242ac110002" # Arbitrary uuid
|
|
|
|
LIST_TASKS_RESPONSE_WATER = {
|
|
"items": [
|
|
{"id": "some-task-id", "title": "Water", "status": "needsAction"},
|
|
],
|
|
}
|
|
LIST_TASKS_RESPONSE_MULTIPLE = {
|
|
"items": [
|
|
{"id": "some-task-id-1", "title": "Water", "status": "needsAction"},
|
|
{"id": "some-task-id-2", "title": "Milk", "status": "needsAction"},
|
|
{"id": "some-task-id-3", "title": "Cheese", "status": "needsAction"},
|
|
],
|
|
}
|
|
|
|
|
|
@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 []
|
|
|
|
|
|
def create_response_object(api_response: dict | list) -> tuple[Response, bytes]:
|
|
"""Create an http response."""
|
|
return (
|
|
Response({"Content-Type": "application/json"}),
|
|
json.dumps(api_response).encode(),
|
|
)
|
|
|
|
|
|
def create_batch_response_object(
|
|
content_ids: list[str], api_responses: list[dict | list | Response]
|
|
) -> tuple[Response, bytes]:
|
|
"""Create a batch response in the multipart/mixed format."""
|
|
assert len(api_responses) == len(content_ids)
|
|
content = []
|
|
for api_response in api_responses:
|
|
status = 200
|
|
body = ""
|
|
if isinstance(api_response, Response):
|
|
status = api_response.status
|
|
else:
|
|
body = json.dumps(api_response)
|
|
content.extend(
|
|
[
|
|
f"--{BOUNDARY}",
|
|
"Content-Type: application/http",
|
|
f"{CONTENT_ID}: {content_ids.pop()}",
|
|
"",
|
|
f"HTTP/1.1 {status} OK",
|
|
"Content-Type: application/json; charset=UTF-8",
|
|
"",
|
|
body,
|
|
]
|
|
)
|
|
content.append(f"--{BOUNDARY}--")
|
|
body = ("\r\n".join(content)).encode()
|
|
return (
|
|
Response(
|
|
{
|
|
"Content-Type": f"multipart/mixed; boundary={BOUNDARY}",
|
|
"Content-ID": "1",
|
|
}
|
|
),
|
|
body,
|
|
)
|
|
|
|
|
|
def create_batch_response_handler(
|
|
api_responses: list[dict | list | Response],
|
|
) -> Callable[[Any], tuple[Response, bytes]]:
|
|
"""Create a fake http2lib response handler that supports generating batch responses.
|
|
|
|
Multi-part response objects are dynamically generated since they
|
|
need to match the Content-ID of the incoming request.
|
|
"""
|
|
|
|
def _handler(url, method, **kwargs) -> tuple[Response, bytes]:
|
|
next_api_response = api_responses.pop(0)
|
|
if method == "POST" and (body := kwargs.get("body")):
|
|
content_ids = [
|
|
line[len(CONTENT_ID) + 2 :]
|
|
for line in body.splitlines()
|
|
if line.startswith(f"{CONTENT_ID}:")
|
|
]
|
|
if content_ids:
|
|
return create_batch_response_object(content_ids, next_api_response)
|
|
return create_response_object(next_api_response)
|
|
|
|
return _handler
|
|
|
|
|
|
@pytest.fixture(name="response_handler")
|
|
def mock_response_handler(api_responses: list[dict | list]) -> list:
|
|
"""Create a mock http2lib response handler."""
|
|
return [create_response_object(api_response) for api_response in api_responses]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_http_response(response_handler: list | Callable) -> Mock:
|
|
"""Fixture to fake out http2lib responses."""
|
|
|
|
with patch("httplib2.Http.request", side_effect=response_handler) as mock_response:
|
|
yield mock_response
|
|
|
|
|
|
@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(
|
|
"response_handler",
|
|
[
|
|
([(Response({"status": HTTPStatus.INTERNAL_SERVER_ERROR}), b"")]),
|
|
],
|
|
)
|
|
async def test_list_items_server_error(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
hass_ws_client: WebSocketGenerator,
|
|
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
|
|
) -> None:
|
|
"""Test an error returned by the server when setting up the platform."""
|
|
|
|
assert await integration_setup()
|
|
|
|
await hass_ws_client(hass)
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"api_responses",
|
|
[
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE,
|
|
]
|
|
],
|
|
)
|
|
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"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"api_responses",
|
|
[
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
ERROR_RESPONSE,
|
|
]
|
|
],
|
|
)
|
|
async def test_task_items_error_response(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
hass_ws_client: WebSocketGenerator,
|
|
ws_get_items: Callable[[], Awaitable[dict[str, str]]],
|
|
) -> None:
|
|
"""Test an error while getting todo list items."""
|
|
|
|
assert await integration_setup()
|
|
|
|
await hass_ws_client(hass)
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "unavailable"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"api_responses",
|
|
[
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE,
|
|
EMPTY_RESPONSE, # create
|
|
LIST_TASKS_RESPONSE, # refresh after delete
|
|
]
|
|
],
|
|
)
|
|
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,
|
|
"add_item",
|
|
{"item": "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_WATER,
|
|
ERROR_RESPONSE,
|
|
]
|
|
],
|
|
)
|
|
async def test_create_todo_list_item_error(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
mock_http_response: Mock,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test for an error response when creating a To-do Item."""
|
|
|
|
assert await integration_setup()
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "1"
|
|
|
|
with pytest.raises(HomeAssistantError, match="Invalid task ID"):
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"add_item",
|
|
{"item": "Soda"},
|
|
target={"entity_id": "todo.my_tasks"},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"api_responses",
|
|
[
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE_WATER,
|
|
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 == "1"
|
|
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"update_item",
|
|
{"item": "some-task-id", "rename": "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_WATER,
|
|
ERROR_RESPONSE, # update fails
|
|
]
|
|
],
|
|
)
|
|
async def test_update_todo_list_item_error(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
mock_http_response: Any,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test for an error response when updating a To-do Item."""
|
|
|
|
assert await integration_setup()
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "1"
|
|
|
|
with pytest.raises(HomeAssistantError, match="Invalid task ID"):
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"update_item",
|
|
{"item": "some-task-id", "rename": "Soda", "status": "completed"},
|
|
target={"entity_id": "todo.my_tasks"},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"api_responses",
|
|
[
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE_WATER,
|
|
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 == "1"
|
|
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"update_item",
|
|
{"item": "some-task-id", "rename": "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_WATER,
|
|
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 == "1"
|
|
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"update_item",
|
|
{"item": "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
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response_handler",
|
|
[
|
|
(
|
|
create_batch_response_handler(
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE_MULTIPLE,
|
|
[EMPTY_RESPONSE, EMPTY_RESPONSE, EMPTY_RESPONSE], # Delete batch
|
|
LIST_TASKS_RESPONSE, # refresh after create
|
|
]
|
|
)
|
|
)
|
|
],
|
|
)
|
|
async def test_delete_todo_list_item(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
mock_http_response: Any,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test for deleting multiple To-do Items."""
|
|
|
|
assert await integration_setup()
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "3"
|
|
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"remove_item",
|
|
{"item": ["some-task-id-1", "some-task-id-2", "some-task-id-3"]},
|
|
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
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response_handler",
|
|
[
|
|
(
|
|
create_batch_response_handler(
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE_MULTIPLE,
|
|
[
|
|
EMPTY_RESPONSE,
|
|
ERROR_RESPONSE, # one item is a failure
|
|
EMPTY_RESPONSE,
|
|
],
|
|
LIST_TASKS_RESPONSE, # refresh after create
|
|
]
|
|
)
|
|
)
|
|
],
|
|
)
|
|
async def test_delete_partial_failure(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
mock_http_response: Any,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test for partial failure when deleting multiple To-do Items."""
|
|
|
|
assert await integration_setup()
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "3"
|
|
|
|
with pytest.raises(HomeAssistantError, match="Invalid task ID"):
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"remove_item",
|
|
{"item": ["some-task-id-1", "some-task-id-2", "some-task-id-3"]},
|
|
target={"entity_id": "todo.my_tasks"},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response_handler",
|
|
[
|
|
(
|
|
create_batch_response_handler(
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE_MULTIPLE,
|
|
[
|
|
"1234-invalid-json",
|
|
],
|
|
]
|
|
)
|
|
)
|
|
],
|
|
)
|
|
async def test_delete_invalid_json_response(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
mock_http_response: Any,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test delete with an invalid json response."""
|
|
|
|
assert await integration_setup()
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "3"
|
|
|
|
with pytest.raises(HomeAssistantError, match="unexpected response"):
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"remove_item",
|
|
{"item": ["some-task-id-1"]},
|
|
target={"entity_id": "todo.my_tasks"},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response_handler",
|
|
[
|
|
(
|
|
create_batch_response_handler(
|
|
[
|
|
LIST_TASK_LIST_RESPONSE,
|
|
LIST_TASKS_RESPONSE_MULTIPLE,
|
|
[Response({"status": HTTPStatus.INTERNAL_SERVER_ERROR})],
|
|
]
|
|
)
|
|
)
|
|
],
|
|
)
|
|
async def test_delete_server_error(
|
|
hass: HomeAssistant,
|
|
setup_credentials: None,
|
|
integration_setup: Callable[[], Awaitable[bool]],
|
|
mock_http_response: Any,
|
|
snapshot: SnapshotAssertion,
|
|
) -> None:
|
|
"""Test delete with an invalid json response."""
|
|
|
|
assert await integration_setup()
|
|
|
|
state = hass.states.get("todo.my_tasks")
|
|
assert state
|
|
assert state.state == "3"
|
|
|
|
with pytest.raises(HomeAssistantError, match="responded with error"):
|
|
await hass.services.async_call(
|
|
TODO_DOMAIN,
|
|
"remove_item",
|
|
{"item": ["some-task-id-1"]},
|
|
target={"entity_id": "todo.my_tasks"},
|
|
blocking=True,
|
|
)
|