Collection of minor changes (#1793)

* Collection of minor changes

* Process smaller comments

* Fix HA version handling

* More type hints

* Review comment from Martin

* Add protection against TypeError in version parsing

supervisor/addons/addon.py

@@ -130,12 +130,9 @@ class Addon(AddonModel):
         return {**self.data[ATTR_OPTIONS], **self.persist[ATTR_OPTIONS]}
 
     @options.setter
-    def options(self, value: Optional[Dict[str, Any]]):
+    def options(self, value: Optional[Dict[str, Any]]) -> None:
         """Store user add-on options."""
-        if value is None:
-            self.persist[ATTR_OPTIONS] = {}
-        else:
-            self.persist[ATTR_OPTIONS] = deepcopy(value)
+        self.persist[ATTR_OPTIONS] = {} if value is None else deepcopy(value)
 
     @property
     def boot(self) -> bool:

supervisor/addons/model.py

@@ -539,12 +539,15 @@ class AddonModel(CoreSysAttributes, ABC):
             return False
 
         # Home Assistant
-        version = config.get(ATTR_HOMEASSISTANT) or self.sys_homeassistant.version
-        if pkg_version.parse(self.sys_homeassistant.version) < pkg_version.parse(
-            version
-        ):
-            return False
+        version = config.get(ATTR_HOMEASSISTANT)
+        if version is None or self.sys_homeassistant.version is None:
+            return True
 
+        try:
+            return pkg_version.parse(
+                self.sys_homeassistant.version
+            ) >= pkg_version.parse(version)
+        except pkg_version.InvalidVersion:
             return True
 
     def _image(self, config) -> str:

supervisor/api/addons.py

@@ -148,9 +148,7 @@ class APIAddons(CoreSysAttributes):
     @api_process
     async def list(self, request: web.Request) -> Dict[str, Any]:
         """Return all add-ons or repositories."""
-        data_addons = []
-        for addon in self.sys_addons.all:
-            data_addons.append(
+        data_addons = [
             {
                 ATTR_NAME: addon.name,
                 ATTR_SLUG: addon.slug,
@@ -167,11 +165,10 @@ class APIAddons(CoreSysAttributes):
                 ATTR_ICON: addon.with_icon,
                 ATTR_LOGO: addon.with_logo,
             }
-            )
+            for addon in self.sys_addons.all
+        ]
 
-        data_repositories = []
-        for repository in self.sys_store.all:
-            data_repositories.append(
+        data_repositories = [
             {
                 ATTR_SLUG: repository.slug,
                 ATTR_NAME: repository.name,
@@ -179,8 +176,8 @@ class APIAddons(CoreSysAttributes):
                 ATTR_URL: repository.url,
                 ATTR_MAINTAINER: repository.maintainer,
             }
-            )
-
+            for repository in self.sys_store.all
+        ]
         return {ATTR_ADDONS: data_addons, ATTR_REPOSITORIES: data_repositories}
 
     @api_process
@@ -449,7 +446,4 @@ def _pretty_devices(addon: AnyAddon) -> List[str]:
 
 def _pretty_services(addon: AnyAddon) -> List[str]:
     """Return a simplified services role list."""
-    services = []
-    for name, access in addon.services_role.items():
-        services.append(f"{name}:{access}")
-    return services
+    return [f"{name}:{access}" for name, access in addon.services_role.items()]

supervisor/api/discovery.py

@@ -5,8 +5,8 @@ from ..const import (
     ATTR_ADDON,
     ATTR_CONFIG,
     ATTR_DISCOVERY,
-    ATTR_SERVICES,
     ATTR_SERVICE,
+    ATTR_SERVICES,
     ATTR_UUID,
     REQUEST_FROM,
 )

supervisor/api/hardware.py

@@ -1,7 +1,7 @@
 """Init file for Supervisor hardware RESTful API."""
 import asyncio
 import logging
-from typing import Any, Dict
+from typing import Any, Awaitable, Dict
 
 from aiohttp import web
 
@@ -52,6 +52,6 @@ class APIHardware(CoreSysAttributes):
         }
 
     @api_process
