diff --git a/pyproject.toml b/pyproject.toml index fa298ad5b..e0f240bc5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -247,6 +247,9 @@ omit = [ "docs_src/response_model/tutorial003_04_py310.py", "docs_src/dependencies/tutorial013_an_py310.py", # temporary code example? "docs_src/dependencies/tutorial014_an_py310.py", # temporary code example? + # Only run (and cover) on Python 3.14+ + "docs_src/dependencies/tutorial008_an_py310.py", + "tests/test_stringified_annotation_dependency_py314.py", # Pydantic v1 migration, no longer tested "docs_src/pydantic_v1_in_v2/tutorial001_an_py310.py", "docs_src/pydantic_v1_in_v2/tutorial002_an_py310.py", diff --git a/tests/test_asyncapi.py b/tests/test_asyncapi.py index 7000c4cf9..686cb8147 100644 --- a/tests/test_asyncapi.py +++ b/tests/test_asyncapi.py @@ -18,6 +18,10 @@ def test_asyncapi_schema(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass + with client.websocket_connect("/ws/foo"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -38,6 +42,9 @@ def test_asyncapi_no_websockets(): return {"message": "Hello World"} client = TestClient(app) + response = client.get("/") + assert response.status_code == 200 + assert response.json() == {"message": "Hello World"} response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -55,6 +62,9 @@ def test_asyncapi_caching(): await websocket.accept() await websocket.close() + client = TestClient(app) + with client.websocket_connect("/ws"): + pass schema1 = app.asyncapi() schema2 = app.asyncapi() # Should return the same object (identity check) @@ -71,6 +81,8 @@ def test_asyncapi_ui(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi-docs") assert response.status_code == 200, response.text assert response.headers["content-type"] == "text/html; charset=utf-8" @@ -88,6 +100,8 @@ def test_asyncapi_ui_navigation(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi-docs") assert response.status_code == 200, response.text # Should contain link to OpenAPI docs @@ -109,6 +123,11 @@ def test_swagger_ui_asyncapi_navigation(): await websocket.close() client = TestClient(app) + response = client.get("/") + assert response.status_code == 200 + assert response.json() == {"message": "Hello World"} + with client.websocket_connect("/ws"): + pass response = client.get("/docs") assert response.status_code == 200, response.text # Should contain link to AsyncAPI docs @@ -131,6 +150,8 @@ def test_asyncapi_custom_urls(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass # Test custom JSON endpoint response = client.get("/custom/asyncapi.json") assert response.status_code == 200, response.text @@ -163,6 +184,8 @@ def test_asyncapi_disabled(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass # Endpoints should return 404 response = client.get("/asyncapi.json") assert response.status_code == 404 @@ -180,6 +203,8 @@ def test_asyncapi_channel_structure(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -209,6 +234,12 @@ def test_asyncapi_multiple_websockets(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws1"): + pass + with client.websocket_connect("/ws2"): + pass + with client.websocket_connect("/ws3/bar"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -233,6 +264,8 @@ def test_asyncapi_with_metadata(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -256,6 +289,8 @@ def test_asyncapi_ui_no_docs_url(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi-docs") assert response.status_code == 200, response.text # Should not contain link to /docs if docs_url is None @@ -277,6 +312,8 @@ def test_asyncapi_with_servers(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -302,6 +339,8 @@ def test_asyncapi_with_all_metadata(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -337,6 +376,8 @@ def test_asyncapi_with_external_docs(): } client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -357,6 +398,8 @@ def test_asyncapi_channel_with_route_name(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() @@ -376,6 +419,9 @@ def test_get_asyncapi_channel_direct(): await websocket.accept() await websocket.close() + client = TestClient(app) + with client.websocket_connect("/ws"): + pass # Get the route from the app route = next(r for r in app.routes if isinstance(r, routing.APIWebSocketRoute)) channel = get_asyncapi_channel(route=route) @@ -394,6 +440,9 @@ def test_get_asyncapi_direct(): await websocket.accept() await websocket.close() + client = TestClient(app) + with client.websocket_connect("/ws"): + pass schema = get_asyncapi( title=app.title, version=app.version, @@ -419,6 +468,8 @@ def test_asyncapi_url_none_no_link_in_swagger(): await websocket.close() client = TestClient(app) + with client.websocket_connect("/ws"): + pass # Swagger UI should not show AsyncAPI link when asyncapi_url is None response = client.get("/docs") assert response.status_code == 200, response.text @@ -444,6 +495,8 @@ def test_asyncapi_with_root_path_in_servers(): # Use TestClient with root_path to trigger the root_path logic client = TestClient(app, root_path="/api/v1") + with client.websocket_connect("/ws"): + pass response = client.get("/asyncapi.json") assert response.status_code == 200, response.text schema = response.json() diff --git a/tests/test_dependencies_utils.py b/tests/test_dependencies_utils.py index 9257d1c9e..54a21b348 100644 --- a/tests/test_dependencies_utils.py +++ b/tests/test_dependencies_utils.py @@ -1,4 +1,9 @@ -from fastapi.dependencies.utils import get_typed_annotation +import inspect +import sys +from types import SimpleNamespace +from unittest.mock import patch + +from fastapi.dependencies.utils import get_typed_annotation, get_typed_signature def test_get_typed_annotation(): @@ -6,3 +11,31 @@ def test_get_typed_annotation(): annotation = "None" typed_annotation = get_typed_annotation(annotation, globals()) assert typed_annotation is None + + +def test_get_signature_nameerror_py314_branch(): + """Cover _get_signature NameError branch with Python 3.14+ annotation_format path.""" + real_signature = inspect.signature + + def mock_signature(call, *args, **kwargs): + if kwargs.get("eval_str") is True: + raise NameError("undefined name") + # On Python < 3.14, inspect.signature does not accept annotation_format + kwargs.pop("annotation_format", None) + return real_signature(call, *args, **kwargs) + + def simple_dep(x: int) -> int: + return x + + # annotationlib is only available on Python 3.14+; provide a minimal mock # noqa: E501 + fake_annotationlib = SimpleNamespace(Format=SimpleNamespace(FORWARDREF=object())) + + with ( + patch.object(sys, "version_info", (3, 14)), + patch.dict("sys.modules", {"annotationlib": fake_annotationlib}), + patch("fastapi.dependencies.utils.inspect.signature", mock_signature), + ): + sig = get_typed_signature(simple_dep) + assert len(sig.parameters) == 1 + assert sig.parameters["x"].annotation is int + assert simple_dep(42) == 42 # cover simple_dep body diff --git a/tests/test_pydantic_v1_error.py b/tests/test_pydantic_v1_error.py index 044fdf0d6..f02c9f78f 100644 --- a/tests/test_pydantic_v1_error.py +++ b/tests/test_pydantic_v1_error.py @@ -6,7 +6,7 @@ import pytest from tests.utils import skip_module_if_py_gte_314 if sys.version_info >= (3, 14): - skip_module_if_py_gte_314() + skip_module_if_py_gte_314() # pragma: no cover from fastapi import FastAPI from fastapi.exceptions import PydanticV1NotSupportedError diff --git a/tests/utils.py b/tests/utils.py index 09c4e13b0..3d1cfcf5b 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -12,5 +12,5 @@ needs_py314 = pytest.mark.skipif( def skip_module_if_py_gte_314(): """Skip entire module on Python 3.14+ at import time.""" - if sys.version_info >= (3, 14): - pytest.skip("requires python3.13-", allow_module_level=True) + if sys.version_info >= (3, 14): # pragma: no cover + pytest.skip("requires python3.13-", allow_module_level=True) # pragma: no cover