From fdde95d849e3385764b820bff3beae453a77d65e Mon Sep 17 00:00:00 2001 From: Mike Degatano Date: Wed, 6 Aug 2025 04:40:48 -0400 Subject: [PATCH] Add an issue for disk lifetime >90% (#6069) --- supervisor/resolution/checks/disk_lifetime.py | 51 +++++++ supervisor/resolution/const.py | 1 + .../check/test_check_disk_lifetime.py | 131 ++++++++++++++++++ 3 files changed, 183 insertions(+) create mode 100644 supervisor/resolution/checks/disk_lifetime.py create mode 100644 tests/resolution/check/test_check_disk_lifetime.py diff --git a/supervisor/resolution/checks/disk_lifetime.py b/supervisor/resolution/checks/disk_lifetime.py new file mode 100644 index 000000000..4caf86b1d --- /dev/null +++ b/supervisor/resolution/checks/disk_lifetime.py @@ -0,0 +1,51 @@ +"""Helpers to check disk lifetime issues.""" + +from ...const import CoreState +from ...coresys import CoreSys +from ..const import ContextType, IssueType +from .base import CheckBase + + +def setup(coresys: CoreSys) -> CheckBase: + """Check setup function.""" + return CheckDiskLifetime(coresys) + + +class CheckDiskLifetime(CheckBase): + """Storage class for check.""" + + async def run_check(self) -> None: + """Run check if not affected by issue.""" + if await self.approve_check(): + self.sys_resolution.create_issue( + IssueType.DISK_LIFETIME, ContextType.SYSTEM + ) + + async def approve_check(self, reference: str | None = None) -> bool: + """Approve check if it is affected by issue.""" + # Get the current data disk device + if not self.sys_dbus.agent.datadisk.current_device: + return False + + # Check disk lifetime + lifetime = await self.sys_hardware.disk.get_disk_life_time( + self.sys_dbus.agent.datadisk.current_device + ) + + # Issue still exists if lifetime is >= 90% + return lifetime is not None and lifetime >= 90 + + @property + def issue(self) -> IssueType: + """Return a IssueType enum.""" + return IssueType.DISK_LIFETIME + + @property + def context(self) -> ContextType: + """Return a ContextType enum.""" + return ContextType.SYSTEM + + @property + def states(self) -> list[CoreState]: + """Return a list of valid states when this check can run.""" + return [CoreState.RUNNING, CoreState.STARTUP] diff --git a/supervisor/resolution/const.py b/supervisor/resolution/const.py index d3e1431d8..a6f2c9058 100644 --- a/supervisor/resolution/const.py +++ b/supervisor/resolution/const.py @@ -84,6 +84,7 @@ class IssueType(StrEnum): DETACHED_ADDON_REMOVED = "detached_addon_removed" DEVICE_ACCESS_MISSING = "device_access_missing" DISABLED_DATA_DISK = "disabled_data_disk" + DISK_LIFETIME = "disk_lifetime" DNS_LOOP = "dns_loop" DUPLICATE_OS_INSTALLATION = "duplicate_os_installation" DNS_SERVER_FAILED = "dns_server_failed" diff --git a/tests/resolution/check/test_check_disk_lifetime.py b/tests/resolution/check/test_check_disk_lifetime.py new file mode 100644 index 000000000..01b37823b --- /dev/null +++ b/tests/resolution/check/test_check_disk_lifetime.py @@ -0,0 +1,131 @@ +"""Test check disk lifetime fixup.""" + +# pylint: disable=import-error,protected-access +from unittest.mock import PropertyMock, patch + +import pytest + +from supervisor.const import CoreState +from supervisor.coresys import CoreSys +from supervisor.resolution.checks.disk_lifetime import CheckDiskLifetime +from supervisor.resolution.const import ContextType, IssueType +from supervisor.resolution.data import Issue + + +async def test_base(coresys: CoreSys): + """Test check basics.""" + disk_lifetime = CheckDiskLifetime(coresys) + assert disk_lifetime.slug == "disk_lifetime" + assert disk_lifetime.enabled + + +async def test_check_no_data_disk(coresys: CoreSys): + """Test check when no data disk is available.""" + disk_lifetime = CheckDiskLifetime(coresys) + await coresys.core.set_state(CoreState.RUNNING) + + # Mock no data disk + with patch.object( + type(coresys.dbus.agent.datadisk), + "current_device", + new=PropertyMock(return_value=None), + ): + await disk_lifetime() + + assert len(coresys.resolution.issues) == 0 + + +@pytest.mark.parametrize( + ("lifetime", "has_issue"), + [(0.0, False), (85.0, False), (90.0, True), (95.0, True), (None, False)], +) +async def test_check_lifetime_threshold( + coresys: CoreSys, lifetime: float | None, has_issue: bool +): + """Test check when disk lifetime at thresholds.""" + disk_lifetime = CheckDiskLifetime(coresys) + await coresys.core.set_state(CoreState.RUNNING) + + # Mock data disk with lifetime + with ( + patch.object( + type(coresys.dbus.agent.datadisk), + "current_device", + new=PropertyMock(return_value="/dev/sda1"), + ), + patch.object( + coresys.hardware.disk, + "get_disk_life_time", + return_value=lifetime, + ), + ): + await disk_lifetime() + + assert ( + Issue(IssueType.DISK_LIFETIME, ContextType.SYSTEM) in coresys.resolution.issues + ) is has_issue + + +async def test_approve_no_data_disk(coresys: CoreSys): + """Test approve when no data disk is available.""" + disk_lifetime = CheckDiskLifetime(coresys) + + # Mock no data disk + with patch.object( + type(coresys.dbus.agent.datadisk), + "current_device", + new=PropertyMock(return_value=None), + ): + assert not await disk_lifetime.approve_check() + + +@pytest.mark.parametrize( + ("lifetime", "approved"), + [(0.0, False), (85.0, False), (90.0, True), (95.0, True), (None, False)], +) +async def test_approve_check_lifetime_threshold( + coresys: CoreSys, lifetime: float | None, approved: bool +): + """Test approve check when disk lifetime at thresholds.""" + disk_lifetime = CheckDiskLifetime(coresys) + await coresys.core.set_state(CoreState.RUNNING) + + # Mock data disk with lifetime + with ( + patch.object( + type(coresys.dbus.agent.datadisk), + "current_device", + new=PropertyMock(return_value="/dev/sda1"), + ), + patch.object( + coresys.hardware.disk, + "get_disk_life_time", + return_value=lifetime, + ), + ): + assert await disk_lifetime.approve_check() is approved + + +async def test_did_run(coresys: CoreSys): + """Test that the check ran as expected.""" + disk_lifetime = CheckDiskLifetime(coresys) + should_run = disk_lifetime.states + should_not_run = [state for state in CoreState if state not in should_run] + assert len(should_run) != 0 + assert len(should_not_run) != 0 + + with patch( + "supervisor.resolution.checks.disk_lifetime.CheckDiskLifetime.run_check", + return_value=None, + ) as check: + for state in should_run: + await coresys.core.set_state(state) + await disk_lifetime() + check.assert_called_once() + check.reset_mock() + + for state in should_not_run: + await coresys.core.set_state(state) + await disk_lifetime() + check.assert_not_called() + check.reset_mock()