-    def trigger(self, request: web.Request) -> None:
+    def trigger(self, request: web.Request) -> Awaitable[None]:
         """Trigger a udev device reload."""
         return asyncio.shield(self.sys_hardware.udev_trigger())

supervisor/api/homeassistant.py

@@ -1,7 +1,7 @@
 """Init file for Supervisor Home Assistant RESTful API."""
 import asyncio
 import logging
-from typing import Any, Coroutine, Dict
+from typing import Any, Awaitable, Dict
 
 from aiohttp import web
 import voluptuous as vol
@@ -141,27 +141,27 @@ class APIHomeAssistant(CoreSysAttributes):
         await asyncio.shield(self.sys_homeassistant.update(version))
 
     @api_process
-    def stop(self, request: web.Request) -> Coroutine:
+    def stop(self, request: web.Request) -> Awaitable[None]:
         """Stop Home Assistant."""
         return asyncio.shield(self.sys_homeassistant.stop())
 
     @api_process
-    def start(self, request: web.Request) -> Coroutine:
+    def start(self, request: web.Request) -> Awaitable[None]:
         """Start Home Assistant."""
         return asyncio.shield(self.sys_homeassistant.start())
 
     @api_process
-    def restart(self, request: web.Request) -> Coroutine:
+    def restart(self, request: web.Request) -> Awaitable[None]:
         """Restart Home Assistant."""
         return asyncio.shield(self.sys_homeassistant.restart())
 
     @api_process
-    def rebuild(self, request: web.Request) -> Coroutine:
+    def rebuild(self, request: web.Request) -> Awaitable[None]:
         """Rebuild Home Assistant."""
         return asyncio.shield(self.sys_homeassistant.rebuild())
 
     @api_process_raw(CONTENT_TYPE_BINARY)
-    def logs(self, request: web.Request) -> Coroutine:
+    def logs(self, request: web.Request) -> Awaitable[bytes]:
         """Return Home Assistant Docker logs."""
         return self.sys_homeassistant.logs()
 

supervisor/bootstrap.py

@@ -46,7 +46,7 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
 MACHINE_ID = Path("/etc/machine-id")
 
 
-async def initialize_coresys() -> None:
+async def initialize_coresys() -> CoreSys:
     """Initialize supervisor coresys/objects."""
     coresys = CoreSys()
 

supervisor/dbus/interface.py

@@ -1,10 +1,11 @@
 """Interface class for D-Bus wrappers."""
+from abc import ABC, abstractmethod
 from typing import Optional
 
 from ..utils.gdbus import DBus
 
 
-class DBusInterface:
+class DBusInterface(ABC):
     """Handle D-Bus interface for hostname/system."""
 
     dbus: Optional[DBus] = None
@@ -14,6 +15,6 @@ class DBusInterface:
         """Return True, if they is connected to D-Bus."""
         return self.dbus is not None
 
+    @abstractmethod
     async def connect(self):
         """Connect to D-Bus."""
-        raise NotImplementedError()

supervisor/docker/__init__.py

