test_views.py

import json
from unittest.mock import patch

import pytest
from django.db import connection

from .models import Pet

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())


def j(**kwargs):
    return json.dumps(kwargs)


def jl(**kwargs):
    return 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({"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(
            {
                "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({"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_inherited")
def test_inherited_class_with_attributes_works(client):
    inherited_url = "/graphql/inherited/"
    # Check schema and pretty attributes work
    response = client.post(url_string(inherited_url, query="{test}"))
    assert response.content.decode() == (
        "{\n" '  "data": {\n' '    "test": "Hello World"\n' "  }\n" "}"
    )

    # Check graphiql works
    response = client.get(url_string(inherited_url), HTTP_ACCEPT="text/html")
    assert response.status_code == 200


@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}],
                "path": ["thrower"],
                "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: Unexpected Name 'syntaxerror'.",
            }
        ]
    }


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 OSError("foo-bar")

    monkeypatch.setattr("django.http.request.HttpRequest.read", mocked_read)

    valid_json = json.dumps({"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"}}


@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": True}
)
def test_form_mutation_multiple_creation_invalid_atomic_request(client):
    query = """
    mutation PetMutations {
        petFormMutation1: petFormMutation(input: { name: "Mia", age: 99 }) {
            errors {
                field
                messages
            }
        }
        petFormMutation2: petFormMutation(input: { name: "Enzo", age: 0 }) {
            errors {
                field
                messages
            }
        }
    }
    """

    response = client.post(url_string(query=query))
    content = response_json(response)

    assert "errors" not in content

    assert content["data"]["petFormMutation1"]["errors"] == [
        {"field": "age", "messages": ["Too old"]}
    ]

    assert content["data"]["petFormMutation2"]["errors"] == []

    assert Pet.objects.count() == 0


@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": True, "ATOMIC_REQUESTS": False}
)
def test_form_mutation_multiple_creation_invalid_atomic_mutation_1(client):
    query = """
    mutation PetMutations {
        petFormMutation1: petFormMutation(input: { name: "Mia", age: 99 }) {
            errors {
                field
                messages
            }
        }
        petFormMutation2: petFormMutation(input: { name: "Enzo", age: 0 }) {
            errors {
                field
                messages
            }
        }
    }
    """

    response = client.post(url_string(query=query))
    content = response_json(response)

    assert "errors" not in content

    assert content["data"]["petFormMutation1"]["errors"] == [
        {"field": "age", "messages": ["Too old"]}
    ]

    assert content["data"]["petFormMutation2"]["errors"] == []

    assert Pet.objects.count() == 0


@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", True)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": False}
)
def test_form_mutation_multiple_creation_invalid_atomic_mutation_2(client):
    query = """
    mutation PetMutations {
        petFormMutation1: petFormMutation(input: { name: "Mia", age: 99 }) {
            errors {
                field
                messages
            }
        }
        petFormMutation2: petFormMutation(input: { name: "Enzo", age: 0 }) {
            errors {
                field
                messages
            }
        }
    }
    """

    response = client.post(url_string(query=query))
    content = response_json(response)

    assert "errors" not in content

    assert content["data"]["petFormMutation1"]["errors"] == [
        {"field": "age", "messages": ["Too old"]}
    ]

    assert content["data"]["petFormMutation2"]["errors"] == []

    assert Pet.objects.count() == 0


@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": False}
)
def test_form_mutation_multiple_creation_invalid_non_atomic(client):
    query = """
    mutation PetMutations {
        petFormMutation1: petFormMutation(input: { name: "Mia", age: 99 }) {
            errors {
                field
                messages
            }
        }
        petFormMutation2: petFormMutation(input: { name: "Enzo", age: 0 }) {
            errors {
                field
                messages
            }
        }
    }
    """

    response = client.post(url_string(query=query))
    content = response_json(response)

    assert "errors" not in content

    assert content["data"]["petFormMutation1"]["errors"] == [
        {"field": "age", "messages": ["Too old"]}
    ]

    assert content["data"]["petFormMutation2"]["errors"] == []

    assert Pet.objects.count() == 1

    pet = Pet.objects.get()
    assert pet.name == "Enzo"
    assert pet.age == 0


@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": True}
)
def test_model_form_mutation_multiple_creation_invalid_atomic_request(client):
    query = """
    mutation PetMutations {
        petMutation1: petMutation(input: { name: "Mia", age: 99 }) {
            pet {
                name
                age
            }
            errors {
                field
                messages
            }
        }
        petMutation2: petMutation(input: { name: "Enzo", age: 0 }) {
            pet {
                name
                age
            }
            errors {
                field
                messages
            }
        }
    }
    """

    response = client.post(url_string(query=query))
    content = response_json(response)

    assert "errors" not in content

    assert content["data"]["petMutation1"]["pet"] is None
    assert content["data"]["petMutation1"]["errors"] == [
        {"field": "age", "messages": ["Too old"]}
    ]

    assert content["data"]["petMutation2"]["pet"] == {"name": "Enzo", "age": 0}

    assert Pet.objects.count() == 0


@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": False}
)
def test_model_form_mutation_multiple_creation_invalid_non_atomic(client):
    query = """
    mutation PetMutations {
        petMutation1: petMutation(input: { name: "Mia", age: 99 }) {
            pet {
                name
                age
            }
            errors {
                field
                messages
            }
        }
        petMutation2: petMutation(input: { name: "Enzo", age: 0 }) {
            pet {
                name
                age
            }
            errors {
                field
                messages
            }
        }
    }
    """

    response = client.post(url_string(query=query))
    content = response_json(response)

    assert "errors" not in content

    assert content["data"]["petMutation1"]["pet"] is None
    assert content["data"]["petMutation1"]["errors"] == [
        {"field": "age", "messages": ["Too old"]}
    ]

    assert content["data"]["petMutation2"]["pet"] == {"name": "Enzo", "age": 0}

    assert Pet.objects.count() == 1

    pet = Pet.objects.get()
    assert pet.name == "Enzo"
    assert pet.age == 0


@patch("graphene_django.utils.utils.transaction.set_rollback")
@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": True}
)
def test_query_errors_atomic_request(set_rollback_mock, client):
    client.get(url_string(query="force error"))
    set_rollback_mock.assert_called_once_with(True)


@patch("graphene_django.utils.utils.transaction.set_rollback")
@patch("graphene_django.settings.graphene_settings.ATOMIC_MUTATIONS", False)
@patch.dict(
    connection.settings_dict, {"ATOMIC_MUTATIONS": False, "ATOMIC_REQUESTS": False}
)
def test_query_errors_non_atomic(set_rollback_mock, client):
    client.get(url_string(query="force error"))
    set_rollback_mock.assert_not_called()