mirror of
https://github.com/graphql-python/graphene-django.git
synced 2024-11-23 01:57:08 +03:00
540 lines
15 KiB
Python
540 lines
15 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
try:
|
|
from urllib import urlencode
|
|
except ImportError:
|
|
from urllib.parse import urlencode
|
|
|
|
|
|
def url_string(string='/graphql', **url_params):
|
|
if url_params:
|
|
string += '?' + urlencode(url_params)
|
|
|
|
return string
|
|
|
|
|
|
def batch_url_string(**url_params):
|
|
return url_string('/graphql/batch', **url_params)
|
|
|
|
|
|
def response_json(response):
|
|
return json.loads(response.content.decode())
|
|
|
|
|
|
j = lambda **kwargs: json.dumps(kwargs)
|
|
jl = lambda **kwargs: json.dumps([kwargs])
|
|
|
|
|
|
def test_graphiql_is_enabled(client):
|
|
response = client.get(url_string(), HTTP_ACCEPT='text/html')
|
|
assert response.status_code == 200
|
|
assert response['Content-Type'].split(';')[0] == 'text/html'
|
|
|
|
def test_qfactor_graphiql(client):
|
|
response = client.get(url_string(query='{test}'), HTTP_ACCEPT='application/json;q=0.8, text/html;q=0.9')
|
|
assert response.status_code == 200
|
|
assert response['Content-Type'].split(';')[0] == 'text/html'
|
|
|
|
def test_qfactor_json(client):
|
|
response = client.get(url_string(query='{test}'), HTTP_ACCEPT='text/html;q=0.8, application/json;q=0.9')
|
|
assert response.status_code == 200
|
|
assert response['Content-Type'].split(';')[0] == 'application/json'
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello World"}
|
|
}
|
|
|
|
|
|
def test_allows_get_with_query_param(client):
|
|
response = client.get(url_string(query='{test}'))
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello World"}
|
|
}
|
|
|
|
|
|
def test_allows_get_with_variable_values(client):
|
|
response = client.get(url_string(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables=json.dumps({'who': "Dolly"})
|
|
))
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
def test_allows_get_with_operation_name(client):
|
|
response = client.get(url_string(
|
|
query='''
|
|
query helloYou { test(who: "You"), ...shared }
|
|
query helloWorld { test(who: "World"), ...shared }
|
|
query helloDolly { test(who: "Dolly"), ...shared }
|
|
fragment shared on QueryRoot {
|
|
shared: test(who: "Everyone")
|
|
}
|
|
''',
|
|
operationName='helloWorld'
|
|
))
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {
|
|
'test': 'Hello World',
|
|
'shared': 'Hello Everyone'
|
|
}
|
|
}
|
|
|
|
|
|
def test_reports_validation_errors(client):
|
|
response = client.get(url_string(
|
|
query='{ test, unknownOne, unknownTwo }'
|
|
))
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [
|
|
{
|
|
'message': 'Cannot query field "unknownOne" on type "QueryRoot".',
|
|
'locations': [{'line': 1, 'column': 9}]
|
|
},
|
|
{
|
|
'message': 'Cannot query field "unknownTwo" on type "QueryRoot".',
|
|
'locations': [{'line': 1, 'column': 21}]
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_errors_when_missing_operation_name(client):
|
|
response = client.get(url_string(
|
|
query='''
|
|
query TestQuery { test }
|
|
mutation TestMutation { writeTest { test } }
|
|
'''
|
|
))
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [
|
|
{
|
|
'message': 'Must provide operation name if query contains multiple operations.'
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_errors_when_sending_a_mutation_via_get(client):
|
|
response = client.get(url_string(
|
|
query='''
|
|
mutation TestMutation { writeTest { test } }
|
|
'''
|
|
))
|
|
assert response.status_code == 405
|
|
assert response_json(response) == {
|
|
'errors': [
|
|
{
|
|
'message': 'Can only perform a mutation operation from a POST request.'
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_errors_when_selecting_a_mutation_within_a_get(client):
|
|
response = client.get(url_string(
|
|
query='''
|
|
query TestQuery { test }
|
|
mutation TestMutation { writeTest { test } }
|
|
''',
|
|
operationName='TestMutation'
|
|
))
|
|
|
|
assert response.status_code == 405
|
|
assert response_json(response) == {
|
|
'errors': [
|
|
{
|
|
'message': 'Can only perform a mutation operation from a POST request.'
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_allows_mutation_to_exist_within_a_get(client):
|
|
response = client.get(url_string(
|
|
query='''
|
|
query TestQuery { test }
|
|
mutation TestMutation { writeTest { test } }
|
|
''',
|
|
operationName='TestQuery'
|
|
))
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello World"}
|
|
}
|
|
|
|
|
|
def test_allows_post_with_json_encoding(client):
|
|
response = client.post(url_string(), j(query='{test}'), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello World"}
|
|
}
|
|
|
|
|
|
def test_batch_allows_post_with_json_encoding(client):
|
|
response = client.post(batch_url_string(), jl(id=1, query='{test}'), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == [{
|
|
'id': 1,
|
|
'data': {'test': "Hello World"},
|
|
'status': 200,
|
|
}]
|
|
|
|
|
|
def test_batch_fails_if_is_empty(client):
|
|
response = client.post(batch_url_string(), '[]', 'application/json')
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'Received an empty list in the batch request.'}]
|
|
}
|
|
|
|
|
|
def test_allows_sending_a_mutation_via_post(client):
|
|
response = client.post(url_string(), j(query='mutation TestMutation { writeTest { test } }'), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'writeTest': {'test': 'Hello World'}}
|
|
}
|
|
|
|
|
|
def test_allows_post_with_url_encoding(client):
|
|
response = client.post(url_string(), urlencode(dict(query='{test}')), 'application/x-www-form-urlencoded')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello World"}
|
|
}
|
|
|
|
|
|
def test_supports_post_json_query_with_string_variables(client):
|
|
response = client.post(url_string(), j(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables=json.dumps({'who': "Dolly"})
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
|
|
def test_batch_supports_post_json_query_with_string_variables(client):
|
|
response = client.post(batch_url_string(), jl(
|
|
id=1,
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables=json.dumps({'who': "Dolly"})
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == [{
|
|
'id': 1,
|
|
'data': {'test': "Hello Dolly"},
|
|
'status': 200,
|
|
}]
|
|
|
|
|
|
def test_supports_post_json_query_with_json_variables(client):
|
|
response = client.post(url_string(), j(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables={'who': "Dolly"}
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
def test_batch_supports_post_json_query_with_json_variables(client):
|
|
response = client.post(batch_url_string(), jl(
|
|
id=1,
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables={'who': "Dolly"}
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == [{
|
|
'id': 1,
|
|
'data': {'test': "Hello Dolly"},
|
|
'status': 200,
|
|
}]
|
|
|
|
|
|
def test_supports_post_url_encoded_query_with_string_variables(client):
|
|
response = client.post(url_string(), urlencode(dict(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables=json.dumps({'who': "Dolly"})
|
|
)), 'application/x-www-form-urlencoded')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
def test_supports_post_json_quey_with_get_variable_values(client):
|
|
response = client.post(url_string(
|
|
variables=json.dumps({'who': "Dolly"})
|
|
), j(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
def test_post_url_encoded_query_with_get_variable_values(client):
|
|
response = client.post(url_string(
|
|
variables=json.dumps({'who': "Dolly"})
|
|
), urlencode(dict(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
)), 'application/x-www-form-urlencoded')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
def test_supports_post_raw_text_query_with_get_variable_values(client):
|
|
response = client.post(url_string(
|
|
variables=json.dumps({'who': "Dolly"})
|
|
),
|
|
'query helloWho($who: String){ test(who: $who) }',
|
|
'application/graphql'
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {'test': "Hello Dolly"}
|
|
}
|
|
|
|
|
|
def test_allows_post_with_operation_name(client):
|
|
response = client.post(url_string(), j(
|
|
query='''
|
|
query helloYou { test(who: "You"), ...shared }
|
|
query helloWorld { test(who: "World"), ...shared }
|
|
query helloDolly { test(who: "Dolly"), ...shared }
|
|
fragment shared on QueryRoot {
|
|
shared: test(who: "Everyone")
|
|
}
|
|
''',
|
|
operationName='helloWorld'
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {
|
|
'test': 'Hello World',
|
|
'shared': 'Hello Everyone'
|
|
}
|
|
}
|
|
|
|
|
|
def test_batch_allows_post_with_operation_name(client):
|
|
response = client.post(batch_url_string(), jl(
|
|
id=1,
|
|
query='''
|
|
query helloYou { test(who: "You"), ...shared }
|
|
query helloWorld { test(who: "World"), ...shared }
|
|
query helloDolly { test(who: "Dolly"), ...shared }
|
|
fragment shared on QueryRoot {
|
|
shared: test(who: "Everyone")
|
|
}
|
|
''',
|
|
operationName='helloWorld'
|
|
), 'application/json')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == [{
|
|
'id': 1,
|
|
'data': {
|
|
'test': 'Hello World',
|
|
'shared': 'Hello Everyone'
|
|
},
|
|
'status': 200,
|
|
}]
|
|
|
|
|
|
def test_allows_post_with_get_operation_name(client):
|
|
response = client.post(url_string(
|
|
operationName='helloWorld'
|
|
), '''
|
|
query helloYou { test(who: "You"), ...shared }
|
|
query helloWorld { test(who: "World"), ...shared }
|
|
query helloDolly { test(who: "Dolly"), ...shared }
|
|
fragment shared on QueryRoot {
|
|
shared: test(who: "Everyone")
|
|
}
|
|
''',
|
|
'application/graphql')
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {
|
|
'test': 'Hello World',
|
|
'shared': 'Hello Everyone'
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.urls('graphene_django.tests.urls_pretty')
|
|
def test_supports_pretty_printing(client):
|
|
response = client.get(url_string(query='{test}'))
|
|
|
|
assert response.content.decode() == (
|
|
'{\n'
|
|
' "data": {\n'
|
|
' "test": "Hello World"\n'
|
|
' }\n'
|
|
'}'
|
|
)
|
|
|
|
|
|
def test_supports_pretty_printing_by_request(client):
|
|
response = client.get(url_string(query='{test}', pretty='1'))
|
|
|
|
assert response.content.decode() == (
|
|
'{\n'
|
|
' "data": {\n'
|
|
' "test": "Hello World"\n'
|
|
' }\n'
|
|
'}'
|
|
)
|
|
|
|
|
|
def test_handles_field_errors_caught_by_graphql(client):
|
|
response = client.get(url_string(query='{thrower}'))
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': None,
|
|
'errors': [{'locations': [{'column': 2, 'line': 1}], 'message': 'Throws!'}]
|
|
}
|
|
|
|
|
|
def test_handles_syntax_errors_caught_by_graphql(client):
|
|
response = client.get(url_string(query='syntaxerror'))
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'locations': [{'column': 1, 'line': 1}],
|
|
'message': 'Syntax Error GraphQL request (1:1) '
|
|
'Unexpected Name "syntaxerror"\n\n1: syntaxerror\n ^\n'}]
|
|
}
|
|
|
|
|
|
def test_handles_errors_caused_by_a_lack_of_query(client):
|
|
response = client.get(url_string())
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'Must provide query string.'}]
|
|
}
|
|
|
|
|
|
def test_handles_not_expected_json_bodies(client):
|
|
response = client.post(url_string(), '[]', 'application/json')
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'The received data is not a valid JSON query.'}]
|
|
}
|
|
|
|
|
|
def test_handles_invalid_json_bodies(client):
|
|
response = client.post(url_string(), '[oh}', 'application/json')
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'POST body sent invalid JSON.'}]
|
|
}
|
|
|
|
|
|
def test_handles_django_request_error(client, monkeypatch):
|
|
def mocked_read(*args):
|
|
raise IOError("foo-bar")
|
|
|
|
monkeypatch.setattr("django.http.request.HttpRequest.read", mocked_read)
|
|
|
|
valid_json = json.dumps(dict(foo='bar'))
|
|
response = client.post(url_string(), valid_json, 'application/json')
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'foo-bar'}]
|
|
}
|
|
|
|
|
|
def test_handles_incomplete_json_bodies(client):
|
|
response = client.post(url_string(), '{"query":', 'application/json')
|
|
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'POST body sent invalid JSON.'}]
|
|
}
|
|
|
|
|
|
def test_handles_plain_post_text(client):
|
|
response = client.post(url_string(
|
|
variables=json.dumps({'who': "Dolly"})
|
|
),
|
|
'query helloWho($who: String){ test(who: $who) }',
|
|
'text/plain'
|
|
)
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'Must provide query string.'}]
|
|
}
|
|
|
|
|
|
def test_handles_poorly_formed_variables(client):
|
|
response = client.get(url_string(
|
|
query='query helloWho($who: String){ test(who: $who) }',
|
|
variables='who:You'
|
|
))
|
|
assert response.status_code == 400
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'Variables are invalid JSON.'}]
|
|
}
|
|
|
|
|
|
def test_handles_unsupported_http_methods(client):
|
|
response = client.put(url_string(query='{test}'))
|
|
assert response.status_code == 405
|
|
assert response['Allow'] == 'GET, POST'
|
|
assert response_json(response) == {
|
|
'errors': [{'message': 'GraphQL only supports GET and POST requests.'}]
|
|
}
|
|
|
|
|
|
def test_passes_request_into_context_request(client):
|
|
response = client.get(url_string(query='{request}', q='testing'))
|
|
|
|
assert response.status_code == 200
|
|
assert response_json(response) == {
|
|
'data': {
|
|
'request': 'testing'
|
|
}
|
|
}
|