Reset throttle in test and refactor helper

tests/common.py

@@ -1,12 +1,15 @@
 """Common test functions."""
 
+from datetime import datetime
 from importlib import import_module
+from inspect import getclosurevars
 import json
 from pathlib import Path
 from typing import Any
 
 from dbus_fast.aio.message_bus import MessageBus
 
+from supervisor.jobs.decorator import Job
 from supervisor.resolution.validate import get_valid_modules
 from supervisor.utils.yaml import read_yaml_file
 
@@ -82,3 +85,17 @@ async def mock_dbus_services(
                 services[module] = service_module.setup(to_mock[module]).export(bus)
 
     return services
+
+
+def get_job_decorator(func) -> Job:
+    """Get Job object of decorated function."""
+    # Access the closure of the wrapper function
+    job = getclosurevars(func).nonlocals["self"]
+    if not isinstance(job, Job):
+        raise TypeError(f"{func.__qualname__} is not a Job")
+    return job
+
+
+def reset_last_call(func, group: str | None = None) -> None:
+    """Reset last call for a function using the Job decorator."""
+    get_job_decorator(func).set_last_call(datetime.min, group)

tests/jobs/test_job_decorator.py

@@ -26,8 +26,11 @@
 from supervisor.os.manager import OSManager
 from supervisor.plugins.audio import PluginAudio
 from supervisor.resolution.const import UnhealthyReason
+from supervisor.supervisor import Supervisor
 from supervisor.utils.dt import utcnow
 
+from tests.common import reset_last_call
+
 
 async def test_healthy(coresys: CoreSys, caplog: pytest.LogCaptureFixture):
     """Test the healty decorator."""
@@ -73,6 +76,7 @@ async def test_internet(
 ):
     """Test the internet decorator."""
     coresys.core.state = CoreState.RUNNING
+    reset_last_call(Supervisor.check_connectivity)
 
     class TestClass:
         """Test class."""

tests/test_supervisor.py

@@ -23,7 +23,7 @@
 from supervisor.resolution.data import Issue
 from supervisor.supervisor import Supervisor
 
-from tests.utils.test_decorator import get_job_decorator
+from tests.common import reset_last_call
 
 
 @pytest.fixture(name="websession", scope="function")
@@ -81,9 +81,7 @@ async def test_connectivity_check_throttling(
     coresys.supervisor.connectivity = None
     websession.head.side_effect = side_effect
 
-    job_decorator = get_job_decorator(Supervisor.check_connectivity)
-    job_decorator._last_call.clear()  # pylint: disable=W0212
-
+    reset_last_call(Supervisor.check_connectivity)
     with (
         travel(datetime.now(), tick=False) as traveller,
     ):