Coverage for tests/test_ws_dependencies.py: 100%

1import json 6 ctx1abcdef

2from typing import List 6 ctx1abcdef

3 

4from fastapi import APIRouter, Depends, FastAPI, WebSocket 6 ctx1abcdef

5from fastapi.testclient import TestClient 6 ctx1abcdef

6from typing_extensions import Annotated 6 ctx1abcdef

7 

8 

9def dependency_list() -> List[str]: 6 ctx1abcdef

10 return [] 18 ctx1ghijklmnopqrstuvwx

11 

12 

13DepList = Annotated[List[str], Depends(dependency_list)] 6 ctx1abcdef

14 

15 

16def create_dependency(name: str): 6 ctx1abcdef

17 def fun(deps: DepList): 6 ctx1abcdef

18 deps.append(name) 18 ctx1ghijklmnopqrstuvwx

19 

20 return Depends(fun) 6 ctx1abcdef

21 

22 

23router = APIRouter(dependencies=[create_dependency("router")]) 6 ctx1abcdef

24prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")]) 6 ctx1abcdef

25app = FastAPI(dependencies=[create_dependency("app")]) 6 ctx1abcdef

26 

27 

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

29async def index(websocket: WebSocket, deps: DepList): 6 ctx1abcdef

30 await websocket.accept() 6 ctx1gjmpsv

31 await websocket.send_text(json.dumps(deps)) 6 ctx1gjmpsv

32 await websocket.close() 6 ctx1gjmpsv

33 

34 

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

36async def routerindex(websocket: WebSocket, deps: DepList): 6 ctx1abcdef

37 await websocket.accept() 6 ctx1hknqtw

38 await websocket.send_text(json.dumps(deps)) 6 ctx1hknqtw

39 await websocket.close() 6 ctx1hknqtw

40 

41 

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

43async def routerprefixindex(websocket: WebSocket, deps: DepList): 6 ctx1abcdef

44 await websocket.accept() 6 ctx1ilorux

45 await websocket.send_text(json.dumps(deps)) 6 ctx1ilorux

46 await websocket.close() 6 ctx1ilorux

47 

48 

49app.include_router(router, dependencies=[create_dependency("router2")]) 6 ctx1abcdef

50app.include_router( 6 ctx1abcdef

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

52) 

53 

54 

55def test_index(): 6 ctx1abcdef

56 client = TestClient(app) 6 ctx1gjmpsv

57 with client.websocket_connect("/") as websocket: 6 ctx1gjmpsv

58 data = json.loads(websocket.receive_text()) 6 ctx1gjmpsv

59 assert data == ["app", "index"] 6 ctx1gjmpsv

60 

61 

62def test_routerindex(): 6 ctx1abcdef

63 client = TestClient(app) 6 ctx1hknqtw

64 with client.websocket_connect("/router") as websocket: 6 ctx1hknqtw

65 data = json.loads(websocket.receive_text()) 6 ctx1hknqtw

66 assert data == ["app", "router2", "router", "routerindex"] 6 ctx1hknqtw

67 

68 

69def test_routerprefixindex(): 6 ctx1abcdef

70 client = TestClient(app) 6 ctx1ilorux

71 with client.websocket_connect("/prefix/") as websocket: 6 ctx1ilorux

72 data = json.loads(websocket.receive_text()) 6 ctx1ilorux

73 assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"] 6 ctx1ilorux