From fa490210cdc35d6651ccbfba746ee9382a4f07d6 Mon Sep 17 00:00:00 2001 From: Stefan Agner Date: Mon, 1 Dec 2025 19:56:05 +0100 Subject: [PATCH] Improve CpuArch type safety across codebase (#6372) Co-authored-by: Claude --- supervisor/addons/build.py | 3 ++- supervisor/addons/model.py | 3 ++- supervisor/arch.py | 44 +++++++++++++++++-------------- supervisor/bootstrap.py | 4 +-- supervisor/coresys.py | 20 +++++++------- supervisor/docker/interface.py | 12 ++++----- tests/addons/test_manager.py | 6 +++-- tests/api/test_addons.py | 14 +++++++--- tests/api/test_store.py | 10 ++++--- tests/store/test_store_manager.py | 10 ++++--- 10 files changed, 73 insertions(+), 53 deletions(-) diff --git a/supervisor/addons/build.py b/supervisor/addons/build.py index ddc2a002e..5fd4204b1 100644 --- a/supervisor/addons/build.py +++ b/supervisor/addons/build.py @@ -20,6 +20,7 @@ from ..const import ( FILE_SUFFIX_CONFIGURATION, META_ADDON, SOCKET_DOCKER, + CpuArch, ) from ..coresys import CoreSys, CoreSysAttributes from ..docker.const import DOCKER_HUB @@ -67,7 +68,7 @@ class AddonBuild(FileConfiguration, CoreSysAttributes): raise RuntimeError() @cached_property - def arch(self) -> str: + def arch(self) -> CpuArch: """Return arch of the add-on.""" return self.sys_arch.match([self.addon.arch]) diff --git a/supervisor/addons/model.py b/supervisor/addons/model.py index b7f8c8bc6..4bf9b5293 100644 --- a/supervisor/addons/model.py +++ b/supervisor/addons/model.py @@ -87,6 +87,7 @@ from ..const import ( AddonBootConfig, AddonStage, AddonStartup, + CpuArch, ) from ..coresys import CoreSys from ..docker.const import Capabilities @@ -548,7 +549,7 @@ class AddonModel(JobGroup, ABC): return self.data.get(ATTR_MACHINE, []) @property - def arch(self) -> str: + def arch(self) -> CpuArch: """Return architecture to use for the addon's image.""" if ATTR_IMAGE in self.data: return self.sys_arch.match(self.data[ATTR_ARCH]) diff --git a/supervisor/arch.py b/supervisor/arch.py index 4d225bd12..1eb888c6d 100644 --- a/supervisor/arch.py +++ b/supervisor/arch.py @@ -4,6 +4,7 @@ import logging from pathlib import Path import platform +from .const import CpuArch from .coresys import CoreSys, CoreSysAttributes from .exceptions import ConfigurationFileError, HassioArchNotFound from .utils.json import read_json_file @@ -12,38 +13,40 @@ _LOGGER: logging.Logger = logging.getLogger(__name__) ARCH_JSON: Path = Path(__file__).parent.joinpath("data/arch.json") -MAP_CPU = { - "armv7": "armv7", - "armv6": "armhf", - "armv8": "aarch64", - "aarch64": "aarch64", - "i686": "i386", - "x86_64": "amd64", +MAP_CPU: dict[str, CpuArch] = { + "armv7": CpuArch.ARMV7, + "armv6": CpuArch.ARMHF, + "armv8": CpuArch.AARCH64, + "aarch64": CpuArch.AARCH64, + "i686": CpuArch.I386, + "x86_64": CpuArch.AMD64, } -class CpuArch(CoreSysAttributes): +class CpuArchManager(CoreSysAttributes): """Manage available architectures.""" def __init__(self, coresys: CoreSys) -> None: """Initialize CPU Architecture handler.""" self.coresys = coresys - self._supported_arch: list[str] = [] - self._supported_set: set[str] = set() - self._default_arch: str + self._supported_arch: list[CpuArch] = [] + self._supported_set: set[CpuArch] = set() + self._default_arch: CpuArch @property - def default(self) -> str: + def default(self) -> CpuArch: """Return system default arch.""" return self._default_arch @property - def supervisor(self) -> str: + def supervisor(self) -> CpuArch: """Return supervisor arch.""" - return self.sys_supervisor.arch or self._default_arch + if self.sys_supervisor.arch: + return CpuArch(self.sys_supervisor.arch) + return self._default_arch @property - def supported(self) -> list[str]: + def supported(self) -> list[CpuArch]: """Return support arch by CPU/Machine.""" return self._supported_arch @@ -65,7 +68,7 @@ class CpuArch(CoreSysAttributes): return # Use configs from arch.json - self._supported_arch.extend(arch_data[self.sys_machine]) + self._supported_arch.extend(CpuArch(a) for a in arch_data[self.sys_machine]) self._default_arch = self.supported[0] # Make sure native support is in supported list @@ -78,14 +81,14 @@ class CpuArch(CoreSysAttributes): """Return True if there is a supported arch by this platform.""" return not self._supported_set.isdisjoint(arch_list) - def match(self, arch_list: list[str]) -> str: + def match(self, arch_list: list[str]) -> CpuArch: """Return best match for this CPU/Platform.""" for self_arch in self.supported: if self_arch in arch_list: return self_arch raise HassioArchNotFound() - def detect_cpu(self) -> str: + def detect_cpu(self) -> CpuArch: """Return the arch type of local CPU.""" cpu = platform.machine() for check, value in MAP_CPU.items(): @@ -96,9 +99,10 @@ class CpuArch(CoreSysAttributes): "Unknown CPU architecture %s, falling back to Supervisor architecture.", cpu, ) - return self.sys_supervisor.arch + return CpuArch(self.sys_supervisor.arch) _LOGGER.warning( "Unknown CPU architecture %s, assuming CPU architecture equals Supervisor architecture.", cpu, ) - return cpu + # Return the cpu string as-is, wrapped in CpuArch (may fail if invalid) + return CpuArch(cpu) diff --git a/supervisor/bootstrap.py b/supervisor/bootstrap.py index 0726bdb84..6a849781d 100644 --- a/supervisor/bootstrap.py +++ b/supervisor/bootstrap.py @@ -13,7 +13,7 @@ from colorlog import ColoredFormatter from .addons.manager import AddonManager from .api import RestAPI -from .arch import CpuArch +from .arch import CpuArchManager from .auth import Auth from .backups.manager import BackupManager from .bus import Bus @@ -71,7 +71,7 @@ async def initialize_coresys() -> CoreSys: coresys.jobs = await JobManager(coresys).load_config() coresys.core = await Core(coresys).post_init() coresys.plugins = await PluginManager(coresys).load_config() - coresys.arch = CpuArch(coresys) + coresys.arch = CpuArchManager(coresys) coresys.auth = await Auth(coresys).load_config() coresys.updater = await Updater(coresys).load_config() coresys.api = RestAPI(coresys) diff --git a/supervisor/coresys.py b/supervisor/coresys.py index daab10f68..d896eb7fd 100644 --- a/supervisor/coresys.py +++ b/supervisor/coresys.py @@ -29,7 +29,7 @@ from .const import ( if TYPE_CHECKING: from .addons.manager import AddonManager from .api import RestAPI - from .arch import CpuArch + from .arch import CpuArchManager from .auth import Auth from .backups.manager import BackupManager from .bus import Bus @@ -78,7 +78,7 @@ class CoreSys: # Internal objects pointers self._docker: DockerAPI | None = None self._core: Core | None = None - self._arch: CpuArch | None = None + self._arch: CpuArchManager | None = None self._auth: Auth | None = None self._homeassistant: HomeAssistant | None = None self._supervisor: Supervisor | None = None @@ -266,17 +266,17 @@ class CoreSys: self._plugins = value @property - def arch(self) -> CpuArch: - """Return CpuArch object.""" + def arch(self) -> CpuArchManager: + """Return CpuArchManager object.""" if self._arch is None: - raise RuntimeError("CpuArch not set!") + raise RuntimeError("CpuArchManager not set!") return self._arch @arch.setter - def arch(self, value: CpuArch) -> None: - """Set a CpuArch object.""" + def arch(self, value: CpuArchManager) -> None: + """Set a CpuArchManager object.""" if self._arch: - raise RuntimeError("CpuArch already set!") + raise RuntimeError("CpuArchManager already set!") self._arch = value @property @@ -733,8 +733,8 @@ class CoreSysAttributes: return self.coresys.plugins @property - def sys_arch(self) -> CpuArch: - """Return CpuArch object.""" + def sys_arch(self) -> CpuArchManager: + """Return CpuArchManager object.""" return self.coresys.arch @property diff --git a/supervisor/docker/interface.py b/supervisor/docker/interface.py index 17453a7d9..60eb9cd33 100644 --- a/supervisor/docker/interface.py +++ b/supervisor/docker/interface.py @@ -52,7 +52,7 @@ from .stats import DockerStats _LOGGER: logging.Logger = logging.getLogger(__name__) -MAP_ARCH: dict[CpuArch | str, str] = { +MAP_ARCH: dict[CpuArch, str] = { CpuArch.ARMV7: "linux/arm/v7", CpuArch.ARMHF: "linux/arm/v6", CpuArch.AARCH64: "linux/arm64", @@ -366,7 +366,7 @@ class DockerInterface(JobGroup, ABC): if not image: raise ValueError("Cannot pull without an image!") - image_arch = str(arch) if arch else self.sys_arch.supervisor + image_arch = arch or self.sys_arch.supervisor listener: EventListener | None = None _LOGGER.info("Downloading docker image %s with tag %s.", image, version) @@ -603,9 +603,7 @@ class DockerInterface(JobGroup, ABC): expected_cpu_arch: CpuArch | None = None, ) -> None: """Check we have expected image with correct arch.""" - expected_image_cpu_arch = ( - str(expected_cpu_arch) if expected_cpu_arch else self.sys_arch.supervisor - ) + arch = expected_cpu_arch or self.sys_arch.supervisor image_name = f"{expected_image}:{version!s}" if self.image == expected_image: try: @@ -623,7 +621,7 @@ class DockerInterface(JobGroup, ABC): # If we have an image and its the right arch, all set # It seems that newer Docker version return a variant for arm64 images. # Make sure we match linux/arm64 and linux/arm64/v8. - expected_image_arch = MAP_ARCH[expected_image_cpu_arch] + expected_image_arch = MAP_ARCH[arch] if image_arch.startswith(expected_image_arch): return _LOGGER.info( @@ -636,7 +634,7 @@ class DockerInterface(JobGroup, ABC): # 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_image_cpu_arch) + await self.install(version, expected_image, arch=arch) @Job( name="docker_interface_update", diff --git a/tests/addons/test_manager.py b/tests/addons/test_manager.py index 5f800da22..d574bdeb3 100644 --- a/tests/addons/test_manager.py +++ b/tests/addons/test_manager.py @@ -10,7 +10,7 @@ from awesomeversion import AwesomeVersion import pytest from supervisor.addons.addon import Addon -from supervisor.arch import CpuArch +from supervisor.arch import CpuArchManager from supervisor.config import CoreConfig from supervisor.const import AddonBoot, AddonStartup, AddonState, BusEvent from supervisor.coresys import CoreSys @@ -54,7 +54,9 @@ async def fixture_mock_arch_disk() -> AsyncGenerator[None]: """Mock supported arch and disk space.""" with ( patch("shutil.disk_usage", return_value=(42, 42, 2 * (1024.0**3))), - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), ): yield diff --git a/tests/api/test_addons.py b/tests/api/test_addons.py index 1ab7bb113..aa0071504 100644 --- a/tests/api/test_addons.py +++ b/tests/api/test_addons.py @@ -9,7 +9,7 @@ import pytest from supervisor.addons.addon import Addon from supervisor.addons.build import AddonBuild -from supervisor.arch import CpuArch +from supervisor.arch import CpuArchManager from supervisor.const import AddonState from supervisor.coresys import CoreSys from supervisor.docker.addon import DockerAddon @@ -236,7 +236,9 @@ async def test_api_addon_rebuild_healthcheck( patch.object(AddonBuild, "is_valid", return_value=True), patch.object(DockerAddon, "is_running", return_value=False), patch.object(Addon, "need_build", new=PropertyMock(return_value=True)), - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), patch.object(DockerAddon, "run", new=container_events_task), patch.object( coresys.docker, @@ -308,7 +310,9 @@ async def test_api_addon_rebuild_force( patch.object( Addon, "need_build", new=PropertyMock(return_value=False) ), # Image-based - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), ): resp = await api_client.post("/addons/local_ssh/rebuild") @@ -326,7 +330,9 @@ async def test_api_addon_rebuild_force( patch.object( Addon, "need_build", new=PropertyMock(return_value=False) ), # Image-based - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), patch.object(DockerAddon, "run", new=container_events_task), patch.object( coresys.docker, diff --git a/tests/api/test_store.py b/tests/api/test_store.py index 332d8a03b..8e9414edf 100644 --- a/tests/api/test_store.py +++ b/tests/api/test_store.py @@ -10,7 +10,7 @@ from awesomeversion import AwesomeVersion import pytest from supervisor.addons.addon import Addon -from supervisor.arch import CpuArch +from supervisor.arch import CpuArchManager from supervisor.backups.manager import BackupManager from supervisor.config import CoreConfig from supervisor.const import AddonState, CoreState @@ -191,7 +191,9 @@ async def test_api_store_update_healthcheck( patch.object(DockerAddon, "run", new=container_events_task), patch.object(DockerInterface, "install"), patch.object(DockerAddon, "is_running", return_value=False), - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), ): resp = await api_client.post(f"/store/addons/{TEST_ADDON_SLUG}/update") @@ -548,7 +550,9 @@ async def test_api_store_addons_addon_availability_arch_not_supported( coresys.addons.data.user[addon_obj.slug] = {"version": AwesomeVersion("0.0.1")} # Mock the system architecture to be different - with patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])): + with patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ): resp = await api_client.request( api_method, f"/store/addons/{addon_obj.slug}/{api_action}" ) diff --git a/tests/store/test_store_manager.py b/tests/store/test_store_manager.py index 568ebfe1f..4a328a928 100644 --- a/tests/store/test_store_manager.py +++ b/tests/store/test_store_manager.py @@ -9,7 +9,7 @@ from awesomeversion import AwesomeVersion import pytest from supervisor.addons.addon import Addon -from supervisor.arch import CpuArch +from supervisor.arch import CpuArchManager from supervisor.backups.manager import BackupManager from supervisor.coresys import CoreSys from supervisor.exceptions import AddonNotSupportedError, StoreJobError @@ -163,7 +163,9 @@ async def test_update_unavailable_addon( with ( patch.object(BackupManager, "do_backup_partial") as backup, patch.object(AddonStore, "data", new=PropertyMock(return_value=addon_config)), - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), patch.object(CoreSys, "machine", new=PropertyMock(return_value="qemux86-64")), patch.object( HomeAssistant, @@ -219,7 +221,9 @@ async def test_install_unavailable_addon( with ( patch.object(AddonStore, "data", new=PropertyMock(return_value=addon_config)), - patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])), + patch.object( + CpuArchManager, "supported", new=PropertyMock(return_value=["amd64"]) + ), patch.object(CoreSys, "machine", new=PropertyMock(return_value="qemux86-64")), patch.object( HomeAssistant,