Allow runtime pinning for plugin update versions

Add plugin-level runtime pinning when a specific plugin version is
requested so automatic plugin updates can be temporarily skipped within
the current Supervisor runtime. While at it, centralize auto-update
decision and logging in PluginBase. Scheduled and job-gated plugin
updates now call through the shared auto_update path with coverage for
the new pinned behavior.

supervisor/jobs/decorator.py

@@ -450,24 +450,27 @@ class Job(CoreSysAttributes):
                 f"'{method_name}' blocked from execution, unsupported system architecture"
             )
 
-        if JobCondition.PLUGINS_UPDATED in used_conditions and (
-            out_of_date := [
+        if JobCondition.PLUGINS_UPDATED in used_conditions:
+            out_of_date = [
                 plugin
                 for plugin in coresys.sys_plugins.all_plugins
                 if plugin.need_update
             ]
-        ):
-            errors = await asyncio.gather(
-                *[plugin.update() for plugin in out_of_date], return_exceptions=True
-            )
-
-            if update_failures := [
-                out_of_date[i].slug for i in range(len(errors)) if errors[i] is not None
-            ]:
-                raise JobConditionException(
-                    f"'{method_name}' blocked from execution, was unable to update plugin(s) {', '.join(update_failures)} and all plugins must be up to date first"
+            if out_of_date:
+                errors = await asyncio.gather(
+                    *[plugin.auto_update() for plugin in out_of_date],
+                    return_exceptions=True,
                 )
 
+                if update_failures := [
+                    out_of_date[i].slug
+                    for i in range(len(errors))
+                    if errors[i] is not None
+                ]:
+                    raise JobConditionException(
+                        f"'{method_name}' blocked from execution, was unable to update plugin(s) {', '.join(update_failures)} and all plugins must be up to date first"
+                    )
+
         if (
             JobCondition.MOUNT_AVAILABLE in used_conditions
             and HostFeature.MOUNT not in coresys.sys_host.features

supervisor/misc/tasks.py

@@ -267,61 +267,27 @@ class Tasks(CoreSysAttributes):
     @Job(name="tasks_update_cli", conditions=PLUGIN_AUTO_UPDATE_CONDITIONS)
     async def _update_cli(self):
         """Check and run update of cli."""
-        if not self.sys_plugins.cli.need_update:
-            return
-
-        _LOGGER.info(
-            "Found new cli version %s, updating", self.sys_plugins.cli.latest_version
-        )
-        await self.sys_plugins.cli.update()
+        await self.sys_plugins.cli.auto_update()
 
     @Job(name="tasks_update_dns", conditions=PLUGIN_AUTO_UPDATE_CONDITIONS)
     async def _update_dns(self):
         """Check and run update of CoreDNS plugin."""
-        if not self.sys_plugins.dns.need_update:
-            return
-
-        _LOGGER.info(
-            "Found new CoreDNS plugin version %s, updating",
-            self.sys_plugins.dns.latest_version,
-        )
-        await self.sys_plugins.dns.update()
+        await self.sys_plugins.dns.auto_update()
 
     @Job(name="tasks_update_audio", conditions=PLUGIN_AUTO_UPDATE_CONDITIONS)
     async def _update_audio(self):
         """Check and run update of PulseAudio plugin."""
-        if not self.sys_plugins.audio.need_update:
-            return
-
-        _LOGGER.info(
-            "Found new PulseAudio plugin version %s, updating",
-            self.sys_plugins.audio.latest_version,
-        )
-        await self.sys_plugins.audio.update()
+        await self.sys_plugins.audio.auto_update()
 
     @Job(name="tasks_update_observer", conditions=PLUGIN_AUTO_UPDATE_CONDITIONS)
     async def _update_observer(self):
         """Check and run update of Observer plugin."""
-        if not self.sys_plugins.observer.need_update:
-            return
-
-        _LOGGER.info(
-            "Found new Observer plugin version %s, updating",
-            self.sys_plugins.observer.latest_version,
-        )
-        await self.sys_plugins.observer.update()
+        await self.sys_plugins.observer.auto_update()
 
     @Job(name="tasks_update_multicast", conditions=PLUGIN_AUTO_UPDATE_CONDITIONS)
     async def _update_multicast(self):
         """Check and run update of multicast."""
-        if not self.sys_plugins.multicast.need_update:
-            return
-
-        _LOGGER.info(
-            "Found new Multicast version %s, updating",
-            self.sys_plugins.multicast.latest_version,
-        )
-        await self.sys_plugins.multicast.update()
+        await self.sys_plugins.multicast.auto_update()
 
     async def _watchdog_observer_application(self):
         """Check running state of application and rebuild if they is not response."""

supervisor/plugins/base.py

@@ -27,6 +27,11 @@ class PluginBase(ABC, FileConfiguration, CoreSysAttributes):
     slug: str
     instance: DockerInterface
 
+    def __init__(self, *args, **kwargs) -> None:
+        """Initialize plugin base state."""
+        super().__init__(*args, **kwargs)
+        self._runtime_update_pin: AwesomeVersion | None = None
+
     @property
     def version(self) -> AwesomeVersion | None:
         """Return current version of the plugin."""
@@ -202,6 +207,24 @@ class PluginBase(ABC, FileConfiguration, CoreSysAttributes):
         self.image = self.default_image
         await self.save_data()
 
+    async def auto_update(self) -> None:
+        """Automatically update system plugin if needed."""
+        if not self.need_update:
+            return
+
+        if self._runtime_update_pin is not None:
+            _LOGGER.warning(
+                "Skipping auto-update of %s plugin due to runtime pin", self.slug
+            )
+            return
+
+        _LOGGER.info(
+            "Plugin %s is not up-to-date, latest version %s, updating",
+            self.slug,
+            self.latest_version,
+        )
+        await self.update()
+
     async def update(self, version: str | None = None) -> None:
         """Update system plugin."""
         to_version = AwesomeVersion(version) if version else self.latest_version
@@ -211,15 +234,24 @@ class PluginBase(ABC, FileConfiguration, CoreSysAttributes):
                 _LOGGER.error,
             )
 
+        # Pin to particular version until next Supervisor restart if user passed it
+        # explicitly. This is useful for regression testing etc.
+        pin_after_update: AwesomeVersion | None = to_version if version else None
+
         old_image = self.image
 
         if to_version == self.version:
             _LOGGER.warning(
                 "Version %s is already installed for %s", to_version, self.slug
             )
+            self._runtime_update_pin = pin_after_update
             return
 
-        await self.instance.update(to_version, image=self.default_image)
+        try:
+            await self.instance.update(to_version, image=self.default_image)
+        finally:
+            self._runtime_update_pin = pin_after_update
+
         self.version = self.instance.version or to_version
         self.image = self.default_image
         await self.save_data()

supervisor/plugins/manager.py

@@ -91,14 +91,8 @@ class PluginManager(CoreSysAttributes):
             # Check if need an update
             if not plugin.need_update:
                 continue
-
-            _LOGGER.info(
-                "Plugin %s is not up-to-date, latest version %s, updating",
-                plugin.slug,
-                plugin.latest_version,
-            )
             try:
-                await plugin.update()
+                await plugin.auto_update()
             except HassioError as ex:
                 _LOGGER.error(
                     "Can't update %s to %s: %s",

tests/jobs/test_job_decorator.py

@@ -6,6 +6,7 @@ from unittest.mock import ANY, AsyncMock, Mock, PropertyMock, patch
 from uuid import uuid4
 
 from aiohttp.client_exceptions import ClientError
+from awesomeversion import AwesomeVersion
 import pytest
 import time_machine
 
@@ -25,6 +26,7 @@ from supervisor.jobs.decorator import Job, JobCondition
 from supervisor.jobs.job_group import JobGroup
 from supervisor.os.manager import OSManager
 from supervisor.plugins.audio import PluginAudio
+from supervisor.plugins.dns import PluginDns
 from supervisor.resolution.const import UnhealthyReason, UnsupportedReason
 from supervisor.supervisor import Supervisor
 from supervisor.utils.dt import utcnow
@@ -528,6 +530,39 @@ async def test_plugins_updated(coresys: CoreSys):
         assert await test.execute()
 
 
+async def test_plugins_updated_skips_runtime_pinned_plugin(
+    coresys: CoreSys, caplog: pytest.LogCaptureFixture
+):
+    """Test PLUGINS_UPDATED skips runtime pinned plugins."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+
+        @Job(
+            name="test_plugins_updated_skips_runtime_pinned_plugin_execute",
+            conditions=[JobCondition.PLUGINS_UPDATED],
+        )
+        async def execute(self) -> bool:
+            """Execute the class method."""
+            return True
+
+    test = TestClass(coresys)
+    coresys.plugins.dns._runtime_update_pin = AwesomeVersion("2025.08.0")
+
+    with (
+        patch.object(PluginDns, "need_update", new=PropertyMock(return_value=True)),
+        patch.object(PluginDns, "update") as update,
+    ):
+        assert await test.execute()
+
+    update.assert_not_called()
+    assert "Skipping auto-update of dns plugin due to runtime pin" in caplog.text
+
+
 async def test_auto_update(coresys: CoreSys):
     """Test the auto update decorator."""
 

tests/misc/test_tasks.py

@@ -16,6 +16,7 @@ from supervisor.homeassistant.api import HomeAssistantAPI
 from supervisor.homeassistant.const import LANDINGPAGE
 from supervisor.homeassistant.core import HomeAssistantCore
 from supervisor.misc.tasks import Tasks
+from supervisor.plugins.dns import PluginDns
 from supervisor.supervisor import Supervisor
 
 from tests.common import MockResponse, get_fixture_path
@@ -178,7 +179,7 @@ async def test_reload_updater_triggers_supervisor_update(
     tasks: Tasks, coresys: CoreSys, mock_update_data: MockResponse
 ):
     """Test an updater reload triggers a supervisor update if there is one."""
-    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    coresys.hardware.disk.get_disk_free_space = lambda path: 5000
     await coresys.core.set_state(CoreState.RUNNING)
 
     with (
@@ -208,7 +209,7 @@ async def test_reload_updater_triggers_supervisor_update(
 async def test_core_backup_cleanup(tasks: Tasks, coresys: CoreSys):
     """Test core backup task cleans up old backup files."""
     await coresys.core.set_state(CoreState.RUNNING)
-    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    coresys.hardware.disk.get_disk_free_space = lambda path: 5000
 
     # Put an old and new backup in folder
     copy(get_fixture_path("backup_example.tar"), coresys.config.path_core_backup)
@@ -261,3 +262,24 @@ async def test_update_addons_auto_update_success(
                 "backup": True,
             }
         )
+
+
+@pytest.mark.usefixtures("no_job_throttle", "supervisor_internet")
+async def test_update_dns_skips_when_runtime_pinned(
+    tasks: Tasks,
+    coresys: CoreSys,
+    caplog: pytest.LogCaptureFixture,
+):
+    """Test that DNS auto update is skipped when runtime pin is set."""
+    await coresys.core.set_state(CoreState.RUNNING)
+    coresys.hardware.disk.get_disk_free_space = lambda path: 5000
+    coresys.plugins.dns._runtime_update_pin = AwesomeVersion("2025.08.0")
+
+    with (
+        patch.object(PluginDns, "need_update", new=PropertyMock(return_value=True)),
+        patch.object(PluginDns, "update") as update,
+    ):
+        await tasks._update_dns()
+
+    update.assert_not_called()
+    assert "Skipping auto-update of dns plugin due to runtime pin" in caplog.text

tests/plugins/test_plugin_base.py

@@ -392,3 +392,53 @@ async def test_default_image_fallback(coresys: CoreSys, plugin: PluginBase):
     """Test default image falls back to hard-coded constant if we fail to fetch version file."""
     assert getattr(coresys.updater, f"image_{plugin.slug}") is None
     assert plugin.default_image == f"ghcr.io/home-assistant/amd64-hassio-{plugin.slug}"
+
+
+async def test_runtime_pin_set_on_specific_manual_update(coresys: CoreSys):
+    """Test plugin update pins runtime when specific version is requested."""
+    plugin = coresys.plugins.cli
+    plugin.version = AwesomeVersion("2025.01.0")
+    plugin._runtime_update_pin = None
+
+    with (
+        patch.object(
+            type(plugin),
+            "latest_version",
+            new=PropertyMock(return_value=AwesomeVersion("2025.03.0")),
+        ),
+        patch.object(type(plugin.instance), "update") as update,
+        patch.object(type(plugin.instance), "cleanup"),
+        patch.object(type(plugin), "start"),
+        patch.object(type(plugin), "save_data"),
+    ):
+        await PluginBase.update(plugin, "2025.02.0")
+
+    update.assert_called_once_with(
+        AwesomeVersion("2025.02.0"), image=plugin.default_image
+    )
+    assert plugin._runtime_update_pin == AwesomeVersion("2025.02.0")
+
+
+async def test_runtime_pin_set_on_update_to_latest(coresys: CoreSys):
+    """Test plugin update pins runtime when targeting latest version explicitly."""
+    plugin = coresys.plugins.cli
+    plugin.version = AwesomeVersion("2025.01.0")
+    plugin._runtime_update_pin = AwesomeVersion("2025.00.0")
+
+    with (
+        patch.object(
+            type(plugin),
+            "latest_version",
+            new=PropertyMock(return_value=AwesomeVersion("2025.03.0")),
+        ),
+        patch.object(type(plugin.instance), "update") as update,
+        patch.object(type(plugin.instance), "cleanup"),
+        patch.object(type(plugin), "start"),
+        patch.object(type(plugin), "save_data"),
+    ):
+        await PluginBase.update(plugin, "2025.03.0")
+
+    update.assert_called_once_with(
+        AwesomeVersion("2025.03.0"), image=plugin.default_image
+    )
+    assert plugin._runtime_update_pin == AwesomeVersion("2025.03.0")