Bump coverage from 7.10.1 to 7.10.2 (#6062)

Bumps [coverage](https://github.com/nedbat/coveragepy) from 7.10.1 to 7.10.2.
- [Release notes](https://github.com/nedbat/coveragepy/releases)
- [Changelog](https://github.com/nedbat/coveragepy/blob/master/CHANGES.rst)
- [Commits](https://github.com/nedbat/coveragepy/compare/7.10.1...7.10.2)

---
updated-dependencies:
- dependency-name: coverage
  dependency-version: 7.10.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/builder.yml

@@ -106,7 +106,7 @@ jobs:
 
       - name: Build wheels
         if: needs.init.outputs.requirements == 'true'
-        uses: home-assistant/wheels@2025.03.0
+        uses: home-assistant/wheels@2025.07.0
         with:
           abi: cp313
           tag: musllinux_1_2
@@ -131,7 +131,7 @@ jobs:
 
       - name: Install Cosign
         if: needs.init.outputs.publish == 'true'
-        uses: sigstore/cosign-installer@v3.9.1
+        uses: sigstore/cosign-installer@v3.9.2
         with:
           cosign-release: "v2.4.3"
 

.github/workflows/ci.yaml

@@ -346,7 +346,7 @@ jobs:
         with:
           python-version: ${{ needs.prepare.outputs.python-version }}
       - name: Install Cosign
-        uses: sigstore/cosign-installer@v3.9.1
+        uses: sigstore/cosign-installer@v3.9.2
         with:
           cosign-release: "v2.4.3"
       - name: Restore Python virtual environment

build.yaml

@@ -1,10 +1,10 @@
 image: ghcr.io/home-assistant/{arch}-hassio-supervisor
 build_from:
-  aarch64: ghcr.io/home-assistant/aarch64-base-python:3.13-alpine3.21
-  armhf: ghcr.io/home-assistant/armhf-base-python:3.13-alpine3.21
-  armv7: ghcr.io/home-assistant/armv7-base-python:3.13-alpine3.21
-  amd64: ghcr.io/home-assistant/amd64-base-python:3.13-alpine3.21
-  i386: ghcr.io/home-assistant/i386-base-python:3.13-alpine3.21
+  aarch64: ghcr.io/home-assistant/aarch64-base-python:3.13-alpine3.22
+  armhf: ghcr.io/home-assistant/armhf-base-python:3.13-alpine3.22
+  armv7: ghcr.io/home-assistant/armv7-base-python:3.13-alpine3.22
+  amd64: ghcr.io/home-assistant/amd64-base-python:3.13-alpine3.22
+  i386: ghcr.io/home-assistant/i386-base-python:3.13-alpine3.22
 codenotary:
   signer: notary@home-assistant.io
   base_image: notary@home-assistant.io

requirements.txt

@@ -1,8 +1,8 @@
 aiodns==3.5.0
-aiohttp==3.12.14
+aiohttp==3.12.15
 atomicwrites-homeassistant==1.4.1
 attrs==25.3.0
-awesomeversion==25.5.0
+awesomeversion==25.8.0
 blockbuster==1.5.25
 brotli==1.1.0
 ciso8601==2.3.2
@@ -14,17 +14,17 @@ deepmerge==2.0
 dirhash==0.5.0
 docker==7.1.0
 faust-cchardet==2.1.19
-gitpython==3.1.44
+gitpython==3.1.45
 jinja2==3.1.6
 log-rate-limit==1.4.2
-orjson==3.11.0
+orjson==3.11.1
 pulsectl==24.12.0
 pyudev==0.24.3
 PyYAML==6.0.2
 requests==2.32.4
 securetar==2025.2.1
-sentry-sdk==2.33.0
+sentry-sdk==2.34.1
 setuptools==80.9.0
 voluptuous==0.15.2
-dbus-fast==2.44.1
+dbus-fast==2.44.3
 zlib-fast==0.2.1

requirements_tests.txt

@@ -1,6 +1,6 @@
 astroid==3.3.11
-coverage==7.9.2
-mypy==1.17.0
+coverage==7.10.2
+mypy==1.17.1
 pre-commit==4.2.0
 pylint==3.3.7
 pytest-aiohttp==1.1.0
@@ -8,7 +8,7 @@ pytest-asyncio==0.25.2
 pytest-cov==6.2.1
 pytest-timeout==2.4.0
 pytest==8.4.1
-ruff==0.12.3
+ruff==0.12.7
 time-machine==2.16.0
 types-docker==7.1.0.20250705
 types-pyyaml==6.0.12.20250516

supervisor/api/docker.py

@@ -6,6 +6,8 @@ from typing import Any
 from aiohttp import web
 import voluptuous as vol
 
+from supervisor.resolution.const import ContextType, IssueType, SuggestionType
+
 from ..const import (
     ATTR_ENABLE_IPV6,
     ATTR_HOSTNAME,
@@ -32,7 +34,7 @@ SCHEMA_DOCKER_REGISTRY = vol.Schema(
 )
 
 # pylint: disable=no-value-for-parameter
-SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_ENABLE_IPV6): vol.Boolean()})
+SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_ENABLE_IPV6): vol.Maybe(vol.Boolean())})
 
 
 class APIDocker(CoreSysAttributes):
@@ -59,8 +61,17 @@ class APIDocker(CoreSysAttributes):
         """Set docker options."""
         body = await api_validate(SCHEMA_OPTIONS, request)
 
-        if ATTR_ENABLE_IPV6 in body:
+        if (
+            ATTR_ENABLE_IPV6 in body
+            and self.sys_docker.config.enable_ipv6 != body[ATTR_ENABLE_IPV6]
+        ):
             self.sys_docker.config.enable_ipv6 = body[ATTR_ENABLE_IPV6]
+            _LOGGER.info("Host system reboot required to apply new IPv6 configuration")
+            self.sys_resolution.create_issue(
+                IssueType.REBOOT_REQUIRED,
+                ContextType.SYSTEM,
+                suggestions=[SuggestionType.EXECUTE_REBOOT],
+            )
 
         await self.sys_docker.config.save_data()
 

supervisor/backups/backup.py

@@ -262,41 +262,35 @@ class Backup(JobGroup):
 
     def __eq__(self, other: Any) -> bool:
         """Return true if backups have same metadata."""
-        if not isinstance(other, Backup):
-            return False
+        return isinstance(other, Backup) and self.slug == other.slug
 
-        # Compare all fields except ones about protection. Current encryption status does not affect equality
-        keys = self._data.keys() | other._data.keys()
-        for k in keys - IGNORED_COMPARISON_FIELDS:
-            if (
-                k not in self._data
-                or k not in other._data
-                or self._data[k] != other._data[k]
-            ):
-                _LOGGER.info(
-                    "Backup %s and %s not equal because %s field has different value: %s and %s",
-                    self.slug,
-                    other.slug,
-                    k,
-                    self._data.get(k),
-                    other._data.get(k),
-                )
-                return False
-        return True
+    def __hash__(self) -> int:
+        """Return hash of backup."""
+        return hash(self.slug)
 
     def consolidate(self, backup: Self) -> None:
         """Consolidate two backups with same slug in different locations."""
-        if self.slug != backup.slug:
+        if self != backup:
             raise ValueError(
                 f"Backup {self.slug} and {backup.slug} are not the same backup"
             )
-        if self != backup:
-            raise BackupInvalidError(
-                f"Backup in {backup.location} and {self.location} both have slug {self.slug} but are not the same!"
-            )
+
+        # Compare all fields except ones about protection. Current encryption status does not affect equality
+        other_data = backup._data  # pylint: disable=protected-access
+        keys = self._data.keys() | other_data.keys()
+        for k in keys - IGNORED_COMPARISON_FIELDS:
+            if (
+                k not in self._data
+                or k not in other_data
+                or self._data[k] != other_data[k]
+            ):
+                raise BackupInvalidError(
+                    f"Cannot consolidate backups in {backup.location} and {self.location} with slug {self.slug} "
+                    f"because field {k} has different values: {self._data.get(k)} and {other_data.get(k)}!",
+                    _LOGGER.error,
+                )
 
         # In case of conflict we always ignore the ones from the first one. But log them to let the user know
