diff --git a/src/plugins/server_status/icon.png b/src/plugins/server_status/icon.png new file mode 100644 index 000000000..25ad14cea Binary files /dev/null and b/src/plugins/server_status/icon.png differ diff --git a/src/plugins/server_status/plugin-info.json b/src/plugins/server_status/plugin-info.json new file mode 100644 index 000000000..9ef6b1467 --- /dev/null +++ b/src/plugins/server_status/plugin-info.json @@ -0,0 +1,9 @@ +{ + "name": "Server Status", + "id": "server_status", + "description": "Shows WireGuard and Pi-hole status", + "version": "1.0.0", + "author": "robi", + "entry_point": "server_status.py", + "class": "ServerStatus" +} diff --git a/src/plugins/server_status/server_status.py b/src/plugins/server_status/server_status.py new file mode 100644 index 000000000..1b7d1124d --- /dev/null +++ b/src/plugins/server_status/server_status.py @@ -0,0 +1,242 @@ +import subprocess +import logging +from datetime import datetime + +from PIL import Image, ImageDraw +from plugins.base_plugin.base_plugin import BasePlugin +import socket +import psutil + + + +logger = logging.getLogger(__name__) + + +class ServerStatus(BasePlugin): + + def __init__(self, config): + super().__init__(config) + self.config = config + + def generate_settings_template(self): + return super().generate_settings_template() + + def generate_image(self, settings, device_config): + + status = self.get_status() + + w, h = device_config.get_resolution() + + img = Image.new("RGB", (w, h), "white") + draw = ImageDraw.Draw(img) + + # Font opzionale + try: + from PIL import ImageFont + font = ImageFont.truetype( + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + 18 + ) + title_font = ImageFont.truetype( + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", + 22 + ) + except Exception: + font = None + title_font = None + + + # Titolo + draw.text( + (20, 15), + status["title"], + fill="black", + font=title_font + ) + + + # Linea separazione + draw.line( + (w // 2, 45, w // 2, h - 30), + fill="black", + width=1 + ) + + + # Colonna sinistra - Sistema + x1 = 15 + y1 = 55 + + system_lines = [ + f"Temp : {status['temperature']} C", + f"CPU : {status['cpu']:.0f} %", + f"RAM : {status['ram']:.0f} %", + f"Disk : {status['disk']:.0f} %", + f"Up : {status['uptime']}", + ] + + for text in system_lines: + draw.text( + (x1, y1), + text, + fill="black", + font=font + ) + y1 += 25 + + + # Colonna destra - Servizi + x2 = w // 2 + 15 + y2 = 55 + + service_lines = [ + f"Net: {status['network']}", + f"IP: {status['ip']}", + "", + f"Pi-hole:", + f" {status['pihole']}", + "", + f"WireGuard:", + f" {status['wireguard']}", + f"Peers: {status['peers']}", + ] + + for text in service_lines: + draw.text( + (x2, y2), + text, + fill="black", + font=font + ) + y2 += 23 + + + # Ora aggiornamento + draw.text( + (20, h - 25), + f"Aggiornato: {status['time']}", + fill="black", + font=font + ) + + + return img + + + + def get_status(self): + + system = self.get_system_info() + + return { + "title": "RASPI STATUS", + + "temperature": system["temperature"], + "cpu": system["cpu"], + "ram": system["ram"], + "disk": system["disk"], + "uptime": system["uptime"], + + "network": system["network"], + "ip": system["ip"], + + "pihole": self.service_status("pihole-FTL"), + "wireguard": self.service_status("wg-quick@wg0"), + "peers": self.wg_peers(), + + "time": datetime.now().strftime("%H:%M") + } + + + def get_system_info(self): + + # Temperatura CPU + try: + with open("/sys/class/thermal/thermal_zone0/temp") as f: + temperature = round(int(f.read()) / 1000, 1) + except Exception: + temperature = None + + # CPU + cpu = psutil.cpu_percent(interval=0.5) + + # RAM + ram = psutil.virtual_memory().percent + + # Disco (/) + disk = psutil.disk_usage("/").percent + + # Uptime + boot = datetime.fromtimestamp(psutil.boot_time()) + uptime = datetime.now() - boot + + days = uptime.days + hours = uptime.seconds // 3600 + + if days > 0: + uptime_str = f"{days}d {hours}h" + else: + uptime_str = f"{hours}h" + + # IP locale + ip = "-" + + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0] + s.close() + network = "OK" + except Exception: + network = "DOWN" + + return { + "temperature": temperature, + "cpu": cpu, + "ram": ram, + "disk": disk, + "uptime": uptime_str, + "network": network, + "ip": ip + } + + + + def service_status(self, service): + try: + result = subprocess.check_output( + ["systemctl", "is-active", service], + text=True + ).strip() + + return "OK" if result == "active" else "DOWN" + + except Exception: + return "DOWN" + + + def wg_peers(self): + try: + data = subprocess.check_output( + ["wg", "show", "wg0", "latest-handshakes"], + text=True + ) + + now = int(datetime.now().timestamp()) + active = 0 + + for line in data.strip().splitlines(): + parts = line.split() + + if len(parts) == 2: + timestamp = int(parts[1]) + + if timestamp > now - 300: + active += 1 + + return active + + except Exception as e: + logger.error(e) + return -1 + diff --git a/src/refresh_task.py b/src/refresh_task.py index f554e2adb..d64ffcbe7 100644 --- a/src/refresh_task.py +++ b/src/refresh_task.py @@ -73,7 +73,7 @@ def _run(self): while True: try: with self.condition: - sleep_time = self.device_config.get_config("plugin_cycle_interval_seconds", default=60*60) + sleep_time = self._get_sleep_time() # Wait for sleep_time or until notified self.condition.wait(timeout=sleep_time) @@ -115,7 +115,17 @@ def _run(self): image_hash = compute_image_hash(image) refresh_info = refresh_action.get_refresh_info() - refresh_info.update({"refresh_time": current_dt.isoformat(), "image_hash": image_hash}) + refresh_info.update({"image_hash": image_hash}) + + # Only update the global rotation time when the plugin actually + # changes (rotation or manual update). In-place refreshes of the + # same plugin (e.g., a clock updating every 60 seconds) must not + # reset the playlist rotation timer. + if self._is_rotation_refresh(refresh_info, latest_refresh): + refresh_info["refresh_time"] = current_dt.isoformat() + else: + refresh_info["refresh_time"] = latest_refresh.refresh_time + # check if image is the same as current image if image_hash != latest_refresh.image_hash: logger.info(f"Updating display. | refresh_info: {refresh_info}") @@ -160,8 +170,60 @@ def _get_current_datetime(self): tz_str = self.device_config.get_config("timezone", default="UTC") return datetime.now(pytz.timezone(tz_str)) + def _get_sleep_time(self): + """Determines how long to sleep before the next refresh check. + + Returns the minimum of the time until the next global playlist rotation + and the time until the currently active plugin instance's next refresh, + so that plugins requiring frequent refreshes (e.g., a clock) are re-rendered + while they remain displayed, and the playlist still advances on schedule. + """ + global_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) + + playlist_manager = self.device_config.get_playlist_manager() + current_dt = self._get_current_datetime() + playlist = playlist_manager.determine_active_playlist(current_dt) + + if not playlist or not playlist.plugins: + return global_interval + + # Time until the next global rotation (plugin change) + time_until_rotation = global_interval + latest_refresh = self.device_config.get_refresh_info() + latest_refresh_dt = latest_refresh.get_refresh_datetime() + if latest_refresh_dt: + if latest_refresh_dt.tzinfo is None: + latest_refresh_dt = latest_refresh_dt.replace(tzinfo=current_dt.tzinfo) + time_since_rotation = (current_dt - latest_refresh_dt).total_seconds() + time_until_rotation = max(0, global_interval - time_since_rotation) + + # Time until the current plugin's next refresh + time_until_plugin_refresh = global_interval + if playlist.current_plugin_index is not None and playlist.current_plugin_index < len(playlist.plugins): + plugin_instance = playlist.plugins[playlist.current_plugin_index] + if "interval" in plugin_instance.refresh: + interval = plugin_instance.refresh.get("interval") + if interval: + latest_refresh_dt = plugin_instance.get_latest_refresh_dt() + if latest_refresh_dt: + if latest_refresh_dt.tzinfo is None: + latest_refresh_dt = latest_refresh_dt.replace(tzinfo=current_dt.tzinfo) + time_since_refresh = (current_dt - latest_refresh_dt).total_seconds() + time_until_plugin_refresh = max(0, interval - time_since_refresh) + else: + time_until_plugin_refresh = interval + + return min(time_until_rotation, time_until_plugin_refresh) + def _determine_next_plugin(self, playlist_manager, latest_refresh_info, current_dt): - """Determines the next plugin to refresh based on the active playlist, plugin cycle interval, and current time.""" + """Determines the next plugin to refresh based on the active playlist, plugin cycle interval, and current time. + + Priority: + 1. If the global rotation interval has elapsed, advance to the next plugin. + 2. Otherwise, if the currently displayed plugin instance has its own refresh + interval that has elapsed, refresh it in place. + 3. Otherwise, do nothing. + """ playlist = playlist_manager.determine_active_playlist(current_dt) if not playlist: playlist_manager.active_playlist = None @@ -173,20 +235,40 @@ def _determine_next_plugin(self, playlist_manager, latest_refresh_info, current_ logger.info(f"Active playlist '{playlist.name}' has no plugins.") return None, None + # 1. Check if it's time to rotate to the next plugin in the playlist. latest_refresh_dt = latest_refresh_info.get_refresh_datetime() plugin_cycle_interval = self.device_config.get_config("plugin_cycle_interval_seconds", default=3600) - should_refresh = PlaylistManager.should_refresh(latest_refresh_dt, plugin_cycle_interval, current_dt) - - if not should_refresh: - latest_refresh_str = latest_refresh_dt.strftime('%Y-%m-%d %H:%M:%S') if latest_refresh_dt else "None" - logger.info(f"Not time to update display. | latest_update: {latest_refresh_str} | plugin_cycle_interval: {plugin_cycle_interval}") - return None, None - - plugin = playlist.get_next_plugin() - logger.info(f"Determined next plugin. | active_playlist: {playlist.name} | plugin_instance: {plugin.name}") + should_rotate = PlaylistManager.should_refresh(latest_refresh_dt, plugin_cycle_interval, current_dt) + + if should_rotate: + plugin = playlist.get_next_plugin() + logger.info(f"Determined next plugin. | active_playlist: {playlist.name} | plugin_instance: {plugin.name}") + return playlist, plugin + + # 2. Otherwise, check if the currently displayed plugin instance needs a + # refresh based on its own settings (e.g., a clock refreshing every 60s). + if playlist.current_plugin_index is not None and playlist.current_plugin_index < len(playlist.plugins): + current_plugin = playlist.plugins[playlist.current_plugin_index] + if current_plugin.should_refresh(current_dt): + logger.info(f"Refreshing currently displayed plugin instance. | plugin_instance: {current_plugin.name}") + return playlist, current_plugin + + # 3. Nothing to do. + latest_refresh_str = latest_refresh_dt.strftime('%Y-%m-%d %H:%M:%S') if latest_refresh_dt else "None" + logger.info(f"Not time to update display. | latest_update: {latest_refresh_str} | plugin_cycle_interval: {plugin_cycle_interval}") + return None, None + + def _is_rotation_refresh(self, refresh_info, latest_refresh): + """Determines whether a refresh action represents a rotation (plugin change) + or a manual update, as opposed to an in-place refresh of the same plugin. + + In-place refreshes (e.g., a clock updating every 60 seconds) must not reset + the global playlist rotation timer. + """ + if refresh_info.get("refresh_type") != "Playlist": + return True + return refresh_info.get("plugin_instance") != latest_refresh.plugin_instance - return playlist, plugin - def log_system_stats(self): metrics = { 'cpu_percent': psutil.cpu_percent(interval=1), diff --git a/tests/test_refresh_task.py b/tests/test_refresh_task.py new file mode 100644 index 000000000..12c1d40cb --- /dev/null +++ b/tests/test_refresh_task.py @@ -0,0 +1,320 @@ +import pytest +import pytz +from datetime import datetime, timedelta +from unittest.mock import MagicMock, patch + +from src.model import Playlist, PlaylistManager, PluginInstance, RefreshInfo +from src.refresh_task import RefreshTask, PlaylistRefresh + + +class MockDeviceConfig: + """Mock device config for testing.""" + def __init__(self, config=None, playlist_manager=None, refresh_info=None): + self.config = config or { + "plugin_cycle_interval_seconds": 600, + "timezone": "UTC", + } + self.playlist_manager = playlist_manager or PlaylistManager() + self.refresh_info = refresh_info or RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=datetime.now(pytz.UTC).isoformat(), + image_hash=0, + playlist="Test Playlist", + plugin_instance="Clock", + ) + self.plugin_image_dir = "/tmp" + self.current_image_file = "/tmp/current.png" + + def get_config(self, key=None, default=None): + if key is None: + return self.config + return self.config.get(key, default) + + def get_playlist_manager(self): + return self.playlist_manager + + def get_refresh_info(self): + return self.refresh_info + + def get_plugin(self, plugin_id): + return {"id": plugin_id, "image_settings": []} + + def write_config(self): + pass + + +class MockDisplayManager: + def display_image(self, image, image_settings=[]): + pass + + +def make_plugin_instance(plugin_id="clock", name="Clock", refresh=None, latest_refresh_time=None): + return PluginInstance( + plugin_id=plugin_id, + name=name, + settings={}, + refresh=refresh if refresh is not None else {"interval": 60}, + latest_refresh_time=latest_refresh_time, + ) + + +def make_playlist(plugins, current_plugin_index=0): + return Playlist( + name="Test Playlist", + start_time="00:00", + end_time="24:00", + plugins=[p.to_dict() for p in plugins], + current_plugin_index=current_plugin_index, + ) + + +def make_refresh_info(plugin_instance="Clock", refresh_time=None, image_hash=0): + if refresh_time is None: + refresh_time = datetime.now(pytz.UTC).isoformat() + return RefreshInfo( + refresh_type="Playlist", + plugin_id="clock", + refresh_time=refresh_time, + image_hash=image_hash, + playlist="Test Playlist", + plugin_instance=plugin_instance, + ) + + +class TestGetSleepTime: + def test_returns_global_interval_when_no_playlist(self): + device_config = MockDeviceConfig() + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == 600 + + def test_returns_global_interval_when_no_plugin_interval(self): + plugin = make_plugin_instance(refresh={}) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == pytest.approx(600, abs=1) + + def test_returns_plugin_interval_when_shorter_than_global(self): + plugin = make_plugin_instance(refresh={"interval": 60}) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == 60 + + def test_returns_time_until_refresh_when_partially_elapsed(self): + # Plugin was refreshed 30 seconds ago with a 60 second interval + latest_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=30)).isoformat() + plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + # Should be ~30 seconds remaining + assert 25 <= sleep_time <= 35 + + def test_returns_global_interval_when_plugin_interval_larger(self): + plugin = make_plugin_instance(refresh={"interval": 7200}) + playlist = make_playlist([plugin]) + manager = PlaylistManager(playlists=[playlist]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + sleep_time = task._get_sleep_time() + assert sleep_time == pytest.approx(600, abs=1) + + +class TestDetermineNextPlugin: + def test_refreshes_current_plugin_when_its_interval_elapsed(self): + """A clock with interval 60 is refreshed in place while it's the current plugin.""" + # Current plugin needs refresh (interval elapsed) + latest_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=120)).isoformat() + plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + # Set global refresh info to be recent so global check would fail + device_config.refresh_info = make_refresh_info(refresh_time=datetime.now(pytz.UTC).isoformat()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + + assert result_playlist == playlist + assert result_plugin.name == plugin.name + # Should NOT have rotated to a different plugin + assert playlist.current_plugin_index == 0 + + def test_rotates_to_next_plugin_when_global_interval_elapsed(self): + """When plugin_cycle_interval_seconds has elapsed, the playlist advances + to the next plugin even if the current plugin has a refresh interval.""" + # Current plugin doesn't need refresh (recently refreshed) + latest_refresh = datetime.now(pytz.UTC).isoformat() + plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=latest_refresh) + plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + # Set global refresh info to be old so global check passes + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock1", + refresh_time=(datetime.now(pytz.UTC) - timedelta(seconds=700)).isoformat(), + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + + assert result_playlist == playlist + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 + + def test_rotates_to_next_plugin_even_when_current_plugin_needs_refresh(self): + """When the global rotation interval has elapsed, the playlist advances + to the next plugin even if the current plugin's own refresh interval + has also elapsed. Rotation takes priority.""" + # Current plugin needs refresh (interval elapsed) + current_plugin_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=120)).isoformat() + plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=current_plugin_refresh) + plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=datetime.now(pytz.UTC).isoformat()) + playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + # Set global refresh info to be old so rotation check passes + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock1", + refresh_time=(datetime.now(pytz.UTC) - timedelta(seconds=700)).isoformat(), + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + + assert result_playlist == playlist + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 + + def test_new_plugin_can_refresh_after_rotation(self): + """After rotation, the new plugin can be refreshed according to its own interval.""" + # Current plugin (Clock1) doesn't need refresh, but global rotation is due + current_plugin_refresh = datetime.now(pytz.UTC).isoformat() + plugin1 = make_plugin_instance(name="Clock1", refresh={"interval": 60}, latest_refresh_time=current_plugin_refresh) + # New plugin (Clock2) needs refresh (interval elapsed) + new_plugin_refresh = (datetime.now(pytz.UTC) - timedelta(seconds=120)).isoformat() + plugin2 = make_plugin_instance(name="Clock2", refresh={"interval": 60}, latest_refresh_time=new_plugin_refresh) + playlist = make_playlist([plugin1, plugin2], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock1", + refresh_time=(datetime.now(pytz.UTC) - timedelta(seconds=700)).isoformat(), + ) + + task = RefreshTask(device_config, MockDisplayManager()) + + # First call: rotation happens + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 + + # Simulate the rotation refresh completing: update refresh_info to point to Clock2 + device_config.refresh_info = make_refresh_info( + plugin_instance="Clock2", + refresh_time=datetime.now(pytz.UTC).isoformat(), + ) + + # Second call: Clock2 needs refresh (its interval elapsed) + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + assert result_plugin.name == plugin2.name + assert playlist.current_plugin_index == 1 # Still on Clock2 + + def test_returns_none_when_no_refresh_needed(self): + """No unnecessary refreshes before expiry.""" + # Current plugin doesn't need refresh, and global interval hasn't elapsed + latest_refresh = datetime.now(pytz.UTC).isoformat() + plugin = make_plugin_instance(refresh={"interval": 60}, latest_refresh_time=latest_refresh) + playlist = make_playlist([plugin], current_plugin_index=0) + manager = PlaylistManager(playlists=[playlist]) + + device_config = MockDeviceConfig(playlist_manager=manager) + device_config.refresh_info = make_refresh_info(refresh_time=datetime.now(pytz.UTC).isoformat()) + + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + + assert result_playlist is None + assert result_plugin is None + + def test_returns_none_when_no_active_playlist(self): + manager = PlaylistManager(playlists=[]) + device_config = MockDeviceConfig(playlist_manager=manager) + task = RefreshTask(device_config, MockDisplayManager()) + + result_playlist, result_plugin = task._determine_next_plugin( + manager, device_config.get_refresh_info(), datetime.now(pytz.UTC) + ) + + assert result_playlist is None + assert result_plugin is None + + +class TestIsRotationRefresh: + def test_rotation_detected_when_plugin_changes(self): + """A rotation to a different plugin is detected.""" + refresh_info = { + "refresh_type": "Playlist", + "plugin_instance": "Clock2", + } + latest_refresh = make_refresh_info(plugin_instance="Clock1") + task = RefreshTask(MockDeviceConfig(), MockDisplayManager()) + + assert task._is_rotation_refresh(refresh_info, latest_refresh) is True + + def test_in_place_refresh_not_rotation(self): + """An in-place refresh of the same plugin is not a rotation.""" + refresh_info = { + "refresh_type": "Playlist", + "plugin_instance": "Clock1", + } + latest_refresh = make_refresh_info(plugin_instance="Clock1") + task = RefreshTask(MockDeviceConfig(), MockDisplayManager()) + + assert task._is_rotation_refresh(refresh_info, latest_refresh) is False + + def test_manual_update_is_rotation(self): + """A manual update is always treated as a rotation (resets rotation timer).""" + refresh_info = { + "refresh_type": "Manual Update", + "plugin_instance": "Clock1", + } + latest_refresh = make_refresh_info(plugin_instance="Clock1") + task = RefreshTask(MockDeviceConfig(), MockDisplayManager()) + + assert task._is_rotation_refresh(refresh_info, latest_refresh) is True