@@ -107,9 +107,9 @@ class DockerAPI:
 
         Need run inside executor.
         """
-        name: str = kwargs.get("name")
-        network_mode: str = kwargs.get("network_mode")
-        hostname: str = kwargs.get("hostname")
+        name: Optional[str] = kwargs.get("name")
+        network_mode: Optional[str] = kwargs.get("network_mode")
+        hostname: Optional[str] = kwargs.get("hostname")
 
         # Setup DNS
         if dns:

supervisor/docker/addon.py

@@ -47,7 +47,7 @@ class DockerAddon(DockerInterface):
         self.addon = addon
 
     @property
-    def image(self) -> str:
+    def image(self) -> Optional[str]:
         """Return name of Docker image."""
         return self.addon.image
 
@@ -102,7 +102,7 @@ class DockerAddon(DockerInterface):
         return not self.addon.protected and self.addon.with_full_access
 
     @property
-    def environment(self) -> Dict[str, str]:
+    def environment(self) -> Dict[str, Optional[str]]:
         """Return environment for Docker add-on."""
         addon_env = self.addon.environment or {}
 
@@ -189,7 +189,7 @@ class DockerAddon(DockerInterface):
         return None
 
     @property
-    def network_mapping(self) -> Dict[str, str]:
+    def network_mapping(self) -> Dict[str, IPv4Address]:
         """Return hosts mapping."""
         return {
             "supervisor": self.sys_docker.network.supervisor,

supervisor/docker/interface.py

@@ -134,11 +134,7 @@ class DockerInterface(CoreSysAttributes):
         except docker.errors.DockerException:
             return False
 
-        # container is not running
-        if docker_container.status != "running":
-            return False
-
-        return True
+        return docker_container.status == "running"
 
     @process_lock
     def attach(self, tag: str):
@@ -301,6 +297,8 @@ class DockerInterface(CoreSysAttributes):
         except docker.errors.DockerException as err:
             _LOGGER.warning("Can't grep logs from %s: %s", self.image, err)
 
+        return b""
+
     @process_lock
     def cleanup(self, old_image: Optional[str] = None) -> Awaitable[None]:
         """Check if old version exists and cleanup."""
@@ -412,10 +410,7 @@ class DockerInterface(CoreSysAttributes):
             return False
 
         # Check return value
-        if int(docker_container.attrs["State"]["ExitCode"]) != 0:
-            return True
-
-        return False
+        return int(docker_container.attrs["State"]["ExitCode"]) != 0
 
     def get_latest_version(self, key: Any = int) -> Awaitable[str]:
         """Return latest version of local Home Asssistant image."""

supervisor/homeassistant.py

@@ -202,12 +202,12 @@ class HomeAssistant(JsonConfig, CoreSysAttributes):
         return self._data[ATTR_UUID]
 
     @property
-    def supervisor_token(self) -> str:
+    def supervisor_token(self) -> Optional[str]:
         """Return an access token for the Supervisor API."""
         return self._data.get(ATTR_ACCESS_TOKEN)
 
     @property
-    def refresh_token(self) -> str:
+    def refresh_token(self) -> Optional[str]:
         """Return the refresh token to authenticate with Home Assistant."""
         return self._data.get(ATTR_REFRESH_TOKEN)
 

supervisor/plugins/__init__.py

@@ -49,7 +49,7 @@ class PluginManager(CoreSysAttributes):
         """Return multicast handler."""
         return self._multicast
 
-    async def load(self):
+    async def load(self) -> None:
         """Load Supervisor plugins."""
         # Sequential to avoid issue on slow IO
         for plugin in (
@@ -94,7 +94,7 @@ class PluginManager(CoreSysAttributes):
                     required_version,
                 )
 
-    async def repair(self):
+    async def repair(self) -> None:
         """Repair Supervisor plugins."""
         await asyncio.wait(
             [

supervisor/services/__init__.py

@@ -1,5 +1,5 @@
 """Handle internal services discovery."""
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from ..coresys import CoreSys, CoreSysAttributes
 from .const import SERVICE_MQTT, SERVICE_MYSQL
@@ -25,7 +25,7 @@ class ServiceManager(CoreSysAttributes):
         """Return a list of services."""
         return list(self.services_obj.values())
 
-    def get(self, slug: str) -> ServiceInterface:
+    def get(self, slug: str) -> Optional[ServiceInterface]:
         """Return service object from slug."""
         return self.services_obj.get(slug)
 

supervisor/utils/tar.py

@@ -41,7 +41,7 @@ class SecureTarFile:
 
         # Encryption/Description
         self._aes: Optional[Cipher] = None
-        self._key: bytes = key
+        self._key: Optional[bytes] = key
 
         # Function helper
         self._decrypt: Optional[CipherContext] = None
@@ -101,7 +101,7 @@ class SecureTarFile:
         return self._name
 
     @property
-    def size(self) -> int:
+    def size(self) -> float:
         """Return snapshot size."""
         if not self._name.is_file():
             return 0

supervisor/validate.py

@@ -36,8 +36,8 @@ from .const import (
     ATTR_VERSION,
     ATTR_WAIT_BOOT,
     ATTR_WATCHDOG,
-    LogLevel,
     SUPERVISOR_VERSION,
+    LogLevel,
     UpdateChannels,
 )
 from .utils.validate import validate_timezone