Nexus Agent
e3efbd4552
Cron matching, audit/alert date filters, and schedule UI align on Asia/Shanghai while JWT/TOTP/heartbeat stay UTC; follow-ups add WS alert timestamps, fixed install timezone copy, and 422 on invalid date filters. Co-authored-by: Cursor <cursoragent@cursor.com>
44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
"""WebSocket alert/recovery messages include server-side UTC timestamp."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from server.api import websocket as ws_mod
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_alert_includes_at(monkeypatch):
|
|
captured: list[dict] = []
|
|
|
|
async def _capture(msg: dict) -> None:
|
|
captured.append(msg)
|
|
|
|
monkeypatch.setattr(ws_mod, "_dispatch_ws_message", _capture)
|
|
monkeypatch.setattr(ws_mod, "_save_alert_log", AsyncMock())
|
|
monkeypatch.setattr(ws_mod, "_should_push_telegram", lambda _key: False)
|
|
|
|
await ws_mod.broadcast_alert(1, "cpu", 95.0, "srv-a")
|
|
|
|
assert len(captured) == 1
|
|
assert "at" in captured[0]
|
|
assert captured[0]["at"].endswith("+00:00") or "T" in captured[0]["at"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_recovery_includes_at(monkeypatch):
|
|
captured: list[dict] = []
|
|
|
|
async def _capture(msg: dict) -> None:
|
|
captured.append(msg)
|
|
|
|
monkeypatch.setattr(ws_mod, "_dispatch_ws_message", _capture)
|
|
monkeypatch.setattr(ws_mod, "_save_alert_log", AsyncMock())
|
|
|
|
await ws_mod.broadcast_recovery(2, "memory", 42.0, "srv-b")
|
|
|
|
assert len(captured) == 1
|
|
assert "at" in captured[0]
|