67 lines
3.1 KiB
Python
67 lines
3.1 KiB
Python
"""A slow browser keeps the latest plot, status and settings independently."""
|
|
|
|
import asyncio
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
|
|
|
from python_app.webui.controller import WebController
|
|
from python_app.webui.routes import ws
|
|
from python_app.webui.streaming import RingBroadcaster
|
|
|
|
|
|
class WebStreamingTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_slow_client_keeps_each_type_and_reconnect_gets_current_settings(self) -> None:
|
|
controller = mock.Mock(spec=WebController)
|
|
controller.status.return_value = {"running": False}
|
|
controller.current_live_settings.return_value = [{"name": "gain", "value": 0}]
|
|
controller.peek_frame.return_value = {"type": "frame", "seq": 0}
|
|
broadcaster = RingBroadcaster(controller)
|
|
app = FastAPI()
|
|
app.state.controller = controller
|
|
app.state.broadcaster = broadcaster
|
|
started, release = asyncio.Event(), asyncio.Event()
|
|
received: list[dict] = []
|
|
|
|
async def send(message: dict) -> None:
|
|
received.append(message)
|
|
if len(received) == 1:
|
|
started.set()
|
|
await release.wait()
|
|
if len(received) == 4:
|
|
raise WebSocketDisconnect()
|
|
|
|
socket = mock.Mock(spec=WebSocket, app=app, accept=mock.AsyncMock(),
|
|
close=mock.AsyncMock(), send_json=mock.AsyncMock(side_effect=send))
|
|
with mock.patch("python_app.webui.routes.asyncio.to_thread", mock.AsyncMock(
|
|
side_effect=lambda func, *args: func(*args))):
|
|
client = asyncio.create_task(ws(socket))
|
|
await asyncio.wait_for(started.wait(), 1)
|
|
for value in range(1, 101):
|
|
controller.status.return_value = {"running": True, "raw_count": value}
|
|
controller.current_live_settings.return_value = [{"name": "gain", "value": value}]
|
|
controller.peek_frame.return_value = {"type": "frame", "seq": value}
|
|
broadcaster._publish({"type": "status", **controller.status()})
|
|
broadcaster._publish({"type": "settings", "schema": controller.current_live_settings()})
|
|
broadcaster._publish(controller.peek_frame())
|
|
release.set()
|
|
await asyncio.wait_for(client, 1)
|
|
latest = {message["type"]: message for message in received[1:]}
|
|
self.assertEqual(latest, {
|
|
"frame": {"type": "frame", "seq": 100},
|
|
"status": {"type": "status", "running": True, "raw_count": 100},
|
|
"settings": {"type": "settings", "schema": [{"name": "gain", "value": 100}]},
|
|
})
|
|
|
|
received.clear()
|
|
|
|
async def reconnect_send(message: dict) -> None:
|
|
received.append(message)
|
|
if len(received) == 3:
|
|
raise WebSocketDisconnect()
|
|
|
|
socket.send_json.side_effect = reconnect_send
|
|
await asyncio.wait_for(ws(socket), 1)
|
|
self.assertEqual({message["type"]: message for message in received}, latest)
|