Check Core version and raise unsupported if older than 2 years

Check the currently installed Core version relative to the current
date, and if its older than 2 years, mark the system unsupported.
Also add a Job condition to prevent automatic refreshing of the update
information in this case.

Author: agners

supervisor/jobs/const.py

@@ -20,6 +20,7 @@ class JobCondition(StrEnum):
     """Job condition enum."""
 
     AUTO_UPDATE = "auto_update"
+    CORE_SUPPORTED = "core_supported"
     FREE_SPACE = "free_space"
     FROZEN = "frozen"
     HAOS = "haos"

supervisor/jobs/decorator.py

@@ -404,6 +404,14 @@ async def check_conditions(
                 f"'{method_name}' blocked from execution, unsupported OS version"
             )
 
+        if (
+            JobCondition.CORE_SUPPORTED in used_conditions
+            and UnsupportedReason.CORE_VERSION in coresys.sys_resolution.unsupported
+        ):
+            raise JobConditionException(
+                f"'{method_name}' blocked from execution, unsupported Core version"
+            )
+
         if (
             JobCondition.HOST_NETWORK in used_conditions
             and not coresys.sys_dbus.network.is_connected

supervisor/resolution/const.py

@@ -40,6 +40,7 @@ class UnsupportedReason(StrEnum):
     CGROUP_VERSION = "cgroup_version"
     CONNECTIVITY_CHECK = "connectivity_check"
     CONTENT_TRUST = "content_trust"
+    CORE_VERSION = "core_version"
     DBUS = "dbus"
     DNS_SERVER = "dns_server"
     DOCKER_CONFIGURATION = "docker_configuration"

supervisor/resolution/evaluations/core_version.py

@@ -0,0 +1,72 @@
+"""Evaluation class for Core version."""
+
+from datetime import datetime, timedelta
+
+from awesomeversion import AwesomeVersion, AwesomeVersionException
+
+from ...const import CoreState
+from ...coresys import CoreSys
+from ..const import UnsupportedReason
+from .base import EvaluateBase
+
+
+def setup(coresys: CoreSys) -> EvaluateBase:
+    """Initialize evaluation-setup function."""
+    return EvaluateCoreVersion(coresys)
+
+
+class EvaluateCoreVersion(EvaluateBase):
+    """Evaluate the Home Assistant Core version."""
+
+    @property
+    def reason(self) -> UnsupportedReason:
+        """Return a UnsupportedReason enum."""
+        return UnsupportedReason.CORE_VERSION
+
+    @property
+    def on_failure(self) -> str:
+        """Return a string that is printed when self.evaluate is True."""
+        return f"Home Assistant Core version '{self.sys_homeassistant.version}' is more than 2 years old!"
+
+    @property
+    def states(self) -> list[CoreState]:
+        """Return a list of valid states when this evaluation can run."""
+        return [CoreState.RUNNING, CoreState.SETUP]
+
+    async def evaluate(self) -> bool:
+        """Run evaluation."""
+        if not (current := self.sys_homeassistant.version) or not (
+            latest := self.sys_homeassistant.latest_version
+        ):
+            return False
+
+        try:
+            # Calculate if the current version was released more than 2 years ago
+            # Home Assistant releases happen monthly, so approximately 24 versions per 2 years
+            # However, we'll be more precise and check based on actual version numbers
+            # Home Assistant follows a versioning scheme like 2024.1, 2024.2, etc.
+
+            # Extract year from current version
+            current_year = int(str(current).split(".")[0])
+
+            # Calculate 2 years ago from now
+            two_years_ago = datetime.now() - timedelta(days=730)  # 2 years = 730 days
+            cutoff_year = two_years_ago.year
+            cutoff_month = two_years_ago.month
+
+            # Create a cutoff version based on the date 2 years ago
+            cutoff_version = AwesomeVersion(f"{cutoff_year}.{cutoff_month}")
+
+            # Compare current version with the cutoff
+            return current < cutoff_version
+
+        except (AwesomeVersionException, ValueError, IndexError):
+            # If we can't parse the version format, fall back to conservative approach
+            # Consider unsupported if current is significantly behind latest
+            try:
+                # If latest version is from current year and current is from 2+ years ago
+                latest_year = int(str(latest).split(".")[0])
+                current_year = int(str(current).split(".")[0])
+                return (latest_year - current_year) >= 2
+            except (ValueError, IndexError):
+                return False

tests/jobs/test_job_decorator.py

@@ -25,7 +25,7 @@
 from supervisor.jobs.job_group import JobGroup
 from supervisor.os.manager import OSManager
 from supervisor.plugins.audio import PluginAudio
-from supervisor.resolution.const import UnhealthyReason
+from supervisor.resolution.const import UnhealthyReason, UnsupportedReason
 from supervisor.supervisor import Supervisor
 from supervisor.utils.dt import utcnow
 
@@ -1384,3 +1384,31 @@ async def nested_method(self) -> None:
 
     assert test.call_count == 2  # Should execute now
     assert test.nested_call_count == 2  # Nested call should also execute
+
+
+async def test_core_supported(coresys: CoreSys, caplog: pytest.LogCaptureFixture):
+    """Test the core_supported decorator."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+
+        @Job(
+            name="test_core_supported_execute", conditions=[JobCondition.CORE_SUPPORTED]
+        )
+        async def execute(self):
+            """Execute the class method."""
+            return True
+
+    test = TestClass(coresys)
+    assert await test.execute()
+
+    coresys.resolution.unsupported.append(UnsupportedReason.CORE_VERSION)
+    assert not await test.execute()
+    assert "blocked from execution, unsupported Core version" in caplog.text
+
+    coresys.jobs.ignore_conditions = [JobCondition.CORE_SUPPORTED]
+    assert await test.execute()

tests/resolution/evaluation/test_evaluate_core_version.py

@@ -0,0 +1,121 @@
+"""Test Core Version evaluation."""
+
+from datetime import datetime
+from unittest.mock import PropertyMock, patch
+
+from awesomeversion import AwesomeVersion
+import pytest
+
+from supervisor.const import CoreState
+from supervisor.coresys import CoreSys
+from supervisor.homeassistant.module import HomeAssistant
+from supervisor.resolution.evaluations.core_version import EvaluateCoreVersion
+
+
+@pytest.mark.parametrize(
+    "current,expected",
+    [
+        ("2022.1.0", True),  # More than 2 years old, should be unsupported
+        ("2023.12.0", False),  # Less than 2 years old, should be supported
+        (f"{datetime.now().year}.1", False),  # Current year, supported
+        (f"{datetime.now().year - 1}.12", False),  # 1 year old, supported
+        (f"{datetime.now().year - 2}.1", True),  # 2 years old, unsupported
+        (f"{datetime.now().year - 3}.1", True),  # 3 years old, unsupported
+        ("2021.6.0", True),  # Very old version, unsupported
+        (None, False),  # No current version info, check skipped
+    ],
+)
+async def test_core_version_evaluation(
+    coresys: CoreSys, current: str | None, expected: bool
+):
+    """Test evaluation logic on Core versions."""
+    evaluation = EvaluateCoreVersion(coresys)
+    await coresys.core.set_state(CoreState.RUNNING)
+
+    with (
+        patch.object(
+            HomeAssistant,
+            "version",
+            new=PropertyMock(return_value=current and AwesomeVersion(current)),
+        ),
+        patch.object(
+            HomeAssistant,
+            "latest_version",
+            new=PropertyMock(
+                return_value=AwesomeVersion("2024.12.0")
+            ),  # Mock latest version
+        ),
+    ):
+        assert evaluation.reason not in coresys.resolution.unsupported
+        await evaluation()
+        assert (evaluation.reason in coresys.resolution.unsupported) is expected
+
+
+async def test_core_version_evaluation_no_latest(coresys: CoreSys):
+    """Test evaluation when no latest version is available."""
+    evaluation = EvaluateCoreVersion(coresys)
+    await coresys.core.set_state(CoreState.RUNNING)
+
+    with (
+        patch.object(
+            HomeAssistant,
+            "version",
+            new=PropertyMock(return_value=AwesomeVersion("2022.1.0")),
+        ),
+        patch.object(
+            HomeAssistant,
+            "latest_version",
+            new=PropertyMock(return_value=None),
+        ),
+    ):
+        assert evaluation.reason not in coresys.resolution.unsupported
+        await evaluation()
+        assert evaluation.reason not in coresys.resolution.unsupported
+
+
+async def test_core_version_invalid_format(coresys: CoreSys):
+    """Test evaluation with invalid version format."""
+    evaluation = EvaluateCoreVersion(coresys)
+    await coresys.core.set_state(CoreState.RUNNING)
+
+    with (
+        patch.object(
+            HomeAssistant,
+            "version",
+            new=PropertyMock(return_value=AwesomeVersion("invalid.version")),
+        ),
+        patch.object(
+            HomeAssistant,
+            "latest_version",
+            new=PropertyMock(return_value=AwesomeVersion("2024.12.0")),
+        ),
+    ):
+        assert evaluation.reason not in coresys.resolution.unsupported
+        await evaluation()
+        # Should handle gracefully and not mark as unsupported
+        assert evaluation.reason not in coresys.resolution.unsupported
+
+
+async def test_did_run(coresys: CoreSys):
+    """Test that the evaluation ran as expected."""
+    evaluation = EvaluateCoreVersion(coresys)
+    should_run = evaluation.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.evaluations.core_version.EvaluateCoreVersion.evaluate",
+        return_value=None,
+    ) as evaluate:
+        for state in should_run:
+            await coresys.core.set_state(state)
+            await evaluation()
+            evaluate.assert_called_once()
+            evaluate.reset_mock()
+
+        for state in should_not_run:
+            await coresys.core.set_state(state)
+            await evaluation()
+            evaluate.assert_not_called()
+            evaluate.reset_mock()