Function test_should_emit_frame_events

tests/async/test_websocket.py:59–106

(page: Page, server: Server)

async def test_should_emit_frame_events(page: Page, server: Server) -> None:
    def _handle_ws_connection(ws: WebSocketProtocol) -> None:
        def _onMessage(payload: bytes, isBinary: bool) -> None:
            ws.sendMessage(b"incoming", False)
            ws.sendClose()

        setattr(ws, "onMessage", _onMessage)

    server.once_web_socket_connection(_handle_ws_connection)
    log = []
    socket_close_future: "asyncio.Future[None]" = asyncio.Future()

    def on_web_socket(ws: WebSocket) -> None:
        log.append("open")

        def _on_framesent(payload: Union[bytes, str]) -> None:
            assert isinstance(payload, str)
            log.append(f"sent<{payload}>")

        ws.on("framesent", _on_framesent)

        def _on_framereceived(payload: Union[bytes, str]) -> None:
            assert isinstance(payload, str)
            log.append(f"received<{payload}>")

        ws.on("framereceived", _on_framereceived)

        def _handle_close(ws: WebSocket) -> None:
            log.append("close")
            socket_close_future.set_result(None)

        ws.on("close", _handle_close)

    page.on("websocket", on_web_socket)
    async with page.expect_event("websocket"):
        await page.evaluate(
            """port => {
            const ws = new WebSocket('ws://localhost:' + port + '/ws');
            ws.addEventListener('open', () => ws.send('outgoing'));
            ws.addEventListener('message', () => ws.close())
        }""",
            server.PORT,
        )
    await socket_close_future
    assert log[0] == "open"
    assert log[3] == "close"
    log.sort()
    assert log == ["close", "open", "received<incoming>", "sent<outgoing>"]


async def test_should_emit_binary_frame_events(page: Page, server: Server) -> None: