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
« prev ^ index » next coverage.py v7.13.3, created at 2026-04-06 01:24 +0000
1import importlib 1abcd
3import pytest 1abcd
4from fastapi.testclient import TestClient 1abcd
5from inline_snapshot import snapshot 1abcd
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
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
28 lines = response.text.strip().split("\n") 1efgh
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 ]
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 ]
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 )