Fix mypy issues in docker, hardware and homeassistant modules (#5805)

* Fix mypy issues in docker and hardware modules

* Fix mypy issues in homeassistant module

* Fix async_send_command typing

* Fixes from feedback

Author: mdegat01

Diff for: supervisor/api/homeassistant.py

@@ -118,7 +118,7 @@ async def options(self, request: web.Request) -> None:
         body = await api_validate(SCHEMA_OPTIONS, request)
 
         if ATTR_IMAGE in body:
-            self.sys_homeassistant.image = body[ATTR_IMAGE]
+            self.sys_homeassistant.set_image(body[ATTR_IMAGE])
             self.sys_homeassistant.override_image = (
                 self.sys_homeassistant.image != self.sys_homeassistant.default_image
             )

Diff for: supervisor/auth.py

@@ -145,13 +145,21 @@ async def change_password(self, username: str, password: str) -> None:
     async def list_users(self) -> list[dict[str, Any]]:
         """List users on the Home Assistant instance."""
         try:
-            return await self.sys_homeassistant.websocket.async_send_command(
+            users: (
+                list[dict[str, Any]] | None
+            ) = await self.sys_homeassistant.websocket.async_send_command(
                 {ATTR_TYPE: "config/auth/list"}
             )
-        except HomeAssistantWSError:
-            _LOGGER.error("Can't request listing users on Home Assistant!")
-
-        raise AuthListUsersError()
+        except HomeAssistantWSError as err:
+            raise AuthListUsersError(
+                f"Can't request listing users on Home Assistant: {err}", _LOGGER.error
+            ) from err
+
+        if users is not None:
+            return users
+        raise AuthListUsersError(
+            "Can't request listing users on Home Assistant!", _LOGGER.error
+        )
 
     @staticmethod
     def _rehash(value: str, salt2: str = "") -> str:

Diff for: supervisor/bootstrap.py

@@ -70,7 +70,7 @@ async def initialize_coresys() -> CoreSys:
     coresys.addons = await AddonManager(coresys).load_config()
     coresys.backups = await BackupManager(coresys).load_config()
     coresys.host = await HostManager(coresys).post_init()
-    coresys.hardware = await HardwareManager(coresys).post_init()
+    coresys.hardware = await HardwareManager.create(coresys)
     coresys.ingress = await Ingress(coresys).load_config()
     coresys.tasks = Tasks(coresys)
     coresys.services = await ServiceManager(coresys).load_config()

Diff for: supervisor/const.py

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass
 from enum import StrEnum
-from ipaddress import ip_network
+from ipaddress import IPv4Network
 from pathlib import Path
 from sys import version_info as systemversion
 from typing import Self
@@ -41,8 +41,8 @@
 SYSTEMD_JOURNAL_VOLATILE = Path("/run/log/journal")
 
 DOCKER_NETWORK = "hassio"
-DOCKER_NETWORK_MASK = ip_network("172.30.32.0/23")
-DOCKER_NETWORK_RANGE = ip_network("172.30.33.0/24")
+DOCKER_NETWORK_MASK = IPv4Network("172.30.32.0/23")
+DOCKER_NETWORK_RANGE = IPv4Network("172.30.33.0/24")
 
 # This needs to match the dockerd --cpu-rt-runtime= argument.
 DOCKER_CPU_RUNTIME_TOTAL = 950_000

Diff for: supervisor/docker/addon.py

@@ -2,13 +2,12 @@
 
 from __future__ import annotations
 
-from collections.abc import Awaitable
 from contextlib import suppress
-from ipaddress import IPv4Address, ip_address
+from ipaddress import IPv4Address
 import logging
 import os
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from attr import evolve
 from awesomeversion import AwesomeVersion
@@ -73,7 +72,7 @@
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
-NO_ADDDRESS = ip_address("0.0.0.0")
+NO_ADDDRESS = IPv4Address("0.0.0.0")
 
 
 class DockerAddon(DockerInterface):
@@ -101,10 +100,12 @@ def ip_address(self) -> IPv4Address:
         """Return IP address of this container."""
         if self.addon.host_network:
             return self.sys_docker.network.gateway
+        if not self._meta:
+            return NO_ADDDRESS
 
         # Extract IP-Address
         try:
-            return ip_address(
+            return IPv4Address(
                 self._meta["NetworkSettings"]["Networks"]["hassio"]["IPAddress"]
             )
         except (KeyError, TypeError, ValueError):
@@ -121,7 +122,7 @@ def version(self) -> AwesomeVersion:
         return self.addon.version
 
     @property
-    def arch(self) -> str:
+    def arch(self) -> str | None:
         """Return arch of Docker image."""
         if self.addon.legacy:
             return self.sys_arch.default
@@ -133,9 +134,9 @@ def name(self) -> str:
         return DockerAddon.slug_to_name(self.addon.slug)
 
     @property
-    def environment(self) -> dict[str, str | None]:
+    def environment(self) -> dict[str, str | int | None]:
         """Return environment for Docker add-on."""
-        addon_env = self.addon.environment or {}
+        addon_env = cast(dict[str, str | int | None], self.addon.environment or {})
 
         # Provide options for legacy add-ons
         if self.addon.legacy:
@@ -336,7 +337,7 @@ def mounts(self) -> list[Mount]:
         """Return mounts for container."""
         addon_mapping = self.addon.map_volumes
 
-        target_data_path = ""
+        target_data_path: str | None = None
         if MappingType.DATA in addon_mapping:
             target_data_path = addon_mapping[MappingType.DATA].path
 
@@ -677,12 +678,12 @@ def build_image():
             )
 
         try:
-            image, log = await self.sys_run_in_executor(build_image)
+            docker_image, log = await self.sys_run_in_executor(build_image)
 
             _LOGGER.debug("Build %s:%s done: %s", self.image, version, log)
 
             # Update meta data
-            self._meta = image.attrs
+            self._meta = docker_image.attrs
 
         except (docker.errors.DockerException, requests.RequestException) as err:
             _LOGGER.error("Can't build %s:%s: %s", self.image, version, err)
@@ -699,9 +700,14 @@ def build_image():
 
         _LOGGER.info("Build %s:%s done", self.image, version)
 
-    def export_image(self, tar_file: Path) -> Awaitable[None]:
-        """Export current images into a tar file."""
-        return self.sys_docker.export_image(self.image, self.version, tar_file)
+    def export_image(self, tar_file: Path) -> None:
+        """Export current images into a tar file.
+
+        Must be run in executor.
+        """
+        if not self.image:
+            raise RuntimeError("Cannot export without image!")
+        self.sys_docker.export_image(self.image, self.version, tar_file)
 
     @Job(
         name="docker_addon_import_image",
@@ -805,15 +811,15 @@ async def stop(self, remove_container: bool = True) -> None:
         ):
             self.sys_resolution.dismiss_issue(self.addon.device_access_missing_issue)
 
-    async def _validate_trust(
-        self, image_id: str, image: str, version: AwesomeVersion
-    ) -> None:
+    async def _validate_trust(self, image_id: str) -> None:
         """Validate trust of content."""
         if not self.addon.signed:
             return
 
         checksum = image_id.partition(":")[2]
-        return await self.sys_security.verify_content(self.addon.codenotary, checksum)
+        return await self.sys_security.verify_content(
+            cast(str, self.addon.codenotary), checksum
+        )
 
     @Job(
         name="docker_addon_hardware_events",
@@ -834,7 +840,8 @@ async def _hardware_events(self, device: Device) -> None:
                 self.sys_docker.containers.get, self.name
             )
         except docker.errors.NotFound:
-            self.sys_bus.remove_listener(self._hw_listener)
+            if self._hw_listener:
+                self.sys_bus.remove_listener(self._hw_listener)
             self._hw_listener = None
             return
         except (docker.errors.DockerException, requests.RequestException) as err:

Diff for: supervisor/docker/homeassistant.py

@@ -248,14 +248,12 @@ def is_initialize(self) -> Awaitable[bool]:
             self.sys_homeassistant.version,
         )
 
-    async def _validate_trust(
-        self, image_id: str, image: str, version: AwesomeVersion
-    ) -> None:
+    async def _validate_trust(self, image_id: str) -> None:
         """Validate trust of content."""
         try:
-            if version != LANDINGPAGE and version < _VERIFY_TRUST:
+            if self.version in {None, LANDINGPAGE} or self.version < _VERIFY_TRUST:
                 return
         except AwesomeVersionCompareException:
             return
 
-        await super()._validate_trust(image_id, image, version)
+        await super()._validate_trust(image_id)

Diff for: supervisor/docker/interface.py

@@ -2,13 +2,14 @@
 
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
 from collections import defaultdict
 from collections.abc import Awaitable
 from contextlib import suppress
 import logging
 import re
 from time import time
-from typing import Any
+from typing import Any, cast
 from uuid import uuid4
 
 from awesomeversion import AwesomeVersion
@@ -79,7 +80,7 @@ def _container_state_from_model(docker_container: Container) -> ContainerState:
     return ContainerState.STOPPED
 
 
-class DockerInterface(JobGroup):
+class DockerInterface(JobGroup, ABC):
     """Docker Supervisor interface."""
 
     def __init__(self, coresys: CoreSys):
@@ -100,9 +101,9 @@ def timeout(self) -> int:
         return 10
 
     @property
-    def name(self) -> str | None:
+    @abstractmethod
+    def name(self) -> str:
         """Return name of Docker container."""
-        return None
 
     @property
     def meta_config(self) -> dict[str, Any]:
@@ -153,7 +154,7 @@ def arch(self) -> str | None:
     @property
     def in_progress(self) -> bool:
         """Return True if a task is in progress."""
-        return self.active_job
+        return self.active_job is not None
 
     @property
     def restart_policy(self) -> RestartPolicy | None:
@@ -230,7 +231,10 @@ async def install(
     ) -> None:
         """Pull docker image."""
         image = image or self.image
-        arch = arch or self.sys_arch.supervisor
+        if not image:
+            raise ValueError("Cannot pull without an image!")
+
+        image_arch = str(arch) if arch else self.sys_arch.supervisor
 
         _LOGGER.info("Downloading docker image %s with tag %s.", image, version)
         try:
@@ -242,12 +246,12 @@ async def install(
             docker_image = await self.sys_run_in_executor(
                 self.sys_docker.images.pull,
                 f"{image}:{version!s}",
-                platform=MAP_ARCH[arch],
+                platform=MAP_ARCH[image_arch],
             )
 
             # Validate content
             try:
-                await self._validate_trust(docker_image.id, image, version)
+                await self._validate_trust(cast(str, docker_image.id))
             except CodeNotaryError:
                 with suppress(docker.errors.DockerException):
                     await self.sys_run_in_executor(
@@ -355,7 +359,7 @@ async def attach(
                 self.sys_bus.fire_event(
                     BusEvent.DOCKER_CONTAINER_STATE_CHANGE,
                     DockerContainerStateEvent(
-                        self.name, state, docker_container.id, int(time())
+                        self.name, state, cast(str, docker_container.id), int(time())
                     ),
                 )
 
@@ -454,7 +458,9 @@ async def check_image(
         expected_arch: CpuArch | None = None,
     ) -> None:
         """Check we have expected image with correct arch."""
-        expected_arch = expected_arch or self.sys_arch.supervisor
+        expected_image_arch = (
+            str(expected_arch) if expected_arch else self.sys_arch.supervisor
+        )
         image_name = f"{expected_image}:{version!s}"
         if self.image == expected_image:
             try:
@@ -472,13 +478,13 @@ async def check_image(
                 image_arch = f"{image_arch}/{image.attrs['Variant']}"
 
             # If we have an image and its the right arch, all set
-            if MAP_ARCH[expected_arch] == image_arch:
+            if MAP_ARCH[expected_image_arch] == image_arch:
                 return
 
         # We're missing the image we need. Stop and clean up what we have then pull the right one
         with suppress(DockerError):
             await self.remove()
-        await self.install(version, expected_image, arch=expected_arch)
+        await self.install(version, expected_image, arch=expected_image_arch)
 
     @Job(
         name="docker_interface_update",
@@ -613,9 +619,7 @@ def run_inside(self, command: str) -> Awaitable[CommandReturn]:
             self.sys_docker.container_run_inside, self.name, command
         )
 
-    async def _validate_trust(
-        self, image_id: str, image: str, version: AwesomeVersion
-    ) -> None:
+    async def _validate_trust(self, image_id: str) -> None:
         """Validate trust of content."""
         checksum = image_id.partition(":")[2]
         return await self.sys_security.verify_own_content(checksum)
@@ -634,4 +638,4 @@ async def check_trust(self) -> None:
         except (docker.errors.DockerException, requests.RequestException):
             return
 
-        await self._validate_trust(image.id, self.image, self.version)
+        await self._validate_trust(cast(str, image.id))