From d84237dbb4a9adc4e057f639aeae76fd32f7562d Mon Sep 17 00:00:00 2001 From: James Smith Date: Thu, 19 Mar 2026 21:47:38 +0000 Subject: [PATCH] feat(satellite): add 24-hour periodic TLE auto-refresh TLE data was only refreshed once at startup. After each refresh, a new 24-hour timer is now scheduled in a finally block so it fires even on refresh failure. threading moved to module-level import. --- routes/satellite.py | 16 +++++++++++++--- tests/test_satellite.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/routes/satellite.py b/routes/satellite.py index 08f6996..78640fe 100644 --- a/routes/satellite.py +++ b/routes/satellite.py @@ -3,6 +3,7 @@ from __future__ import annotations import math +import threading import time import urllib.request from datetime import datetime, timedelta @@ -295,9 +296,15 @@ def _start_satellite_tracker(): time.sleep(1) +_TLE_REFRESH_INTERVAL_SECONDS = 24 * 60 * 60 # 24 hours + + def init_tle_auto_refresh(): """Initialize TLE auto-refresh. Called by app.py after initialization.""" - import threading + def _schedule_next_tle_refresh(delay: float = _TLE_REFRESH_INTERVAL_SECONDS) -> None: + t = threading.Timer(delay, _auto_refresh_tle) + t.daemon = True + t.start() def _auto_refresh_tle(): try: @@ -307,10 +314,13 @@ def init_tle_auto_refresh(): logger.info(f"Auto-refreshed TLE data for: {', '.join(updated)}") except Exception as e: logger.warning(f"Auto TLE refresh failed: {e}") + finally: + # Schedule next refresh regardless of success/failure + _schedule_next_tle_refresh() - # Start auto-refresh in background + # First refresh 2 seconds after startup, then every 24 hours threading.Timer(2.0, _auto_refresh_tle).start() - logger.info("TLE auto-refresh scheduled") + logger.info("TLE auto-refresh scheduled (24h interval)") # Start live position tracker thread tracker_thread = threading.Thread( diff --git a/tests/test_satellite.py b/tests/test_satellite.py index 2eeb0e5..7d62a9f 100644 --- a/tests/test_satellite.py +++ b/tests/test_satellite.py @@ -121,6 +121,39 @@ def test_tracker_position_has_no_observer_fields(): assert required in pos, f"SSE tracker must emit '{required}'" +@patch('routes.satellite.refresh_tle_data', return_value=['ISS']) +@patch('routes.satellite._load_db_satellites_into_cache') +def test_tle_auto_refresh_schedules_daily_repeat(mock_load_db, mock_refresh): + """After the first TLE refresh, a 24-hour follow-up timer must be scheduled.""" + import threading as real_threading + + scheduled_delays = [] + + class CapturingTimer: + def __init__(self, delay, fn, *a, **kw): + scheduled_delays.append(delay) + self._fn = fn + self._delay = delay + + def start(self): + # Execute the startup timer inline so we can capture the follow-up + if self._delay <= 5: + self._fn() + + with patch('routes.satellite.threading') as mock_threading: + mock_threading.Timer = CapturingTimer + mock_threading.Thread = real_threading.Thread + + from routes.satellite import init_tle_auto_refresh + init_tle_auto_refresh() + + # First timer: startup delay (≤5s); second timer: 24h repeat (≥86400s) + assert any(d <= 5 for d in scheduled_delays), \ + f"Expected startup delay timer; got delays: {scheduled_delays}" + assert any(d >= 86400 for d in scheduled_delays), \ + f"Expected ~24h repeat timer; got delays: {scheduled_delays}" + + # Logic Integration Test (Simulating prediction) def test_predict_passes_empty_cache(client): """Verify that if the satellite is not in cache, no passes are returned."""