Question

我用FastAPI在用户和服务器之间通信的网页上书写互联网游戏背后机械。为测试框架,我选择了Py Test。我的目标是编写一份检查关系关闭的测试。

在试图与非现有房间连接之后,这不应深入执行(管理人员不允许与不存在的东西连接)。

我完全不知道如何检验这一点。我试图利用正式文件找到任何解决办法,但我没有发现。

我的测试守则如下:

import pytest
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect

from src.game_manager import GameManager
from src.main import app

async def override_manager() -> GameManager:
    try:
        yield override_manager.manager
    except AttributeError:
        manager = GameManager()
        manager.add_new_game(max_players=2, room_name="foo", password=None)
        manager.add_new_game(max_players=2, room_name="bar", password="123")
        override_manager.manager = manager
        yield override_manager.manager


app.dependency_overrides[get_manager] = override_manager

client = TestClient(app)


class TestWebsocketConnection:
    def test_connect_to_non_existing_room(self):
        with pytest.raises(WebSocketDisconnect) as e_info:
            with client.websocket_connect("/ws/non_existing_room") as ws:
                ws.send_json({"message": "Hello world"})

回返

FAILED tests/test_websockets.py::TestWebsocketConnection::test_connect_to_non_existing_room - Failed: DID NOT RAISE <class  starlette.websockets.WebSocketDisconnect >

最终点看:

@router.websocket("/ws/{room_name}")
@router.websocket("/ws/{room_name}/{password}")
async def websocket_endpoint(
    websocket: WebSocket,
    manager: Annotated[GameManager, Depends(get_manager)],
):
    room_name = websocket.path_params["room_name"]
    password = websocket.path_params.get("password", None)

    try:
        await manager.connect(websocket, room_name, password)
        client_id = websocket.scope["client_id"]
        while True:
            data = await websocket.receive_json()
            await manager.handle_message(room_name, client_id, data)
    except WebSocketDisconnect:
        await manager.remove(websocket)

<NOTE:manager.link(websocket, Room_name, Password),WebSocketDisjoint,条件是不能允许链接。

如果联系已经关闭,请你帮助我检验。

Answer 1

奥凯找到了解决办法。

如果在打电话<代码>websocket.accept()后没有将任何名词发送到网上,并且在特定条件下终止联系(websocket.close(<>),我们可以尝试使用ws.receive_json()。页: 1 WebSocketDislink。

对像我这样有类似问题的人来说,这是一个工作考验:

import pytest
from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect

...
client = TestClient(app)


class TestWebsocketConnection:
    def test_connect_to_non_existing_room(self):
        with pytest.raises(WebSocketDisconnect):
            with client.websocket_connect("/ws/non_existing_room") as ws:
                ws.receive_json()

友情链接