Coverage for tests/test_ws_dependencies.py: 100%

1import json 7 ctx1abcdefg

2from typing import List 7 ctx1abcdefg

3 

4from fastapi import APIRouter, Depends, FastAPI, WebSocket 7 ctx1abcdefg

5from fastapi.testclient import TestClient 7 ctx1abcdefg

6from typing_extensions import Annotated 7 ctx1abcdefg

7 

8 

9def dependency_list() -> List[str]: 7 ctx1abcdefg

10 return [] 21 ctx1hijklmnopqrstuvwxyzAB

11 

12 

13DepList = Annotated[List[str], Depends(dependency_list)] 7 ctx1abcdefg

14 

15 

16def create_dependency(name: str): 7 ctx1abcdefg

17 def fun(deps: DepList): 7 ctx1abcdefg

18 deps.append(name) 21 ctx1hijklmnopqrstuvwxyzAB

19 

20 return Depends(fun) 7 ctx1abcdefg

21 

22 

23router = APIRouter(dependencies=[create_dependency("router")]) 7 ctx1abcdefg

24prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")]) 7 ctx1abcdefg

25app = FastAPI(dependencies=[create_dependency("app")]) 7 ctx1abcdefg

26 

27 

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

29async def index(websocket: WebSocket, deps: DepList): 7 ctx1abcdefg

30 await websocket.accept() 7 ctx1hknqtwz

31 await websocket.send_text(json.dumps(deps)) 7 ctx1hknqtwz

32 await websocket.close() 7 ctx1hknqtwz

33 

34 

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

36async def routerindex(websocket: WebSocket, deps: DepList): 7 ctx1abcdefg

37 await websocket.accept() 7 ctx1iloruxA

38 await websocket.send_text(json.dumps(deps)) 7 ctx1iloruxA

39 await websocket.close() 7 ctx1iloruxA

40 

41 

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

43async def routerprefixindex(websocket: WebSocket, deps: DepList): 7 ctx1abcdefg

44 await websocket.accept() 7 ctx1jmpsvyB

45 await websocket.send_text(json.dumps(deps)) 7 ctx1jmpsvyB

46 await websocket.close() 7 ctx1jmpsvyB

47 

48 

49app.include_router(router, dependencies=[create_dependency("router2")]) 7 ctx1abcdefg

50app.include_router( 7 ctx1abcdefg

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

52) 

53 

54 

55def test_index(): 7 ctx1abcdefg

56 client = TestClient(app) 7 ctx1hknqtwz

57 with client.websocket_connect("/") as websocket: 7 ctx1hknqtwz

58 data = json.loads(websocket.receive_text()) 7 ctx1hknqtwz

59 assert data == ["app", "index"] 7 ctx1hknqtwz

60 

61 

62def test_routerindex(): 7 ctx1abcdefg

63 client = TestClient(app) 7 ctx1iloruxA

64 with client.websocket_connect("/router") as websocket: 7 ctx1iloruxA

65 data = json.loads(websocket.receive_text()) 7 ctx1iloruxA

66 assert data == ["app", "router2", "router", "routerindex"] 7 ctx1iloruxA

67 

68 

69def test_routerprefixindex(): 7 ctx1abcdefg

70 client = TestClient(app) 7 ctx1jmpsvyB

71 with client.websocket_connect("/prefix/") as websocket: 7 ctx1jmpsvyB

72 data = json.loads(websocket.receive_text()) 7 ctx1jmpsvyB

73 assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"] 7 ctx1jmpsvyB