其实代码没有问题，但我们忽略了一点，那就是在正常的开发中，我们一定会遇到这样的情况，我们频繁地有客户端链接，断了链接，我们需要统一管理，那么我们应该如何管理呢？事实上，在这个时候，我们必须声明一个类来管理我们的链接。我们应该如何优化它？
[En]
In fact, there is no problem with the code, but we overlook one point, that is, in normal development, we must encounter such a situation, we frequently have client links, break links, we need unified management, so how should we manage it? in fact, at this time, we have to declare a class to manage our links. How should we optimize it?
定义一个处理所有链接的链接管理类。
[En]
Define a link management class that handles all our links.
class ConnectionManager:
def __init__(self):
# 存放**的链接
self.active_connections: List[Dict[str, WebSocket]] = []
async def connect(self, user: str, ws: WebSocket):
# 链接
await ws.accept()
self.active_connections.append({"user": user, "ws": ws})
def disconnect(self, user: str, ws: WebSocket):
# 关闭时 移除ws对象
self.active_connections.remove({"user": user, "ws": ws})
我们添加了链接，删除了链接，所以我们检查了项目代码。
[En]
We added links, removed links, so we checked our project code.
manager = ConnectionManager()
@app.websocket("/items/ws")
async def websocket_endpoint(
websocket: WebSocket,
cookie_or_token: str = Depends(get_cookie_or_token),
):
await manager.connect(cookie_or_token, websocket)
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"消息是: {data}")
except WebSocketDisconnect as e:
manager.disconnect(cookie_or_token, websocket)
这样，我们就可以正常地处理连接。前面的错误是由于我们没有正常关闭连接造成的，所以现在让我们来看一下测试代码。
[En]
In this way, we can handle the connection normally. The previous error was caused by the fact that we did not close the connection normally, so let’s take a look at the test code at this time.
def test_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"
此时，我们将再次执行它，并发现该代码不应报告错误。
[En]
At this point, we will execute it again and find that the code should not report an error.
我们在用例优化下
class FastApiTestWeb(unittest.TestCase):
def setUp(self) -> None:
self.client = TestClient(app)
def tearDown(self) -> None:
self.client = None
def test_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"
def test_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()
通过这种方式，我们的一个测试用例更加完整。当我们执行正常的性能时，我们没有报告错误。
[En]
In this way, one of our test cases is more complete. We did not report wrong when we carried out the normal performance.
我们想要看下代码的覆盖率，应该如何看呢。我是用的coverage。
然后我们再去report，
我们想看html测试报告，可以运行下 coverage html。
然后看index.html
因为我的main.py还有其他的方法，我们还需要点进去看我们对应方法的覆盖率。
我们可以看到，我们的关键代码还差了一行没有覆盖到。就是不带token的访问，我们在增加下一条case去覆盖下。
def test_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)
包含此行代码。但是我们需要在我们的用例中判断这个例外。
[En]
Covers this line of code. But we need to judge this exception in our use case.
到这里，我们对于WebSockets接口测试完毕，但是还有问题，我们真正的聊天中，还需要上线进行通知，下线进行通知，我们应该如何实现呢，且听下回分解。
第一个帖子是在官方账号上。欢迎关注我们。
[En]
The first post is on the official account. Welcome to follow us.
Original: https://www.cnblogs.com/leiziv5/p/15416888.html
Author: 北漂的雷子
Title: FastAPI 学习之路（四十九）WebSockets（五）修复接口测试中的问题
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/511006/
转载文章受原作者版权保护。转载请注明原作者出处！