Coverage for tests/test_http_connection_injection.py: 100%

1from fastapi import Depends, FastAPI 7 ctx1abcdefg

2from fastapi.requests import HTTPConnection 7 ctx1abcdefg

3from fastapi.testclient import TestClient 7 ctx1abcdefg

4from starlette.websockets import WebSocket 7 ctx1abcdefg

5 

6app = FastAPI() 7 ctx1abcdefg

7app.state.value = 42 7 ctx1abcdefg

8 

9 

10async def extract_value_from_http_connection(conn: HTTPConnection): 7 ctx1abcdefg

11 return conn.app.state.value 14 ctx1ohpiqjrksltmun

12 

13 

14@app.get("/http") 7 ctx1abcdefg

15async def get_value_by_http(value: int = Depends(extract_value_from_http_connection)): 7 ctx1abcdefg

16 return value 7 ctx1opqrstu

17 

18 

19@app.websocket("/ws") 7 ctx1abcdefg

20async def get_value_by_ws( 7 ctx1abcdefg

21 websocket: WebSocket, value: int = Depends(extract_value_from_http_connection) 

22): 

23 await websocket.accept() 7 ctx1hijklmn

24 await websocket.send_json(value) 7 ctx1hijklmn

25 await websocket.close() 7 ctx1hijklmn

26 

27 

28client = TestClient(app) 7 ctx1abcdefg

29 

30 

31def test_value_extracting_by_http(): 7 ctx1abcdefg

32 response = client.get("/http") 7 ctx1opqrstu

33 assert response.status_code == 200 7 ctx1opqrstu

34 assert response.json() == 42 7 ctx1opqrstu

35 

36 

37def test_value_extracting_by_ws(): 7 ctx1abcdefg

38 with client.websocket_connect("/ws") as websocket: 7 ctx1hijklmn

39 assert websocket.receive_json() == 42 7 ctx1hijklmn