摘要：
之前报错是因为我们没有正常的关闭连接导致的，那么我们这个时候再来去看下测试代码deftest_websocket():client=TestClientwithclient.websocket_connect("/items/ws?token="+"leizishuoceshi")aswebsocket:websocket.send_textdata=websocket.receive_text()printassertstr=="消息是:HelloWebSocket"此时我们再去执行，发现代码应该不会报错我们在用例优化下classFastApiTestWeb:defsetUp-˃None:self.client=TestClientdeftearDown-˃None:self.client=Nonedeftest_websocket:withself.client.websocket_connect("/items/ws?token="+"leizishuoceshi")aswebsocket:websocket.send_textdata=websocket.receive_text()printassertstr=="消息是:Hello123"if__name__=="__main__":unittest.main()这样我们的一个测试用例就更加的完整了。就是不带token的访问，我们在增加下一条case去覆盖下。deftest_websocket_notoken:try:withself.client.websocket_connectaswebsocket:websocket.send_textdata=websocket.receive_text()self.assertEqualexcept:self.assertTrue覆盖了这行代码。到这里，我们对于WebSockets接口测试完毕，但是还有问题，我们真正的聊天中，还需要上线进行通知，下线进行通知，我们应该如何实现呢，且听下回分解。
其实代码没有问题，但是我们忽略了一点，就是我们在正常的开发中，肯定是遇到这样的情况，我们频繁的有客户端链接，断开链接，我们需要统一的管理起来，那么我们应该如何去管理呢，其实这个时候，我们要去声明一个类去管理我们的这些链接。我们应该如何优化呢。
定义一个链接管理类，处理我们所有的链接。
classConnectionManager: def __init__(self): #存放**的链接 self.active_connections: List[Dict[str, WebSocket]] =[] async defconnect(self, user: str, ws: WebSocket): #链接 await ws.accept() self.active_connections.append({"user": user, "ws": ws}) defdisconnect(self, user: str, ws: WebSocket): #关闭时 移除ws对象 self.active_connections.remove({"user": user, "ws": ws})
我们增加了链接，移除链接的操作，那么我们垓下我们的工程代码
manager =ConnectionManager() @app.websocket("/items/ws") async defwebsocket_endpoint( websocket: WebSocket, cookie_or_token: str =Depends(get_cookie_or_token), ): await manager.connect(cookie_or_token, websocket) try: whileTrue: data =await websocket.receive_text() await websocket.send_text(f"消息是: {data}") exceptWebSocketDisconnect as e: manager.disconnect(cookie_or_token, websocket)
这样我们在连接处理的时候就可以正常处理了。之前报错是因为我们没有正常的关闭连接导致的，那么我们这个时候再来去看下测试代码
deftest_websocket(): client =TestClient(app) with client.websocket_connect("/items/ws?token="+"leizishuoceshi") as websocket: websocket.send_text("Hello WebSocket") data =websocket.receive_text() print(data) assert str(data) =="消息是: Hello WebSocket"
此时我们再去执行，发现代码应该不会报错
我们在用例优化下
classFastApiTestWeb(unittest.TestCase): def setUp(self) ->None: self.client =TestClient(app) def tearDown(self) ->None: self.client =None deftest_websocket(self): with self.client.websocket_connect("/items/ws?token=" + "leizishuoceshi") as websocket: websocket.send_text("Hello WebSocket") data =websocket.receive_text() print(data) assert str(data) == "消息是: Hello WebSocket" deftest_websocket_two(self): with self.client.websocket_connect("/items/ws?token=" + "leizishuoceshi") as websocket: websocket.send_text("Hello 123") data =websocket.receive_text() print(data) assert str(data) == "消息是: Hello 123" if __name__ == "__main__": unittest.main()
这样我们的一个测试用例就更加的完整了。我们执行正常是没有报错的
我们想要看下代码的覆盖率，应该如何看呢。我是用的coverage。
然后我们再去report，
我们想看html测试报告，可以运行下 coverage html。
然后看index.html
因为我的main.py还有其他的方法，我们还需要点进去看我们对应方法的覆盖率。
我们可以看到，我们的关键代码还差了一行没有覆盖到。就是不带token的访问，我们在增加下一条case去覆盖下。
deftest_websocket_notoken(self): try: with self.client.websocket_connect("/items/ws") as websocket: websocket.send_text("Hello 123") data =websocket.receive_text() self.assertEqual("消息是: Hello 123",data) except: self.assertTrue(True)
覆盖了这行代码。但是我们用例需要判断这个异常。
到这里，我们对于WebSockets接口测试完毕，但是还有问题，我们真正的聊天中，还需要上线进行通知，下线进行通知，我们应该如何实现呢，且听下回分解。
文章首发在公众号，欢迎关注。
