Browse Source

fix coverage

pull/15006/head
rechain 5 months ago
parent
commit
5040c2986c
  1. 3
      pyproject.toml
  2. 53
      tests/test_asyncapi.py
  3. 35
      tests/test_dependencies_utils.py
  4. 2
      tests/test_pydantic_v1_error.py
  5. 4
      tests/utils.py

3
pyproject.toml

@ -247,6 +247,9 @@ omit = [
"docs_src/response_model/tutorial003_04_py310.py", "docs_src/response_model/tutorial003_04_py310.py",
"docs_src/dependencies/tutorial013_an_py310.py", # temporary code example? "docs_src/dependencies/tutorial013_an_py310.py", # temporary code example?
"docs_src/dependencies/tutorial014_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 # Pydantic v1 migration, no longer tested
"docs_src/pydantic_v1_in_v2/tutorial001_an_py310.py", "docs_src/pydantic_v1_in_v2/tutorial001_an_py310.py",
"docs_src/pydantic_v1_in_v2/tutorial002_an_py310.py", "docs_src/pydantic_v1_in_v2/tutorial002_an_py310.py",

53
tests/test_asyncapi.py

@ -18,6 +18,10 @@ def test_asyncapi_schema():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
with client.websocket_connect("/ws/foo"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -38,6 +42,9 @@ def test_asyncapi_no_websockets():
return {"message": "Hello World"} return {"message": "Hello World"}
client = TestClient(app) client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -55,6 +62,9 @@ def test_asyncapi_caching():
await websocket.accept() await websocket.accept()
await websocket.close() await websocket.close()
client = TestClient(app)
with client.websocket_connect("/ws"):
pass
schema1 = app.asyncapi() schema1 = app.asyncapi()
schema2 = app.asyncapi() schema2 = app.asyncapi()
# Should return the same object (identity check) # Should return the same object (identity check)
@ -71,6 +81,8 @@ def test_asyncapi_ui():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi-docs") response = client.get("/asyncapi-docs")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
assert response.headers["content-type"] == "text/html; charset=utf-8" assert response.headers["content-type"] == "text/html; charset=utf-8"
@ -88,6 +100,8 @@ def test_asyncapi_ui_navigation():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi-docs") response = client.get("/asyncapi-docs")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
# Should contain link to OpenAPI docs # Should contain link to OpenAPI docs
@ -109,6 +123,11 @@ def test_swagger_ui_asyncapi_navigation():
await websocket.close() await websocket.close()
client = TestClient(app) 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") response = client.get("/docs")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
# Should contain link to AsyncAPI docs # Should contain link to AsyncAPI docs
@ -131,6 +150,8 @@ def test_asyncapi_custom_urls():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
# Test custom JSON endpoint # Test custom JSON endpoint
response = client.get("/custom/asyncapi.json") response = client.get("/custom/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
@ -163,6 +184,8 @@ def test_asyncapi_disabled():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
# Endpoints should return 404 # Endpoints should return 404
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 404 assert response.status_code == 404
@ -180,6 +203,8 @@ def test_asyncapi_channel_structure():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -209,6 +234,12 @@ def test_asyncapi_multiple_websockets():
await websocket.close() await websocket.close()
client = TestClient(app) 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") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -233,6 +264,8 @@ def test_asyncapi_with_metadata():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -256,6 +289,8 @@ def test_asyncapi_ui_no_docs_url():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi-docs") response = client.get("/asyncapi-docs")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
# Should not contain link to /docs if docs_url is None # Should not contain link to /docs if docs_url is None
@ -277,6 +312,8 @@ def test_asyncapi_with_servers():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -302,6 +339,8 @@ def test_asyncapi_with_all_metadata():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -337,6 +376,8 @@ def test_asyncapi_with_external_docs():
} }
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -357,6 +398,8 @@ def test_asyncapi_channel_with_route_name():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()
@ -376,6 +419,9 @@ def test_get_asyncapi_channel_direct():
await websocket.accept() await websocket.accept()
await websocket.close() await websocket.close()
client = TestClient(app)
with client.websocket_connect("/ws"):
pass
# Get the route from the app # Get the route from the app
route = next(r for r in app.routes if isinstance(r, routing.APIWebSocketRoute)) route = next(r for r in app.routes if isinstance(r, routing.APIWebSocketRoute))
channel = get_asyncapi_channel(route=route) channel = get_asyncapi_channel(route=route)
@ -394,6 +440,9 @@ def test_get_asyncapi_direct():
await websocket.accept() await websocket.accept()
await websocket.close() await websocket.close()
client = TestClient(app)
with client.websocket_connect("/ws"):
pass
schema = get_asyncapi( schema = get_asyncapi(
title=app.title, title=app.title,
version=app.version, version=app.version,
@ -419,6 +468,8 @@ def test_asyncapi_url_none_no_link_in_swagger():
await websocket.close() await websocket.close()
client = TestClient(app) client = TestClient(app)
with client.websocket_connect("/ws"):
pass
# Swagger UI should not show AsyncAPI link when asyncapi_url is None # Swagger UI should not show AsyncAPI link when asyncapi_url is None
response = client.get("/docs") response = client.get("/docs")
assert response.status_code == 200, response.text 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 # Use TestClient with root_path to trigger the root_path logic
client = TestClient(app, root_path="/api/v1") client = TestClient(app, root_path="/api/v1")
with client.websocket_connect("/ws"):
pass
response = client.get("/asyncapi.json") response = client.get("/asyncapi.json")
assert response.status_code == 200, response.text assert response.status_code == 200, response.text
schema = response.json() schema = response.json()

35
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(): def test_get_typed_annotation():
@ -6,3 +11,31 @@ def test_get_typed_annotation():
annotation = "None" annotation = "None"
typed_annotation = get_typed_annotation(annotation, globals()) typed_annotation = get_typed_annotation(annotation, globals())
assert typed_annotation is None 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

2
tests/test_pydantic_v1_error.py

@ -6,7 +6,7 @@ import pytest
from tests.utils import skip_module_if_py_gte_314 from tests.utils import skip_module_if_py_gte_314
if sys.version_info >= (3, 14): 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 import FastAPI
from fastapi.exceptions import PydanticV1NotSupportedError from fastapi.exceptions import PydanticV1NotSupportedError

4
tests/utils.py

@ -12,5 +12,5 @@ needs_py314 = pytest.mark.skipif(
def skip_module_if_py_gte_314(): def skip_module_if_py_gte_314():
"""Skip entire module on Python 3.14+ at import time.""" """Skip entire module on Python 3.14+ at import time."""
if sys.version_info >= (3, 14): if sys.version_info >= (3, 14): # pragma: no cover
pytest.skip("requires python3.13-", allow_module_level=True) pytest.skip("requires python3.13-", allow_module_level=True) # pragma: no cover

Loading…
Cancel
Save