mirror of
https://github.com/home-assistant/core
synced 2024-10-04 20:02:40 +00:00
Joakim Sørensen
GitHub
parent
46d49336a1
commit
6b3b21bcfd
|
|
@ -94,7 +94,6 @@ components: &components
|
|||
- homeassistant/components/tag/*
|
||||
- homeassistant/components/template/*
|
||||
- homeassistant/components/timer/*
|
||||
- homeassistant/components/update/*
|
||||
- homeassistant/components/usb/*
|
||||
- homeassistant/components/webhook/*
|
||||
- homeassistant/components/websocket_api/*
|
||||
|
|
|
|||
|
|
@ -207,7 +207,6 @@ homeassistant.components.tts.*
|
|||
homeassistant.components.twentemilieu.*
|
||||
homeassistant.components.unifiprotect.*
|
||||
homeassistant.components.upcloud.*
|
||||
homeassistant.components.update.*
|
||||
homeassistant.components.uptime.*
|
||||
homeassistant.components.uptimerobot.*
|
||||
homeassistant.components.usb.*
|
||||
|
|
|
|||
|
|
@ -1057,8 +1057,6 @@ tests/components/upb/* @gwww
|
|||
homeassistant/components/upc_connect/* @pvizeli @fabaff
|
||||
homeassistant/components/upcloud/* @scop
|
||||
tests/components/upcloud/* @scop
|
||||
homeassistant/components/update/* @home-assistant/core
|
||||
tests/components/update/* @home-assistant/core
|
||||
homeassistant/components/updater/* @home-assistant/core
|
||||
tests/components/updater/* @home-assistant/core
|
||||
homeassistant/components/upnp/* @StevenLooman @ehendrix23
|
||||
|
|
|
|||
|
|
@ -31,11 +31,10 @@
|
|||
"tag",
|
||||
"timer",
|
||||
"usb",
|
||||
"update",
|
||||
"webhook",
|
||||
"zeroconf",
|
||||
"zone"
|
||||
],
|
||||
"codeowners": ["@home-assistant/core"],
|
||||
"quality_scale": "internal"
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,273 +0,0 @@
|
|||
"""Support for Update."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import dataclasses
|
||||
import logging
|
||||
from typing import Any, Protocol
|
||||
|
||||
import async_timeout
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import integration_platform, storage
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = "update"
|
||||
INFO_CALLBACK_TIMEOUT = 5
|
||||
STORAGE_VERSION = 1
|
||||
|
||||
|
||||
class IntegrationUpdateFailed(HomeAssistantError):
|
||||
"""Error to indicate an update has failed."""
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the Update integration."""
|
||||
hass.data[DOMAIN] = UpdateManager(hass=hass)
|
||||
websocket_api.async_register_command(hass, handle_info)
|
||||
websocket_api.async_register_command(hass, handle_update)
|
||||
websocket_api.async_register_command(hass, handle_skip)
|
||||
return True
|
||||
|
||||
|
||||
@websocket_api.websocket_command({vol.Required("type"): "update/info"})
|
||||
@websocket_api.async_response
|
||||
async def handle_info(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
"""Get pending updates from all platforms."""
|
||||
manager: UpdateManager = hass.data[DOMAIN]
|
||||
updates = await manager.gather_updates()
|
||||
connection.send_result(msg["id"], updates)
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required("type"): "update/skip",
|
||||
vol.Required("domain"): str,
|
||||
vol.Required("identifier"): str,
|
||||
vol.Required("version"): str,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
async def handle_skip(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
"""Skip an update."""
|
||||
manager: UpdateManager = hass.data[DOMAIN]
|
||||
|
||||
if not await manager.domain_is_valid(msg["domain"]):
|
||||
connection.send_error(
|
||||
msg["id"], websocket_api.ERR_NOT_FOUND, "Domain not supported"
|
||||
)
|
||||
return
|
||||
|
||||
manager.skip_update(msg["domain"], msg["identifier"], msg["version"])
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required("type"): "update/update",
|
||||
vol.Required("domain"): str,
|
||||
vol.Required("identifier"): str,
|
||||
vol.Required("version"): str,
|
||||
vol.Optional("backup"): bool,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
async def handle_update(
|
||||
hass: HomeAssistant,
|
||||
connection: websocket_api.ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
) -> None:
|
||||
"""Handle an update."""
|
||||
manager: UpdateManager = hass.data[DOMAIN]
|
||||
|
||||
if not await manager.domain_is_valid(msg["domain"]):
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
websocket_api.ERR_NOT_FOUND,
|
||||
f"{msg['domain']} is not a supported domain",
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
await manager.perform_update(
|
||||
domain=msg["domain"],
|
||||
identifier=msg["identifier"],
|
||||
version=msg["version"],
|
||||
backup=msg.get("backup"),
|
||||
)
|
||||
except IntegrationUpdateFailed as err:
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
"update_failed",
|
||||
str(err),
|
||||
)
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception(
|
||||
"Update of %s to version %s failed",
|
||||
msg["identifier"],
|
||||
msg["version"],
|
||||
)
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
"update_failed",
|
||||
"Unknown Error",
|
||||
)
|
||||
else:
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
|
||||
class UpdatePlatformProtocol(Protocol):
|
||||
"""Define the format that update platforms can have."""
|
||||
|
||||
async def async_list_updates(self, hass: HomeAssistant) -> list[UpdateDescription]:
|
||||
"""List all updates available in the integration."""
|
||||
|
||||
async def async_perform_update(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
identifier: str,
|
||||
version: str,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""Perform an update."""
|
||||
|
||||
|
||||
@dataclasses.dataclass()
|
||||
class UpdateDescription:
|
||||
"""Describe an update update."""
|
||||
|
||||
identifier: str
|
||||
name: str
|
||||
current_version: str
|
||||
available_version: str
|
||||
changelog_content: str | None = None
|
||||
changelog_url: str | None = None
|
||||
icon_url: str | None = None
|
||||
supports_backup: bool = False
|
||||
|
||||
|
||||
class UpdateManager:
|
||||
"""Update manager for the update integration."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant) -> None:
|
||||
"""Initialize the update manager."""
|
||||
self._hass = hass
|
||||
self._store = storage.Store(
|
||||
hass=hass,
|
||||
version=STORAGE_VERSION,
|
||||
key=DOMAIN,
|
||||
)
|
||||
self._skip: set[str] = set()
|
||||
self._platforms: dict[str, UpdatePlatformProtocol] = {}
|
||||
self._loaded = False
|
||||
|
||||
async def add_platform(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
integration_domain: str,
|
||||
platform: UpdatePlatformProtocol,
|
||||
) -> None:
|
||||
"""Add a platform to the update manager."""
|
||||
self._platforms[integration_domain] = platform
|
||||
|
||||
async def _load(self) -> None:
|
||||
"""Load platforms and data from storage."""
|
||||
await integration_platform.async_process_integration_platforms(
|
||||
self._hass, DOMAIN, self.add_platform
|
||||
)
|
||||
from_storage = await self._store.async_load()
|
||||
if isinstance(from_storage, dict):
|
||||
self._skip = set(from_storage["skipped"])
|
||||
|
||||
self._loaded = True
|
||||
|
||||
async def gather_updates(self) -> list[dict[str, Any]]:
|
||||
"""Gather updates."""
|
||||
if not self._loaded:
|
||||
await self._load()
|
||||
|
||||
updates: dict[str, list[UpdateDescription] | None] = {}
|
||||
|
||||
for domain, update_descriptions in zip(
|
||||
self._platforms,
|
||||
await asyncio.gather(
|
||||
*(
|
||||
self._get_integration_info(integration_domain, registration)
|
||||
for integration_domain, registration in self._platforms.items()
|
||||
)
|
||||
),
|
||||
):
|
||||
updates[domain] = update_descriptions
|
||||
|
||||
return [
|
||||
{
|
||||
"domain": integration_domain,
|
||||
**dataclasses.asdict(description),
|
||||
}
|
||||
for integration_domain, update_descriptions in updates.items()
|
||||
if update_descriptions is not None
|
||||
for description in update_descriptions
|
||||
if f"{integration_domain}_{description.identifier}_{description.available_version}"
|
||||
not in self._skip
|
||||
]
|
||||
|
||||
async def domain_is_valid(self, domain: str) -> bool:
|
||||
"""Return if the domain is valid."""
|
||||
if not self._loaded:
|
||||
await self._load()
|
||||
return domain in self._platforms
|
||||
|
||||
@callback
|
||||
def _data_to_save(self) -> dict[str, Any]:
|
||||
"""Schedule storing the data."""
|
||||
return {"skipped": list(self._skip)}
|
||||
|
||||
async def perform_update(
|
||||
self,
|
||||
domain: str,
|
||||
identifier: str,
|
||||
version: str,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""Perform an update."""
|
||||
await self._platforms[domain].async_perform_update(
|
||||
hass=self._hass,
|
||||
identifier=identifier,
|
||||
version=version,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
@callback
|
||||
def skip_update(self, domain: str, identifier: str, version: str) -> None:
|
||||
"""Skip an update."""
|
||||
self._skip.add(f"{domain}_{identifier}_{version}")
|
||||
self._store.async_delay_save(self._data_to_save, 60)
|
||||
|
||||
async def _get_integration_info(
|
||||
self,
|
||||
integration_domain: str,
|
||||
platform: UpdatePlatformProtocol,
|
||||
) -> list[UpdateDescription] | None:
|
||||
"""Get integration update details."""
|
||||
|
||||
try:
|
||||
async with async_timeout.timeout(INFO_CALLBACK_TIMEOUT):
|
||||
return await platform.async_list_updates(hass=self._hass)
|
||||
except asyncio.TimeoutError:
|
||||
_LOGGER.warning("Timeout while getting updates from %s", integration_domain)
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Error fetching info from %s", integration_domain)
|
||||
return None
|
||||
|
|
@ -1,10 +0,0 @@
|
|||
{
|
||||
"domain": "update",
|
||||
"name": "Update",
|
||||
"documentation": "https://www.home-assistant.io/integrations/update",
|
||||
"codeowners": [
|
||||
"@home-assistant/core"
|
||||
],
|
||||
"quality_scale": "internal",
|
||||
"iot_class": "calculated"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "Update"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "Actualitza"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "Aktualisieren"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "\u0395\u03bd\u03b7\u03bc\u03ad\u03c1\u03c9\u03c3\u03b7"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "Update"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "Aggiornamento"
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"title": "Atualiza\u00e7\u00e3o"
|
||||
}
|
||||
11
mypy.ini
11
mypy.ini
|
|
@ -2078,17 +2078,6 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.update.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.uptime.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
|
|||
|
|
@ -1 +0,0 @@
|
|||
"""Tests for the Update integration."""
|
||||
|
|
@ -1,347 +0,0 @@
|
|||
"""Tests for the Update integration init."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from aiohttp import ClientWebSocketResponse
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.update import (
|
||||
DOMAIN,
|
||||
IntegrationUpdateFailed,
|
||||
UpdateDescription,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import mock_platform
|
||||
|
||||
|
||||
async def setup_mock_domain(
|
||||
hass: HomeAssistant,
|
||||
async_list_updates: Callable[[HomeAssistant], Awaitable[list[UpdateDescription]]]
|
||||
| None = None,
|
||||
async_perform_update: Callable[[HomeAssistant, str, str], Awaitable[bool]]
|
||||
| None = None,
|
||||
) -> None:
|
||||
"""Set up a mock domain."""
|
||||
|
||||
async def _mock_async_list_updates(hass: HomeAssistant) -> list[UpdateDescription]:
|
||||
return [
|
||||
UpdateDescription(
|
||||
identifier="lorem_ipsum",
|
||||
name="Lorem Ipsum",
|
||||
current_version="1.0.0",
|
||||
available_version="1.0.1",
|
||||
)
|
||||
]
|
||||
|
||||
async def _mock_async_perform_update(
|
||||
hass: HomeAssistant,
|
||||
identifier: str,
|
||||
version: str,
|
||||
**kwargs: Any,
|
||||
) -> bool:
|
||||
return True
|
||||
|
||||
mock_platform(
|
||||
hass,
|
||||
"some_domain.update",
|
||||
Mock(
|
||||
async_list_updates=async_list_updates or _mock_async_list_updates,
|
||||
async_perform_update=async_perform_update or _mock_async_perform_update,
|
||||
),
|
||||
)
|
||||
|
||||
assert await async_setup_component(hass, "some_domain", {})
|
||||
|
||||
|
||||
async def gather_update_info(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> list[dict]:
|
||||
"""Gather all info."""
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json({"id": 1, "type": "update/info"})
|
||||
resp = await client.receive_json()
|
||||
return resp["result"]
|
||||
|
||||
|
||||
async def test_update_updates(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test getting updates."""
|
||||
await setup_mock_domain(hass)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.update.storage.Store.async_load",
|
||||
return_value={"skipped": []},
|
||||
):
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
data = data[0] == {
|
||||
"domain": "some_domain",
|
||||
"identifier": "lorem_ipsum",
|
||||
"name": "Lorem Ipsum",
|
||||
"current_version": "1.0.0",
|
||||
"available_version": "1.0.1",
|
||||
"changelog_url": None,
|
||||
"icon_url": None,
|
||||
}
|
||||
|
||||
|
||||
async def test_update_updates_with_timeout_error(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test timeout while getting updates."""
|
||||
|
||||
async def mock_async_list_updates(hass: HomeAssistant) -> list[UpdateDescription]:
|
||||
raise asyncio.TimeoutError()
|
||||
|
||||
await setup_mock_domain(hass, async_list_updates=mock_async_list_updates)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 0
|
||||
|
||||
|
||||
async def test_update_updates_with_exception(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test exception while getting updates."""
|
||||
|
||||
async def mock_async_list_updates(hass: HomeAssistant) -> list[UpdateDescription]:
|
||||
raise Exception()
|
||||
|
||||
await setup_mock_domain(hass, async_list_updates=mock_async_list_updates)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 0
|
||||
|
||||
|
||||
async def test_update_update(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test performing an update."""
|
||||
await setup_mock_domain(hass)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
update = data[0]
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/update",
|
||||
"domain": update["domain"],
|
||||
"identifier": update["identifier"],
|
||||
"version": update["available_version"],
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert resp["success"]
|
||||
|
||||
|
||||
async def test_skip_update(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test skipping updates."""
|
||||
await setup_mock_domain(hass)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
update = data[0]
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/skip",
|
||||
"domain": update["domain"],
|
||||
"identifier": update["identifier"],
|
||||
"version": update["available_version"],
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert resp["success"]
|
||||
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
assert len(data) == 0
|
||||
|
||||
|
||||
async def test_skip_non_existing_update(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test skipping non-existing updates."""
|
||||
await setup_mock_domain(hass)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/skip",
|
||||
"domain": "non_existing",
|
||||
"identifier": "non_existing",
|
||||
"version": "non_existing",
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert not resp["success"]
|
||||
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
assert len(data) == 1
|
||||
|
||||
|
||||
async def test_update_update_non_existing(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test that we fail when trying to update something that does not exist."""
|
||||
await setup_mock_domain(hass)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/update",
|
||||
"domain": "does_not_exist",
|
||||
"identifier": "does_not_exist",
|
||||
"version": "non_existing",
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert not resp["success"]
|
||||
assert resp["error"]["code"] == "not_found"
|
||||
|
||||
|
||||
async def test_update_update_failed(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test that we correctly handle failed updates."""
|
||||
|
||||
async def mock_async_perform_update(
|
||||
hass: HomeAssistant,
|
||||
identifier: str,
|
||||
version: str,
|
||||
**kwargs,
|
||||
) -> bool:
|
||||
raise IntegrationUpdateFailed("Test update failed")
|
||||
|
||||
await setup_mock_domain(hass, async_perform_update=mock_async_perform_update)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
update = data[0]
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/update",
|
||||
"domain": update["domain"],
|
||||
"identifier": update["identifier"],
|
||||
"version": update["available_version"],
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert not resp["success"]
|
||||
assert resp["error"]["code"] == "update_failed"
|
||||
assert resp["error"]["message"] == "Test update failed"
|
||||
|
||||
|
||||
async def test_update_update_failed_generic(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test that we correctly handle failed updates."""
|
||||
|
||||
async def mock_async_perform_update(
|
||||
hass: HomeAssistant,
|
||||
identifier: str,
|
||||
version: str,
|
||||
**kwargs,
|
||||
) -> bool:
|
||||
raise TypeError("Test update failed")
|
||||
|
||||
await setup_mock_domain(hass, async_perform_update=mock_async_perform_update)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
data = await gather_update_info(hass, hass_ws_client)
|
||||
|
||||
assert len(data) == 1
|
||||
update = data[0]
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/update",
|
||||
"domain": update["domain"],
|
||||
"identifier": update["identifier"],
|
||||
"version": update["available_version"],
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert not resp["success"]
|
||||
assert resp["error"]["code"] == "update_failed"
|
||||
assert resp["error"]["message"] == "Unknown Error"
|
||||
assert "Test update failed" in caplog.text
|
||||
|
||||
|
||||
async def test_update_before_info(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
|
||||
) -> None:
|
||||
"""Test that we fail when trying to update something that does not exist."""
|
||||
await setup_mock_domain(hass)
|
||||
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 1,
|
||||
"type": "update/update",
|
||||
"domain": "does_not_exist",
|
||||
"identifier": "does_not_exist",
|
||||
"version": "non_existing",
|
||||
}
|
||||
)
|
||||
resp = await client.receive_json()
|
||||
assert not resp["success"]
|
||||
assert resp["error"]["code"] == "not_found"
|
||||
Loading…
Reference in a new issue