Coverage for tests / test_ws_dependencies.py: 100%

46 statements  

« prev     ^ index     » next       coverage.py v7.13.3, created at 2026-04-06 01:24 +0000

1import json 1abcd

2from typing import Annotated 1abcd

3 

4from fastapi import APIRouter, Depends, FastAPI, WebSocket 1abcd

5from fastapi.testclient import TestClient 1abcd

6 

7 

8def dependency_list() -> list[str]: 1abcd

9 return [] 1efghijklmno

10 

11 

12DepList = Annotated[list[str], Depends(dependency_list)] 1abcd

13 

14 

15def create_dependency(name: str): 1abcd

16 def fun(deps: DepList): 1abcd

17 deps.append(name) 1efghijklmno

18 

19 return Depends(fun) 1abcd

20 

21 

22router = APIRouter(dependencies=[create_dependency("router")]) 1abcd

23prefix_router = APIRouter(dependencies=[create_dependency("prefix_router")]) 1abcd

24app = FastAPI(dependencies=[create_dependency("app")]) 1abcd

25 

26 

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

28async def index(websocket: WebSocket, deps: DepList): 1abcd

29 await websocket.accept() 1ehjm

30 await websocket.send_text(json.dumps(deps)) 1ehjm

31 await websocket.close() 1ehjm

32 

33 

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

35async def routerindex(websocket: WebSocket, deps: DepList): 1abcd

36 await websocket.accept() 1fpkn

37 await websocket.send_text(json.dumps(deps)) 1fpkn

38 await websocket.close() 1fpkn

39 

40 

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

42async def routerprefixindex(websocket: WebSocket, deps: DepList): 1abcd

43 await websocket.accept() 1gilo

44 await websocket.send_text(json.dumps(deps)) 1gilo

45 await websocket.close() 1gilo

46 

47 

48app.include_router(router, dependencies=[create_dependency("router2")]) 1abcd

49app.include_router( 1abcd

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

51) 

52 

53 

54def test_index(): 1abcd

55 client = TestClient(app) 1ehjm

56 with client.websocket_connect("/") as websocket: 1ehjm

57 data = json.loads(websocket.receive_text()) 1ehjm

58 assert data == ["app", "index"] 1ehjm

59 

60 

61def test_routerindex(): 1abcd

62 client = TestClient(app) 1fpkn

63 with client.websocket_connect("/router") as websocket: 1fpkn

64 data = json.loads(websocket.receive_text()) 1fpkn

65 assert data == ["app", "router2", "router", "routerindex"] 1fpkn

66 

67 

68def test_routerprefixindex(): 1abcd

69 client = TestClient(app) 1gilo

70 with client.websocket_connect("/prefix/") as websocket: 1gilo

71 data = json.loads(websocket.receive_text()) 1gilo

72 assert data == ["app", "prefix_router2", "prefix_router", "routerprefixindex"] 1gilo