Coverage for tests/test_ws_dependencies.py: 100%

1import json 5 ctx1abcde

2from typing import List 5 ctx1abcde

3 

4from fastapi import APIRouter, Depends, FastAPI, WebSocket 5 ctx1abcde

5from fastapi.testclient import TestClient 5 ctx1abcde

6from typing_extensions import Annotated 5 ctx1abcde

7 

8 

9def dependency_list() -> List[str]: 5 ctx1abcde

10 return [] 5 ctx1abcde

11 

12 

13DepList = Annotated[List[str], Depends(dependency_list)] 5 ctx1abcde

14 

15 

16def create_dependency(name: str): 5 ctx1abcde

17 def fun(deps: DepList): 5 ctx1abcde

18 deps.append(name) 5 ctx1abcde

19 

20 return Depends(fun) 5 ctx1abcde

21 

22 

23router = APIRouter(dependencies=[create_dependency("router")]) 5 ctx1abcde

24prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")]) 5 ctx1abcde

25app = FastAPI(dependencies=[create_dependency("app")]) 5 ctx1abcde

26 

27 

28@app.websocket("/", dependencies=[create_dependency("index")]) 5 ctx1abcde

29async def index(websocket: WebSocket, deps: DepList): 5 ctx1abcde

30 await websocket.accept() 5 ctx1abcde

31 await websocket.send_text(json.dumps(deps)) 5 ctx1abcde

32 await websocket.close() 5 ctx1abcde

33 

34 

35@router.websocket("/router", dependencies=[create_dependency("routerindex")]) 5 ctx1abcde

36async def routerindex(websocket: WebSocket, deps: DepList): 5 ctx1abcde

37 await websocket.accept() 5 ctx1abcde

38 await websocket.send_text(json.dumps(deps)) 5 ctx1abcde

39 await websocket.close() 5 ctx1abcde

40 

41 

42@prefix_router.websocket("/", dependencies=[create_dependency("routerprefixindex")]) 5 ctx1abcde

43async def routerprefixindex(websocket: WebSocket, deps: DepList): 5 ctx1abcde

44 await websocket.accept() 5 ctx1abcde

45 await websocket.send_text(json.dumps(deps)) 5 ctx1abcde

46 await websocket.close() 5 ctx1abcde

47 

48 

49app.include_router(router, dependencies=[create_dependency("router2")]) 5 ctx1abcde

50app.include_router( 5 ctx1abcde

51 prefix_router, prefix="/prefix", dependencies=[create_dependency("prefix_router2")] 

52) 

53 

54 

55def test_index(): 5 ctx1abcde

56 client = TestClient(app) 5 ctx1abcde

57 with client.websocket_connect("/") as websocket: 5 ctx1abcde

58 data = json.loads(websocket.receive_text()) 5 ctx1abcde

59 assert data == ["app", "index"] 5 ctx1abcde

60 

61 

62def test_routerindex(): 5 ctx1abcde

63 client = TestClient(app) 5 ctx1abcde

64 with client.websocket_connect("/router") as websocket: 5 ctx1abcde

65 data = json.loads(websocket.receive_text()) 5 ctx1abcde

66 assert data == ["app", "router2", "router", "routerindex"] 5 ctx1abcde

67 

68 

69def test_routerprefixindex(): 5 ctx1abcde

70 client = TestClient(app) 5 ctx1abcde

71 with client.websocket_connect("/prefix/") as websocket: 5 ctx1abcde

72 data = json.loads(websocket.receive_text()) 5 ctx1abcde

73 assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"] 5 ctx1abcde