-
         if conflict := {
             loc: val.path
             for loc, val in self.all_locations.items()

supervisor/dbus/const.py

@@ -32,6 +32,7 @@ DBUS_IFACE_HOSTNAME = "org.freedesktop.hostname1"
 DBUS_IFACE_IP4CONFIG = "org.freedesktop.NetworkManager.IP4Config"
 DBUS_IFACE_IP6CONFIG = "org.freedesktop.NetworkManager.IP6Config"
 DBUS_IFACE_NM = "org.freedesktop.NetworkManager"
+DBUS_IFACE_NVME_CONTROLLER = "org.freedesktop.UDisks2.NVMe.Controller"
 DBUS_IFACE_PARTITION = "org.freedesktop.UDisks2.Partition"
 DBUS_IFACE_PARTITION_TABLE = "org.freedesktop.UDisks2.PartitionTable"
 DBUS_IFACE_RAUC_INSTALLER = "de.pengutronix.rauc.Installer"
@@ -87,6 +88,7 @@ DBUS_ATTR_CONNECTIVITY = "Connectivity"
 DBUS_ATTR_CURRENT_DEVICE = "CurrentDevice"
 DBUS_ATTR_CURRENT_DNS_SERVER = "CurrentDNSServer"
 DBUS_ATTR_CURRENT_DNS_SERVER_EX = "CurrentDNSServerEx"
+DBUS_ATTR_CONTROLLER_ID = "ControllerID"
 DBUS_ATTR_DEFAULT = "Default"
 DBUS_ATTR_DEPLOYMENT = "Deployment"
 DBUS_ATTR_DESCRIPTION = "Description"
@@ -111,6 +113,7 @@ DBUS_ATTR_DRIVER = "Driver"
 DBUS_ATTR_EJECTABLE = "Ejectable"
 DBUS_ATTR_FALLBACK_DNS = "FallbackDNS"
 DBUS_ATTR_FALLBACK_DNS_EX = "FallbackDNSEx"
+DBUS_ATTR_FGUID = "FGUID"
 DBUS_ATTR_FINISH_TIMESTAMP = "FinishTimestamp"
 DBUS_ATTR_FIRMWARE_TIMESTAMP_MONOTONIC = "FirmwareTimestampMonotonic"
 DBUS_ATTR_FREQUENCY = "Frequency"
@@ -147,6 +150,7 @@ DBUS_ATTR_NAMESERVERS = "Nameservers"
 DBUS_ATTR_NTP = "NTP"
 DBUS_ATTR_NTPSYNCHRONIZED = "NTPSynchronized"
 DBUS_ATTR_NUMBER = "Number"
+DBUS_ATTR_NVME_REVISION = "NVMeRevision"
 DBUS_ATTR_OFFSET = "Offset"
 DBUS_ATTR_OPERATING_SYSTEM_PRETTY_NAME = "OperatingSystemPrettyName"
 DBUS_ATTR_OPERATION = "Operation"
@@ -161,15 +165,24 @@ DBUS_ATTR_REMOVABLE = "Removable"
 DBUS_ATTR_RESOLV_CONF_MODE = "ResolvConfMode"
 DBUS_ATTR_REVISION = "Revision"
 DBUS_ATTR_RCMANAGER = "RcManager"
+DBUS_ATTR_SANITIZE_PERCENT_REMAINING = "SanitizePercentRemaining"
+DBUS_ATTR_SANITIZE_STATUS = "SanitizeStatus"
 DBUS_ATTR_SEAT = "Seat"
 DBUS_ATTR_SERIAL = "Serial"
 DBUS_ATTR_SIZE = "Size"
+DBUS_ATTR_SMART_CRITICAL_WARNING = "SmartCriticalWarning"
+DBUS_ATTR_SMART_POWER_ON_HOURS = "SmartPowerOnHours"
+DBUS_ATTR_SMART_SELFTEST_PERCENT_REMAINING = "SmartSelftestPercentRemaining"
+DBUS_ATTR_SMART_SELFTEST_STATUS = "SmartSelftestStatus"
+DBUS_ATTR_SMART_TEMPERATURE = "SmartTemperature"
+DBUS_ATTR_SMART_UPDATED = "SmartUpdated"
 DBUS_ATTR_SSID = "Ssid"
 DBUS_ATTR_STATE = "State"
 DBUS_ATTR_STATE_FLAGS = "StateFlags"
 DBUS_ATTR_STATIC_HOSTNAME = "StaticHostname"
 DBUS_ATTR_STATIC_OPERATING_SYSTEM_CPE_NAME = "OperatingSystemCPEName"
 DBUS_ATTR_STRENGTH = "Strength"
+DBUS_ATTR_SUBSYSTEM_NQN = "SubsystemNQN"
 DBUS_ATTR_SUPPORTED_FILESYSTEMS = "SupportedFilesystems"
 DBUS_ATTR_SYMLINKS = "Symlinks"
 DBUS_ATTR_SWAP_SIZE = "SwapSize"
@@ -180,6 +193,7 @@ DBUS_ATTR_TIMEUSEC = "TimeUSec"
 DBUS_ATTR_TIMEZONE = "Timezone"
 DBUS_ATTR_TRANSACTION_STATISTICS = "TransactionStatistics"
 DBUS_ATTR_TYPE = "Type"
+DBUS_ATTR_UNALLOCATED_CAPACITY = "UnallocatedCapacity"
 DBUS_ATTR_USER_LED = "UserLED"
 DBUS_ATTR_USERSPACE_TIMESTAMP_MONOTONIC = "UserspaceTimestampMonotonic"
 DBUS_ATTR_UUID_UPPERCASE = "UUID"

supervisor/dbus/udisks2/__init__.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+from pathlib import Path
 from typing import Any
 
 from awesomeversion import AwesomeVersion
@@ -132,7 +133,10 @@ class UDisks2Manager(DBusInterfaceProxy):
                 for drive in drives
             }
 
-            # Update existing drives
+            # For existing drives, need to check their type and call update
+            await asyncio.gather(
+                *[self._drives[path].check_type() for path in unchanged_drives]
+            )
             await asyncio.gather(
                 *[self._drives[path].update() for path in unchanged_drives]
             )
@@ -160,20 +164,33 @@ class UDisks2Manager(DBusInterfaceProxy):
         return list(self._drives.values())
 
     @dbus_connected
-    def get_drive(self, drive_path: str) -> UDisks2Drive:
+    def get_drive(self, object_path: str) -> UDisks2Drive:
         """Get additional info on drive from object path."""
-        if drive_path not in self._drives:
-            raise DBusObjectError(f"Drive {drive_path} not found")
+        if object_path not in self._drives:
+            raise DBusObjectError(f"Drive {object_path} not found")
 
-        return self._drives[drive_path]
+        return self._drives[object_path]
 
     @dbus_connected
-    def get_block_device(self, device_path: str) -> UDisks2Block:
+    def get_block_device(self, object_path: str) -> UDisks2Block:
         """Get additional info on block device from object path."""
-        if device_path not in self._block_devices:
-            raise DBusObjectError(f"Block device {device_path} not found")
+        if object_path not in self._block_devices:
+            raise DBusObjectError(f"Block device {object_path} not found")
 
-        return self._block_devices[device_path]
+        return self._block_devices[object_path]
+
+    @dbus_connected
+    def get_block_device_by_path(self, device_path: Path) -> UDisks2Block:
+        """Get additional info on block device from device path.
+
+        Uses cache only. Use `resolve_device` to force a call for fresh data.
+        """
+        for device in self._block_devices.values():
+            if device.device == device_path:
+                return device
+        raise DBusObjectError(
+            f"Block device not found with device path {device_path.as_posix()}"
+        )
 
     @dbus_connected
     async def resolve_device(self, devspec: DeviceSpecification) -> list[UDisks2Block]:

supervisor/dbus/udisks2/drive.py

@@ -1,6 +1,7 @@
 """Interface to UDisks2 Drive over D-Bus."""
 
 from datetime import UTC, datetime
+from typing import Any
 
 from dbus_fast.aio import MessageBus
 
@@ -18,11 +19,13 @@ from ..const import (
     DBUS_ATTR_VENDOR,
     DBUS_ATTR_WWN,
     DBUS_IFACE_DRIVE,
+    DBUS_IFACE_NVME_CONTROLLER,
     DBUS_NAME_UDISKS2,
 )
 from ..interface import DBusInterfaceProxy, dbus_property
 from ..utils import dbus_connected
 from .const import UDISKS2_DEFAULT_OPTIONS
+from .nvme_controller import UDisks2NVMeController
 
 
 class UDisks2Drive(DBusInterfaceProxy):
@@ -35,11 +38,18 @@ class UDisks2Drive(DBusInterfaceProxy):
     bus_name: str = DBUS_NAME_UDISKS2
     properties_interface: str = DBUS_IFACE_DRIVE
 
+    _nvme_controller: UDisks2NVMeController | None = None
+
     def __init__(self, object_path: str) -> None:
         """Initialize object."""
         self._object_path = object_path
         super().__init__()
 
+    async def connect(self, bus: MessageBus) -> None:
+        """Connect to bus."""
+        await super().connect(bus)
+        await self._reload_interfaces()
+
     @staticmethod
     async def new(object_path: str, bus: MessageBus) -> "UDisks2Drive":
         """Create and connect object."""
@@ -52,6 +62,11 @@ class UDisks2Drive(DBusInterfaceProxy):
         """Object path for dbus object."""
         return self._object_path
 
+    @property
+    def nvme_controller(self) -> UDisks2NVMeController | None:
+        """NVMe controller interface if drive is one."""
+        return self._nvme_controller
+
     @property
     @dbus_property
     def vendor(self) -> str:
@@ -130,3 +145,40 @@ class UDisks2Drive(DBusInterfaceProxy):
     async def eject(self) -> None:
         """Eject media from drive."""
         await self.connected_dbus.Drive.call("eject", UDISKS2_DEFAULT_OPTIONS)
+
+    @dbus_connected
+    async def update(self, changed: dict[str, Any] | None = None) -> None:
+        """Update properties via D-Bus."""
+        await super().update(changed)
+
+        if not changed and self.nvme_controller:
+            await self.nvme_controller.update()
+
+    @dbus_connected
+    async def check_type(self) -> None:
+        """Check if type of drive has changed and adjust interfaces if so."""
+        introspection = await self.connected_dbus.introspect()
+        interfaces = {intr.name for intr in introspection.interfaces}
+
+        # If interfaces changed, update the proxy from introspection and reload interfaces
+        if interfaces != set(self.connected_dbus.proxies.keys()):
+            await self.connected_dbus.init_proxy(introspection=introspection)
+            await self._reload_interfaces()
+
+    @dbus_connected
+    async def _reload_interfaces(self) -> None:
+        """Reload interfaces from introspection as necessary."""
+        # Check if drive is an nvme controller
+        if (
+            not self.nvme_controller
+            and DBUS_IFACE_NVME_CONTROLLER in self.connected_dbus.proxies
+        ):
+            self._nvme_controller = UDisks2NVMeController(self.object_path)
+            await self._nvme_controller.initialize(self.connected_dbus)
+
+        elif (
+            self.nvme_controller
+            and DBUS_IFACE_NVME_CONTROLLER not in self.connected_dbus.proxies
+        ):
+            self.nvme_controller.stop_sync_property_changes()
+            self._nvme_controller = None

supervisor/dbus/udisks2/nvme_controller.py

@@ -0,0 +1,200 @@
+"""Interface to UDisks2 NVME Controller over D-Bus."""
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, cast
+
+from dbus_fast.aio import MessageBus
+
+from ..const import (
+    DBUS_ATTR_CONTROLLER_ID,
+    DBUS_ATTR_FGUID,
+    DBUS_ATTR_NVME_REVISION,
+    DBUS_ATTR_SANITIZE_PERCENT_REMAINING,
+    DBUS_ATTR_SANITIZE_STATUS,
+    DBUS_ATTR_SMART_CRITICAL_WARNING,
+    DBUS_ATTR_SMART_POWER_ON_HOURS,
+    DBUS_ATTR_SMART_SELFTEST_PERCENT_REMAINING,
+    DBUS_ATTR_SMART_SELFTEST_STATUS,
+    DBUS_ATTR_SMART_TEMPERATURE,
+    DBUS_ATTR_SMART_UPDATED,
+    DBUS_ATTR_STATE,
+    DBUS_ATTR_SUBSYSTEM_NQN,
+    DBUS_ATTR_UNALLOCATED_CAPACITY,
+    DBUS_IFACE_NVME_CONTROLLER,
+    DBUS_NAME_UDISKS2,
+)
+from ..interface import DBusInterfaceProxy, dbus_property
+from ..utils import dbus_connected
+from .const import UDISKS2_DEFAULT_OPTIONS
+
+
+@dataclass(frozen=True, slots=True)
+class SmartStatus:
+    """Smart status information for NVMe devices.
+
+    https://storaged.org/doc/udisks2-api/latest/gdbus-org.freedesktop.UDisks2.NVMe.Controller.html#gdbus-method-org-freedesktop-UDisks2-NVMe-Controller.SmartGetAttributes
+    """
+
+    available_spare: int
+    spare_threshold: int
+    percent_used: int
+    total_data_read: int
+    total_data_written: int
+    controller_busy_minutes: int
+    power_cycles: int
+    unsafe_shutdowns: int
+    media_errors: int
+    number_error_log_entries: int
+    temperature_sensors: list[int]
+    warning_composite_temperature: int
+    critical_composite_temperature: int
+    warning_temperature_minutes: int
+    critical_temperature_minutes: int
+
+    @classmethod
+    def from_smart_get_attributes_resp(cls, resp: dict[str, Any]):
+        """Convert SmartGetAttributes response dictionary to instance."""
+        return cls(
+            available_spare=resp["avail_spare"],
+            spare_threshold=resp["spare_thresh"],
+            percent_used=resp["percent_used"],
+            total_data_read=resp["total_data_read"],
+            total_data_written=resp["total_data_written"],
+            controller_busy_minutes=resp["ctrl_busy_time"],
+            power_cycles=resp["power_cycles"],
+            unsafe_shutdowns=resp["unsafe_shutdowns"],
+            media_errors=resp["media_errors"],
+            number_error_log_entries=resp["num_err_log_entries"],
+            temperature_sensors=resp["temp_sensors"],
+            warning_composite_temperature=resp["wctemp"],
+            critical_composite_temperature=resp["cctemp"],
+            warning_temperature_minutes=resp["warning_temp_time"],
+            critical_temperature_minutes=resp["critical_temp_time"],
+        )
+
+
+class UDisks2NVMeController(DBusInterfaceProxy):
+    """Handle D-Bus interface for NVMe Controller object.
+
+    https://storaged.org/doc/udisks2-api/latest/gdbus-org.freedesktop.UDisks2.NVMe.Controller.html
+    """
+
+    name: str = DBUS_IFACE_NVME_CONTROLLER
+    bus_name: str = DBUS_NAME_UDISKS2
+    properties_interface: str = DBUS_IFACE_NVME_CONTROLLER
+
+    def __init__(self, object_path: str) -> None:
+        """Initialize object."""
+        self._object_path = object_path
+        super().__init__()
+
+    @staticmethod
+    async def new(object_path: str, bus: MessageBus) -> "UDisks2NVMeController":
+        """Create and connect object."""
+        obj = UDisks2NVMeController(object_path)
+        await obj.connect(bus)
+        return obj
+
+    @property
+    def object_path(self) -> str:
+        """Object path for dbus object."""
+        return self._object_path
+
+    @property
+    @dbus_property
+    def state(self) -> str:
+        """Return NVMe controller state."""
+        return self.properties[DBUS_ATTR_STATE]
+
+    @property
+    @dbus_property
+    def controller_id(self) -> int:
+        """Return controller ID."""
+        return self.properties[DBUS_ATTR_CONTROLLER_ID]
+
+    @property
+    @dbus_property
+    def subsystem_nqn(self) -> str:
+        """Return NVM Subsystem NVMe Qualified Name."""
+        return cast(bytes, self.properties[DBUS_ATTR_SUBSYSTEM_NQN]).decode("utf-8")
+
+    @property
+    @dbus_property
+    def fguid(self) -> str:
+        """Return FRU GUID."""
+        return self.properties[DBUS_ATTR_FGUID]
+
+    @property
+    @dbus_property
+    def nvme_revision(self) -> str:
+        """Return NVMe version information."""
+        return self.properties[DBUS_ATTR_NVME_REVISION]
+
+    @property
+    @dbus_property
+    def unallocated_capacity(self) -> int:
+        """Return unallocated capacity."""
+        return self.properties[DBUS_ATTR_UNALLOCATED_CAPACITY]
+
+    @property
+    @dbus_property
+    def smart_updated(self) -> datetime | None:
+        """Return last time smart information was updated (or None if it hasn't been).
+
+        If this is None other smart properties are not meaningful.
+        """
+        if not (ts := self.properties[DBUS_ATTR_SMART_UPDATED]):
+            return None
+        return datetime.fromtimestamp(ts, UTC)
+
+    @property
+    @dbus_property
+    def smart_critical_warning(self) -> list[str]:
+        """Return critical warnings issued for current state of controller."""
+        return self.properties[DBUS_ATTR_SMART_CRITICAL_WARNING]
+
+    @property
+    @dbus_property
+    def smart_power_on_hours(self) -> int:
+        """Return hours the disk has been powered on."""
+        return self.properties[DBUS_ATTR_SMART_POWER_ON_HOURS]
+
+    @property
+    @dbus_property
+    def smart_temperature(self) -> int:
+        """Return current composite temperature of controller in Kelvin."""
+        return self.properties[DBUS_ATTR_SMART_TEMPERATURE]
+
+    @property
+    @dbus_property
+    def smart_selftest_status(self) -> str:
+        """Return status of last sel-test."""
+        return self.properties[DBUS_ATTR_SMART_SELFTEST_STATUS]
+
+    @property
+    @dbus_property
+    def smart_selftest_percent_remaining(self) -> int:
+        """Return percent remaining of self-test."""
+        return self.properties[DBUS_ATTR_SMART_SELFTEST_PERCENT_REMAINING]
+
+    @property
+    @dbus_property
+    def sanitize_status(self) -> str:
+        """Return status of last sanitize operation."""
+        return self.properties[DBUS_ATTR_SANITIZE_STATUS]
+
+    @property
+    @dbus_property
+    def sanitize_percent_remaining(self) -> int:
+        """Return percent remaining of sanitize operation."""
+        return self.properties[DBUS_ATTR_SANITIZE_PERCENT_REMAINING]
+
+    @dbus_connected
+    async def smart_get_attributes(self) -> SmartStatus:
+        """Return smart/health information of controller."""
+        return SmartStatus.from_smart_get_attributes_resp(
+            await self.connected_dbus.NVMe.Controller.call(
+                "smart_get_attributes", UDISKS2_DEFAULT_OPTIONS
+            )
+        )

supervisor/docker/manager.py

@@ -95,12 +95,12 @@ class DockerConfig(FileConfiguration):
         super().__init__(FILE_HASSIO_DOCKER, SCHEMA_DOCKER_CONFIG)
 
     @property
-    def enable_ipv6(self) -> bool:
+    def enable_ipv6(self) -> bool | None:
         """Return IPv6 configuration for docker network."""
-        return self._data.get(ATTR_ENABLE_IPV6, False)
+        return self._data.get(ATTR_ENABLE_IPV6, None)
 
     @enable_ipv6.setter
-    def enable_ipv6(self, value: bool) -> None:
+    def enable_ipv6(self, value: bool | None) -> None:
         """Set IPv6 configuration for docker network."""
         self._data[ATTR_ENABLE_IPV6] = value
 

supervisor/docker/network.py

@@ -47,6 +47,8 @@ DOCKER_NETWORK_PARAMS = {
     "options": {"com.docker.network.bridge.name": DOCKER_NETWORK},
 }
 
+DOCKER_ENABLE_IPV6_DEFAULT = True
+
 
 class DockerNetwork:
     """Internal Supervisor Network.
@@ -59,7 +61,7 @@ class DockerNetwork:
         self.docker: docker.DockerClient = docker_client
         self._network: docker.models.networks.Network
 
-    async def post_init(self, enable_ipv6: bool = False) -> Self:
+    async def post_init(self, enable_ipv6: bool | None = None) -> Self:
         """Post init actions that must be done in event loop."""
         self._network = await asyncio.get_running_loop().run_in_executor(
             None, self._get_network, enable_ipv6
@@ -111,16 +113,24 @@ class DockerNetwork:
         """Return observer of the network."""
         return DOCKER_IPV4_NETWORK_MASK[6]
 
-    def _get_network(self, enable_ipv6: bool = False) -> docker.models.networks.Network:
+    def _get_network(
+        self, enable_ipv6: bool | None = None
+    ) -> docker.models.networks.Network:
         """Get supervisor network."""
         try:
             if network := self.docker.networks.get(DOCKER_NETWORK):
-                if network.attrs.get(DOCKER_ENABLEIPV6) == enable_ipv6:
+                current_ipv6 = network.attrs.get(DOCKER_ENABLEIPV6, False)
+                # If the network exists and we don't have an explicit setting,
+                # simply stick with what we have.
+                if enable_ipv6 is None or current_ipv6 == enable_ipv6:
                     return network
+
+                # We have an explicit setting which differs from the current state.
                 _LOGGER.info(
                     "Migrating Supervisor network to %s",
                     "IPv4/IPv6 Dual-Stack" if enable_ipv6 else "IPv4-Only",
                 )
+
                 if (containers := network.containers) and (
                     containers_all := all(
                         container.name in (OBSERVER_DOCKER_NAME, SUPERVISOR_DOCKER_NAME)
@@ -134,6 +144,7 @@ class DockerNetwork:
                             requests.RequestException,
                         ):
                             network.disconnect(container, force=True)
+
                 if not containers or containers_all:
                     try:
                         network.remove()
@@ -151,7 +162,9 @@ class DockerNetwork:
             _LOGGER.info("Can't find Supervisor network, creating a new network")
 
         network_params = DOCKER_NETWORK_PARAMS.copy()
-        network_params[ATTR_ENABLE_IPV6] = enable_ipv6
+        network_params[ATTR_ENABLE_IPV6] = (
+            DOCKER_ENABLE_IPV6_DEFAULT if enable_ipv6 is None else enable_ipv6
+        )
 
         try:
             self._network = self.docker.networks.create(**network_params)  # type: ignore

supervisor/hardware/disk.py

@@ -5,7 +5,7 @@ from pathlib import Path
 import shutil
 
 from ..coresys import CoreSys, CoreSysAttributes
-from ..exceptions import HardwareNotFound
+from ..exceptions import DBusError, DBusObjectError, HardwareNotFound
 from .const import UdevSubsystem
 from .data import Device
 
@@ -14,6 +14,7 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
 _MOUNTINFO: Path = Path("/proc/self/mountinfo")
 _BLOCK_DEVICE_CLASS = "/sys/class/block/{}"
 _BLOCK_DEVICE_EMMC_LIFE_TIME = "/sys/block/{}/device/life_time"
+_DEVICE_PATH = "/dev/{}"
 
 
 class HwDisk(CoreSysAttributes):
@@ -92,8 +93,67 @@ class HwDisk(CoreSysAttributes):
             optionsep += 1
         return mountinfoarr[optionsep + 2]
 
+    def _get_mount_source_device_name(self, path: str | Path) -> str | None:
+        """Get mount source device name.
+
+        Must be run in executor.
+        """
+        mount_source = self._get_mount_source(str(path))
+        if not mount_source or mount_source == "overlay":
+            return None
+
+        mount_source_path = Path(mount_source)
+        if not mount_source_path.is_block_device():
+            return None
+
+        # This looks a bit funky but it is more or less what lsblk is doing to get
+        # the parent dev reliably
+
+        # Get class device...
+        mount_source_device_part = Path(
+            _BLOCK_DEVICE_CLASS.format(mount_source_path.name)
+        )
+
+        # ... resolve symlink and get parent device from that path.
+        return mount_source_device_part.resolve().parts[-2]
+
+    async def _try_get_nvme_lifetime(self, device_name: str) -> float | None:
+        """Get NVMe device lifetime."""
+        device_path = Path(_DEVICE_PATH.format(device_name))
+        try:
+            block_device = self.sys_dbus.udisks2.get_block_device_by_path(device_path)
+            drive = self.sys_dbus.udisks2.get_drive(block_device.drive)
+        except DBusObjectError:
+            _LOGGER.warning(
+                "Unable to find UDisks2 drive for device at %s", device_path.as_posix()
+            )
+            return None
+
+        # Exit if this isn't an NVMe device
+        if not drive.nvme_controller:
+            return None
+
+        try:
+            smart_log = await drive.nvme_controller.smart_get_attributes()
+        except DBusError as err:
+            _LOGGER.warning(
+                "Unable to get smart log for drive %s due to %s", drive.id, err
+            )
+            return None
+
+        # UDisks2 documentation specifies that value can exceed 100
+        if smart_log.percent_used >= 100:
+            _LOGGER.warning(
+                "NVMe controller reports that its estimated life-time has been exceeded!"
+            )
+            return 100.0
+        return smart_log.percent_used
+
     def _try_get_emmc_life_time(self, device_name: str) -> float | None:
-        # Get eMMC life_time
+        """Get eMMC life_time.
+
+        Must be run in executor.
+        """
         life_time_path = Path(_BLOCK_DEVICE_EMMC_LIFE_TIME.format(device_name))
 
         if not life_time_path.exists():
@@ -121,29 +181,20 @@ class HwDisk(CoreSysAttributes):
         # Return the pessimistic estimate (0x02 -> 10%-20%, return 20%)
         return life_time_value * 10.0
 
-    def get_disk_life_time(self, path: str | Path) -> float | None:
-        """Return life time estimate of the underlying SSD drive.
-
-        Must be run in executor.
-        """
-        mount_source = self._get_mount_source(str(path))
-        if not mount_source or mount_source == "overlay":
-            return None
-
-        mount_source_path = Path(mount_source)
-        if not mount_source_path.is_block_device():
-            return None
-
-        # This looks a bit funky but it is more or less what lsblk is doing to get
-        # the parent dev reliably
-
-        # Get class device...
-        mount_source_device_part = Path(
-            _BLOCK_DEVICE_CLASS.format(mount_source_path.name)
+    async def get_disk_life_time(self, path: str | Path) -> float | None:
+        """Return life time estimate of the underlying SSD drive."""
+        mount_source_device_name = await self.sys_run_in_executor(
+            self._get_mount_source_device_name, path
         )
+        if mount_source_device_name is None:
+            return None
 
-        # ... resolve symlink and get parent device from that path.
-        mount_source_device_name = mount_source_device_part.resolve().parts[-2]
+        # First check if its an NVMe device and get lifetime information that way
+        nvme_lifetime = await self._try_get_nvme_lifetime(mount_source_device_name)
+        if nvme_lifetime is not None:
+            return nvme_lifetime
 
-        # Currently only eMMC block devices supported
-        return self._try_get_emmc_life_time(mount_source_device_name)
+        # Else try to get lifetime information for eMMC devices. Other types of devices will return None
+        return await self.sys_run_in_executor(
+            self._try_get_emmc_life_time, mount_source_device_name
+        )

supervisor/host/info.py

@@ -135,9 +135,8 @@ class InfoCenter(CoreSysAttributes):
 
     async def disk_life_time(self) -> float | None:
         """Return the estimated life-time usage (in %) of the SSD storing the data directory."""
-        return await self.sys_run_in_executor(
-            self.sys_hardware.disk.get_disk_life_time,
-            self.coresys.config.path_supervisor,
+        return await self.sys_hardware.disk.get_disk_life_time(
+            self.coresys.config.path_supervisor
         )
 
     async def get_dmesg(self) -> bytes:

supervisor/jobs/const.py

@@ -34,8 +34,60 @@ class JobCondition(StrEnum):
     SUPERVISOR_UPDATED = "supervisor_updated"
 
 
+class JobConcurrency(StrEnum):
+    """Job concurrency control.
+
+    Controls how many instances of a job can run simultaneously.
+
+    Individual Concurrency (applies to each method separately):
+    - REJECT: Fail immediately if another instance is already running
+    - QUEUE: Wait for the current instance to finish, then run
+
+    Group Concurrency (applies across all methods on a JobGroup):
+    - GROUP_REJECT: Fail if ANY job is running on the JobGroup
+    - GROUP_QUEUE: Wait for ANY running job on the JobGroup to finish
+
+    JobGroup Behavior:
+    - All methods on the same JobGroup instance share a single lock
+    - Methods can call other methods on the same group without deadlock
+    - Uses the JobGroup.group_name for coordination
+    - Requires the class to inherit from JobGroup
+    """
+
+    REJECT = "reject"  # Fail if already running (was ONCE)
+    QUEUE = "queue"  # Wait if already running (was SINGLE_WAIT)
+    GROUP_REJECT = "group_reject"  # Was GROUP_ONCE
+    GROUP_QUEUE = "group_queue"  # Was GROUP_WAIT
+
+
+class JobThrottle(StrEnum):
+    """Job throttling control.
+
+    Controls how frequently jobs can be executed.
+
+    Individual Throttling (each method has its own throttle state):
+    - THROTTLE: Skip execution if called within throttle_period
+    - RATE_LIMIT: Allow up to throttle_max_calls within throttle_period, then fail
+
+    Group Throttling (all methods on a JobGroup share throttle state):
+    - GROUP_THROTTLE: Skip if ANY method was called within throttle_period
+    - GROUP_RATE_LIMIT: Allow up to throttle_max_calls total across ALL methods
+
+    JobGroup Behavior:
+    - All methods on the same JobGroup instance share throttle counters/timers
+    - Uses the JobGroup.group_name as the key for tracking state
+    - If one method is throttled, other methods may also be throttled
+    - Requires the class to inherit from JobGroup
+    """
+
+    THROTTLE = "throttle"  # Skip if called too frequently
+    RATE_LIMIT = "rate_limit"  # Rate limiting with max calls per period
+    GROUP_THROTTLE = "group_throttle"  # Group version of THROTTLE
+    GROUP_RATE_LIMIT = "group_rate_limit"  # Group version of RATE_LIMIT
+
+
 class JobExecutionLimit(StrEnum):
-    """Job Execution limits."""
+    """Job Execution limits - DEPRECATED: Use JobConcurrency and JobThrottle instead."""
 
     ONCE = "once"
     SINGLE_WAIT = "single_wait"

supervisor/jobs/decorator.py

@@ -20,7 +20,7 @@ from ..host.const import HostFeature
 from ..resolution.const import MINIMUM_FREE_SPACE_THRESHOLD, ContextType, IssueType
 from ..utils.sentry import async_capture_exception
 from . import SupervisorJob
-from .const import JobCondition, JobExecutionLimit
+from .const import JobConcurrency, JobCondition, JobExecutionLimit, JobThrottle
 from .job_group import JobGroup
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
@@ -36,13 +36,16 @@ class Job(CoreSysAttributes):
         conditions: list[JobCondition] | None = None,
         cleanup: bool = True,
         on_condition: type[JobException] | None = None,
-        limit: JobExecutionLimit | None = None,
+        concurrency: JobConcurrency | None = None,
+        throttle: JobThrottle | None = None,
         throttle_period: timedelta
         | Callable[[CoreSys, datetime, list[datetime] | None], timedelta]
         | None = None,
         throttle_max_calls: int | None = None,
         internal: bool = False,
-    ):
+        # Backward compatibility - DEPRECATED
+        limit: JobExecutionLimit | None = None,
+    ):  # pylint: disable=too-many-positional-arguments
         """Initialize the Job decorator.
 
         Args:
@@ -50,13 +53,15 @@ class Job(CoreSysAttributes):
             conditions (list[JobCondition] | None): List of conditions that must be met before the job runs.
             cleanup (bool): Whether to clean up the job after execution. Defaults to True. If set to False, the job will remain accessible through the Supervisor API until the next restart.
             on_condition (type[JobException] | None): Exception type to raise if a job condition fails. If None, logs the failure.
-            limit (JobExecutionLimit | None): Execution limit policy for the job (e.g., throttle, once, group-based).
-            throttle_period (timedelta | Callable | None): Throttle period as a timedelta or a callable returning a timedelta (for rate-limited jobs).
+            concurrency (JobConcurrency | None): Concurrency control policy (e.g., reject, queue, group-based).
+            throttle (JobThrottle | None): Throttling policy (e.g., throttle, rate_limit, group-based).
+            throttle_period (timedelta | Callable | None): Throttle period as a timedelta or a callable returning a timedelta (for throttled jobs).
             throttle_max_calls (int | None): Maximum number of calls allowed within the throttle period (for rate-limited jobs).
             internal (bool): Whether the job is internal (not exposed through the Supervisor API). Defaults to False.
+            limit (JobExecutionLimit | None): DEPRECATED - Use concurrency and throttle instead.
 
         Raises:
