evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Move zwave_js addon manager to hassio integration (#81354)

Browse source
This commit is contained in:
Martin Hjelmare 2022-11-10 10:09:52 +01:00 • committed by GitHub
parent 0bd04068de
commit 9ded232522
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 1747 additions and 598 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

218
homeassistant/components/hassio/__init__.py
View file

@ -47,6 +47,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from homeassistant.loader import bind_hass
from homeassistant.util.dt import utcnow
from .addon_manager import AddonError, AddonInfo, AddonManager, AddonState # noqa: F401
from .addon_panel import async_setup_addon_panel
from .auth import async_setup_auth_view
from .const import (
@ -55,7 +56,6 @@ from .const import (
ATTR_AUTO_UPDATE,
ATTR_CHANGELOG,
ATTR_COMPRESSED,
ATTR_DISCOVERY,
ATTR_FOLDERS,
ATTR_HOMEASSISTANT,
ATTR_INPUT,
@ -74,7 +74,25 @@ from .const import (
SupervisorEntityModel,
)
from .discovery import HassioServiceInfo, async_setup_discovery_view # noqa: F401
from .handler import HassIO, HassioAPIError, api_data
from .handler import ( # noqa: F401
HassIO,
HassioAPIError,
async_create_backup,
async_get_addon_discovery_info,
async_get_addon_info,
async_get_addon_store_info,
async_install_addon,
async_restart_addon,
async_set_addon_options,
async_start_addon,
async_stop_addon,
async_uninstall_addon,
async_update_addon,
async_update_core,
async_update_diagnostics,
async_update_os,
async_update_supervisor,
)
from .http import HassIOView
from .ingress import async_setup_ingress_view
from .repairs import SupervisorRepairs
@ -221,202 +239,6 @@ HARDWARE_INTEGRATIONS = {
}
@bind_hass
async def async_get_addon_info(hass: HomeAssistant, slug: str) -> dict:
"""Return add-on info.
The add-on must be installed.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
return await hassio.get_addon_info(slug)
@api_data
async def async_get_addon_store_info(hass: HomeAssistant, slug: str) -> dict:
"""Return add-on store info.
The caller of the function should handle HassioAPIError.
"""
hassio: HassIO = hass.data[DOMAIN]
command = f"/store/addons/{slug}"
return await hassio.send_command(command, method="get")
@bind_hass
async def async_update_diagnostics(hass: HomeAssistant, diagnostics: bool) -> dict:
"""Update Supervisor diagnostics toggle.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
return await hassio.update_diagnostics(diagnostics)
@bind_hass
@api_data
async def async_install_addon(hass: HomeAssistant, slug: str) -> dict:
"""Install add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/install"
return await hassio.send_command(command, timeout=None)
@bind_hass
@api_data
async def async_uninstall_addon(hass: HomeAssistant, slug: str) -> dict:
"""Uninstall add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/uninstall"
return await hassio.send_command(command, timeout=60)
@bind_hass
@api_data
async def async_update_addon(
hass: HomeAssistant,
slug: str,
backup: bool = False,
) -> dict:
"""Update add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/update"
return await hassio.send_command(
command,
payload={"backup": backup},
timeout=None,
)
@bind_hass
@api_data
async def async_start_addon(hass: HomeAssistant, slug: str) -> dict:
"""Start add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/start"
return await hassio.send_command(command, timeout=60)
@bind_hass
@api_data
async def async_restart_addon(hass: HomeAssistant, slug: str) -> dict:
"""Restart add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/restart"
return await hassio.send_command(command, timeout=None)
@bind_hass
@api_data
async def async_stop_addon(hass: HomeAssistant, slug: str) -> dict:
"""Stop add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/stop"
return await hassio.send_command(command, timeout=60)
@bind_hass
@api_data
async def async_set_addon_options(
hass: HomeAssistant, slug: str, options: dict
) -> dict:
"""Set add-on options.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/options"
return await hassio.send_command(command, payload=options)
@bind_hass
async def async_get_addon_discovery_info(hass: HomeAssistant, slug: str) -> dict | None:
"""Return discovery data for an add-on."""
hassio = hass.data[DOMAIN]
data = await hassio.retrieve_discovery_messages()
discovered_addons = data[ATTR_DISCOVERY]
return next((addon for addon in discovered_addons if addon["addon"] == slug), None)
@bind_hass
@api_data
async def async_create_backup(
hass: HomeAssistant, payload: dict, partial: bool = False
) -> dict:
"""Create a full or partial backup.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
backup_type = "partial" if partial else "full"
command = f"/backups/new/{backup_type}"
return await hassio.send_command(command, payload=payload, timeout=None)
@bind_hass
@api_data
async def async_update_os(hass: HomeAssistant, version: str | None = None) -> dict:
"""Update Home Assistant Operating System.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = "/os/update"
return await hassio.send_command(
command,
payload={"version": version},
timeout=None,
)
@bind_hass
@api_data
async def async_update_supervisor(hass: HomeAssistant) -> dict:
"""Update Home Assistant Supervisor.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = "/supervisor/update"
return await hassio.send_command(command, timeout=None)
@bind_hass
@api_data
async def async_update_core(
hass: HomeAssistant, version: str | None = None, backup: bool = False
) -> dict:
"""Update Home Assistant Core.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = "/core/update"
return await hassio.send_command(
command,
payload={"version": version, "backup": backup},
timeout=None,
)
@callback
@bind_hass
def get_info(hass):

373
homeassistant/components/hassio/addon_manager.py Normal file
View file

@ -0,0 +1,373 @@
"""Provide add-on management."""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from enum import Enum
from functools import partial, wraps
import logging
from typing import Any, TypeVar
from typing_extensions import Concatenate, ParamSpec
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from .handler import (
HassioAPIError,
async_create_backup,
async_get_addon_discovery_info,
async_get_addon_info,
async_get_addon_store_info,
async_install_addon,
async_restart_addon,
async_set_addon_options,
async_start_addon,
async_stop_addon,
async_uninstall_addon,
async_update_addon,
)
_AddonManagerT = TypeVar("_AddonManagerT", bound="AddonManager")
_R = TypeVar("_R")
_P = ParamSpec("_P")
def api_error(
error_message: str,
) -> Callable[
[Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]],
Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]],
]:
"""Handle HassioAPIError and raise a specific AddonError."""
def handle_hassio_api_error(
func: Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]:
"""Handle a HassioAPIError."""
@wraps(func)
async def wrapper(
self: _AddonManagerT, *args: _P.args, **kwargs: _P.kwargs
) -> _R:
"""Wrap an add-on manager method."""
try:
return_value = await func(self, *args, **kwargs)
except HassioAPIError as err:
raise AddonError(
f"{error_message.format(addon_name=self.addon_name)}: {err}"
) from err
return return_value
return wrapper
return handle_hassio_api_error
@dataclass
class AddonInfo:
"""Represent the current add-on info state."""
options: dict[str, Any]
state: AddonState
update_available: bool
version: str | None
class AddonState(Enum):
"""Represent the current state of the add-on."""
NOT_INSTALLED = "not_installed"
INSTALLING = "installing"
UPDATING = "updating"
NOT_RUNNING = "not_running"
RUNNING = "running"
class AddonManager:
"""Manage the add-on.
Methods may raise AddonError.
Only one instance of this class may exist per add-on
to keep track of running add-on tasks.
"""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
addon_name: str,
addon_slug: str,
) -> None:
"""Set up the add-on manager."""
self.addon_name = addon_name
self.addon_slug = addon_slug
self._hass = hass
self._logger = logger
self._install_task: asyncio.Task | None = None
self._restart_task: asyncio.Task | None = None
self._start_task: asyncio.Task | None = None
self._update_task: asyncio.Task | None = None
def task_in_progress(self) -> bool:
"""Return True if any of the add-on tasks are in progress."""
return any(
task and not task.done()
for task in (
self._restart_task,
self._install_task,
self._start_task,
self._update_task,
)
)
@api_error("Failed to get the {addon_name} add-on discovery info")
async def async_get_addon_discovery_info(self) -> dict:
"""Return add-on discovery info."""
discovery_info = await async_get_addon_discovery_info(
self._hass, self.addon_slug
)
if not discovery_info:
raise AddonError(f"Failed to get {self.addon_name} add-on discovery info")
discovery_info_config: dict = discovery_info["config"]
return discovery_info_config
@api_error("Failed to get the {addon_name} add-on info")
async def async_get_addon_info(self) -> AddonInfo:
"""Return and cache manager add-on info."""
addon_store_info = await async_get_addon_store_info(self._hass, self.addon_slug)
self._logger.debug("Add-on store info: %s", addon_store_info)
if not addon_store_info["installed"]:
return AddonInfo(
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
addon_info = await async_get_addon_info(self._hass, self.addon_slug)
addon_state = self.async_get_addon_state(addon_info)
return AddonInfo(
options=addon_info["options"],
state=addon_state,
update_available=addon_info["update_available"],
version=addon_info["version"],
)
@callback
def async_get_addon_state(self, addon_info: dict[str, Any]) -> AddonState:
"""Return the current state of the managed add-on."""
addon_state = AddonState.NOT_RUNNING
if addon_info["state"] == "started":
addon_state = AddonState.RUNNING
if self._install_task and not self._install_task.done():
addon_state = AddonState.INSTALLING
if self._update_task and not self._update_task.done():
addon_state = AddonState.UPDATING
return addon_state
@api_error("Failed to set the {addon_name} add-on options")
async def async_set_addon_options(self, config: dict) -> None:
"""Set manager add-on options."""
options = {"options": config}
await async_set_addon_options(self._hass, self.addon_slug, options)
@api_error("Failed to install the {addon_name} add-on")
async def async_install_addon(self) -> None:
"""Install the managed add-on."""
await async_install_addon(self._hass, self.addon_slug)
@api_error("Failed to uninstall the {addon_name} add-on")
async def async_uninstall_addon(self) -> None:
"""Uninstall the managed add-on."""
await async_uninstall_addon(self._hass, self.addon_slug)
@api_error("Failed to update the {addon_name} add-on")
async def async_update_addon(self) -> None:
"""Update the managed add-on if needed."""
addon_info = await self.async_get_addon_info()
if addon_info.state is AddonState.NOT_INSTALLED:
raise AddonError(f"{self.addon_name} add-on is not installed")
if not addon_info.update_available:
return
await self.async_create_backup()
await async_update_addon(self._hass, self.addon_slug)
@api_error("Failed to start the {addon_name} add-on")
async def async_start_addon(self) -> None:
"""Start the managed add-on."""
await async_start_addon(self._hass, self.addon_slug)
@api_error("Failed to restart the {addon_name} add-on")
async def async_restart_addon(self) -> None:
"""Restart the managed add-on."""
await async_restart_addon(self._hass, self.addon_slug)
@api_error("Failed to stop the {addon_name} add-on")
async def async_stop_addon(self) -> None:
"""Stop the managed add-on."""
await async_stop_addon(self._hass, self.addon_slug)
@api_error("Failed to create a backup of the {addon_name} add-on")
async def async_create_backup(self) -> None:
"""Create a partial backup of the managed add-on."""
addon_info = await self.async_get_addon_info()
name = f"addon_{self.addon_slug}_{addon_info.version}"
self._logger.debug("Creating backup: %s", name)
await async_create_backup(
self._hass,
{"name": name, "addons": [self.addon_slug]},
partial=True,
)
async def async_configure_addon(
self,
addon_config: dict[str, Any],
) -> None:
"""Configure the manager add-on, if needed."""
addon_info = await self.async_get_addon_info()
if addon_info.state is AddonState.NOT_INSTALLED:
raise AddonError(f"{self.addon_name} add-on is not installed")
if addon_config != addon_info.options:
await self.async_set_addon_options(addon_config)
@callback
def async_schedule_install_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that installs the managed add-on.
Only schedule a new install task if the there's no running task.
"""
if not self._install_task or self._install_task.done():
self._logger.info(
"%s add-on is not installed. Installing add-on", self.addon_name
)
self._install_task = self._async_schedule_addon_operation(
self.async_install_addon, catch_error=catch_error
)
return self._install_task
@callback
def async_schedule_install_setup_addon(
self,
addon_config: dict[str, Any],
catch_error: bool = False,
) -> asyncio.Task:
"""Schedule a task that installs and sets up the managed add-on.
Only schedule a new install task if the there's no running task.
"""
if not self._install_task or self._install_task.done():
self._logger.info(
"%s add-on is not installed. Installing add-on", self.addon_name
)
self._install_task = self._async_schedule_addon_operation(
self.async_install_addon,
partial(
self.async_configure_addon,
addon_config,
),
self.async_start_addon,
catch_error=catch_error,
)
return self._install_task
@callback
def async_schedule_update_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that updates and sets up the managed add-on.
Only schedule a new update task if the there's no running task.
"""
if not self._update_task or self._update_task.done():
self._logger.info("Trying to update the %s add-on", self.addon_name)
self._update_task = self._async_schedule_addon_operation(
self.async_update_addon,
catch_error=catch_error,
)
return self._update_task
@callback
def async_schedule_start_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that starts the managed add-on.
Only schedule a new start task if the there's no running task.
"""
if not self._start_task or self._start_task.done():
self._logger.info(
"%s add-on is not running. Starting add-on", self.addon_name
)
self._start_task = self._async_schedule_addon_operation(
self.async_start_addon, catch_error=catch_error
)
return self._start_task
@callback
def async_schedule_restart_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that restarts the managed add-on.
Only schedule a new restart task if the there's no running task.
"""
if not self._restart_task or self._restart_task.done():
self._logger.info("Restarting %s add-on", self.addon_name)
self._restart_task = self._async_schedule_addon_operation(
self.async_restart_addon, catch_error=catch_error
)
return self._restart_task
@callback
def async_schedule_setup_addon(
self,
addon_config: dict[str, Any],
catch_error: bool = False,
) -> asyncio.Task:
"""Schedule a task that configures and starts the managed add-on.
Only schedule a new setup task if there's no running task.
"""
if not self._start_task or self._start_task.done():
self._logger.info(
"%s add-on is not running. Starting add-on", self.addon_name
)
self._start_task = self._async_schedule_addon_operation(
partial(
self.async_configure_addon,
addon_config,
),
self.async_start_addon,
catch_error=catch_error,
)
return self._start_task
@callback
def _async_schedule_addon_operation(
self, *funcs: Callable, catch_error: bool = False
) -> asyncio.Task:
"""Schedule an add-on task."""
async def addon_operation() -> None:
"""Do the add-on operation and catch AddonError."""
for func in funcs:
try:
await func()
except AddonError as err:
if not catch_error:
raise
self._logger.error(err)
break
return self._hass.async_create_task(addon_operation())
class AddonError(HomeAssistantError):
"""Represent an error with the managed add-on."""

202
homeassistant/components/hassio/handler.py
View file

@ -1,4 +1,6 @@
"""Handler for Hass.io."""
from __future__ import annotations
import asyncio
from http import HTTPStatus
import logging
@ -12,6 +14,10 @@ from homeassistant.components.http import (
CONF_SSL_CERTIFICATE,
)
from homeassistant.const import SERVER_PORT
from homeassistant.core import HomeAssistant
from homeassistant.loader import bind_hass
from .const import ATTR_DISCOVERY, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -47,6 +53,202 @@ def api_data(funct):
return _wrapper
@bind_hass
async def async_get_addon_info(hass: HomeAssistant, slug: str) -> dict:
"""Return add-on info.
The add-on must be installed.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
return await hassio.get_addon_info(slug)
@api_data
async def async_get_addon_store_info(hass: HomeAssistant, slug: str) -> dict:
"""Return add-on store info.
The caller of the function should handle HassioAPIError.
"""
hassio: HassIO = hass.data[DOMAIN]
command = f"/store/addons/{slug}"
return await hassio.send_command(command, method="get")
@bind_hass
async def async_update_diagnostics(hass: HomeAssistant, diagnostics: bool) -> dict:
"""Update Supervisor diagnostics toggle.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
return await hassio.update_diagnostics(diagnostics)
@bind_hass
@api_data
async def async_install_addon(hass: HomeAssistant, slug: str) -> dict:
"""Install add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/install"
return await hassio.send_command(command, timeout=None)
@bind_hass
@api_data
async def async_uninstall_addon(hass: HomeAssistant, slug: str) -> dict:
"""Uninstall add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/uninstall"
return await hassio.send_command(command, timeout=60)
@bind_hass
@api_data
async def async_update_addon(
hass: HomeAssistant,
slug: str,
backup: bool = False,
) -> dict:
"""Update add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/update"
return await hassio.send_command(
command,
payload={"backup": backup},
timeout=None,
)
@bind_hass
@api_data
async def async_start_addon(hass: HomeAssistant, slug: str) -> dict:
"""Start add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/start"
return await hassio.send_command(command, timeout=60)
@bind_hass
@api_data
async def async_restart_addon(hass: HomeAssistant, slug: str) -> dict:
"""Restart add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/restart"
return await hassio.send_command(command, timeout=None)
@bind_hass
@api_data
async def async_stop_addon(hass: HomeAssistant, slug: str) -> dict:
"""Stop add-on.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/stop"
return await hassio.send_command(command, timeout=60)
@bind_hass
@api_data
async def async_set_addon_options(
hass: HomeAssistant, slug: str, options: dict
) -> dict:
"""Set add-on options.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = f"/addons/{slug}/options"
return await hassio.send_command(command, payload=options)
@bind_hass
async def async_get_addon_discovery_info(hass: HomeAssistant, slug: str) -> dict | None:
"""Return discovery data for an add-on."""
hassio = hass.data[DOMAIN]
data = await hassio.retrieve_discovery_messages()
discovered_addons = data[ATTR_DISCOVERY]
return next((addon for addon in discovered_addons if addon["addon"] == slug), None)
@bind_hass
@api_data
async def async_create_backup(
hass: HomeAssistant, payload: dict, partial: bool = False
) -> dict:
"""Create a full or partial backup.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
backup_type = "partial" if partial else "full"
command = f"/backups/new/{backup_type}"
return await hassio.send_command(command, payload=payload, timeout=None)
@bind_hass
@api_data
async def async_update_os(hass: HomeAssistant, version: str | None = None) -> dict:
"""Update Home Assistant Operating System.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = "/os/update"
return await hassio.send_command(
command,
payload={"version": version},
timeout=None,
)
@bind_hass
@api_data
async def async_update_supervisor(hass: HomeAssistant) -> dict:
"""Update Home Assistant Supervisor.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = "/supervisor/update"
return await hassio.send_command(command, timeout=None)
@bind_hass
@api_data
async def async_update_core(
hass: HomeAssistant, version: str | None = None, backup: bool = False
) -> dict:
"""Update Home Assistant Core.
The caller of the function should handle HassioAPIError.
"""
hassio = hass.data[DOMAIN]
command = "/core/update"
return await hassio.send_command(
command,
payload={"version": version, "backup": backup},
timeout=None,
)
class HassIO:
"""Small API wrapper for Hass.io."""

3
homeassistant/components/zwave_js/__init__.py
View file

@ -20,6 +20,7 @@ from zwave_js_server.model.notification import (
)
from zwave_js_server.model.value import Value, ValueNotification
from homeassistant.components.hassio import AddonError, AddonManager, AddonState
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_DEVICE_ID,
@ -41,7 +42,7 @@ from homeassistant.helpers.issue_registry import (
)
from homeassistant.helpers.typing import UNDEFINED, ConfigType
from .addon import AddonError, AddonManager, AddonState, get_addon_manager
from .addon import get_addon_manager
from .api import async_register_api
from .const import (
ATTR_ACKNOWLEDGED_FRAMES,

358
homeassistant/components/zwave_js/addon.py
View file

@ -1,39 +1,12 @@
"""Provide add-on management."""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from enum import Enum
from functools import partial, wraps
from typing import Any, TypeVar
from typing_extensions import Concatenate, ParamSpec
from homeassistant.components.hassio import (
async_create_backup,
async_get_addon_discovery_info,
async_get_addon_info,
async_get_addon_store_info,
async_install_addon,
async_restart_addon,
async_set_addon_options,
async_start_addon,
async_stop_addon,
async_uninstall_addon,
async_update_addon,
)
from homeassistant.components.hassio.handler import HassioAPIError
from homeassistant.components.hassio import AddonManager
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.singleton import singleton
from .const import ADDON_SLUG, DOMAIN, LOGGER
_AddonManagerT = TypeVar("_AddonManagerT", bound="AddonManager")
_R = TypeVar("_R")
_P = ParamSpec("_P")
DATA_ADDON_MANAGER = f"{DOMAIN}_addon_manager"
@ -41,331 +14,4 @@ DATA_ADDON_MANAGER = f"{DOMAIN}_addon_manager"
@callback
def get_addon_manager(hass: HomeAssistant) -> AddonManager:
"""Get the add-on manager."""
return AddonManager(hass, "Z-Wave JS", ADDON_SLUG)
def api_error(
error_message: str,
) -> Callable[
[Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]],
Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]],
]:
"""Handle HassioAPIError and raise a specific AddonError."""
def handle_hassio_api_error(
func: Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]:
"""Handle a HassioAPIError."""
@wraps(func)
async def wrapper(
self: _AddonManagerT, *args: _P.args, **kwargs: _P.kwargs
) -> _R:
"""Wrap an add-on manager method."""
try:
return_value = await func(self, *args, **kwargs)
except HassioAPIError as err:
raise AddonError(
f"{error_message.format(addon_name=self.addon_name)}: {err}"
) from err
return return_value
return wrapper
return handle_hassio_api_error
@dataclass
class AddonInfo:
"""Represent the current add-on info state."""
options: dict[str, Any]
state: AddonState
update_available: bool
version: str | None
class AddonState(Enum):
"""Represent the current state of the add-on."""
NOT_INSTALLED = "not_installed"
INSTALLING = "installing"
UPDATING = "updating"
NOT_RUNNING = "not_running"
RUNNING = "running"
class AddonManager:
"""Manage the add-on.
Methods may raise AddonError.
Only one instance of this class may exist per add-on
to keep track of running add-on tasks.
"""
def __init__(self, hass: HomeAssistant, addon_name: str, addon_slug: str) -> None:
"""Set up the add-on manager."""
self.addon_name = addon_name
self.addon_slug = addon_slug
self._hass = hass
self._install_task: asyncio.Task | None = None
self._restart_task: asyncio.Task | None = None
self._start_task: asyncio.Task | None = None
self._update_task: asyncio.Task | None = None
def task_in_progress(self) -> bool:
"""Return True if any of the add-on tasks are in progress."""
return any(
task and not task.done()
for task in (
self._install_task,
self._start_task,
self._update_task,
)
)
@api_error("Failed to get {addon_name} add-on discovery info")
async def async_get_addon_discovery_info(self) -> dict:
"""Return add-on discovery info."""
discovery_info = await async_get_addon_discovery_info(
self._hass, self.addon_slug
)
if not discovery_info:
raise AddonError(f"Failed to get {self.addon_name} add-on discovery info")
discovery_info_config: dict = discovery_info["config"]
return discovery_info_config
@api_error("Failed to get the {addon_name} add-on info")
async def async_get_addon_info(self) -> AddonInfo:
"""Return and cache manager add-on info."""
addon_store_info = await async_get_addon_store_info(self._hass, self.addon_slug)
LOGGER.debug("Add-on store info: %s", addon_store_info)
if not addon_store_info["installed"]:
return AddonInfo(
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
addon_info = await async_get_addon_info(self._hass, self.addon_slug)
addon_state = self.async_get_addon_state(addon_info)
return AddonInfo(
options=addon_info["options"],
state=addon_state,
update_available=addon_info["update_available"],
version=addon_info["version"],
)
@callback
def async_get_addon_state(self, addon_info: dict[str, Any]) -> AddonState:
"""Return the current state of the managed add-on."""
addon_state = AddonState.NOT_RUNNING
if addon_info["state"] == "started":
addon_state = AddonState.RUNNING
if self._install_task and not self._install_task.done():
addon_state = AddonState.INSTALLING
if self._update_task and not self._update_task.done():
addon_state = AddonState.UPDATING
return addon_state
@api_error("Failed to set the {addon_name} add-on options")
async def async_set_addon_options(self, config: dict) -> None:
"""Set manager add-on options."""
options = {"options": config}
await async_set_addon_options(self._hass, self.addon_slug, options)
@api_error("Failed to install the {addon_name} add-on")
async def async_install_addon(self) -> None:
"""Install the managed add-on."""
await async_install_addon(self._hass, self.addon_slug)
@callback
def async_schedule_install_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that installs the managed add-on.
Only schedule a new install task if the there's no running task.
"""
if not self._install_task or self._install_task.done():
LOGGER.info(
"%s add-on is not installed. Installing add-on", self.addon_name
)
self._install_task = self._async_schedule_addon_operation(
self.async_install_addon, catch_error=catch_error
)
return self._install_task
@callback
def async_schedule_install_setup_addon(
self,
addon_config: dict[str, Any],
catch_error: bool = False,
) -> asyncio.Task:
"""Schedule a task that installs and sets up the managed add-on.
Only schedule a new install task if the there's no running task.
"""
if not self._install_task or self._install_task.done():
LOGGER.info(
"%s add-on is not installed. Installing add-on", self.addon_name
)
self._install_task = self._async_schedule_addon_operation(
self.async_install_addon,
partial(
self.async_configure_addon,
addon_config,
),
self.async_start_addon,
catch_error=catch_error,
)
return self._install_task
@api_error("Failed to uninstall the {addon_name} add-on")
async def async_uninstall_addon(self) -> None:
"""Uninstall the managed add-on."""
await async_uninstall_addon(self._hass, self.addon_slug)
@api_error("Failed to update the {addon_name} add-on")
async def async_update_addon(self) -> None:
"""Update the managed add-on if needed."""
addon_info = await self.async_get_addon_info()
if addon_info.state is AddonState.NOT_INSTALLED:
raise AddonError(f"{self.addon_name} add-on is not installed")
if not addon_info.update_available:
return
await self.async_create_backup()
await async_update_addon(self._hass, self.addon_slug)
@callback
def async_schedule_update_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that updates and sets up the managed add-on.
Only schedule a new update task if the there's no running task.
"""
if not self._update_task or self._update_task.done():
LOGGER.info("Trying to update the %s add-on", self.addon_name)
self._update_task = self._async_schedule_addon_operation(
self.async_update_addon,
catch_error=catch_error,
)
return self._update_task
@api_error("Failed to start the {addon_name} add-on")
async def async_start_addon(self) -> None:
"""Start the managed add-on."""
await async_start_addon(self._hass, self.addon_slug)
@api_error("Failed to restart the {addon_name} add-on")
async def async_restart_addon(self) -> None:
"""Restart the managed add-on."""
await async_restart_addon(self._hass, self.addon_slug)
@callback
def async_schedule_start_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that starts the managed add-on.
Only schedule a new start task if the there's no running task.
"""
if not self._start_task or self._start_task.done():
LOGGER.info("%s add-on is not running. Starting add-on", self.addon_name)
self._start_task = self._async_schedule_addon_operation(
self.async_start_addon, catch_error=catch_error
)
return self._start_task
@callback
def async_schedule_restart_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that restarts the managed add-on.
Only schedule a new restart task if the there's no running task.
"""
if not self._restart_task or self._restart_task.done():
LOGGER.info("Restarting %s add-on", self.addon_name)
self._restart_task = self._async_schedule_addon_operation(
self.async_restart_addon, catch_error=catch_error
)
return self._restart_task
@api_error("Failed to stop the {addon_name} add-on")
async def async_stop_addon(self) -> None:
"""Stop the managed add-on."""
await async_stop_addon(self._hass, self.addon_slug)
async def async_configure_addon(
self,
addon_config: dict[str, Any],
) -> None:
"""Configure and start manager add-on."""
addon_info = await self.async_get_addon_info()
if addon_info.state is AddonState.NOT_INSTALLED:
raise AddonError(f"{self.addon_name} add-on is not installed")
if addon_config != addon_info.options:
await self.async_set_addon_options(addon_config)
@callback
def async_schedule_setup_addon(
self,
addon_config: dict[str, Any],
catch_error: bool = False,
) -> asyncio.Task:
"""Schedule a task that configures and starts the managed add-on.
Only schedule a new setup task if there's no running task.
"""
if not self._start_task or self._start_task.done():
LOGGER.info("%s add-on is not running. Starting add-on", self.addon_name)
self._start_task = self._async_schedule_addon_operation(
partial(
self.async_configure_addon,
addon_config,
),
self.async_start_addon,
catch_error=catch_error,
)
return self._start_task
@api_error("Failed to create a backup of the {addon_name} add-on.")
async def async_create_backup(self) -> None:
"""Create a partial backup of the managed add-on."""
addon_info = await self.async_get_addon_info()
name = f"addon_{self.addon_slug}_{addon_info.version}"
LOGGER.debug("Creating backup: %s", name)
await async_create_backup(
self._hass,
{"name": name, "addons": [self.addon_slug]},
partial=True,
)
@callback
def _async_schedule_addon_operation(
self, *funcs: Callable, catch_error: bool = False
) -> asyncio.Task:
"""Schedule an add-on task."""
async def addon_operation() -> None:
"""Do the add-on operation and catch AddonError."""
for func in funcs:
try:
await func()
except AddonError as err:
if not catch_error:
raise
LOGGER.error(err)
break
return self._hass.async_create_task(addon_operation())
class AddonError(HomeAssistantError):
"""Represent an error with the managed add-on."""
return AddonManager(hass, LOGGER, "Z-Wave JS", ADDON_SLUG)

11
homeassistant/components/zwave_js/config_flow.py
View file

@ -14,7 +14,14 @@ from zwave_js_server.version import VersionInfo, get_server_version
from homeassistant import config_entries, exceptions
from homeassistant.components import usb
from homeassistant.components.hassio import HassioServiceInfo, is_hassio
from homeassistant.components.hassio import (
AddonError,
AddonInfo,
AddonManager,
AddonState,
HassioServiceInfo,
is_hassio,
)
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.const import CONF_NAME, CONF_URL
from homeassistant.core import HomeAssistant, callback
@ -27,7 +34,7 @@ from homeassistant.data_entry_flow import (
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from . import disconnect_client
from .addon import AddonError, AddonInfo, AddonManager, AddonState, get_addon_manager
from .addon import get_addon_manager
from .const import (
ADDON_SLUG,
CONF_ADDON_DEVICE,

1128
tests/components/hassio/test_addon_manager.py Normal file
View file

File diff suppressed because it is too large Load diff

20
tests/components/zwave_js/conftest.py
View file

@ -26,7 +26,7 @@ def addon_info_side_effect_fixture():
def mock_addon_info(addon_info_side_effect):
"""Mock Supervisor add-on info."""
with patch(
"homeassistant.components.zwave_js.addon.async_get_addon_info",
"homeassistant.components.hassio.addon_manager.async_get_addon_info",
side_effect=addon_info_side_effect,
) as addon_info:
addon_info.return_value = {
@ -48,7 +48,7 @@ def addon_store_info_side_effect_fixture():
def mock_addon_store_info(addon_store_info_side_effect):
"""Mock Supervisor add-on info."""
with patch(
"homeassistant.components.zwave_js.addon.async_get_addon_store_info",
"homeassistant.components.hassio.addon_manager.async_get_addon_store_info",
side_effect=addon_store_info_side_effect,
) as addon_store_info:
addon_store_info.return_value = {
@ -112,7 +112,7 @@ def set_addon_options_side_effect_fixture(addon_options):
def mock_set_addon_options(set_addon_options_side_effect):
"""Mock set add-on options."""
with patch(
"homeassistant.components.zwave_js.addon.async_set_addon_options",
"homeassistant.components.hassio.addon_manager.async_set_addon_options",
side_effect=set_addon_options_side_effect,
) as set_options:
yield set_options
@ -139,7 +139,7 @@ def install_addon_side_effect_fixture(addon_store_info, addon_info):
def mock_install_addon(install_addon_side_effect):
"""Mock install add-on."""
with patch(
"homeassistant.components.zwave_js.addon.async_install_addon",
"homeassistant.components.hassio.addon_manager.async_install_addon",
side_effect=install_addon_side_effect,
) as install_addon:
yield install_addon
@ -149,7 +149,7 @@ def mock_install_addon(install_addon_side_effect):
def mock_update_addon():
"""Mock update add-on."""
with patch(
"homeassistant.components.zwave_js.addon.async_update_addon"
"homeassistant.components.hassio.addon_manager.async_update_addon"
) as update_addon:
yield update_addon
@ -174,7 +174,7 @@ def start_addon_side_effect_fixture(addon_store_info, addon_info):
def mock_start_addon(start_addon_side_effect):
"""Mock start add-on."""
with patch(
"homeassistant.components.zwave_js.addon.async_start_addon",
"homeassistant.components.hassio.addon_manager.async_start_addon",
side_effect=start_addon_side_effect,
) as start_addon:
yield start_addon
@ -184,7 +184,7 @@ def mock_start_addon(start_addon_side_effect):
def stop_addon_fixture():
"""Mock stop add-on."""
with patch(
"homeassistant.components.zwave_js.addon.async_stop_addon"
"homeassistant.components.hassio.addon_manager.async_stop_addon"
) as stop_addon:
yield stop_addon
@ -199,7 +199,7 @@ def restart_addon_side_effect_fixture():
def mock_restart_addon(restart_addon_side_effect):
"""Mock restart add-on."""
with patch(
"homeassistant.components.zwave_js.addon.async_restart_addon",
"homeassistant.components.hassio.addon_manager.async_restart_addon",
side_effect=restart_addon_side_effect,
) as restart_addon:
yield restart_addon
@ -209,7 +209,7 @@ def mock_restart_addon(restart_addon_side_effect):
def uninstall_addon_fixture():
"""Mock uninstall add-on."""
with patch(
"homeassistant.components.zwave_js.addon.async_uninstall_addon"
"homeassistant.components.hassio.addon_manager.async_uninstall_addon"
) as uninstall_addon:
yield uninstall_addon
@ -218,7 +218,7 @@ def uninstall_addon_fixture():
def create_backup_fixture():
"""Mock create backup."""
with patch(
"homeassistant.components.zwave_js.addon.async_create_backup"
"homeassistant.components.hassio.addon_manager.async_create_backup"
) as create_backup:
yield create_backup

30
tests/components/zwave_js/test_addon.py
View file

@ -1,30 +0,0 @@
"""Tests for Z-Wave JS addon module."""
import pytest
from homeassistant.components.zwave_js.addon import AddonError, get_addon_manager
from homeassistant.components.zwave_js.const import (
CONF_ADDON_DEVICE,
CONF_ADDON_S0_LEGACY_KEY,
CONF_ADDON_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY,
CONF_ADDON_S2_UNAUTHENTICATED_KEY,
)
async def test_not_installed_raises_exception(hass, addon_not_installed):
"""Test addon not installed raises exception."""
addon_manager = get_addon_manager(hass)
addon_config = {
CONF_ADDON_DEVICE: "/test",
CONF_ADDON_S0_LEGACY_KEY: "123",
CONF_ADDON_S2_ACCESS_CONTROL_KEY: "456",
CONF_ADDON_S2_AUTHENTICATED_KEY: "789",
CONF_ADDON_S2_UNAUTHENTICATED_KEY: "012",
}
with pytest.raises(AddonError):
await addon_manager.async_configure_addon(addon_config)
with pytest.raises(AddonError):
await addon_manager.async_update_addon()

2
tests/components/zwave_js/test_config_flow.py
View file

@ -88,7 +88,7 @@ def discovery_info_side_effect_fixture():
def mock_get_addon_discovery_info(discovery_info, discovery_info_side_effect):
"""Mock get add-on discovery info."""
with patch(
"homeassistant.components.zwave_js.addon.async_get_addon_discovery_info",
"homeassistant.components.hassio.addon_manager.async_get_addon_discovery_info",
side_effect=discovery_info_side_effect,
return_value=discovery_info,
) as get_addon_discovery_info: