/
test_ws_dependencies.py
70 lines (48 loc) · 2.22 KB
/
test_ws_dependencies.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import json
from typing import Optional
from fastapi import APIRouter, Depends, FastAPI, WebSocket
from fastapi.testclient import TestClient
def dependency_list(deps: Optional[str] = None) -> list[str]:
return [deps] if deps else []
def create_dependency(name: str):
def fun(deps: list[str] = Depends(dependency_list)):
print(f"create_dependency.fun({name})")
deps.append(name)
return Depends(fun)
router = APIRouter(dependencies=[create_dependency("router")])
prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")])
app = FastAPI(dependencies=[create_dependency("app")])
@app.websocket("/", dependencies=[create_dependency("index")])
async def index(websocket: WebSocket, deps=Depends(dependency_list)):
await websocket.accept()
await websocket.send_text(json.dumps(deps))
await websocket.close()
@router.websocket("/router", dependencies=[create_dependency("routerindex")])
async def routerindex(websocket: WebSocket, deps=Depends(dependency_list)):
await websocket.accept()
await websocket.send_text(json.dumps(deps))
await websocket.close()
@prefix_router.websocket("/", dependencies=[create_dependency("routerprefixindex")])
async def routerprefixindex(websocket: WebSocket, deps=Depends(dependency_list)):
await websocket.accept()
await websocket.send_text(json.dumps(deps))
await websocket.close()
app.include_router(router, dependencies=[create_dependency("router2")])
app.include_router(
prefix_router, prefix="/prefix", dependencies=[create_dependency("prefix_router2")]
)
def test_index():
client = TestClient(app)
with client.websocket_connect("/") as websocket:
data = json.loads(websocket.receive_text())
assert data == ["app", "index"]
def test_routerindex():
client = TestClient(app)
with client.websocket_connect("/router") as websocket:
data = json.loads(websocket.receive_text())
assert data == ["app", "router2", "router", "routerindex"]
def test_routerprefixindex():
client = TestClient(app)
with client.websocket_connect("/prefix/") as websocket:
data = json.loads(websocket.receive_text())
assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"]