Bind libraries to different files and refactor images.pull

Author: mdegat01

supervisor/bus.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from asyncio import Task
 from collections.abc import Callable, Coroutine
 import logging
 from typing import Any
@@ -38,11 +39,13 @@ def register_event(
         self._listeners.setdefault(event, []).append(listener)
         return listener
 
-    def fire_event(self, event: BusEvent, reference: Any) -> None:
+    def fire_event(self, event: BusEvent, reference: Any) -> list[Task]:
         """Fire an event to the bus."""
         _LOGGER.debug("Fire event '%s' with '%s'", event, reference)
+        tasks: list[Task] = []
         for listener in self._listeners.get(event, []):
-            self.sys_create_task(listener.callback(reference))
+            tasks.append(self.sys_create_task(listener.callback(reference)))
+        return tasks
 
     def remove_listener(self, listener: EventListener) -> None:
         """Unregister an listener."""

supervisor/docker/interface.py

@@ -452,11 +452,23 @@ async def process_pull_image_log(reference: PullLogEntry) -> None:
                     suggestions=[SuggestionType.REGISTRY_LOGIN],
                 )
                 raise DockerHubRateLimitExceeded(_LOGGER.error) from err
+            await async_capture_exception(err)
+            raise DockerError(
+                f"Can't install {image}:{version!s}: {err}", _LOGGER.error
+            ) from err
+        except aiodocker.DockerError as err:
+            if err.status == HTTPStatus.TOO_MANY_REQUESTS:
+                self.sys_resolution.create_issue(
+                    IssueType.DOCKER_RATELIMIT,
+                    ContextType.SYSTEM,
+                    suggestions=[SuggestionType.REGISTRY_LOGIN],
+                )
+                raise DockerHubRateLimitExceeded(_LOGGER.error) from err
+            await async_capture_exception(err)
             raise DockerError(
                 f"Can't install {image}:{version!s}: {err}", _LOGGER.error
             ) from err
         except (
-            aiodocker.DockerError,
             docker.errors.DockerException,
             requests.RequestException,
         ) as err:

supervisor/docker/manager.py

@@ -17,6 +17,7 @@
 
 import aiodocker
 from aiodocker.images import DockerImages
+from aiohttp import ClientSession, ClientTimeout, UnixConnector
 import attr
 from awesomeversion import AwesomeVersion, AwesomeVersionCompareException
 from docker import errors as docker_errors
@@ -211,7 +212,10 @@ def __init__(self, coresys: CoreSys):
         # We keep both until we can fully refactor to aiodocker
         self._dockerpy: DockerClient | None = None
         self.docker: aiodocker.Docker = aiodocker.Docker(
-            url=f"unix:/{str(SOCKET_DOCKER)}", api_version="auto"
+            url="unix://localhost",  # dummy hostname for URL composition
+            connector=(connector := UnixConnector(SOCKET_DOCKER.as_posix())),
+            session=ClientSession(connector=connector, timeout=ClientTimeout(900)),
+            api_version="auto",
         )
 
         self._network: DockerNetwork | None = None
@@ -221,11 +225,13 @@ def __init__(self, coresys: CoreSys):
 
     async def post_init(self) -> Self:
         """Post init actions that must be done in event loop."""
