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

75 statements  

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

1import importlib 1xyzA

2 

3import pytest 1xyzA

4from fastapi import FastAPI 1xyzA

5from fastapi.testclient import TestClient 1xyzA

6from fastapi.websockets import WebSocketDisconnect 1xyzA

7 

8from ...utils import needs_py310 1xyzA

9 

10 

11@pytest.fixture( 1xyzA

12 name="app", 

13 params=[ 

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

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

16 ], 

17) 

18def get_app(request: pytest.FixtureRequest): 1xyzA

19 mod = importlib.import_module(f"docs_src.websockets_.{request.param}") 1Z0123456789!#$%'()*+,-./:;=?@[]^_`{|}~

20 

21 return mod.app 1Z0123456789!#$%'()*+,-./:;=?@[]^_`{|}~

22 

23 

24def test_main(app: FastAPI): 1xyzA

25 client = TestClient(app) 1BCDEFGHI

26 response = client.get("/") 1BCDEFGHI

27 assert response.status_code == 200, response.text 1BCDEFGHI

28 assert b"<!DOCTYPE html>" in response.content 1BCDEFGHI

29 

30 

31def test_websocket_with_cookie(app: FastAPI): 1xyzA

32 client = TestClient(app, cookies={"session": "fakesession"}) 1ijklmnop

33 with pytest.raises(WebSocketDisconnect): 1ijklmnop

34 with client.websocket_connect("/items/foo/ws") as websocket: 1ijklmnop

35 message = "Message one" 1ijklmnop

36 websocket.send_text(message) 1ijklmnop

37 data = websocket.receive_text() 1ijklmnop

38 assert data == "Session cookie or query token value is: fakesession" 1ijklmnop

39 data = websocket.receive_text() 1ijklmnop

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

41 message = "Message two" 1ijklmnop

42 websocket.send_text(message) 1ijklmnop

43 data = websocket.receive_text() 1ijklmnop

44 assert data == "Session cookie or query token value is: fakesession" 1ijklmnop

45 data = websocket.receive_text() 1ijklmnop

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

47 

48 

49def test_websocket_with_header(app: FastAPI): 1xyzA

50 client = TestClient(app) 1qrstuvw

51 with pytest.raises(WebSocketDisconnect): 1qrstuvw

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

53 message = "Message one" 1qrstuvw

54 websocket.send_text(message) 1qrstuvw

55 data = websocket.receive_text() 1qrstuvw

56 assert data == "Session cookie or query token value is: some-token" 1qrstuvw

57 data = websocket.receive_text() 1qrstuvw

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

59 message = "Message two" 1qrstuvw

60 websocket.send_text(message) 1qrstuvw

61 data = websocket.receive_text() 1qrstuvw

62 assert data == "Session cookie or query token value is: some-token" 1qrstuvw

63 data = websocket.receive_text() 1qrstuvw

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

65 

66 

67def test_websocket_with_header_and_query(app: FastAPI): 1xyzA

68 client = TestClient(app) 1abcdefgh

69 with pytest.raises(WebSocketDisconnect): 1abcdefgh

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

71 message = "Message one" 1abcdefgh

72 websocket.send_text(message) 1abcdefgh

73 data = websocket.receive_text() 1abcdefgh

74 assert data == "Session cookie or query token value is: some-token" 1abcdefgh

75 data = websocket.receive_text() 1abcdefgh

76 assert data == "Query parameter q is: 3" 1abcdefgh

77 data = websocket.receive_text() 1abcdefgh

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

79 message = "Message two" 1abcdefgh

80 websocket.send_text(message) 1abcdefgh

81 data = websocket.receive_text() 1abcdefgh

82 assert data == "Session cookie or query token value is: some-token" 1abcdefgh

83 data = websocket.receive_text() 1abcdefgh

84 assert data == "Query parameter q is: 3" 1abcdefgh

85 data = websocket.receive_text() 1abcdefgh

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

87 

88 

89def test_websocket_no_credentials(app: FastAPI): 1xyzA

90 client = TestClient(app) 1JKLMNOPQ

91 with pytest.raises(WebSocketDisconnect): 1JKLMNOPQ

92 with client.websocket_connect("/items/foo/ws"): 1JKLMNOPQ

93 pytest.fail( 

94 "did not raise WebSocketDisconnect on __enter__" 

95 ) # pragma: no cover 

96 

97 

98def test_websocket_invalid_data(app: FastAPI): 1xyzA

99 client = TestClient(app) 1RSTUVWXY

100 with pytest.raises(WebSocketDisconnect): 1RSTUVWXY

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

102 pytest.fail( 

103 "did not raise WebSocketDisconnect on __enter__" 

104 ) # pragma: no cover