diff --git a/supervisor/addons/addon.py b/supervisor/addons/addon.py index 09cbb5f50..87be7d012 100644 --- a/supervisor/addons/addon.py +++ b/supervisor/addons/addon.py @@ -143,7 +143,7 @@ class Addon(AddonModel): return self.persist.get(ATTR_BOOT, super().boot) @boot.setter - def boot(self, value: bool): + def boot(self, value: bool) -> None: """Store user boot options.""" self.persist[ATTR_BOOT] = value @@ -153,7 +153,7 @@ class Addon(AddonModel): return self.persist.get(ATTR_AUTO_UPDATE, super().auto_update) @auto_update.setter - def auto_update(self, value: bool): + def auto_update(self, value: bool) -> None: """Set auto update.""" self.persist[ATTR_AUTO_UPDATE] = value @@ -190,7 +190,7 @@ class Addon(AddonModel): return self.persist[ATTR_PROTECTED] @protected.setter - def protected(self, value: bool): + def protected(self, value: bool) -> None: """Set add-on in protected mode.""" self.persist[ATTR_PROTECTED] = value @@ -200,7 +200,7 @@ class Addon(AddonModel): return self.persist.get(ATTR_NETWORK, super().ports) @ports.setter - def ports(self, value: Optional[Dict[str, Optional[int]]]): + def ports(self, value: Optional[Dict[str, Optional[int]]]) -> None: """Set custom ports of add-on.""" if value is None: self.persist.pop(ATTR_NETWORK, None) @@ -232,6 +232,8 @@ class Addon(AddonModel): if not url: return None webui = RE_WEBUI.match(url) + if not webui: + return None # extract arguments t_port = webui.group("t_port") @@ -274,7 +276,7 @@ class Addon(AddonModel): return self.persist[ATTR_INGRESS_PANEL] @ingress_panel.setter - def ingress_panel(self, value: bool): + def ingress_panel(self, value: bool) -> None: """Return True if the add-on access support ingress.""" self.persist[ATTR_INGRESS_PANEL] = value @@ -310,50 +312,50 @@ class Addon(AddonModel): return input_data @audio_input.setter - def audio_input(self, value: Optional[str]): + def audio_input(self, value: Optional[str]) -> None: """Set audio input settings.""" self.persist[ATTR_AUDIO_INPUT] = value @property - def image(self): + def image(self) -> Optional[str]: """Return image name of add-on.""" return self.persist.get(ATTR_IMAGE) @property - def need_build(self): + def need_build(self) -> bool: """Return True if this add-on need a local build.""" return ATTR_IMAGE not in self.data @property - def path_data(self): + def path_data(self) -> Path: """Return add-on data path inside Supervisor.""" return Path(self.sys_config.path_addons_data, self.slug) @property - def path_extern_data(self): + def path_extern_data(self) -> PurePath: """Return add-on data path external for Docker.""" return PurePath(self.sys_config.path_extern_addons_data, self.slug) @property - def path_options(self): + def path_options(self) -> Path: """Return path to add-on options.""" return Path(self.path_data, "options.json") @property - def path_pulse(self): + def path_pulse(self) -> Path: """Return path to asound config.""" return Path(self.sys_config.path_tmp, f"{self.slug}_pulse") @property - def path_extern_pulse(self): + def path_extern_pulse(self) -> Path: """Return path to asound config for Docker.""" return Path(self.sys_config.path_extern_tmp, f"{self.slug}_pulse") - def save_persist(self): + def save_persist(self) -> None: """Save data of add-on.""" self.sys_addons.data.save_data() - async def write_options(self): + async def write_options(self) -> None: """Return True if add-on options is written to data.""" schema = self.schema options = self.options @@ -378,7 +380,7 @@ class Addon(AddonModel): raise AddonsError() - async def remove_data(self): + async def remove_data(self) -> None: """Remove add-on data.""" if not self.path_data.is_dir(): return @@ -386,7 +388,7 @@ class Addon(AddonModel): _LOGGER.info("Remove add-on data folder %s", self.path_data) await remove_data(self.path_data) - def write_pulse(self): + def write_pulse(self) -> None: """Write asound config to file and return True on success.""" pulse_config = self.sys_plugins.audio.pulse_client( input_profile=self.audio_input, output_profile=self.audio_output @@ -518,7 +520,7 @@ class Addon(AddonModel): except DockerAPIError: raise AddonsError() from None - async def write_stdin(self, data): + async def write_stdin(self, data) -> None: """Write data to add-on stdin. Return a coroutine. diff --git a/supervisor/addons/model.py b/supervisor/addons/model.py index 94731ae10..8e7e8c5c2 100644 --- a/supervisor/addons/model.py +++ b/supervisor/addons/model.py @@ -185,7 +185,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_VERSION] @property - def version(self) -> str: + def version(self) -> Optional[str]: """Return version of add-on.""" return self.data[ATTR_VERSION] @@ -288,7 +288,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_HOST_DBUS] @property - def devices(self) -> Optional[List[str]]: + def devices(self) -> List[str]: """Return devices of add-on.""" return self.data.get(ATTR_DEVICES, []) @@ -452,7 +452,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data.get(ATTR_MACHINE, []) @property - def image(self) -> str: + def image(self) -> Optional[str]: """Generate image name from data.""" return self._image(self.data) diff --git a/supervisor/api/addons.py b/supervisor/api/addons.py index e48b60b74..f827e506b 100644 --- a/supervisor/api/addons.py +++ b/supervisor/api/addons.py @@ -122,9 +122,7 @@ SCHEMA_SECURITY = vol.Schema({vol.Optional(ATTR_PROTECTED): vol.Boolean()}) class APIAddons(CoreSysAttributes): """Handle RESTful API for add-on functions.""" - def _extract_addon( - self, request: web.Request, check_installed: bool = True - ) -> AnyAddon: + def _extract_addon(self, request: web.Request) -> AnyAddon: """Return addon, throw an exception it it doesn't exist.""" addon_slug: str = request.match_info.get("addon") @@ -139,9 +137,12 @@ class APIAddons(CoreSysAttributes): if not addon: raise APIError(f"Addon {addon_slug} does not exist") - if check_installed and not addon.is_installed: - raise APIError(f"Addon {addon.slug} is not installed") + return addon + def _extract_addon_installed(self, request: web.Request) -> Addon: + addon = self._extract_addon(request) + if not isinstance(addon, Addon) or not addon.is_installed: + raise APIError("Addon is not installed") return addon @api_process @@ -190,7 +191,7 @@ class APIAddons(CoreSysAttributes): @api_process async def info(self, request: web.Request) -> Dict[str, Any]: """Return add-on information.""" - addon: AnyAddon = self._extract_addon(request, check_installed=False) + addon: AnyAddon = self._extract_addon(request) data = { ATTR_NAME: addon.name, @@ -257,7 +258,7 @@ class APIAddons(CoreSysAttributes): ATTR_INGRESS_PANEL: None, } - if addon.is_installed: + if isinstance(addon, Addon) and addon.is_installed: data.update( { ATTR_STATE: await addon.state(), @@ -279,7 +280,7 @@ class APIAddons(CoreSysAttributes): @api_process async def options(self, request: web.Request) -> None: """Store user options for add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) # Update secrets for validation await self.sys_secrets.reload() @@ -312,7 +313,7 @@ class APIAddons(CoreSysAttributes): @api_process async def security(self, request: web.Request) -> None: """Store security options for add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) body: Dict[str, Any] = await api_validate(SCHEMA_SECURITY, request) if ATTR_PROTECTED in body: @@ -324,7 +325,8 @@ class APIAddons(CoreSysAttributes): @api_process async def stats(self, request: web.Request) -> Dict[str, Any]: """Return resource information.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) + stats: DockerStats = await addon.stats() return { @@ -341,62 +343,55 @@ class APIAddons(CoreSysAttributes): @api_process def install(self, request: web.Request) -> Awaitable[None]: """Install add-on.""" - addon: AnyAddon = self._extract_addon(request, check_installed=False) + addon = self._extract_addon(request) return asyncio.shield(addon.install()) @api_process def uninstall(self, request: web.Request) -> Awaitable[None]: """Uninstall add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) return asyncio.shield(addon.uninstall()) @api_process def start(self, request: web.Request) -> Awaitable[None]: """Start add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) return asyncio.shield(addon.start()) @api_process def stop(self, request: web.Request) -> Awaitable[None]: """Stop add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) return asyncio.shield(addon.stop()) @api_process def update(self, request: web.Request) -> Awaitable[None]: """Update add-on.""" - addon: AnyAddon = self._extract_addon(request) - - if addon.latest_version == addon.version: - raise APIError("No update available!") - + addon: Addon = self._extract_addon_installed(request) return asyncio.shield(addon.update()) @api_process def restart(self, request: web.Request) -> Awaitable[None]: """Restart add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon: Addon = self._extract_addon_installed(request) return asyncio.shield(addon.restart()) @api_process def rebuild(self, request: web.Request) -> Awaitable[None]: """Rebuild local build add-on.""" - addon: AnyAddon = self._extract_addon(request) - if not addon.need_build: - raise APIError("Only local build addons are supported") - + addon = self._extract_addon_installed(request) return asyncio.shield(addon.rebuild()) @api_process_raw(CONTENT_TYPE_BINARY) def logs(self, request: web.Request) -> Awaitable[bytes]: """Return logs from add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) return addon.logs() @api_process_raw(CONTENT_TYPE_PNG) async def icon(self, request: web.Request) -> bytes: """Return icon from add-on.""" - addon: AnyAddon = self._extract_addon(request, check_installed=False) + addon = self._extract_addon(request) if not addon.with_icon: raise APIError(f"No icon found for add-on {addon.slug}!") @@ -406,7 +401,7 @@ class APIAddons(CoreSysAttributes): @api_process_raw(CONTENT_TYPE_PNG) async def logo(self, request: web.Request) -> bytes: """Return logo from add-on.""" - addon: AnyAddon = self._extract_addon(request, check_installed=False) + addon = self._extract_addon(request) if not addon.with_logo: raise APIError(f"No logo found for add-on {addon.slug}!") @@ -416,7 +411,7 @@ class APIAddons(CoreSysAttributes): @api_process_raw(CONTENT_TYPE_TEXT) async def changelog(self, request: web.Request) -> str: """Return changelog from add-on.""" - addon: AnyAddon = self._extract_addon(request, check_installed=False) + addon = self._extract_addon(request) if not addon.with_changelog: raise APIError(f"No changelog found for add-on {addon.slug}!") @@ -426,7 +421,7 @@ class APIAddons(CoreSysAttributes): @api_process_raw(CONTENT_TYPE_TEXT) async def documentation(self, request: web.Request) -> str: """Return documentation from add-on.""" - addon: AnyAddon = self._extract_addon(request, check_installed=False) + addon = self._extract_addon(request) if not addon.with_documentation: raise APIError(f"No documentation found for add-on {addon.slug}!") @@ -436,7 +431,7 @@ class APIAddons(CoreSysAttributes): @api_process async def stdin(self, request: web.Request) -> None: """Write to stdin of add-on.""" - addon: AnyAddon = self._extract_addon(request) + addon = self._extract_addon_installed(request) if not addon.with_stdin: raise APIError(f"STDIN not supported the {addon.slug} add-on") @@ -448,7 +443,7 @@ def _pretty_devices(addon: AnyAddon) -> List[str]: """Return a simplified device list.""" dev_list = addon.devices if not dev_list: - return None + return [] return [row.split(":")[0] for row in dev_list]