mirror of
https://github.com/kennethreitz/responder.git
synced 2026-07-21 18:39:29 +00:00
kennethreitz
ea5ba0d26b
The first 8.0 sweep. Flagship: responder.Router records route declarations without an API instance (Flask-blueprint style) and api.include_router() replays them with prefix/tags/dependencies/auth composition, nesting, and conflict guards. Correctness: lifespan= now composes with on_event (background draining runs again), view-style default routes dispatch instead of 500ing, CBV HEAD falls back to on_get, tolerant cookie parsing, case-insensitive resp.headers, async background tasks, async GraphQL resolvers, Jinja builtins survive context assignment, identity-based dependency cycle detection, WSGI mounts wrap once, WebSocket crashes send a 1011 close frame, and dependency metadata matches actual imports. Security: Content-Disposition filename escaping in resp.download(); bounded, lock-guarded MemorySessionBackend with expiry-first eviction. 8.0 behavior changes: no implicit static/ mkdir, url_for raises RouteNotFoundError and percent-encodes values, redirect() defaults to 307 (permanent=True for 308), charset= on encoded text responses (no more fake 'Encoding' header), and RFC 9512 application/yaml. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
# ruff: noqa: E402
|
|
"""GraphQL execution is async-native: ``async def`` resolvers work, and
|
|
resolvers run via ``schema.execute_async`` so a slow resolver doesn't need
|
|
to block the event loop with a sync ``schema.execute`` call.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
graphene = pytest.importorskip("graphene")
|
|
|
|
from responder.ext.graphql import GraphQLView
|
|
|
|
|
|
@pytest.fixture
|
|
def async_schema():
|
|
class Query(graphene.ObjectType):
|
|
hello = graphene.String(name=graphene.String(default_value="stranger"))
|
|
|
|
async def resolve_hello(self, info, name):
|
|
await asyncio.sleep(0)
|
|
return f"Hello {name}"
|
|
|
|
return graphene.Schema(query=Query)
|
|
|
|
|
|
@pytest.fixture
|
|
def mixed_schema():
|
|
class Query(graphene.ObjectType):
|
|
sync_field = graphene.String()
|
|
async_field = graphene.String()
|
|
|
|
def resolve_sync_field(self, info):
|
|
return "sync"
|
|
|
|
async def resolve_async_field(self, info):
|
|
await asyncio.sleep(0)
|
|
return "async"
|
|
|
|
return graphene.Schema(query=Query)
|
|
|
|
|
|
def test_async_resolver_over_post(api, async_schema):
|
|
api.add_route("/", GraphQLView(schema=async_schema, api=api))
|
|
r = api.requests.post("http://;/", json={"query": "{ hello }"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"data": {"hello": "Hello stranger"}}
|
|
|
|
|
|
def test_async_resolver_over_get(api, async_schema):
|
|
api.add_route("/", GraphQLView(schema=async_schema, api=api))
|
|
r = api.requests.get("http://;/?query={ hello }", headers={"Accept": "json"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"data": {"hello": "Hello stranger"}}
|
|
|
|
|
|
def test_async_resolver_with_variables(api, async_schema):
|
|
api.add_route("/", GraphQLView(schema=async_schema, api=api))
|
|
r = api.requests.post(
|
|
"http://;/",
|
|
json={
|
|
"query": "query H($name: String) { hello(name: $name) }",
|
|
"variables": {"name": "world"},
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json() == {"data": {"hello": "Hello world"}}
|
|
|
|
|
|
def test_mixed_sync_and_async_resolvers(api, mixed_schema):
|
|
api.add_route("/", GraphQLView(schema=mixed_schema, api=api))
|
|
r = api.requests.post("http://;/", json={"query": "{ syncField asyncField }"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"data": {"syncField": "sync", "asyncField": "async"}}
|
|
|
|
|
|
def test_sync_resolver_still_works(api):
|
|
class Query(graphene.ObjectType):
|
|
hello = graphene.String()
|
|
|
|
def resolve_hello(self, info):
|
|
return "hi"
|
|
|
|
api.add_route("/", GraphQLView(schema=graphene.Schema(query=Query), api=api))
|
|
r = api.requests.post("http://;/", json={"query": "{ hello }"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"data": {"hello": "hi"}}
|
|
|
|
|
|
def test_syntax_error_still_returns_400(api, async_schema):
|
|
api.add_route("/", GraphQLView(schema=async_schema, api=api))
|
|
r = api.requests.post("http://;/", json={"query": "{ hello"})
|
|
assert r.status_code == 400
|
|
assert "errors" in r.json()
|
|
|
|
|
|
def test_async_mutation(api):
|
|
class Query(graphene.ObjectType):
|
|
hello = graphene.String()
|
|
|
|
def resolve_hello(self, info):
|
|
return "hi"
|
|
|
|
class CreateUser(graphene.Mutation):
|
|
class Arguments:
|
|
name = graphene.String(required=True)
|
|
|
|
ok = graphene.Boolean()
|
|
|
|
async def mutate(self, info, name):
|
|
await asyncio.sleep(0)
|
|
return CreateUser(ok=True)
|
|
|
|
class Mutation(graphene.ObjectType):
|
|
create_user = CreateUser.Field()
|
|
|
|
schema = graphene.Schema(query=Query, mutation=Mutation)
|
|
api.add_route("/", GraphQLView(schema=schema, api=api))
|
|
r = api.requests.post(
|
|
"http://;/", json={"query": 'mutation { createUser(name: "eve") { ok } }'}
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json() == {"data": {"createUser": {"ok": True}}}
|