test: add tests for status WebSocket and scheduler
- test_status.py: 8 tests covering WS auth rejection/acceptance, broadcast_status (dead connection removal, no response_time, no connections), broadcast_scan_update - test_scheduler.py: 9 tests covering _load_interval variants, _run_status_checks DB update/last_seen/error handling, start/stop lifecycle - scheduler.py: reinitialize AsyncIOScheduler on each start() to avoid stale event loop across test restarts
This commit is contained in:
@@ -13,7 +13,7 @@ from app.services.status_checker import check_node
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
scheduler = AsyncIOScheduler()
|
||||
scheduler: AsyncIOScheduler = AsyncIOScheduler()
|
||||
|
||||
|
||||
async def _run_status_checks() -> None:
|
||||
@@ -56,6 +56,8 @@ def _load_interval() -> int:
|
||||
|
||||
|
||||
def start_scheduler() -> None:
|
||||
global scheduler
|
||||
scheduler = AsyncIOScheduler()
|
||||
interval = _load_interval()
|
||||
scheduler.add_job(_run_status_checks, "interval", seconds=interval, id="status_checks")
|
||||
scheduler.start()
|
||||
|
||||
Reference in New Issue
Block a user