Coverage for tests / test_ws_dependencies.py: 100%

1import json 4 ctx1abcd

2from typing import Annotated 4 ctx1abcd

3 

4from fastapi import APIRouter, Depends, FastAPI, WebSocket 4 ctx1abcd

5from fastapi.testclient import TestClient 4 ctx1abcd

6 

7 

8def dependency_list() -> list[str]: 4 ctx1abcd

9 return [] 9 ctx1efghijklm

10 

11 

12DepList = Annotated[list[str], Depends(dependency_list)] 4 ctx1abcd

13 

14 

15def create_dependency(name: str): 4 ctx1abcd

16 def fun(deps: DepList): 4 ctx1abcd

17 deps.append(name) 9 ctx1efghijklm

18 

19 return Depends(fun) 4 ctx1abcd

20 

21 

22router = APIRouter(dependencies=[create_dependency("router")]) 4 ctx1abcd

23prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")]) 4 ctx1abcd

24app = FastAPI(dependencies=[create_dependency("app")]) 4 ctx1abcd

25 

26 

27@app.websocket("/", dependencies=[create_dependency("index")]) 4 ctx1abcd

28async def index(websocket: WebSocket, deps: DepList): 4 ctx1abcd

29 await websocket.accept() 3 ctx1ehk

30 await websocket.send_text(json.dumps(deps)) 3 ctx1ehk

31 await websocket.close() 3 ctx1ehk

32 

33 

34@router.websocket("/router", dependencies=[create_dependency("routerindex")]) 4 ctx1abcd

35async def routerindex(websocket: WebSocket, deps: DepList): 4 ctx1abcd

36 await websocket.accept() 3 ctx1fil

37 await websocket.send_text(json.dumps(deps)) 3 ctx1fil

38 await websocket.close() 3 ctx1fil

39 

40 

41@prefix_router.websocket("/", dependencies=[create_dependency("routerprefixindex")]) 4 ctx1abcd

42async def routerprefixindex(websocket: WebSocket, deps: DepList): 4 ctx1abcd

43 await websocket.accept() 3 ctx1gjm

44 await websocket.send_text(json.dumps(deps)) 3 ctx1gjm

45 await websocket.close() 3 ctx1gjm

46 

47 

48app.include_router(router, dependencies=[create_dependency("router2")]) 4 ctx1abcd

49app.include_router( 4 ctx1abcd

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

51) 

52 

53 

54def test_index(): 4 ctx1abcd

55 client = TestClient(app) 3 ctx1ehk

56 with client.websocket_connect("/") as websocket: 3 ctx1ehk

57 data = json.loads(websocket.receive_text()) 3 ctx1ehk

58 assert data == ["app", "index"] 3 ctx1ehk

59 

60 

61def test_routerindex(): 4 ctx1abcd

62 client = TestClient(app) 3 ctx1fil

63 with client.websocket_connect("/router") as websocket: 3 ctx1fil

64 data = json.loads(websocket.receive_text()) 3 ctx1fil

65 assert data == ["app", "router2", "router", "routerindex"] 3 ctx1fil

66 

67 

68def test_routerprefixindex(): 4 ctx1abcd

69 client = TestClient(app) 3 ctx1gjm

70 with client.websocket_connect("/prefix/") as websocket: 3 ctx1gjm

71 data = json.loads(websocket.receive_text()) 3 ctx1gjm

72 assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"] 3 ctx1gjm