-            RuntimeError: If job name is not unique, or required throttle parameters are missing for the selected limit.
+            RuntimeError: If job name is not unique, or required throttle parameters are missing for the selected throttle policy.
 
         """
         if name in _JOB_NAMES:
@@ -67,7 +72,6 @@ class Job(CoreSysAttributes):
         self.conditions = conditions
         self.cleanup = cleanup
         self.on_condition = on_condition
-        self.limit = limit
         self._throttle_period = throttle_period
         self._throttle_max_calls = throttle_max_calls
         self._lock: asyncio.Semaphore | None = None
@@ -75,34 +79,91 @@ class Job(CoreSysAttributes):
         self._rate_limited_calls: dict[str | None, list[datetime]] | None = None
         self._internal = internal
 
+        # Handle backward compatibility with limit parameter
+        if limit is not None:
+            if concurrency is not None or throttle is not None:
+                raise RuntimeError(
+                    f"Job {name} cannot specify both 'limit' (deprecated) and 'concurrency'/'throttle' parameters!"
+                )
+            # Map old limit values to new parameters
+            concurrency, throttle = self._map_limit_to_new_params(limit)
+
+        self.concurrency = concurrency
+        self.throttle = throttle
+
         # Validate Options
+        self._validate_parameters()
+
+    def _map_limit_to_new_params(
+        self, limit: JobExecutionLimit
+    ) -> tuple[JobConcurrency | None, JobThrottle | None]:
+        """Map old limit parameter to new concurrency and throttle parameters."""
+        mapping = {
+            JobExecutionLimit.ONCE: (JobConcurrency.REJECT, None),
+            JobExecutionLimit.SINGLE_WAIT: (JobConcurrency.QUEUE, None),
+            JobExecutionLimit.THROTTLE: (None, JobThrottle.THROTTLE),
+            JobExecutionLimit.THROTTLE_WAIT: (
+                JobConcurrency.QUEUE,
+                JobThrottle.THROTTLE,
+            ),
+            JobExecutionLimit.THROTTLE_RATE_LIMIT: (None, JobThrottle.RATE_LIMIT),
+            JobExecutionLimit.GROUP_ONCE: (JobConcurrency.GROUP_REJECT, None),
+            JobExecutionLimit.GROUP_WAIT: (JobConcurrency.GROUP_QUEUE, None),
+            JobExecutionLimit.GROUP_THROTTLE: (None, JobThrottle.GROUP_THROTTLE),
+            JobExecutionLimit.GROUP_THROTTLE_WAIT: (
+                # Seems a bit counter intuitive, but GROUP_QUEUE deadlocks
+                # tests/jobs/test_job_decorator.py::test_execution_limit_group_throttle_wait
+                # The reason this deadlocks is because when using GROUP_QUEUE and the
+                # throttle limit is hit, the group lock is trying to be unlocked outside
+                # of the job context. The current implementation doesn't allow to unlock
+                # the group lock when the job is not running.
+                JobConcurrency.QUEUE,
+                JobThrottle.GROUP_THROTTLE,
+            ),
+            JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT: (
+                None,
+                JobThrottle.GROUP_RATE_LIMIT,
+            ),
+        }
+        return mapping.get(limit, (None, None))
+
+    def _validate_parameters(self) -> None:
+        """Validate job parameters."""
+        # Validate throttle parameters
         if (
-            self.limit
+            self.throttle
             in (
-                JobExecutionLimit.THROTTLE,
-                JobExecutionLimit.THROTTLE_WAIT,
-                JobExecutionLimit.THROTTLE_RATE_LIMIT,
-                JobExecutionLimit.GROUP_THROTTLE,
-                JobExecutionLimit.GROUP_THROTTLE_WAIT,
-                JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
+                JobThrottle.THROTTLE,
+                JobThrottle.GROUP_THROTTLE,
+                JobThrottle.RATE_LIMIT,
+                JobThrottle.GROUP_RATE_LIMIT,
             )
             and self._throttle_period is None
         ):
             raise RuntimeError(
-                f"Job {name} is using execution limit {limit} without a throttle period!"
+                f"Job {self.name} is using throttle {self.throttle} without a throttle period!"
             )
 
-        if self.limit in (
-            JobExecutionLimit.THROTTLE_RATE_LIMIT,
-            JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
+        if self.throttle in (
+            JobThrottle.RATE_LIMIT,
+            JobThrottle.GROUP_RATE_LIMIT,
         ):
             if self._throttle_max_calls is None:
                 raise RuntimeError(
-                    f"Job {name} is using execution limit {limit} without throttle max calls!"
+                    f"Job {self.name} is using throttle {self.throttle} without throttle max calls!"
                 )
-
             self._rate_limited_calls = {}
 
+        if self.throttle is not None and self.concurrency in (
+            JobConcurrency.GROUP_REJECT,
+            JobConcurrency.GROUP_QUEUE,
+        ):
+            # We cannot release group locks when Job is not running (e.g. throttled)
+            # which makes these combinations impossible to use currently.
+            raise RuntimeError(
+                f"Job {self.name} is using throttling ({self.throttle}) with group concurrency ({self.concurrency}), which is not allowed!"
+            )
+
     @property
     def throttle_max_calls(self) -> int:
         """Return max calls for throttle."""
@@ -131,7 +192,7 @@ class Job(CoreSysAttributes):
         """Return rate limited calls if used."""
         if self._rate_limited_calls is None:
             raise RuntimeError(
-                f"Rate limited calls not available for limit type {self.limit}"
+                "Rate limited calls not available for this throttle type"
             )
 
         return self._rate_limited_calls.get(group_name, [])
@@ -142,7 +203,7 @@ class Job(CoreSysAttributes):
         """Add a rate limited call to list if used."""
         if self._rate_limited_calls is None:
             raise RuntimeError(
-                f"Rate limited calls not available for limit type {self.limit}"
+                "Rate limited calls not available for this throttle type"
             )
 
         if group_name in self._rate_limited_calls:
@@ -156,7 +217,7 @@ class Job(CoreSysAttributes):
         """Set rate limited calls if used."""
         if self._rate_limited_calls is None:
             raise RuntimeError(
-                f"Rate limited calls not available for limit type {self.limit}"
+                "Rate limited calls not available for this throttle type"
             )
 
         self._rate_limited_calls[group_name] = value
@@ -193,16 +254,24 @@ class Job(CoreSysAttributes):
             if obj.acquire and obj.release:  # type: ignore
                 job_group = cast(JobGroup, obj)
 
-        if not job_group and self.limit in (
-            JobExecutionLimit.GROUP_ONCE,
-            JobExecutionLimit.GROUP_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE,
-            JobExecutionLimit.GROUP_THROTTLE_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
-        ):
-            raise RuntimeError(
-                f"Job on {self.name} need to be a JobGroup to use group based limits!"
-            ) from None
+        # Check for group-based parameters
+        if not job_group:
+            if self.concurrency in (
+                JobConcurrency.GROUP_REJECT,
+                JobConcurrency.GROUP_QUEUE,
+            ):
+                raise RuntimeError(
+                    f"Job {self.name} uses group concurrency ({self.concurrency}) but is not on a JobGroup! "
+                    f"The class must inherit from JobGroup to use GROUP_REJECT or GROUP_QUEUE."
+                ) from None
+            if self.throttle in (
+                JobThrottle.GROUP_THROTTLE,
+                JobThrottle.GROUP_RATE_LIMIT,
+            ):
+                raise RuntimeError(
+                    f"Job {self.name} uses group throttling ({self.throttle}) but is not on a JobGroup! "
+                    f"The class must inherit from JobGroup to use GROUP_THROTTLE or GROUP_RATE_LIMIT."
+                ) from None
 
         return job_group
 
@@ -255,71 +324,15 @@ class Job(CoreSysAttributes):
                     except JobConditionException as err:
                         return self._handle_job_condition_exception(err)
 
-                # Handle exection limits
-                if self.limit in (
-                    JobExecutionLimit.SINGLE_WAIT,
-                    JobExecutionLimit.ONCE,
-                ):
-                    await self._acquire_exection_limit()
-                elif self.limit in (
-                    JobExecutionLimit.GROUP_ONCE,
-                    JobExecutionLimit.GROUP_WAIT,
-                ):
-                    try:
-                        await cast(JobGroup, job_group).acquire(
-                            job, self.limit == JobExecutionLimit.GROUP_WAIT
-                        )
-                    except JobGroupExecutionLimitExceeded as err:
-                        if self.on_condition:
-                            raise self.on_condition(str(err)) from err
-                        raise err
-                elif self.limit in (
-                    JobExecutionLimit.THROTTLE,
-                    JobExecutionLimit.GROUP_THROTTLE,
-                ):
-                    time_since_last_call = datetime.now() - self.last_call(group_name)
-                    if time_since_last_call < self.throttle_period(group_name):
-                        return
-                elif self.limit in (
-                    JobExecutionLimit.THROTTLE_WAIT,
-                    JobExecutionLimit.GROUP_THROTTLE_WAIT,
-                ):
-                    await self._acquire_exection_limit()
-                    time_since_last_call = datetime.now() - self.last_call(group_name)
-                    if time_since_last_call < self.throttle_period(group_name):
-                        self._release_exception_limits()
-                        return
-                elif self.limit in (
-                    JobExecutionLimit.THROTTLE_RATE_LIMIT,
-                    JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
-                ):
-                    # Only reprocess array when necessary (at limit)
-                    if (
-                        len(self.rate_limited_calls(group_name))
-                        >= self.throttle_max_calls
-                    ):
-                        self.set_rate_limited_calls(
-                            [
-                                call
-                                for call in self.rate_limited_calls(group_name)
-                                if call
-                                > datetime.now() - self.throttle_period(group_name)
-                            ],
-                            group_name,
-                        )
-
-                    if (
-                        len(self.rate_limited_calls(group_name))
-                        >= self.throttle_max_calls
-                    ):
-                        on_condition = (
-                            JobException
-                            if self.on_condition is None
-                            else self.on_condition
-                        )
-                        raise on_condition(
-                            f"Rate limit exceeded, more than {self.throttle_max_calls} calls in {self.throttle_period(group_name)}",
-                        )
+                # Handle execution limits
+                await self._handle_concurrency_control(job_group, job)
+                try:
+                    if not await self._handle_throttling(group_name):
+                        self._release_concurrency_control(job_group)
+                        return  # Job was throttled, exit early
+                except Exception:
+                    self._release_concurrency_control(job_group)
+                    raise
 
                 # Execute Job
                 with job.start():
@@ -345,12 +358,7 @@ class Job(CoreSysAttributes):
                         await async_capture_exception(err)
                         raise JobException() from err
                     finally:
-                        self._release_exception_limits()
-                        if job_group and self.limit in (
-                            JobExecutionLimit.GROUP_ONCE,
-                            JobExecutionLimit.GROUP_WAIT,
-                        ):
-                            job_group.release()
+                        self._release_concurrency_control(job_group)
 
             # Jobs that weren't started are always cleaned up. Also clean up done jobs if required
             finally:
@@ -492,31 +500,75 @@ class Job(CoreSysAttributes):
                 f"'{method_name}' blocked from execution, mounting not supported on system"
             )
 
-    async def _acquire_exection_limit(self) -> None:
-        """Process exection limits."""
-        if self.limit not in (
-            JobExecutionLimit.SINGLE_WAIT,
-            JobExecutionLimit.ONCE,
-            JobExecutionLimit.THROTTLE_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE_WAIT,
+    def _release_concurrency_control(self, job_group: JobGroup | None) -> None:
+        """Release concurrency control locks."""
+        if self.concurrency == JobConcurrency.REJECT:
+            if self.lock.locked():
+                self.lock.release()
+        elif self.concurrency == JobConcurrency.QUEUE:
+            if self.lock.locked():
+                self.lock.release()
+        elif self.concurrency in (
+            JobConcurrency.GROUP_REJECT,
+            JobConcurrency.GROUP_QUEUE,
         ):
-            return
+            if job_group and job_group.has_lock:
+                job_group.release()
 
-        if self.limit == JobExecutionLimit.ONCE and self.lock.locked():
-            on_condition = (
-                JobException if self.on_condition is None else self.on_condition
-            )
-            raise on_condition("Another job is running")
+    async def _handle_concurrency_control(
+        self, job_group: JobGroup | None, job: SupervisorJob
+    ) -> None:
+        """Handle concurrency control limits."""
+        if self.concurrency == JobConcurrency.REJECT:
+            if self.lock.locked():
+                on_condition = (
+                    JobException if self.on_condition is None else self.on_condition
+                )
+                raise on_condition("Another job is running")
+            await self.lock.acquire()
+        elif self.concurrency == JobConcurrency.QUEUE:
+            await self.lock.acquire()
+        elif self.concurrency == JobConcurrency.GROUP_REJECT:
+            try:
+                await cast(JobGroup, job_group).acquire(job, wait=False)
+            except JobGroupExecutionLimitExceeded as err:
+                if self.on_condition:
+                    raise self.on_condition(str(err)) from err
+                raise err
+        elif self.concurrency == JobConcurrency.GROUP_QUEUE:
+            try:
+                await cast(JobGroup, job_group).acquire(job, wait=True)
+            except JobGroupExecutionLimitExceeded as err:
+                if self.on_condition:
+                    raise self.on_condition(str(err)) from err
+                raise err
 
-        await self.lock.acquire()
+    async def _handle_throttling(self, group_name: str | None) -> bool:
+        """Handle throttling limits. Returns True if job should continue, False if throttled."""
+        if self.throttle in (JobThrottle.THROTTLE, JobThrottle.GROUP_THROTTLE):
+            time_since_last_call = datetime.now() - self.last_call(group_name)
+            throttle_period = self.throttle_period(group_name)
+            if time_since_last_call < throttle_period:
+                # Always return False when throttled (skip execution)
+                return False
+        elif self.throttle in (JobThrottle.RATE_LIMIT, JobThrottle.GROUP_RATE_LIMIT):
+            # Only reprocess array when necessary (at limit)
+            if len(self.rate_limited_calls(group_name)) >= self.throttle_max_calls:
+                self.set_rate_limited_calls(
+                    [
+                        call
+                        for call in self.rate_limited_calls(group_name)
+                        if call > datetime.now() - self.throttle_period(group_name)
+                    ],
+                    group_name,
+                )
 
-    def _release_exception_limits(self) -> None:
-        """Release possible exception limits."""
-        if self.limit not in (
-            JobExecutionLimit.SINGLE_WAIT,
-            JobExecutionLimit.ONCE,
-            JobExecutionLimit.THROTTLE_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE_WAIT,
-        ):
-            return
-        self.lock.release()
+            if len(self.rate_limited_calls(group_name)) >= self.throttle_max_calls:
+                on_condition = (
+                    JobException if self.on_condition is None else self.on_condition
+                )
+                raise on_condition(
+                    f"Rate limit exceeded, more than {self.throttle_max_calls} calls in {self.throttle_period(group_name)}",
+                )
+
+        return True

supervisor/mounts/mount.py

@@ -164,10 +164,14 @@ class Mount(CoreSysAttributes, ABC):
         """Return true if successfully mounted and available."""
         return self.state == UnitActiveState.ACTIVE
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         """Return true if mounts are the same."""
         return isinstance(other, Mount) and self.name == other.name
 
+    def __hash__(self) -> int:
+        """Return hash of mount."""
+        return hash(self.name)
+
     async def load(self) -> None:
         """Initialize object."""
         # If there's no mount unit, mount it to make one

supervisor/os/manager.py

@@ -272,6 +272,7 @@ class OSManager(CoreSysAttributes):
         name="os_manager_update",
         conditions=[
             JobCondition.HAOS,
+            JobCondition.HEALTHY,
             JobCondition.INTERNET_SYSTEM,
             JobCondition.RUNNING,
             JobCondition.SUPERVISOR_UPDATED,

supervisor/utils/sentry.py

@@ -12,6 +12,7 @@ from sentry_sdk.integrations.dedupe import DedupeIntegration
 from sentry_sdk.integrations.excepthook import ExcepthookIntegration
 from sentry_sdk.integrations.logging import LoggingIntegration
 from sentry_sdk.integrations.threading import ThreadingIntegration
+from sentry_sdk.scrubber import DEFAULT_DENYLIST, EventScrubber
 
 from ..const import SUPERVISOR_VERSION
 from ..coresys import CoreSys
@@ -26,6 +27,7 @@ def init_sentry(coresys: CoreSys) -> None:
     """Initialize sentry client."""
     if not sentry_sdk.is_initialized():
         _LOGGER.info("Initializing Supervisor Sentry")
+        denylist = DEFAULT_DENYLIST + ["psk", "ssid"]
         # Don't use AsyncioIntegration(). We commonly handle task exceptions
         # outside of tasks. This would cause exception we gracefully handle to
         # be captured by sentry.
@@ -34,6 +36,7 @@ def init_sentry(coresys: CoreSys) -> None:
             before_send=partial(filter_data, coresys),
             auto_enabling_integrations=False,
             default_integrations=False,
+            event_scrubber=EventScrubber(denylist=denylist),
             integrations=[
                 AioHttpIntegration(
                     failed_request_status_codes=frozenset(range(500, 600))

supervisor/validate.py

@@ -182,7 +182,7 @@ SCHEMA_DOCKER_CONFIG = vol.Schema(
                 }
             }
         ),
-        vol.Optional(ATTR_ENABLE_IPV6): vol.Boolean(),
+        vol.Optional(ATTR_ENABLE_IPV6, default=None): vol.Maybe(vol.Boolean()),
     }
 )
 

tests/api/test_docker.py

@@ -19,7 +19,7 @@ async def test_api_docker_info(api_client: TestClient):
 
 async def test_api_network_enable_ipv6(coresys: CoreSys, api_client: TestClient):
     """Test setting docker network for enabled IPv6."""
-    assert coresys.docker.config.enable_ipv6 is False
+    assert coresys.docker.config.enable_ipv6 is None
 
     resp = await api_client.post("/docker/options", json={"enable_ipv6": True})
     assert resp.status == 200

tests/api/test_host.py

@@ -23,7 +23,11 @@ DEFAULT_RANGE_FOLLOW = "entries=:-99:18446744073709551615"
 @pytest.fixture(name="coresys_disk_info")
 async def fixture_coresys_disk_info(coresys: CoreSys) -> AsyncGenerator[CoreSys]:
     """Mock basic disk information for host APIs."""
-    coresys.hardware.disk.get_disk_life_time = lambda _: 0
+
+    async def mock_disk_lifetime(_):
+        return 0
+
+    coresys.hardware.disk.get_disk_life_time = mock_disk_lifetime
     coresys.hardware.disk.get_disk_free_space = lambda _: 5000
     coresys.hardware.disk.get_disk_total_space = lambda _: 50000
     coresys.hardware.disk.get_disk_used_space = lambda _: 45000

tests/backups/test_backup.py

@@ -185,6 +185,33 @@ async def test_consolidate(
     }
 
 
+@pytest.mark.usefixtures("tmp_supervisor_data")
+async def test_consolidate_failure(coresys: CoreSys, tmp_path: Path):
+    """Test consolidate with two backups that are not the same."""
+    (mount_dir := coresys.config.path_mounts / "backup_test").mkdir()
+    tar1 = Path(copy(get_fixture_path("test_consolidate_unc.tar"), tmp_path))
+    backup1 = Backup(coresys, tar1, "test", None)
+    await backup1.load()
+
+    tar2 = Path(copy(get_fixture_path("backup_example.tar"), mount_dir))
+    backup2 = Backup(coresys, tar2, "test", "backup_test")
+    await backup2.load()
+
+    with pytest.raises(
+        ValueError,
+        match=f"Backup {backup1.slug} and {backup2.slug} are not the same backup",
+    ):
+        backup1.consolidate(backup2)
+
+    # Force slugs to be the same to run the fields check
+    backup1._data["slug"] = backup2.slug  # pylint: disable=protected-access
+    with pytest.raises(
+        BackupInvalidError,
+        match=f"Cannot consolidate backups in {backup2.location} and {backup1.location} with slug {backup1.slug}",
+    ):
+        backup1.consolidate(backup2)
+
+
 @pytest.mark.parametrize(
     (
         "tarfile_side_effect",

tests/dbus/udisks2/test_nvme_controller.py

@@ -0,0 +1,72 @@
+"""Test UDisks2 NVMe Controller."""
+
+from datetime import UTC, datetime
+
+from dbus_fast.aio import MessageBus
+import pytest
+
+from supervisor.dbus.udisks2.nvme_controller import UDisks2NVMeController
+
+from tests.common import mock_dbus_services
+from tests.dbus_service_mocks.udisks2_nvme_controller import (
+    NVMeController as NVMeControllerService,
+)
+
+
+@pytest.fixture(name="nvme_controller_service")
+async def fixture_nvme_controller_service(
+    dbus_session_bus: MessageBus,
+) -> NVMeControllerService:
+    """Mock NVMe Controller service."""
+    yield (
+        await mock_dbus_services(
+            {
+                "udisks2_nvme_controller": "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC"
+            },
+            dbus_session_bus,
+        )
+    )["udisks2_nvme_controller"]
+
+
+async def test_nvme_controller_info(
+    nvme_controller_service: NVMeControllerService, dbus_session_bus: MessageBus
+):
+    """Test NVMe Controller info."""
+    controller = UDisks2NVMeController(
+        "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC"
+    )
+
+    assert controller.state is None
+    assert controller.unallocated_capacity is None
+    assert controller.smart_updated is None
+    assert controller.smart_temperature is None
+
+    await controller.connect(dbus_session_bus)
+
+    assert controller.state == "live"
+    assert controller.unallocated_capacity == 0
+    assert controller.smart_updated == datetime.fromtimestamp(1753906112, UTC)
+    assert controller.smart_temperature == 311
+
+    nvme_controller_service.emit_properties_changed({"SmartTemperature": 300})
+    await nvme_controller_service.ping()
+    await nvme_controller_service.ping()
+
+    assert controller.smart_temperature == 300
+
+
+@pytest.mark.usefixtures("nvme_controller_service")
+async def test_nvme_controller_smart_get_attributes(dbus_session_bus: MessageBus):
+    """Test NVMe Controller smart get attributes."""
+    controller = UDisks2NVMeController(
+        "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC"
+    )
+    await controller.connect(dbus_session_bus)
+
+    smart_log = await controller.smart_get_attributes()
+    assert smart_log.available_spare == 100
+    assert smart_log.percent_used == 1
+    assert smart_log.total_data_read == 22890461184000
+    assert smart_log.total_data_written == 27723431936000
+    assert smart_log.controller_busy_minutes == 2682
+    assert smart_log.temperature_sensors == [310, 305, 0, 0, 0, 0, 0, 0]

tests/dbus_service_mocks/udisks2_block.py

@@ -410,6 +410,33 @@ FIXTURES: dict[str, BlockFixture] = {
         HintSymbolicIconName="",
         UserspaceMountOptions=[],
     ),
+    "/org/freedesktop/UDisks2/block_devices/nvme0n1": BlockFixture(
+        Device=b"/dev/nvme0n1",
+        PreferredDevice=b"/dev/nvme0n1",
+        Symlinks=[],
+        DeviceNumber=66304,
+        Id="by-id-nvme-Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC",
+        Size=33554432,
+        ReadOnly=False,
+        Drive="/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC",
+        MDRaid="/",
+        MDRaidMember="/",
+        IdUsage="",
+        IdType="",
+        IdVersion="",
+        IdLabel="",
+        IdUUID="",
+        Configuration=[],
+        CryptoBackingDevice="/",
+        HintPartitionable=True,
+        HintSystem=True,
+        HintIgnore=False,
+        HintAuto=False,
+        HintName="",
+        HintIconName="",
+        HintSymbolicIconName="",
+        UserspaceMountOptions=[],
+    ),
 }
 
 

tests/dbus_service_mocks/udisks2_drive.py

@@ -177,6 +177,37 @@ FIXTURES: dict[str, DriveFixture] = {
         CanPowerOff=True,
         SiblingId="",
     ),
+    "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC": DriveFixture(
+        Vendor="",
+        Model="Samsung SSD 970 EVO Plus 2TB",
+        Revision="2B2QEXM7",
+        Serial="S40123456789ABC",
+        WWN="",
+        Id="Samsung-SSD-970-EVO-Plus-2TB-S40123456789ABC",
+        Configuration={},
+        Media="",
+        MediaCompatibility=[],
+        MediaRemovable=False,
+        MediaAvailable=True,
+        MediaChangeDetected=True,
+        Size=0,
+        TimeDetected=0,
+        TimeMediaDetected=0,
+        Optical=False,
+        OpticalBlank=False,
+        OpticalNumTracks=0,
+        OpticalNumAudioTracks=0,
+        OpticalNumDataTracks=0,
+        OpticalNumSessions=0,
+        RotationRate=0,
+        ConnectionBus="usb",
+        Seat="seat0",
+        Removable=True,
+        Ejectable=False,
+        SortKey="",
+        CanPowerOff=True,
+        SiblingId="",
+    ),
 }
 
 

tests/dbus_service_mocks/udisks2_manager.py

@@ -20,21 +20,25 @@ class UDisks2Manager(DBusServiceMock):
 
     interface = "org.freedesktop.UDisks2.Manager"
     object_path = "/org/freedesktop/UDisks2/Manager"
-    block_devices = [
-        "/org/freedesktop/UDisks2/block_devices/loop0",
-        "/org/freedesktop/UDisks2/block_devices/mmcblk1",
-        "/org/freedesktop/UDisks2/block_devices/mmcblk1p1",
-        "/org/freedesktop/UDisks2/block_devices/mmcblk1p2",
-        "/org/freedesktop/UDisks2/block_devices/mmcblk1p3",
-        "/org/freedesktop/UDisks2/block_devices/sda",
-        "/org/freedesktop/UDisks2/block_devices/sda1",
-        "/org/freedesktop/UDisks2/block_devices/sdb",
-        "/org/freedesktop/UDisks2/block_devices/sdb1",
-        "/org/freedesktop/UDisks2/block_devices/zram1",
-    ]
-    resolved_devices: list[list[str]] | list[str] = [
-        "/org/freedesktop/UDisks2/block_devices/sda1"
-    ]
+
+    def __init__(self):
+        """Initialize object."""
+        super().__init__()
+        self.block_devices = [
+            "/org/freedesktop/UDisks2/block_devices/loop0",
+            "/org/freedesktop/UDisks2/block_devices/mmcblk1",
+            "/org/freedesktop/UDisks2/block_devices/mmcblk1p1",
+            "/org/freedesktop/UDisks2/block_devices/mmcblk1p2",
+            "/org/freedesktop/UDisks2/block_devices/mmcblk1p3",
+            "/org/freedesktop/UDisks2/block_devices/sda",
+            "/org/freedesktop/UDisks2/block_devices/sda1",
+            "/org/freedesktop/UDisks2/block_devices/sdb",
+            "/org/freedesktop/UDisks2/block_devices/sdb1",
+            "/org/freedesktop/UDisks2/block_devices/zram1",
+        ]
+        self.resolved_devices: list[list[str]] | list[str] = [
+            "/org/freedesktop/UDisks2/block_devices/sda1"
+        ]
 
     @dbus_property(access=PropertyAccess.READ)
     def Version(self) -> "s":

tests/dbus_service_mocks/udisks2_nvme_controller.py

@@ -0,0 +1,138 @@
+"""Mock of UDisks2 Drive service."""
+
+from dbus_fast import Variant
+from dbus_fast.service import PropertyAccess, dbus_property
+
+from .base import DBusServiceMock, dbus_method
+
+BUS_NAME = "org.freedesktop.UDisks2"
+DEFAULT_OBJECT_PATH = (
+    "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC"
+)
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return NVMeController(object_path if object_path else DEFAULT_OBJECT_PATH)
+
+
+class NVMeController(DBusServiceMock):
+    """NVMe Controller mock.
+
+    gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/drives/id
+    """
+
+    interface = "org.freedesktop.UDisks2.NVMe.Controller"
+
+    def __init__(self, object_path: str):
+        """Initialize object."""
+        super().__init__()
+        self.object_path = object_path
+        self.smart_get_attributes_response = {
+            "avail_spare": Variant("y", 0x64),
+            "spare_thresh": Variant("y", 0x0A),
+            "percent_used": Variant("y", 0x01),
+            "total_data_read": Variant("t", 22890461184000),
+            "total_data_written": Variant("t", 27723431936000),
+            "ctrl_busy_time": Variant("t", 2682),
+            "power_cycles": Variant("t", 652),
+            "unsafe_shutdowns": Variant("t", 107),
+            "media_errors": Variant("t", 0),
+            "num_err_log_entries": Variant("t", 1069),
+            "temp_sensors": Variant("aq", [310, 305, 0, 0, 0, 0, 0, 0]),
+            "wctemp": Variant("q", 358),
+            "cctemp": Variant("q", 358),
+            "warning_temp_time": Variant("i", 0),
+            "critical_temp_time": Variant("i", 0),
+        }
+
+    @dbus_property(access=PropertyAccess.READ)
+    def State(self) -> "s":
+        """Get State."""
+        return "live"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def ControllerID(self) -> "q":
+        """Get ControllerID."""
+        return 4
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SubsystemNQN(self) -> "ay":
+        """Get SubsystemNQN."""
+        return b"nqn.2014.08.org.nvmexpress:144d144dS4J4NM0RB05961P     Samsung SSD 970 EVO Plus 2TB"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def FGUID(self) -> "s":
+        """Get FGUID."""
+        return ""
+
+    @dbus_property(access=PropertyAccess.READ)
+    def NVMeRevision(self) -> "s":
+        """Get NVMeRevision."""
+        return "1.3"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def UnallocatedCapacity(self) -> "t":
+        """Get UnallocatedCapacity."""
+        return 0
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SmartUpdated(self) -> "t":
+        """Get SmartUpdated."""
+        return 1753906112
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SmartCriticalWarning(self) -> "as":
+        """Get SmartCriticalWarning."""
+        return []
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SmartPowerOnHours(self) -> "t":
+        """Get SmartPowerOnHours."""
+        return 3208
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SmartTemperature(self) -> "q":
+        """Get SmartTemperature."""
+        return 311
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SmartSelftestStatus(self) -> "s":
+        """Get SmartSelftestStatus."""
+        return "success"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SmartSelftestPercentRemaining(self) -> "i":
+        """Get SmartSelftestPercentRemaining."""
+        return -1
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SanitizeStatus(self) -> "s":
+        """Get SanitizeStatus."""
+        return ""
+
+    @dbus_property(access=PropertyAccess.READ)
+    def SanitizePercentRemaining(self) -> "i":
+        """Get SanitizePercentRemaining."""
+        return -1
+
+    @dbus_method()
+    def SmartUpdate(self, options: "a{sv}") -> None:
+        """Do SmartUpdate."""
+
+    @dbus_method()
+    def SmartGetAttributes(self, options: "a{sv}") -> "a{sv}":
+        """Do SmartGetAttributes."""
+        return self.smart_get_attributes_response
+
+    @dbus_method()
+    def SmartSelftestStart(self, type_: "s", options: "a{sv}") -> None:
+        """Do SmartSelftestStart."""
+
+    @dbus_method()
+    def SmartSelftestAbort(self, options: "a{sv}") -> None:
+        """Do SmartSelftestAbort."""
+
+    @dbus_method()
+    def SanitizeStart(self, action: "s", options: "a{sv}") -> None:
+        """Do SanitizeStart."""

tests/docker/test_network.py

@@ -111,3 +111,39 @@ async def test_network_recreation(
             network_params[ATTR_ENABLE_IPV6] = new_enable_ipv6
 
             mock_create.assert_called_with(**network_params)
+
+
+async def test_network_default_ipv6_for_new_installations():
+    """Test that IPv6 is enabled by default when no user setting is provided (None)."""
+    with (
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker",
+            new_callable=PropertyMock,
+            return_value=MagicMock(),
+            create=True,
+        ),
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker.networks",
+            new_callable=PropertyMock,
+            return_value=MagicMock(),
+            create=True,
+        ),
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker.networks.get",
+            side_effect=docker.errors.NotFound("Network not found"),
+        ),
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker.networks.create",
+            return_value=MockNetwork(False, None, True),
+        ) as mock_create,
+    ):
+        # Pass None as enable_ipv6 to simulate no user setting
+        network = (await DockerNetwork(MagicMock()).post_init(None)).network
+
+        assert network is not None
+        assert network.attrs.get(DOCKER_ENABLEIPV6) is True
+
+        # Verify that create was called with IPv6 enabled by default
+        expected_params = DOCKER_NETWORK_PARAMS.copy()
+        expected_params[ATTR_ENABLE_IPV6] = True
+        mock_create.assert_called_with(**expected_params)

tests/hardware/test_disk.py

@@ -4,9 +4,71 @@
 from pathlib import Path
 from unittest.mock import patch
 
+from dbus_fast.aio import MessageBus
+import pytest
+
 from supervisor.coresys import CoreSys
 from supervisor.hardware.data import Device
 
+from tests.common import mock_dbus_services
+from tests.dbus_service_mocks.base import DBusServiceMock
+from tests.dbus_service_mocks.udisks2_manager import (
+    UDisks2Manager as UDisks2ManagerService,
+)
+from tests.dbus_service_mocks.udisks2_nvme_controller import (
+    NVMeController as NVMeControllerService,
+)
+
+MOCK_MOUNTINFO = """790 750 259:8 /supervisor /data rw,relatime master:118 - ext4 /dev/nvme0n1p8 rw,commit=30
+810 750 0:24 /systemd-journal-gatewayd.sock /run/systemd-journal-gatewayd.sock rw,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+811 750 0:24 /supervisor /run/os rw,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+813 750 0:24 /udev /run/udev ro,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+814 750 0:24 /machine-id /etc/machine-id ro - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+815 750 0:24 /docker.sock /run/docker.sock rw,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+816 750 0:24 /dbus /run/dbus ro,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+820 750 0:24 /containerd/containerd.sock /run/containerd/containerd.sock rw,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+821 750 0:24 /systemd/journal/socket /run/systemd/journal/socket rw,nosuid,nodev - tmpfs tmpfs rw,size=405464k,nr_inodes=819200,mode=755
+"""
+
+
+@pytest.fixture(name="nvme_data_disk")
+async def fixture_nvme_data_disk(
+    udisks2_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
+    coresys: CoreSys,
+    dbus_session_bus: MessageBus,
+) -> NVMeControllerService:
+    """Mock using an NVMe data disk."""
+    nvme_service = (
+        await mock_dbus_services(
+            {
+                "udisks2_block": "/org/freedesktop/UDisks2/block_devices/nvme0n1",
+                "udisks2_drive": "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC",
+                "udisks2_nvme_controller": "/org/freedesktop/UDisks2/drives/Samsung_SSD_970_EVO_Plus_2TB_S40123456789ABC",
+            },
+            dbus_session_bus,
+        )
+    )["udisks2_nvme_controller"]
+    udisks2_manager: UDisks2ManagerService = udisks2_services["udisks2_manager"]
+    udisks2_manager.block_devices.append(
+        "/org/freedesktop/UDisks2/block_devices/nvme0n1"
+    )
+    await coresys.dbus.udisks2.update()
+
+    with (
+        patch(
+            "supervisor.hardware.disk.Path.read_text",
+            return_value=MOCK_MOUNTINFO,
+        ),
+        patch("supervisor.hardware.disk.Path.is_block_device", return_value=True),
+        patch(
+            "supervisor.hardware.disk.Path.resolve",
+            return_value=Path(
+                "/sys/devices/platform/soc/ffe07000.nvme/nvme_host/nvme0/nvme0:0000/block/nvme0n1/nvme0n1p8"
+            ),
+        ),
+    ):
+        yield nvme_service
+
 
 def test_system_partition_disk(coresys: CoreSys):
     """Test if it is a system disk/partition."""
@@ -99,3 +161,19 @@ def test_try_get_emmc_life_time(coresys, tmp_path):
     ):
         value = coresys.hardware.disk._try_get_emmc_life_time("mmcblk0")
     assert value == 20.0
+
+
+async def test_try_get_nvme_life_time(
+    coresys: CoreSys, nvme_data_disk: NVMeControllerService
+):
+    """Test getting lifetime info from an NVMe."""
+    lifetime = await coresys.hardware.disk.get_disk_life_time(
+        coresys.config.path_supervisor
+    )
+    assert lifetime == 1
+
+    nvme_data_disk.smart_get_attributes_response["percent_used"].value = 50
+    lifetime = await coresys.hardware.disk.get_disk_life_time(
+        coresys.config.path_supervisor
+    )
+    assert lifetime == 50

tests/jobs/test_job_decorator.py

@@ -20,7 +20,7 @@ from supervisor.exceptions import (
 from supervisor.host.const import HostFeature
 from supervisor.host.manager import HostManager
 from supervisor.jobs import JobSchedulerOptions, SupervisorJob
-from supervisor.jobs.const import JobExecutionLimit
+from supervisor.jobs.const import JobConcurrency, JobExecutionLimit, JobThrottle
 from supervisor.jobs.decorator import Job, JobCondition
 from supervisor.jobs.job_group import JobGroup
 from supervisor.os.manager import OSManager
@@ -1212,3 +1212,93 @@ async def test_job_scheduled_at(coresys: CoreSys):
     assert job.name == "test_job_scheduled_at_job_task"
     assert job.stage == "work"
     assert job.parent_id is None
+
+
+async def test_concurency_reject_and_throttle(coresys: CoreSys):
+    """Test the concurrency rejct and throttle job execution limit."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+            self.run = asyncio.Lock()
+            self.call = 0
+
+        @Job(
+            name="test_concurency_reject_and_throttle_execute",
+            concurrency=JobConcurrency.REJECT,
+            throttle=JobThrottle.THROTTLE,
+            throttle_period=timedelta(hours=1),
+        )
+        async def execute(self, sleep: float):
+            """Execute the class method."""
+            assert not self.run.locked()
+            async with self.run:
+                await asyncio.sleep(sleep)
+            self.call += 1
+
+    test = TestClass(coresys)
+
+    results = await asyncio.gather(
+        *[test.execute(0.1), test.execute(0.1), test.execute(0.1)],
+        return_exceptions=True,
+    )
+    assert results[0] is None
+    assert isinstance(results[1], JobException)
+    assert isinstance(results[2], JobException)
+    assert test.call == 1
+
+    await asyncio.gather(*[test.execute(0.1)])
+    assert test.call == 1
+
+
+@pytest.mark.parametrize("error", [None, PluginJobError])
+async def test_concurency_reject_and_rate_limit(
+    coresys: CoreSys, error: JobException | None
+):
+    """Test the concurrency rejct and rate limit job execution limit."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+            self.run = asyncio.Lock()
+            self.call = 0
+
+        @Job(
+            name=f"test_concurency_reject_and_rate_limit_execute_{uuid4().hex}",
+            concurrency=JobConcurrency.REJECT,
+            throttle=JobThrottle.RATE_LIMIT,
+            throttle_period=timedelta(hours=1),
+            throttle_max_calls=1,
+            on_condition=error,
+        )
+        async def execute(self, sleep: float = 0):
+            """Execute the class method."""
+            async with self.run:
+                await asyncio.sleep(sleep)
+            self.call += 1
+
+    test = TestClass(coresys)
+
+    results = await asyncio.gather(
+        *[test.execute(0.1), test.execute(), test.execute()], return_exceptions=True
+    )
+    assert results[0] is None
+    assert isinstance(results[1], JobException)
+    assert isinstance(results[2], JobException)
+    assert test.call == 1
+
+    with pytest.raises(JobException if error is None else error):
+        await test.execute()
+
+    assert test.call == 1
+
+    with time_machine.travel(utcnow() + timedelta(hours=1)):
+        await test.execute()
+
+    assert test.call == 2

tests/os/test_manager.py

@@ -9,6 +9,7 @@ import pytest
 from supervisor.const import CoreState
 from supervisor.coresys import CoreSys
 from supervisor.exceptions import HassOSJobError
+from supervisor.resolution.const import UnhealthyReason
 
 from tests.common import MockResponse
 from tests.dbus_service_mocks.base import DBusServiceMock
@@ -85,6 +86,21 @@ async def test_update_fails_if_out_of_date(
         await coresys.os.update()
 
 
+async def test_update_fails_if_unhealthy(
+    coresys: CoreSys,
+) -> None:
+    """Test update of OS fails if Supervisor is unhealthy."""
+    await coresys.core.set_state(CoreState.RUNNING)
+    coresys.resolution.add_unhealthy_reason(UnhealthyReason.DUPLICATE_OS_INSTALLATION)
+    with (
+        patch.object(
+            type(coresys.os), "available", new=PropertyMock(return_value=True)
+        ),
+        pytest.raises(HassOSJobError),
+    ):
+        await coresys.os.update()
+
+
 async def test_board_name_supervised(coresys: CoreSys) -> None:
     """Test board name is supervised when not on haos."""
     with patch("supervisor.os.manager.CPE.get_product", return_value=["not-hassos"]):