graphene-django/graphene_django/tests/test_views.py

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'
}
}