mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

✅ Add tests for SSE docs src

pull/15030/head
Sebastián Ramírez 5 months ago
parent
4e7abc7871
commit
a8df17664b
6 changed files with 652 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 0
      tests/test_tutorial/test_server_sent_events/__init__.py
  2. 191
      tests/test_tutorial/test_server_sent_events/test_tutorial001.py
  3. 83
      tests/test_tutorial/test_server_sent_events/test_tutorial002.py
  4. 73
      tests/test_tutorial/test_server_sent_events/test_tutorial003.py
  5. 164
      tests/test_tutorial/test_server_sent_events/test_tutorial004.py
  6. 141
      tests/test_tutorial/test_server_sent_events/test_tutorial005.py

0
tests/test_tutorial/test_server_sent_events/__init__.py
View File

191
tests/test_tutorial/test_server_sent_events/test_tutorial001.py
View File

@ -0,0 +1,191 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial001_py310"),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}")
client = TestClient(mod.app)
return client
@pytest.mark.parametrize(
"path",
[
"/items/stream",
"/items/stream-no-async",
"/items/stream-no-annotation",
"/items/stream-no-async-no-annotation",
],
)
def test_stream_items(client: TestClient, path: str):
response = client.get(path)
assert response.status_code == 200, response.text
assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
data_lines = [
line for line in response.text.strip().split("\n") if line.startswith("data: ")
]
assert len(data_lines) == 3
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/stream": {
"get": {
"summary": "Sse Items",
"operationId": "sse_items_items_stream_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {
"type": "string",
"contentMediaType": "application/json",
"contentSchema": {
"$ref": "#/components/schemas/Item"
},
},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
"required": ["data"],
}
}
},
}
},
}
},
"/items/stream-no-async": {
"get": {
"summary": "Sse Items No Async",
"operationId": "sse_items_no_async_items_stream_no_async_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {
"type": "string",
"contentMediaType": "application/json",
"contentSchema": {
"$ref": "#/components/schemas/Item"
},
},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
"required": ["data"],
}
}
},
}
},
}
},
"/items/stream-no-annotation": {
"get": {
"summary": "Sse Items No Annotation",
"operationId": "sse_items_no_annotation_items_stream_no_annotation_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {"type": "string"},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
}
}
},
}
},
}
},
"/items/stream-no-async-no-annotation": {
"get": {
"summary": "Sse Items No Async No Annotation",
"operationId": "sse_items_no_async_no_annotation_items_stream_no_async_no_annotation_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {"type": "string"},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
}
}
},
}
},
}
},
},
"components": {
"schemas": {
"Item": {
"properties": {
"name": {"type": "string", "title": "Name"},
"description": {
"anyOf": [
{"type": "string"},
{"type": "null"},
],
"title": "Description",
},
},
"type": "object",
"required": ["name", "description"],
"title": "Item",
}
}
},
}
)

83
tests/test_tutorial/test_server_sent_events/test_tutorial002.py
View File

@ -0,0 +1,83 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial002_py310"),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}")
client = TestClient(mod.app)
return client
def test_stream_items(client: TestClient):
response = client.get("/items/stream")
assert response.status_code == 200, response.text
assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
lines = response.text.strip().split("\n")
# First event is a comment-only event
assert lines[0] == ": stream of item updates"
# Remaining lines contain event:, data:, id:, retry: fields
event_lines = [line for line in lines if line.startswith("event: ")]
assert len(event_lines) == 3
assert all(line == "event: item_update" for line in event_lines)
data_lines = [line for line in lines if line.startswith("data: ")]
assert len(data_lines) == 3
id_lines = [line for line in lines if line.startswith("id: ")]
assert id_lines == ["id: 1", "id: 2", "id: 3"]
retry_lines = [line for line in lines if line.startswith("retry: ")]
assert len(retry_lines) == 3
assert all(line == "retry: 5000" for line in retry_lines)
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/stream": {
"get": {
"summary": "Stream Items",
"operationId": "stream_items_items_stream_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {"type": "string"},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
}
}
},
}
},
}
}
},
}
)

73
tests/test_tutorial/test_server_sent_events/test_tutorial003.py
View File