+        # Use /var/run/docker.sock for this one so aiodocker and dockerpy don't
+        # share the same handle. Temporary fix while refactoring this client out
         self._dockerpy = await asyncio.get_running_loop().run_in_executor(
             None,
             partial(
                 DockerClient,
-                base_url=f"unix:/{str(SOCKET_DOCKER)}",
+                base_url=f"unix://var{SOCKET_DOCKER.as_posix()}",
                 version="auto",
                 timeout=900,
             ),
@@ -433,20 +439,16 @@ async def pull_image(
         raises only if the get fails afterwards. Additionally it fires progress reports for the pull
         on the bus so listeners can use that to update status for users.
         """
-
-        def api_pull():
-            pull_log = self.dockerpy.api.pull(
-                repository, tag=tag, platform=platform, stream=True, decode=True
+        async for e in self.images.pull(
+            repository, tag=tag, platform=platform, stream=True
+        ):
+            entry = PullLogEntry.from_pull_log_dict(job_id, e)
+            if entry.error:
+                raise entry.exception
+            await asyncio.gather(
+                *self.sys_bus.fire_event(BusEvent.DOCKER_IMAGE_PULL_UPDATE, entry)
             )
-            for e in pull_log:
-                entry = PullLogEntry.from_pull_log_dict(job_id, e)
-                if entry.error:
-                    raise entry.exception
-                self.sys_loop.call_soon_threadsafe(
-                    self.sys_bus.fire_event, BusEvent.DOCKER_IMAGE_PULL_UPDATE, entry
-                )
 
-        await self.sys_run_in_executor(api_pull)
         sep = "@" if tag.startswith("sha256:") else ":"
         return await self.images.inspect(f"{repository}{sep}{tag}")
 

tests/addons/test_addon.py

@@ -9,7 +9,7 @@
 
 import aiodocker
 from awesomeversion import AwesomeVersion
-from docker.errors import DockerException, NotFound
+from docker.errors import APIError, DockerException, NotFound
 import pytest
 from securetar import SecureTarFile
 
@@ -931,7 +931,7 @@ async def test_addon_loads_missing_image(
 
 @pytest.mark.parametrize(
     "pull_image_exc",
-    [DockerException(), aiodocker.DockerError(400, {"message": "error"})],
+    [APIError("error"), aiodocker.DockerError(400, {"message": "error"})],
 )
 @pytest.mark.usefixtures("container", "mock_amd64_arch_supported")
 async def test_addon_load_succeeds_with_docker_errors(
@@ -973,7 +973,7 @@ async def test_addon_load_succeeds_with_docker_errors(
     caplog.clear()
     with patch.object(DockerAPI, "pull_image", side_effect=pull_image_exc):
         await install_addon_ssh.load()
-    assert "Unknown error with test/amd64-addon-ssh:9.2.1" in caplog.text
+    assert "Can't install test/amd64-addon-ssh:9.2.1:" in caplog.text
 
 
 async def test_addon_manual_only_boot(coresys: CoreSys, install_addon_example: Addon):

tests/api/test_homeassistant.py

@@ -19,7 +19,7 @@
 from supervisor.homeassistant.module import HomeAssistant
 
 from tests.api import common_test_api_advanced_logs
-from tests.common import load_json_fixture
+from tests.common import AsyncIterator, load_json_fixture
 
 
 @pytest.mark.parametrize("legacy_route", [True, False])
@@ -283,9 +283,9 @@ async def test_api_progress_updates_home_assistant_update(
     """Test progress updates sent to Home Assistant for updates."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
     coresys.core.set_state(CoreState.RUNNING)
-    coresys.docker.dockerpy.api.pull.return_value = load_json_fixture(
-        "docker_pull_image_log.json"
-    )
+
+    logs = load_json_fixture("docker_pull_image_log.json")
+    coresys.docker.images.pull.return_value = AsyncIterator(logs)
     coresys.homeassistant.version = AwesomeVersion("2025.8.0")
 
     with (

tests/api/test_store.py

@@ -24,7 +24,7 @@
 from supervisor.store.addon import AddonStore
 from supervisor.store.repository import Repository
 
-from tests.common import load_json_fixture
+from tests.common import AsyncIterator, load_json_fixture
 from tests.const import TEST_ADDON_SLUG
 
 REPO_URL = "https://github.com/awesome-developer/awesome-repo"
@@ -732,9 +732,10 @@ async def test_api_progress_updates_addon_install_update(
     """Test progress updates sent to Home Assistant for installs/updates."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
     coresys.core.set_state(CoreState.RUNNING)
-    coresys.docker.dockerpy.api.pull.return_value = load_json_fixture(
-        "docker_pull_image_log.json"
-    )
+
+    logs = load_json_fixture("docker_pull_image_log.json")
+    coresys.docker.images.pull.return_value = AsyncIterator(logs)
+
     coresys.arch._supported_arch = ["amd64"]  # pylint: disable=protected-access
     install_addon_example.data_store["version"] = AwesomeVersion("2.0.0")
 

tests/api/test_supervisor.py

@@ -19,7 +19,7 @@
 from supervisor.updater import Updater
 
 from tests.api import common_test_api_advanced_logs
-from tests.common import load_json_fixture
+from tests.common import AsyncIterator, load_json_fixture
 from tests.dbus_service_mocks.base import DBusServiceMock
 from tests.dbus_service_mocks.os_agent import OSAgent as OSAgentService
 
@@ -332,9 +332,9 @@ async def test_api_progress_updates_supervisor_update(
     """Test progress updates sent to Home Assistant for updates."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
     coresys.core.set_state(CoreState.RUNNING)
-    coresys.docker.dockerpy.api.pull.return_value = load_json_fixture(
-        "docker_pull_image_log.json"
-    )
+
+    logs = load_json_fixture("docker_pull_image_log.json")
+    coresys.docker.images.pull.return_value = AsyncIterator(logs)
 
     with (
         patch.object(

tests/common.py

@@ -1,13 +1,14 @@
 """Common test functions."""
 
 import asyncio
+from collections.abc import Sequence
 from datetime import datetime
 from functools import partial
 from importlib import import_module
 from inspect import getclosurevars
 import json
 from pathlib import Path
-from typing import Any
+from typing import Any, Self
 
 from dbus_fast.aio.message_bus import MessageBus
 
@@ -145,3 +146,22 @@ async def __aenter__(self):
 
     async def __aexit__(self, exc_type, exc, tb):
         """Exit the context manager."""
+
+
+class AsyncIterator:
+    """Make list/fixture into async iterator for test mocks."""
+
+    def __init__(self, seq: Sequence[Any]) -> None:
+        """Initialize with sequence."""
+        self.iter = iter(seq)
+
+    def __aiter__(self) -> Self:
+        """Implement aiter."""
+        return self
+
+    async def __anext__(self) -> Any:
+        """Return next in sequence."""
+        try:
+            return next(self.iter)
+        except StopIteration:
+            raise StopAsyncIteration() from None

tests/conftest.py

@@ -56,6 +56,7 @@
 from supervisor.utils.dt import utcnow
 
 from .common import (
+    AsyncIterator,
     MockResponse,
     load_binary_fixture,
     load_fixture,
@@ -125,14 +126,8 @@ async def docker() -> DockerAPI:
         patch(
             "supervisor.docker.manager.DockerAPI.containers", return_value=MagicMock()
         ),
-        patch(
-            "supervisor.docker.manager.DockerAPI.api",
-            return_value=(api_mock := MagicMock()),
-        ),
-        patch(
-            "supervisor.docker.manager.DockerAPI.info",
-            return_value=MagicMock(),
-        ),
+        patch("supervisor.docker.manager.DockerAPI.api", return_value=MagicMock()),
+        patch("supervisor.docker.manager.DockerAPI.info", return_value=MagicMock()),
         patch("supervisor.docker.manager.DockerAPI.unload"),
         patch("supervisor.docker.manager.aiodocker.Docker", return_value=MagicMock()),
         patch(
@@ -153,13 +148,12 @@ async def docker() -> DockerAPI:
             {"stream": "Loaded image: test:latest\n"}
         ]
 
+        docker_images.pull.return_value = AsyncIterator([{}])
+
         docker_obj.info.logging = "journald"
         docker_obj.info.storage = "overlay2"
         docker_obj.info.version = AwesomeVersion("1.0.0")
 
-        # Need an iterable for logs
-        api_mock.pull.return_value = []
-
         yield docker_obj