Coverage for tests / test_tutorial / test_server_sent_events / test_tutorial005.py: 100%

22 statements  

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

1import importlib 1abcd

2 

3import pytest 1abcd

4from fastapi.testclient import TestClient 1abcd

5from inline_snapshot import snapshot 1abcd

6 

7 

8@pytest.fixture( 1abcd

9 name="client", 

10 params=[ 

11 pytest.param("tutorial005_py310"), 

12 ], 

13) 

14def get_client(request: pytest.FixtureRequest): 1abcd

15 mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}") 1ijklmnop

16 client = TestClient(mod.app) 1ijklmnop

17 return client 1ijklmnop

18 

19 

20def test_stream_chat(client: TestClient): 1abcd

21 response = client.post( 1efgh

22 "/chat/stream", 

23 json={"text": "hello world"}, 

24 ) 

25 assert response.status_code == 200, response.text 1efgh

26 assert response.headers["content-type"] == "text/event-stream; charset=utf-8" 1efgh

27 

28 lines = response.text.strip().split("\n") 1efgh

29 

30 event_lines = [line for line in lines if line.startswith("event: ")] 1efgh

31 assert event_lines == [ 1efgh

32 "event: token", 

33 "event: token", 

34 "event: done", 

35 ] 

36 

37 data_lines = [line for line in lines if line.startswith("data: ")] 1efgh

38 assert data_lines == [ 1efgh

39 'data: "hello"', 

40 'data: "world"', 

41 "data: [DONE]", 

42 ] 

43 

44 

45def test_openapi_schema(client: TestClient): 1abcd

46 response = client.get("/openapi.json") 1qrst

47 assert response.status_code == 200, response.text 1qrst

48 assert response.json() == snapshot( 1qrst

49 { 

50 "openapi": "3.1.0", 

51 "info": {"title": "FastAPI", "version": "0.1.0"}, 

52 "paths": { 

53 "/chat/stream": { 

54 "post": { 

55 "summary": "Stream Chat", 

56 "operationId": "stream_chat_chat_stream_post", 

57 "requestBody": { 

58 "content": { 

59 "application/json": { 

60 "schema": {"$ref": "#/components/schemas/Prompt"} 

61 } 

62 }, 

63 "required": True, 

64 }, 

65 "responses": { 

66 "200": { 

67 "description": "Successful Response", 

68 "content": { 

69 "text/event-stream": { 

70 "itemSchema": { 

71 "type": "object", 

72 "properties": { 

73 "data": {"type": "string"}, 

74 "event": {"type": "string"}, 

75 "id": {"type": "string"}, 

76 "retry": { 

77 "type": "integer", 

78 "minimum": 0, 

79 }, 

80 }, 

81 } 

82 } 

83 }, 

84 }, 

85 "422": { 

86 "description": "Validation Error", 

87 "content": { 

88 "application/json": { 

89 "schema": { 

90 "$ref": "#/components/schemas/HTTPValidationError" 

91 } 

92 } 

93 }, 

94 }, 

95 }, 

96 } 

97 } 

98 }, 

99 "components": { 

100 "schemas": { 

101 "HTTPValidationError": { 

102 "properties": { 

103 "detail": { 

104 "items": { 

105 "$ref": "#/components/schemas/ValidationError" 

106 }, 

107 "type": "array", 

108 "title": "Detail", 

109 } 

110 }, 

111 "type": "object", 

112 "title": "HTTPValidationError", 

113 }, 

114 "Prompt": { 

115 "properties": {"text": {"type": "string", "title": "Text"}}, 

116 "type": "object", 

117 "required": ["text"], 

118 "title": "Prompt", 

119 }, 

120 "ValidationError": { 

121 "properties": { 

122 "loc": { 

123 "items": { 

124 "anyOf": [{"type": "string"}, {"type": "integer"}] 

125 }, 

126 "type": "array", 

127 "title": "Location", 

128 }, 

129 "msg": {"type": "string", "title": "Message"}, 

130 "type": {"type": "string", "title": "Error Type"}, 

131 "input": {"title": "Input"}, 

132 "ctx": {"type": "object", "title": "Context"}, 

133 }, 

134 "type": "object", 

135 "required": ["loc", "msg", "type"], 

136 "title": "ValidationError", 

137 }, 

138 } 

139 }, 

140 } 

141 )