diff --git a/tests/test_tutorial/test_server_sent_events/__init__.py b/tests/test_tutorial/test_server_sent_events/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/test_tutorial/test_server_sent_events/test_tutorial001.py b/tests/test_tutorial/test_server_sent_events/test_tutorial001.py new file mode 100644 index 0000000000..75fdda3601 --- /dev/null +++ b/tests/test_tutorial/test_server_sent_events/test_tutorial001.py @@ -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", + } + } + }, + } + ) diff --git a/tests/test_tutorial/test_server_sent_events/test_tutorial002.py b/tests/test_tutorial/test_server_sent_events/test_tutorial002.py new file mode 100644 index 0000000000..b9cbf43854 --- /dev/null +++ b/tests/test_tutorial/test_server_sent_events/test_tutorial002.py @@ -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, + }, + }, + } + } + }, + } + }, + } + } + }, + } + ) diff --git a/tests/test_tutorial/test_server_sent_events/test_tutorial003.py b/tests/test_tutorial/test_server_sent_events/test_tutorial003.py new file mode 100644 index 0000000000..6277a27c90 --- /dev/null +++ b/tests/test_tutorial/test_server_sent_events/test_tutorial003.py @@ -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, + }, + }, + } + } + }, + } + }, + } + } + }, + } + ) diff --git a/tests/test_tutorial/test_server_sent_events/test_tutorial004.py b/tests/test_tutorial/test_server_sent_events/test_tutorial004.py new file mode 100644 index 0000000000..38ce888c1c --- /dev/null +++ b/tests/test_tutorial/test_server_sent_events/test_tutorial004.py @@ -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", + }, + } + }, + } + ) diff --git a/tests/test_tutorial/test_server_sent_events/test_tutorial005.py b/tests/test_tutorial/test_server_sent_events/test_tutorial005.py new file mode 100644 index 0000000000..1b5c3492f7 --- /dev/null +++ b/tests/test_tutorial/test_server_sent_events/test_tutorial005.py @@ -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", + }, + } + }, + } + )