Codex Release Bot
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""Tests for dedicated push-progress WebSocket channel."""
|
|
|
|
import pytest
|
|
|
|
|
|
def test_sync_progress_uses_sync_dispatch(monkeypatch):
|
|
"""broadcast_sync_progress must not publish to the alert channel."""
|
|
from server.api import websocket as ws
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
async def capture_sync(msg):
|
|
calls.append(("sync", msg))
|
|
|
|
async def capture_alert(msg):
|
|
calls.append(("alert", msg))
|
|
|
|
monkeypatch.setattr(ws, "_dispatch_sync_message", capture_sync)
|
|
monkeypatch.setattr(ws, "_dispatch_ws_message", capture_alert)
|
|
|
|
import asyncio
|
|
|
|
asyncio.run(
|
|
ws.broadcast_sync_progress(
|
|
batch_id="b1",
|
|
server_id=1,
|
|
server_name="srv",
|
|
status="running",
|
|
completed=0,
|
|
failed=0,
|
|
total=1,
|
|
),
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0][0] == "sync"
|
|
assert calls[0][1]["type"] == "sync_progress"
|
|
assert calls[0][1]["batch_id"] == "b1"
|
|
|
|
|
|
def test_ws_sync_path_is_public():
|
|
from server.api.auth_jwt import _is_public_path
|
|
|
|
assert _is_public_path("/ws/sync") is True
|