Метод websocket_connect

Метод websocket_connect применяется к экземпляру класса TestClient и открывает WebSocket-соединение с приложением FastAPI. Первым параметром передаётся путь URL, например '/ws'. Метод возвращает контекстный менеджер, внутри которого доступен объект соединения для отправки и получения сообщений.

Внутри блока with у объекта соединения вызываются методы send_text, send_json, receive_text и receive_json. Это позволяет проверить работу WebSocket-эндпоинта без запуска реального сервера.

Синтаксис

with client.websocket_connect(path) as websocket:
	websocket.send_text(data)
	res = websocket.receive_text()

Пример

Давайте создадим приложение с WebSocket-эндпоинтом и проверим его через websocket_connect:

from fastapi import FastAPI, WebSocket
from fastapi.testclient import TestClient

app = FastAPI()

@app.websocket('/ws')
async def ws_endpoint(websocket: WebSocket):
	await websocket.accept()
	data = await websocket.receive_text()
	await websocket.send_text('echo: ' + data)
	await websocket.close()

client = TestClient(app)

with client.websocket_connect('/ws') as websocket:
	websocket.send_text('hello')
	res = websocket.receive_text()
	print(res)

Результат выполнения кода:

"echo: hello"

Пример

Давайте передадим данные в формате JSON и получим ответ тоже в формате JSON:

<+python+> from fastapi import FastAPI, WebSocket from fastapi.testclient import TestClient app = FastAPI() @app.websocket('/ws') async def ws_endpoint(websocket: WebSocket): await websocket.accept() data = await websocket.receive_json() await websocket.send_json({'user': data['user']}) await websocket.close() client = TestClient(app) with client.websocket_connect('/ws') as websocket: websocket.send_json({'user': 'abcde'}) res = websocket.receive_json() print(res) <-python+>

Результат выполнения кода:

{'user': 'abcde'}

Пример

Давайте проверим, что сервер отправляет несколько сообщений подряд, и прочитаем их все:

<+python+> from fastapi import FastAPI, WebSocket from fastapi.testclient import TestClient app = FastAPI() @app.websocket('/ws') async def ws_endpoint(websocket: WebSocket): await websocket.accept() await websocket.send_text('article') await websocket.send_text('user') await websocket.close() client = TestClient(app) with client.websocket_connect('/ws') as websocket: res1 = websocket.receive_text() res2 = websocket.receive_text() print(res1) print(res2) <-python+>

Результат выполнения кода:

"article"
"user"

Смотрите также

класс TestClient,
который используется для тестирования приложения
метод get,
который выполняет GET-запрос в тестах
метод post,
который выполняет POST-запрос в тестах
атрибут app,
который хранит тестируемое приложение