Coverage for tests/test_tutorial/test_websockets/test_tutorial002.py: 100%

75 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2025-12-04 08:29 +0000

1import importlib 1hijklmn

2 

3import pytest 1hijklmn

4from fastapi import FastAPI 1hijklmn

5from fastapi.testclient import TestClient 1hijklmn

6from fastapi.websockets import WebSocketDisconnect 1hijklmn

7 

8from ...utils import needs_py39, needs_py310 1hijklmn

9 

10 

11@pytest.fixture( 1hijklmn

12 name="app", 

13 params=[ 

14 "tutorial002", 

15 pytest.param("tutorial002_py310", marks=needs_py310), 

16 "tutorial002_an", 

17 pytest.param("tutorial002_an_py39", marks=needs_py39), 

18 pytest.param("tutorial002_an_py310", marks=needs_py310), 

19 ], 

20) 

21def get_app(request: pytest.FixtureRequest): 1hijklmn

22 mod = importlib.import_module(f"docs_src.websockets.{request.param}") 1hijklmn

23 

24 return mod.app 1hijklmn

25 

26 

27def test_main(app: FastAPI): 1hijklmn

28 client = TestClient(app) 1CDEFGHI

29 response = client.get("/") 1CDEFGHI

30 assert response.status_code == 200, response.text 1CDEFGHI

31 assert b"<!DOCTYPE html>" in response.content 1CDEFGHI

32 

33 

34def test_websocket_with_cookie(app: FastAPI): 1hijklmn

35 client = TestClient(app, cookies={"session": "fakesession"}) 1opqrstu

36 with pytest.raises(WebSocketDisconnect): 1opqrstu

37 with client.websocket_connect("/items/foo/ws") as websocket: 1opqrstu

38 message = "Message one" 1opqrstu

39 websocket.send_text(message) 1opqrstu

40 data = websocket.receive_text() 1opqrstu

41 assert data == "Session cookie or query token value is: fakesession" 1opqrstu

42 data = websocket.receive_text() 1opqrstu

43 assert data == f"Message text was: {message}, for item ID: foo" 1opqrstu

44 message = "Message two" 1opqrstu

45 websocket.send_text(message) 1opqrstu

46 data = websocket.receive_text() 1opqrstu

47 assert data == "Session cookie or query token value is: fakesession" 1opqrstu

48 data = websocket.receive_text() 1opqrstu

49 assert data == f"Message text was: {message}, for item ID: foo" 1opqrstu

50 

51 

52def test_websocket_with_header(app: FastAPI): 1hijklmn

53 client = TestClient(app) 1vwxyzAB

54 with pytest.raises(WebSocketDisconnect): 1vwxyzAB

55 with client.websocket_connect("/items/bar/ws?token=some-token") as websocket: 1vwxyzAB

56 message = "Message one" 1vwxyzAB

57 websocket.send_text(message) 1vwxyzAB

58 data = websocket.receive_text() 1vwxyzAB

59 assert data == "Session cookie or query token value is: some-token" 1vwxyzAB

60 data = websocket.receive_text() 1vwxyzAB

61 assert data == f"Message text was: {message}, for item ID: bar" 1vwxyzAB

62 message = "Message two" 1vwxyzAB

63 websocket.send_text(message) 1vwxyzAB

64 data = websocket.receive_text() 1vwxyzAB

65 assert data == "Session cookie or query token value is: some-token" 1vwxyzAB

66 data = websocket.receive_text() 1vwxyzAB

67 assert data == f"Message text was: {message}, for item ID: bar" 1vwxyzAB

68 

69 

70def test_websocket_with_header_and_query(app: FastAPI): 1hijklmn

71 client = TestClient(app) 1abcdefg

72 with pytest.raises(WebSocketDisconnect): 1abcdefg

73 with client.websocket_connect("/items/2/ws?q=3&token=some-token") as websocket: 1abcdefg

74 message = "Message one" 1abcdefg

75 websocket.send_text(message) 1abcdefg

76 data = websocket.receive_text() 1abcdefg

77 assert data == "Session cookie or query token value is: some-token" 1abcdefg

78 data = websocket.receive_text() 1abcdefg

79 assert data == "Query parameter q is: 3" 1abcdefg

80 data = websocket.receive_text() 1abcdefg

81 assert data == f"Message text was: {message}, for item ID: 2" 1abcdefg

82 message = "Message two" 1abcdefg

83 websocket.send_text(message) 1abcdefg

84 data = websocket.receive_text() 1abcdefg

85 assert data == "Session cookie or query token value is: some-token" 1abcdefg

86 data = websocket.receive_text() 1abcdefg

87 assert data == "Query parameter q is: 3" 1abcdefg

88 data = websocket.receive_text() 1abcdefg

89 assert data == f"Message text was: {message}, for item ID: 2" 1abcdefg

90 

91 

92def test_websocket_no_credentials(app: FastAPI): 1hijklmn

93 client = TestClient(app) 1LJMNOPQ

94 with pytest.raises(WebSocketDisconnect): 1LJMNOPQ

95 with client.websocket_connect("/items/foo/ws"): 1LJMNOPQ

96 pytest.fail( 1J

97 "did not raise WebSocketDisconnect on __enter__" 

98 ) # pragma: no cover 

99 

100 

101def test_websocket_invalid_data(app: FastAPI): 1hijklmn

102 client = TestClient(app) 1RKSTUVW

103 with pytest.raises(WebSocketDisconnect): 1RKSTUVW

104 with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"): 1RKSTUVW

105 pytest.fail( 1K

106 "did not raise WebSocketDisconnect on __enter__" 

107 ) # pragma: no cover