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

75 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2025-05-05 00:03 +0000

1import importlib 1ghijkl

2 

3import pytest 1ghijkl

4from fastapi import FastAPI 1ghijkl

5from fastapi.testclient import TestClient 1ghijkl

6from fastapi.websockets import WebSocketDisconnect 1ghijkl

7 

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

9 

10 

11@pytest.fixture( 1ghijkl

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): 1ghijkl

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

23 

24 return mod.app 1ghijkl

25 

26 

27def test_main(app: FastAPI): 1ghijkl

28 client = TestClient(app) 1yzABCD

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

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

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

32 

33 

34def test_websocket_with_cookie(app: FastAPI): 1ghijkl

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

36 with pytest.raises(WebSocketDisconnect): 1mnopqr

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

38 message = "Message one" 1mnopqr

39 websocket.send_text(message) 1mnopqr

40 data = websocket.receive_text() 1mnopqr

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

42 data = websocket.receive_text() 1mnopqr

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

44 message = "Message two" 1mnopqr

45 websocket.send_text(message) 1mnopqr

46 data = websocket.receive_text() 1mnopqr

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

48 data = websocket.receive_text() 1mnopqr

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

50 

51 

52def test_websocket_with_header(app: FastAPI): 1ghijkl

53 client = TestClient(app) 1stuvwx

54 with pytest.raises(WebSocketDisconnect): 1stuvwx

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

56 message = "Message one" 1stuvwx

57 websocket.send_text(message) 1stuvwx

58 data = websocket.receive_text() 1stuvwx

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

60 data = websocket.receive_text() 1stuvwx

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

62 message = "Message two" 1stuvwx

63 websocket.send_text(message) 1stuvwx

64 data = websocket.receive_text() 1stuvwx

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

66 data = websocket.receive_text() 1stuvwx

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

68 

69 

70def test_websocket_with_header_and_query(app: FastAPI): 1ghijkl

71 client = TestClient(app) 1abcdef

72 with pytest.raises(WebSocketDisconnect): 1abcdef

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

74 message = "Message one" 1abcdef

75 websocket.send_text(message) 1abcdef

76 data = websocket.receive_text() 1abcdef

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

78 data = websocket.receive_text() 1abcdef

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

80 data = websocket.receive_text() 1abcdef

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

82 message = "Message two" 1abcdef

83 websocket.send_text(message) 1abcdef

84 data = websocket.receive_text() 1abcdef

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

86 data = websocket.receive_text() 1abcdef

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

88 data = websocket.receive_text() 1abcdef

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

90 

91 

92def test_websocket_no_credentials(app: FastAPI): 1ghijkl

93 client = TestClient(app) 1GEHIJK

94 with pytest.raises(WebSocketDisconnect): 1GEHIJK

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

96 pytest.fail( 1E

97 "did not raise WebSocketDisconnect on __enter__" 

98 ) # pragma: no cover 

99 

100 

101def test_websocket_invalid_data(app: FastAPI): 1ghijkl

102 client = TestClient(app) 1LFMNOP

103 with pytest.raises(WebSocketDisconnect): 1LFMNOP

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

105 pytest.fail( 1F

106 "did not raise WebSocketDisconnect on __enter__" 

107 ) # pragma: no cover