github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-01 05:30:36 +02:00
Code Releases Activity

Refactor zwave_js add-on manager (#80883)

Browse Source 
* Make addon slug an instance attribute

* Extract addon name and addon config

* Update docstrings
This commit is contained in:
Martin Hjelmare 2022-10-24 18:21:05 +02:00 committed by GitHub
parent 4279d73800
commit 838691f22f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 112 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
homeassistant/components/zwave_js/__init__.py
View File

@ -854,24 +854,24 @@ async def async_ensure_addon_running(hass: HomeAssistant, entry: ConfigEntry) ->
s2_unauthenticated_key: str = entry.data.get(CONF_S2_UNAUTHENTICATED_KEY, "")
addon_state = addon_info.state
addon_config = {
CONF_ADDON_DEVICE: usb_path,
CONF_ADDON_S0_LEGACY_KEY: s0_legacy_key,
CONF_ADDON_S2_ACCESS_CONTROL_KEY: s2_access_control_key,
CONF_ADDON_S2_AUTHENTICATED_KEY: s2_authenticated_key,
CONF_ADDON_S2_UNAUTHENTICATED_KEY: s2_unauthenticated_key,
}
if addon_state == AddonState.NOT_INSTALLED:
addon_manager.async_schedule_install_setup_addon(
usb_path,
s0_legacy_key,
s2_access_control_key,
s2_authenticated_key,
s2_unauthenticated_key,
addon_config,
catch_error=True,
)
raise ConfigEntryNotReady
if addon_state == AddonState.NOT_RUNNING:
addon_manager.async_schedule_setup_addon(
usb_path,
s0_legacy_key,
s2_access_control_key,
s2_authenticated_key,
s2_unauthenticated_key,
addon_config,
catch_error=True,
)
raise ConfigEntryNotReady

192
homeassistant/components/zwave_js/addon.py
View File

@ -5,10 +5,10 @@ import asyncio
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from enum import Enum
from functools import partial
from functools import partial, wraps
from typing import Any, TypeVar
from typing_extensions import ParamSpec
from typing_extensions import Concatenate, ParamSpec
from homeassistant.components.hassio import (
async_create_backup,
@ -28,17 +28,9 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.singleton import singleton
from .const import (
ADDON_SLUG,
CONF_ADDON_DEVICE,
CONF_ADDON_S0_LEGACY_KEY,
CONF_ADDON_S2_ACCESS_CONTROL_KEY,
CONF_ADDON_S2_AUTHENTICATED_KEY,
CONF_ADDON_S2_UNAUTHENTICATED_KEY,
DOMAIN,
LOGGER,
)
from .const import ADDON_SLUG, DOMAIN, LOGGER
_AddonManagerT = TypeVar("_AddonManagerT", bound="AddonManager")
_R = TypeVar("_R")
_P = ParamSpec("_P")
@ -49,25 +41,33 @@ DATA_ADDON_MANAGER = f"{DOMAIN}_addon_manager"
@callback
def get_addon_manager(hass: HomeAssistant) -> AddonManager:
"""Get the add-on manager."""
return AddonManager(hass)
return AddonManager(hass, "Z-Wave JS", ADDON_SLUG)
def api_error(
error_message: str,
) -> Callable[[Callable[_P, Awaitable[_R]]], Callable[_P, Coroutine[Any, Any, _R]]]:
) -> 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[_P, Awaitable[_R]]
) -> Callable[_P, Coroutine[Any, Any, _R]]:
func: Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]:
"""Handle a HassioAPIError."""
async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
@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(*args, **kwargs)
return_value = await func(self, *args, **kwargs)
except HassioAPIError as err:
raise AddonError(f"{error_message}: {err}") from err
raise AddonError(
f"{error_message.format(addon_name=self.addon_name)}: {err}"
) from err
return return_value
@ -100,12 +100,14 @@ class AddonManager:
"""Manage the add-on.
Methods may raise AddonError.
Only one instance of this class may exist
Only one instance of this class may exist per add-on
to keep track of running add-on tasks.
"""
def __init__(self, hass: HomeAssistant) -> None:
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
@ -123,21 +125,23 @@ class AddonManager:
)
)
@api_error("Failed to get Z-Wave JS add-on discovery info")
@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, ADDON_SLUG)
discovery_info = await async_get_addon_discovery_info(
self._hass, self.addon_slug
)
if not discovery_info:
raise AddonError("Failed to get Z-Wave JS add-on 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 Z-Wave JS add-on info")
@api_error("Failed to get the {addon_name} add-on info")
async def async_get_addon_info(self) -> AddonInfo:
"""Return and cache Z-Wave JS add-on info."""
addon_store_info = await async_get_addon_store_info(self._hass, ADDON_SLUG)
"""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(
@ -147,7 +151,7 @@ class AddonManager:
version=None,
)
addon_info = await async_get_addon_info(self._hass, ADDON_SLUG)
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"],
@ -158,7 +162,7 @@ class AddonManager:
@callback
def async_get_addon_state(self, addon_info: dict[str, Any]) -> AddonState:
"""Return the current state of the Z-Wave JS add-on."""
"""Return the current state of the managed add-on."""
addon_state = AddonState.NOT_RUNNING
if addon_info["state"] == "started":
@ -170,25 +174,27 @@ class AddonManager:
return addon_state
@api_error("Failed to set the Z-Wave JS add-on options")
@api_error("Failed to set the {addon_name} add-on options")
async def async_set_addon_options(self, config: dict) -> None:
"""Set Z-Wave JS add-on options."""
"""Set manager add-on options."""
options = {"options": config}
await async_set_addon_options(self._hass, ADDON_SLUG, options)
await async_set_addon_options(self._hass, self.addon_slug, options)
@api_error("Failed to install the Z-Wave JS add-on")
@api_error("Failed to install the {addon_name} add-on")
async def async_install_addon(self) -> None:
"""Install the Z-Wave JS add-on."""
await async_install_addon(self._hass, ADDON_SLUG)
"""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 Z-Wave JS add-on.
"""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("Z-Wave JS add-on is not installed. Installing add-on")
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
)
@ -197,85 +203,79 @@ class AddonManager:
@callback
def async_schedule_install_setup_addon(
self,
usb_path: str,
s0_legacy_key: str,
s2_access_control_key: str,
s2_authenticated_key: str,
s2_unauthenticated_key: str,
addon_config: dict[str, Any],
catch_error: bool = False,
) -> asyncio.Task:
"""Schedule a task that installs and sets up the Z-Wave JS add-on.
"""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("Z-Wave JS add-on is not installed. Installing add-on")
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,
usb_path,
s0_legacy_key,
s2_access_control_key,
s2_authenticated_key,
s2_unauthenticated_key,
addon_config,
),
self.async_start_addon,
catch_error=catch_error,
)
return self._install_task
@api_error("Failed to uninstall the Z-Wave JS add-on")
@api_error("Failed to uninstall the {addon_name} add-on")
async def async_uninstall_addon(self) -> None:
"""Uninstall the Z-Wave JS add-on."""
await async_uninstall_addon(self._hass, ADDON_SLUG)
"""Uninstall the managed add-on."""
await async_uninstall_addon(self._hass, self.addon_slug)
@api_error("Failed to update the Z-Wave JS add-on")
@api_error("Failed to update the {addon_name} add-on")
async def async_update_addon(self) -> None:
"""Update the Z-Wave JS add-on if needed."""
"""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("Z-Wave JS add-on is 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, ADDON_SLUG)
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 Z-Wave JS add-on.
"""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 Z-Wave JS add-on")
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 Z-Wave JS add-on")
@api_error("Failed to start the {addon_name} add-on")
async def async_start_addon(self) -> None:
"""Start the Z-Wave JS add-on."""
await async_start_addon(self._hass, ADDON_SLUG)
"""Start the managed add-on."""
await async_start_addon(self._hass, self.addon_slug)
@api_error("Failed to restart the Z-Wave JS add-on")
@api_error("Failed to restart the {addon_name} add-on")
async def async_restart_addon(self) -> None:
"""Restart the Z-Wave JS add-on."""
await async_restart_addon(self._hass, ADDON_SLUG)
"""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 Z-Wave JS add-on.
"""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("Z-Wave JS add-on is not running. Starting add-on")
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
)
@ -283,87 +283,67 @@ class AddonManager:
@callback
def async_schedule_restart_addon(self, catch_error: bool = False) -> asyncio.Task:
"""Schedule a task that restarts the Z-Wave JS add-on.
"""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 Z-Wave JS add-on")
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 Z-Wave JS add-on")
@api_error("Failed to stop the {addon_name} add-on")
async def async_stop_addon(self) -> None:
"""Stop the Z-Wave JS add-on."""
await async_stop_addon(self._hass, ADDON_SLUG)
"""Stop the managed add-on."""
await async_stop_addon(self._hass, self.addon_slug)
async def async_configure_addon(
self,
usb_path: str,
s0_legacy_key: str,
s2_access_control_key: str,
s2_authenticated_key: str,
s2_unauthenticated_key: str,
addon_config: dict[str, Any],
) -> None:
"""Configure and start Z-Wave JS add-on."""
"""Configure and start manager add-on."""
addon_info = await self.async_get_addon_info()
if addon_info.state is AddonState.NOT_INSTALLED:
raise AddonError("Z-Wave JS add-on is not installed")
raise AddonError(f"{self.addon_name} add-on is not installed")
new_addon_options = {
CONF_ADDON_DEVICE: usb_path,
CONF_ADDON_S0_LEGACY_KEY: s0_legacy_key,
CONF_ADDON_S2_ACCESS_CONTROL_KEY: s2_access_control_key,
CONF_ADDON_S2_AUTHENTICATED_KEY: s2_authenticated_key,
CONF_ADDON_S2_UNAUTHENTICATED_KEY: s2_unauthenticated_key,
}
if new_addon_options != addon_info.options:
await self.async_set_addon_options(new_addon_options)
if addon_config != addon_info.options:
await self.async_set_addon_options(addon_config)
@callback
def async_schedule_setup_addon(
self,
usb_path: str,
s0_legacy_key: str,
s2_access_control_key: str,
s2_authenticated_key: str,
s2_unauthenticated_key: str,
addon_config: dict[str, Any],
catch_error: bool = False,
) -> asyncio.Task:
"""Schedule a task that configures and starts the Z-Wave JS add-on.
"""Schedule a task that configures and starts the managed add-on.
Only schedule a new setup task if the there's no running task.
Only schedule a new setup task if there's no running task.
"""
if not self._start_task or self._start_task.done():
LOGGER.info("Z-Wave JS add-on is not running. Starting add-on")
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,
usb_path,
s0_legacy_key,
s2_access_control_key,
s2_authenticated_key,
s2_unauthenticated_key,
addon_config,
),
self.async_start_addon,
catch_error=catch_error,
)
return self._start_task
@api_error("Failed to create a backup of the Z-Wave JS add-on.")
@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 Z-Wave JS add-on."""
"""Create a partial backup of the managed add-on."""
addon_info = await self.async_get_addon_info()
name = f"addon_{ADDON_SLUG}_{addon_info.version}"
name = f"addon_{self.addon_slug}_{addon_info.version}"
LOGGER.debug("Creating backup: %s", name)
await async_create_backup(
self._hass,
{"name": name, "addons": [ADDON_SLUG]},
{"name": name, "addons": [self.addon_slug]},
partial=True,
)
@ -388,4 +368,4 @@ class AddonManager:
class AddonError(HomeAssistantError):
"""Represent an error with Z-Wave JS add-on."""
"""Represent an error with the managed add-on."""

17
tests/components/zwave_js/test_addon.py
View File

@ -2,14 +2,29 @@
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("/test", "123", "456", "789", "012")
await addon_manager.async_configure_addon(addon_config)
with pytest.raises(AddonError):
await addon_manager.async_update_addon()