Add /supervisor/available_updates endpoint (#3300)

supervisor/api/__init__.py

@@ -284,6 +284,9 @@ class RestAPI(CoreSysAttributes):
 
         self.webapp.add_routes(
             [
+                web.get(
+                    "/supervisor/available_updates", api_supervisor.available_updates
+                ),
                 web.get("/supervisor/ping", api_supervisor.ping),
                 web.get("/supervisor/info", api_supervisor.info),
                 web.get("/supervisor/stats", api_supervisor.stats),

supervisor/api/const.py

@@ -10,3 +10,6 @@ ATTR_STARTUP_TIME = "startup_time"
 ATTR_USE_NTP = "use_ntp"
 ATTR_USE_RTC = "use_rtc"
 ATTR_APPARMOR_VERSION = "apparmor_version"
+ATTR_PANEL_PATH = "panel_path"
+ATTR_UPDATE_TYPE = "update_type"
+ATTR_AVAILABLE_UPDATES = "available_updates"

supervisor/api/supervisor.py

@@ -50,6 +50,7 @@ from ..coresys import CoreSysAttributes
 from ..exceptions import APIError
 from ..utils.validate import validate_timezone
 from ..validate import repositories, version_tag, wait_boot
+from .const import ATTR_AVAILABLE_UPDATES, ATTR_PANEL_PATH, ATTR_UPDATE_TYPE
 from .utils import api_process, api_process_raw, api_validate
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -239,3 +240,53 @@ class APISupervisor(CoreSysAttributes):
     def logs(self, request: web.Request) -> Awaitable[bytes]:
         """Return supervisor Docker logs."""
         return self.sys_supervisor.logs()
+
+    @api_process
+    async def available_updates(self, request: web.Request) -> dict[str, Any]:
+        """Return a list of items with available updates."""
+        available_updates = []
+
+        # Core
+        if self.sys_homeassistant.need_update:
+            available_updates.append(
+                {
+                    ATTR_UPDATE_TYPE: "core",
+                    ATTR_PANEL_PATH: "/update-available/core",
+                    ATTR_VERSION_LATEST: self.sys_homeassistant.latest_version,
+                }
+            )
+
+        # Supervisor
+        if self.sys_supervisor.need_update:
+            available_updates.append(
+                {
+                    ATTR_UPDATE_TYPE: "supervisor",
+                    ATTR_PANEL_PATH: "/update-available/supervisor",
+                    ATTR_VERSION_LATEST: self.sys_supervisor.latest_version,
+                }
+            )
+
+        # OS
+        if self.sys_os.need_update:
+            available_updates.append(
+                {
+                    ATTR_UPDATE_TYPE: "os",
+                    ATTR_PANEL_PATH: "/update-available/os",
+                    ATTR_VERSION_LATEST: self.sys_os.latest_version,
+                }
+            )
+
+        # Add-ons
+        available_updates.extend(
+            {
+                ATTR_UPDATE_TYPE: "addon",
+                ATTR_NAME: addon.name,
+                ATTR_ICON: f"/addons/{addon.slug}/icon" if addon.with_icon else None,
+                ATTR_PANEL_PATH: f"/update-available/{addon.slug}",
+                ATTR_VERSION_LATEST: addon.latest_version,
+            }
+            for addon in self.sys_addons.installed
+            if addon.need_update
+        )
+
+        return {ATTR_AVAILABLE_UPDATES: available_updates}

tests/api/test_supervisor.py

@@ -1,9 +1,12 @@
 """Test Supervisor API."""
-
+# pylint: disable=protected-access
 import pytest
 
+from supervisor.api.const import ATTR_AVAILABLE_UPDATES
 from supervisor.coresys import CoreSys
 
+from tests.const import TEST_ADDON_SLUG
+
 
 @pytest.mark.asyncio
 async def test_api_supervisor_options_debug(api_client, coresys: CoreSys):
@@ -13,3 +16,49 @@ async def test_api_supervisor_options_debug(api_client, coresys: CoreSys):
     await api_client.post("/supervisor/options", json={"debug": True})
 
     assert coresys.config.debug
+
+
+@pytest.mark.asyncio
+async def test_api_supervisor_available_updates(
+    install_addon_ssh,
+    api_client,
+    coresys: CoreSys,
+):
+    """Test available_updates."""
+    installed_addon = coresys.addons.get(TEST_ADDON_SLUG)
+    installed_addon.persist["version"] = "1.2.3"
+
+    async def available_updates():
+        return (await (await api_client.get("/supervisor/available_updates")).json())[
+            "data"
+        ][ATTR_AVAILABLE_UPDATES]
+
+    updates = await available_updates()
+    assert len(updates) == 1
+    assert updates[-1] == {
+        "icon": None,
+        "name": "Terminal & SSH",
+        "panel_path": "/update-available/local_ssh",
+        "update_type": "addon",
+        "version_latest": "9.2.1",
+    }
+
+    coresys.updater._data["hassos"] = "321"
+    coresys.os._version = "123"
+    updates = await available_updates()
+    assert len(updates) == 2
+    assert updates[0] == {
+        "panel_path": "/update-available/os",
+        "update_type": "os",
+        "version_latest": "321",
+    }
+
+    coresys.updater._data["homeassistant"] = "321"
+    coresys.homeassistant.version = "123"
+    updates = await available_updates()
+    assert len(updates) == 3
+    assert updates[0] == {
+        "panel_path": "/update-available/core",
+        "update_type": "core",
+        "version_latest": "321",
+    }

tests/conftest.py

@@ -42,7 +42,7 @@ def docker() -> DockerAPI:
     """Mock DockerAPI."""
     images = [MagicMock(tags=["ghcr.io/home-assistant/amd64-hassio-supervisor:latest"])]
 
-    with patch("docker.DockerClient", return_value=MagicMock()), patch(
+    with patch("supervisor.docker.DockerClient", return_value=MagicMock()), patch(
         "supervisor.docker.DockerAPI.images", return_value=MagicMock()
     ), patch("supervisor.docker.DockerAPI.containers", return_value=MagicMock()), patch(
         "supervisor.docker.DockerAPI.api", return_value=MagicMock()