@ -0,0 +1,73 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial003_py310"),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}")
client = TestClient(mod.app)
return client
def test_stream_logs(client: TestClient):
response = client.get("/logs/stream")
assert response.status_code == 200, response.text
assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
data_lines = [
line for line in response.text.strip().split("\n") if line.startswith("data: ")
]
assert len(data_lines) == 3
# raw_data is sent without JSON encoding (no quotes around the string)
assert data_lines[0] == "data: 2025-01-01 INFO Application started"
assert data_lines[1] == "data: 2025-01-01 DEBUG Connected to database"
assert data_lines[2] == "data: 2025-01-01 WARN High memory usage detected"
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/logs/stream": {
"get": {
"summary": "Stream Logs",
"operationId": "stream_logs_logs_stream_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {"type": "string"},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
}
}
},
}
},
}
}
},
}
)

164
tests/test_tutorial/test_server_sent_events/test_tutorial004.py
View File

@ -0,0 +1,164 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial004_py310"),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}")
client = TestClient(mod.app)
return client
def test_stream_all_items(client: TestClient):
response = client.get("/items/stream")
assert response.status_code == 200, response.text
data_lines = [
line for line in response.text.strip().split("\n") if line.startswith("data: ")
]
assert len(data_lines) == 3
id_lines = [
line for line in response.text.strip().split("\n") if line.startswith("id: ")
]
assert id_lines == ["id: 0", "id: 1", "id: 2"]
def test_resume_from_last_event_id(client: TestClient):
response = client.get(
"/items/stream",
headers={"last-event-id": "0"},
)
assert response.status_code == 200, response.text
data_lines = [
line for line in response.text.strip().split("\n") if line.startswith("data: ")
]
assert len(data_lines) == 2
id_lines = [
line for line in response.text.strip().split("\n") if line.startswith("id: ")
]
assert id_lines == ["id: 1", "id: 2"]
def test_resume_from_last_item(client: TestClient):
response = client.get(
"/items/stream",
headers={"last-event-id": "1"},
)
assert response.status_code == 200, response.text
data_lines = [
line for line in response.text.strip().split("\n") if line.startswith("data: ")
]
assert len(data_lines) == 1
id_lines = [
line for line in response.text.strip().split("\n") if line.startswith("id: ")
]
assert id_lines == ["id: 2"]
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/items/stream": {
"get": {
"summary": "Stream Items",
"operationId": "stream_items_items_stream_get",
"parameters": [
{
"name": "last-event-id",
"in": "header",
"required": False,
"schema": {
"anyOf": [{"type": "integer"}, {"type": "null"}],
"title": "Last-Event-Id",
},
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {"type": "string"},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
"input": {"title": "Input"},
"ctx": {"type": "object", "title": "Context"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)

141
tests/test_tutorial/test_server_sent_events/test_tutorial005.py
View File

@ -0,0 +1,141 @@
import importlib
import pytest
from fastapi.testclient import TestClient
from inline_snapshot import snapshot
@pytest.fixture(
name="client",
params=[
pytest.param("tutorial005_py310"),
],
)
def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}")
client = TestClient(mod.app)
return client
def test_stream_chat(client: TestClient):
response = client.post(
"/chat/stream",
json={"text": "hello world"},
)
assert response.status_code == 200, response.text
assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
lines = response.text.strip().split("\n")
event_lines = [line for line in lines if line.startswith("event: ")]
assert event_lines == [
"event: token",
"event: token",
"event: done",
]
data_lines = [line for line in lines if line.startswith("data: ")]
assert data_lines == [
'data: "hello"',
'data: "world"',
"data: [DONE]",
]
def test_openapi_schema(client: TestClient):
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == snapshot(
{
"openapi": "3.1.0",
"info": {"title": "FastAPI", "version": "0.1.0"},
"paths": {
"/chat/stream": {
"post": {
"summary": "Stream Chat",
"operationId": "stream_chat_chat_stream_post",
"requestBody": {
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Prompt"}
}
},
"required": True,
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"text/event-stream": {
"itemSchema": {
"type": "object",
"properties": {
"data": {"type": "string"},
"event": {"type": "string"},
"id": {"type": "string"},
"retry": {
"type": "integer",
"minimum": 0,
},
},
}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
}
}
},
"components": {
"schemas": {
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail",
}
},
"type": "object",
"title": "HTTPValidationError",
},
"Prompt": {
"properties": {"text": {"type": "string", "title": "Text"}},
"type": "object",
"required": ["text"],
"title": "Prompt",
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [{"type": "string"}, {"type": "integer"}]
},
"type": "array",
"title": "Location",
},
"msg": {"type": "string", "title": "Message"},
"type": {"type": "string", "title": "Error Type"},
"input": {"title": "Input"},
"ctx": {"type": "object", "title": "Context"},
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError",
},
}
},
}
)
Write Preview
Loading…
Cancel
Save