import asyncio
from typing import Any, Awaitable, Optional, cast

from graphql.error import GraphQLError
from graphql.execution import execute, execute_sync
from graphql.language import FieldNode, OperationDefinitionNode, parse
from graphql.pyutils import Undefined, inspect
from graphql.type import (
    GraphQLArgument,
    GraphQLBoolean,
    GraphQLField,
    GraphQLInt,
    GraphQLInterfaceType,
    GraphQLList,
    GraphQLNonNull,
    GraphQLObjectType,
    GraphQLResolveInfo,
    GraphQLScalarType,
    GraphQLSchema,
    GraphQLString,
    GraphQLUnionType,
    ResponsePath,
)
from pytest import mark, raises


def describe_execute_handles_basic_execution_tasks():
    # noinspection PyTypeChecker
    def throws_if_no_document_is_provided():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        with raises(TypeError) as exc_info:
            assert execute_sync(schema=schema, document=None)  # type: ignore

        assert str(exc_info.value) == "Must provide document."

    # noinspection PyTypeChecker
    def throws_if_no_schema_is_provided():
        document = parse("{ field }")

        with raises(TypeError) as exc_info:
            assert execute_sync(schema=None, document=document)  # type: ignore

        assert str(exc_info.value) == "Expected None to be a GraphQL schema."

    def throws_on_invalid_variables():
        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type",
                {
                    "fieldA": GraphQLField(
                        GraphQLString, args={"argA": GraphQLArgument(GraphQLInt)}
                    )
                },
            )
        )
        document = parse(
            """
            query ($a: Int) {
              fieldA(argA: $a)
            }
            """
        )
        variable_values = "{'a': 1}"

        with raises(TypeError) as exc_info:
            assert execute_sync(
                schema=schema,
                document=document,
                variable_values=variable_values,  # type: ignore
            )

        assert str(exc_info.value) == (
            "Variable values must be provided as a dictionary"
            " with variable names as keys. Perhaps look to see"
            " if an unparsed JSON string was provided."
        )

    def accepts_positional_arguments():
        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type",
                {"a": GraphQLField(GraphQLString, resolve=lambda obj, *args: obj)},
            )
        )

        result = execute_sync(schema, parse("{ a }"), "rootValue")

        assert result == ({"a": "rootValue"}, None)

    @mark.asyncio
    async def executes_arbitrary_code():
        # noinspection PyMethodMayBeStatic,PyMethodMayBeStatic
        class Data:
            def a(self, _info):
                return "Apple"

            def b(self, _info):
                return "Banana"

            def c(self, _info):
                return "Cookie"

            def d(self, _info):
                return "Donut"

            def e(self, _info):
                return "Egg"

            f = "Fish"

            # Called only by DataType::pic static resolver
            def pic(self, _info, size: int):
                return f"Pic of size: {size}"

            def deep(self, _info):
                return DeepData()

            def promise(self, _info):
                return promise_data()

        # noinspection PyMethodMayBeStatic,PyMethodMayBeStatic
        class DeepData:
            def a(self, _info):
                return "Already Been Done"

            def b(self, _info):
                return "Boring"

            def c(self, _info):
                return ["Contrived", None, "Confusing"]

            def deeper(self, _info):
                return [Data(), None, Data()]

        async def promise_data():
            await asyncio.sleep(0)
            return Data()

        DeepDataType: GraphQLObjectType

        DataType = GraphQLObjectType(
            "DataType",
            lambda: {
                "a": GraphQLField(GraphQLString),
                "b": GraphQLField(GraphQLString),
                "c": GraphQLField(GraphQLString),
                "d": GraphQLField(GraphQLString),
                "e": GraphQLField(GraphQLString),
                "f": GraphQLField(GraphQLString),
                "pic": GraphQLField(
                    GraphQLString,
                    args={"size": GraphQLArgument(GraphQLInt)},
                    resolve=lambda obj, info, size: obj.pic(info, size),
                ),
                "deep": GraphQLField(DeepDataType),
                "promise": GraphQLField(DataType),
            },
        )

        DeepDataType = GraphQLObjectType(
            "DeepDataType",
            {
                "a": GraphQLField(GraphQLString),
                "b": GraphQLField(GraphQLString),
                "c": GraphQLField(GraphQLList(GraphQLString)),
                "deeper": GraphQLField(GraphQLList(DataType)),
            },
        )

        document = parse(
            """
            query ($size: Int) {
              a,
              b,
              x: c
              ...c
              f
              ...on DataType {
                pic(size: $size)
                promise {
                  a
                }
              }
              deep {
                a
                b
                c
                deeper {
                  a
                  b
                }
              }
            }

            fragment c on DataType {
              d
              e
            }
            """
        )

        awaitable_result = execute(
            GraphQLSchema(DataType), document, Data(), variable_values={"size": 100}
        )
        assert isinstance(awaitable_result, Awaitable)
        result = await awaitable_result

        assert result == (
            {
                "a": "Apple",
                "b": "Banana",
                "x": "Cookie",
                "d": "Donut",
                "e": "Egg",
                "f": "Fish",
                "pic": "Pic of size: 100",
                "promise": {"a": "Apple"},
                "deep": {
                    "a": "Already Been Done",
                    "b": "Boring",
                    "c": ["Contrived", None, "Confusing"],
                    "deeper": [
                        {"a": "Apple", "b": "Banana"},
                        None,
                        {"a": "Apple", "b": "Banana"},
                    ],
                },
            },
            None,
        )

    def merges_parallel_fragments():
        Type = GraphQLObjectType(
            "Type",
            lambda: {
                "a": GraphQLField(GraphQLString, resolve=lambda *_args: "Apple"),
                "b": GraphQLField(GraphQLString, resolve=lambda *_args: "Banana"),
                "c": GraphQLField(GraphQLString, resolve=lambda *_args: "Cherry"),
                "deep": GraphQLField(Type, resolve=lambda *_args: {}),
            },
        )
        schema = GraphQLSchema(Type)

        ast = parse(
            """
            { a, ...FragOne, ...FragTwo }

            fragment FragOne on Type {
              b
              deep { b, deeper: deep { b } }
            }

            fragment FragTwo on Type {
              c
              deep { c, deeper: deep { c } }
            }
            """
        )

        result = execute_sync(schema, ast)
        assert result == (
            {
                "a": "Apple",
                "b": "Banana",
                "c": "Cherry",
                "deep": {
                    "b": "Banana",
                    "c": "Cherry",
                    "deeper": {"b": "Banana", "c": "Cherry"},
                },
            },
            None,
        )

    def provides_info_about_current_execution_state():
        resolved_infos = []

        def resolve(_obj, info):
            resolved_infos.append(info)

        test_type = GraphQLObjectType(
            "Test", {"test": GraphQLField(GraphQLString, resolve=resolve)}
        )

        schema = GraphQLSchema(test_type)

        document = parse("query ($var: String) { result: test }")
        root_value = {"root": "val"}
        variable_values = {"var": "abc"}
        execute_sync(schema, document, root_value, variable_values=variable_values)

        assert len(resolved_infos) == 1
        operation = cast(OperationDefinitionNode, document.definitions[0])
        assert operation and operation.kind == "operation_definition"
        field = cast(FieldNode, operation.selection_set.selections[0])

        assert resolved_infos[0] == GraphQLResolveInfo(
            field_name="test",
            field_nodes=[field],
            return_type=GraphQLString,
            parent_type=cast(GraphQLObjectType, schema.query_type),
            path=ResponsePath(None, "result", "Test"),
            schema=schema,
            fragments={},
            root_value=root_value,
            operation=operation,
            variable_values=variable_values,
            context=None,
            is_awaitable=resolved_infos[0].is_awaitable,
        )

    def it_populates_path_correctly_with_complex_types():
        path: Optional[ResponsePath] = None

        def resolve(_val, info):
            nonlocal path
            path = info.path

        def resolve_type(_val, _info, _type):
            return "SomeObject"

        some_object = GraphQLObjectType(
            "SomeObject", {"test": GraphQLField(GraphQLString, resolve=resolve)}
        )
        some_union = GraphQLUnionType(
            "SomeUnion", [some_object], resolve_type=resolve_type
        )
        test_type = GraphQLObjectType(
            "SomeQuery",
            {
                "test": GraphQLField(
                    GraphQLNonNull(GraphQLList(GraphQLNonNull(some_union)))
                )
            },
        )
        schema = GraphQLSchema(test_type)
        root_value: Any = {"test": [{}]}
        document = parse(
            """
            query {
              l1: test {
                ... on SomeObject {
                  l2: test
                }
              }
            }
            """
        )

        execute_sync(schema, document, root_value)

        assert path is not None
        prev, key, typename = path
        assert key == "l2"
        assert typename == "SomeObject"
        prev, key, typename = prev
        assert key == 0
        assert typename is None
        prev, key, typename = prev
        assert key == "l1"
        assert typename == "SomeQuery"
        assert prev is None

    def threads_root_value_context_correctly():
        resolved_values = []

        class Data:
            context_thing = "thing"

        def resolve(obj, _info):
            resolved_values.append(obj)

        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type", {"a": GraphQLField(GraphQLString, resolve=resolve)}
            )
        )

        document = parse("query Example { a }")
        root_value = Data()
        execute_sync(schema, document, root_value)

        assert len(resolved_values) == 1
        assert resolved_values[0] is root_value

    def correctly_threads_arguments():
        resolved_args = []

        def resolve(_obj, _info, **args):
            resolved_args.append(args)

        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type",
                {
                    "b": GraphQLField(
                        GraphQLString,
                        args={
                            "numArg": GraphQLArgument(GraphQLInt),
                            "stringArg": GraphQLArgument(GraphQLString),
                        },
                        resolve=resolve,
                    )
                },
            )
        )

        document = parse(
            """
            query Example {
              b(numArg: 123, stringArg: "foo")
            }
            """
        )

        execute_sync(schema, document)

        assert len(resolved_args) == 1
        assert resolved_args[0] == {"numArg": 123, "stringArg": "foo"}

    @mark.asyncio
    async def nulls_out_error_subtrees():
        document = parse(
            """
            {
              syncOk
              syncError
              syncRawError
              syncReturnError
              syncReturnErrorList
              asyncOk
              asyncError
              asyncRawError
              asyncReturnError
              asyncReturnErrorWithExtensions
            }
            """
        )

        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type",
                {
                    "syncOk": GraphQLField(GraphQLString),
                    "syncError": GraphQLField(GraphQLString),
                    "syncRawError": GraphQLField(GraphQLString),
                    "syncReturnError": GraphQLField(GraphQLString),
                    "syncReturnErrorList": GraphQLField(GraphQLList(GraphQLString)),
                    "asyncOk": GraphQLField(GraphQLString),
                    "asyncError": GraphQLField(GraphQLString),
                    "asyncErrorWithExtensions": GraphQLField(GraphQLString),
                    "asyncRawError": GraphQLField(GraphQLString),
                    "asyncReturnError": GraphQLField(GraphQLString),
                    "asyncReturnErrorWithExtensions": GraphQLField(GraphQLString),
                },
            )
        )

        # noinspection PyPep8Naming,PyMethodMayBeStatic
        class Data:
            def syncOk(self, _info):
                return "sync ok"

            def syncError(self, _info):
                raise GraphQLError("Error getting syncError")

            def syncRawError(self, _info):
                raise Exception("Error getting syncRawError")

            def syncReturnError(self, _info):
                return Exception("Error getting syncReturnError")

            def syncReturnErrorList(self, _info):
                return [
                    "sync0",
                    Exception("Error getting syncReturnErrorList1"),
                    "sync2",
                    Exception("Error getting syncReturnErrorList3"),
                ]

            async def asyncOk(self, _info):
                return "async ok"

            async def asyncError(self, _info):
                raise GraphQLError("Error getting asyncError")

            async def asyncRawError(self, _info):
                raise Exception("Error getting asyncRawError")

            async def asyncReturnError(self, _info):
                return GraphQLError("Error getting asyncReturnError")

            async def asyncReturnErrorWithExtensions(self, _info):
                return GraphQLError(
                    "Error getting asyncReturnErrorWithExtensions",
                    extensions={"foo": "bar"},
                )

        awaitable_result = execute(schema, document, Data())
        assert isinstance(awaitable_result, Awaitable)
        result = await awaitable_result

        assert result == (
            {
                "syncOk": "sync ok",
                "syncError": None,
                "syncRawError": None,
                "syncReturnError": None,
                "syncReturnErrorList": ["sync0", None, "sync2", None],
                "asyncOk": "async ok",
                "asyncError": None,
                "asyncRawError": None,
                "asyncReturnError": None,
                "asyncReturnErrorWithExtensions": None,
            },
            [
                {
                    "message": "Error getting syncError",
                    "locations": [(4, 15)],
                    "path": ["syncError"],
                },
                {
                    "message": "Error getting syncRawError",
                    "locations": [(5, 15)],
                    "path": ["syncRawError"],
                },
                {
                    "message": "Error getting syncReturnError",
                    "locations": [(6, 15)],
                    "path": ["syncReturnError"],
                },
                {
                    "message": "Error getting syncReturnErrorList1",
                    "locations": [(7, 15)],
                    "path": ["syncReturnErrorList", 1],
                },
                {
                    "message": "Error getting syncReturnErrorList3",
                    "locations": [(7, 15)],
                    "path": ["syncReturnErrorList", 3],
                },
                {
                    "message": "Error getting asyncError",
                    "locations": [(9, 15)],
                    "path": ["asyncError"],
                },
                {
                    "message": "Error getting asyncRawError",
                    "locations": [(10, 15)],
                    "path": ["asyncRawError"],
                },
                {
                    "message": "Error getting asyncReturnError",
                    "locations": [(11, 15)],
                    "path": ["asyncReturnError"],
                },
                {
                    "message": "Error getting asyncReturnErrorWithExtensions",
                    "locations": [(12, 15)],
                    "path": ["asyncReturnErrorWithExtensions"],
                    "extensions": {"foo": "bar"},
                },
            ],
        )

    @mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
    def handles_sync_errors_combined_with_async_ones():
        is_async_resolver_finished = False

        async def async_resolver(_obj, _info):
            nonlocal is_async_resolver_finished
            is_async_resolver_finished = True  # pragma: no cover

        schema = GraphQLSchema(
            GraphQLObjectType(
                "Query",
                {
                    "syncNullError": GraphQLField(
                        GraphQLNonNull(GraphQLString), resolve=lambda _obj, _info: None
                    ),
                    "asyncNullError": GraphQLField(
                        GraphQLNonNull(GraphQLString), resolve=async_resolver
                    ),
                },
            )
        )

        document = parse(
            """
            {
              asyncNullError
              syncNullError
            }
            """
        )

        result = execute(schema, document)

        assert is_async_resolver_finished is False

        assert result == (
            None,
            [
                {
                    "message": "Cannot return null"
                    " for non-nullable field Query.syncNullError.",
                    "locations": [(4, 15)],
                    "path": ["syncNullError"],
                }
            ],
        )

    @mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
    def full_response_path_is_included_for_non_nullable_fields():
        def resolve_ok(*_args):
            return {}

        def resolve_error(*_args):
            raise Exception("Catch me if you can")

        A = GraphQLObjectType(
            "A",
            lambda: {
                "nullableA": GraphQLField(A, resolve=resolve_ok),
                "nonNullA": GraphQLField(GraphQLNonNull(A), resolve=resolve_ok),
                "throws": GraphQLField(GraphQLNonNull(A), resolve=resolve_error),
            },
        )

        schema = GraphQLSchema(
            GraphQLObjectType(
                "query", lambda: {"nullableA": GraphQLField(A, resolve=resolve_ok)}
            )
        )

        document = parse(
            """
            query {
              nullableA {
                aliasedA: nullableA {
                  nonNullA {
                    anotherA: nonNullA {
                      throws
                    }
                  }
                }
              }
            }
            """
        )

        assert execute_sync(schema, document) == (
            {"nullableA": {"aliasedA": None}},
            [
                {
                    "message": "Catch me if you can",
                    "locations": [(7, 23)],
                    "path": ["nullableA", "aliasedA", "nonNullA", "anotherA", "throws"],
                }
            ],
        )

    @mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
    def uses_the_inline_operation_if_no_operation_name_is_provided():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse("{ a }")

        class Data:
            a = "b"

        result = execute_sync(schema, document, Data())
        assert result == ({"a": "b"}, None)

    @mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
    def uses_the_only_operation_if_no_operation_name_is_provided():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse("query Example { a }")

        class Data:
            a = "b"

        result = execute_sync(schema, document, Data())
        assert result == ({"a": "b"}, None)

    @mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
    def uses_the_named_operation_if_operation_name_is_provided():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse(
            """
            query Example { first: a }
            query OtherExample { second: a }
            """
        )

        class Data:
            a = "b"

        result = execute_sync(schema, document, Data(), operation_name="OtherExample")
        assert result == ({"second": "b"}, None)

    def provides_error_if_no_operation_is_provided():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse("fragment Example on Type { a }")

        class Data:
            a = "b"

        result = execute_sync(schema, document, Data())
        assert result == (None, [{"message": "Must provide an operation."}])

    def errors_if_no_operation_name_is_provided_with_multiple_operations():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse(
            """
            query Example { a }
            query OtherExample { a }
            """
        )

        result = execute_sync(schema, document)
        assert result == (
            None,
            [
                {
                    "message": "Must provide operation name if query contains"
                    " multiple operations."
                }
            ],
        )

    def errors_if_unknown_operation_name_is_provided():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse(
            """
            query Example { a }
            query OtherExample { a }
            """
        )

        result = execute_sync(schema, document, operation_name="UnknownExample")
        assert result == (
            None,
            [{"message": "Unknown operation named 'UnknownExample'."}],
        )

    def errors_if_empty_string_is_provided_as_operation_name():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse("{ a }")

        result = execute_sync(schema, document, operation_name="")
        assert result == (
            None,
            [{"message": "Unknown operation named ''."}],
        )

    def uses_the_query_schema_for_queries():
        schema = GraphQLSchema(
            GraphQLObjectType("Q", {"a": GraphQLField(GraphQLString)}),
            GraphQLObjectType("M", {"c": GraphQLField(GraphQLString)}),
            GraphQLObjectType("S", {"a": GraphQLField(GraphQLString)}),
        )

        document = parse(
            """
            query Q { a }
            mutation M { c }
            subscription S { a }
            """
        )

        class Data:
            a = "b"
            c = "d"

        result = execute_sync(schema, document, Data(), operation_name="Q")
        assert result == ({"a": "b"}, None)

    def uses_the_mutation_schema_for_mutations():
        schema = GraphQLSchema(
            GraphQLObjectType("Q", {"a": GraphQLField(GraphQLString)}),
            GraphQLObjectType("M", {"c": GraphQLField(GraphQLString)}),
        )

        document = parse(
            """
            query Q { a }
            mutation M { c }
            """
        )

        class Data:
            a = "b"
            c = "d"

        result = execute_sync(schema, document, Data(), operation_name="M")
        assert result == ({"c": "d"}, None)

    def uses_the_subscription_schema_for_subscriptions():
        schema = GraphQLSchema(
            query=GraphQLObjectType("Q", {"a": GraphQLField(GraphQLString)}),
            subscription=GraphQLObjectType("S", {"a": GraphQLField(GraphQLString)}),
        )

        document = parse(
            """
            query Q { a }
            subscription S { a }
            """
        )

        class Data:
            a = "b"
            c = "d"

        result = execute_sync(schema, document, Data(), operation_name="S")
        assert result == ({"a": "b"}, None)

    def resolves_to_an_error_if_schema_does_not_support_operation():
        schema = GraphQLSchema(assume_valid=True)

        document = parse(
            """
            query Q { __typename }
            mutation M { __typename }
            subscription S { __typename }
            """
        )

        assert execute_sync(schema, document, operation_name="Q") == (
            None,
            [
                {
                    "message": "Schema is not configured to execute query operation.",
                    "locations": [(2, 13)],
                }
            ],
        )

        assert execute_sync(schema, document, operation_name="M") == (
            None,
            [
                {
                    "message": "Schema is not configured to execute"
                    " mutation operation.",
                    "locations": [(3, 13)],
                }
            ],
        )

        assert execute_sync(schema, document, operation_name="S") == (
            None,
            [
                {
                    "message": "Schema is not configured to execute"
                    " subscription operation.",
                    "locations": [(4, 13)],
                }
            ],
        )

    @mark.asyncio
    async def correct_field_ordering_despite_execution_order():
        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type",
                {
                    "a": GraphQLField(GraphQLString),
                    "b": GraphQLField(GraphQLString),
                    "c": GraphQLField(GraphQLString),
                    "d": GraphQLField(GraphQLString),
                    "e": GraphQLField(GraphQLString),
                },
            )
        )

        document = parse("{ a, b, c, d, e}")

        # noinspection PyMethodMayBeStatic,PyMethodMayBeStatic
        class Data:
            def a(self, _info):
                return "a"

            async def b(self, _info):
                return "b"

            def c(self, _info):
                return "c"

            async def d(self, _info):
                return "d"

            def e(self, _info):
                return "e"

        awaitable_result = execute(schema, document, Data())
        assert isinstance(awaitable_result, Awaitable)
        result = await awaitable_result

        assert result == ({"a": "a", "b": "b", "c": "c", "d": "d", "e": "e"}, None)

    def avoids_recursion():
        schema = GraphQLSchema(
            GraphQLObjectType("Type", {"a": GraphQLField(GraphQLString)})
        )

        document = parse(
            """
            query Q {
              a
              ...Frag
              ...Frag
            }

            fragment Frag on Type {
              a,
              ...Frag
            }
            """
        )

        class Data:
            a = "b"

        result = execute_sync(schema, document, Data(), operation_name="Q")

        assert result == ({"a": "b"}, None)

    def ignores_missing_sub_selections_on_fields():
        some_type = GraphQLObjectType("SomeType", {"b": GraphQLField(GraphQLString)})
        schema = GraphQLSchema(
            GraphQLObjectType("Query", {"a": GraphQLField(some_type)})
        )
        document = parse("{ a }")
        root_value = {"a": {"b": "c"}}

        result = execute_sync(schema, document, root_value)
        assert result == ({"a": {}}, None)

    def does_not_include_illegal_fields_in_output():
        schema = GraphQLSchema(
            GraphQLObjectType("Q", {"a": GraphQLField(GraphQLString)})
        )

        document = parse("{ thisIsIllegalDoNotIncludeMe }")

        result = execute_sync(schema, document)

        assert result == ({}, None)

    def does_not_include_arguments_that_were_not_set():
        schema = GraphQLSchema(
            GraphQLObjectType(
                "Type",
                {
                    "field": GraphQLField(
                        GraphQLString,
                        args={
                            "a": GraphQLArgument(GraphQLBoolean),
                            "b": GraphQLArgument(GraphQLBoolean),
                            "c": GraphQLArgument(GraphQLBoolean),
                            "d": GraphQLArgument(GraphQLInt),
                            "e": GraphQLArgument(GraphQLInt),
                        },
                        resolve=lambda _source, _info, **args: inspect(args),
                    )
                },
            )
        )

        document = parse("{ field(a: true, c: false, e: 0) }")

        assert execute_sync(schema, document) == (
            {"field": "{'a': True, 'c': False, 'e': 0}"},
            None,
        )

    @mark.asyncio
    async def fails_when_is_type_of_check_is_not_met():
        class Special:
            value: str

            def __init__(self, value):
                self.value = value

        class NotSpecial:
            value: str

            def __init__(self, value):
                self.value = value

        def is_type_of_special(obj, _info):
            is_special = isinstance(obj, Special)
            if not _info.context["async"]:
                return is_special

            async def async_is_special():
                return is_special

            return async_is_special()

        SpecialType = GraphQLObjectType(
            "SpecialType",
            {"value": GraphQLField(GraphQLString)},
            is_type_of=is_type_of_special,
        )

        schema = GraphQLSchema(
            GraphQLObjectType(
                "Query", {"specials": GraphQLField(GraphQLList(SpecialType))}
            )
        )

        document = parse("{ specials { value } }")
        root_value = {"specials": [Special("foo"), NotSpecial("bar")]}

        result = execute_sync(schema, document, root_value, {"async": False})
        assert not isinstance(result, Awaitable)
        assert result == (
            {"specials": [{"value": "foo"}, None]},
            [
                {
                    "message": "Expected value of type 'SpecialType' but got:"
                    " <NotSpecial instance>.",
                    "locations": [(1, 3)],
                    "path": ["specials", 1],
                }
            ],
        )

        async_result = execute(schema, document, root_value, {"async": True})
        assert isinstance(async_result, Awaitable)
        awaited_result = await async_result
        assert awaited_result == result

    def fails_when_serialize_of_custom_scalar_does_not_return_a_value():
        custom_scalar = GraphQLScalarType(
            "CustomScalar", serialize=lambda _value: Undefined  # returns nothing
        )
        schema = GraphQLSchema(
            GraphQLObjectType(
                "Query",
                {
                    "customScalar": GraphQLField(
                        custom_scalar, resolve=lambda *_args: "CUSTOM_VALUE"
                    )
                },
            )
        )

        result = execute_sync(schema, parse("{ customScalar }"))
        assert result == (
            {"customScalar": None},
            [
                {
                    "message": "Expected `CustomScalar.serialize('CUSTOM_VALUE')`"
                    " to return non-nullable value, returned: Undefined",
                    "locations": [(1, 3)],
                    "path": ["customScalar"],
                }
            ],
        )

    def executes_ignoring_invalid_non_executable_definitions():
        schema = GraphQLSchema(
            GraphQLObjectType("Query", {"foo": GraphQLField(GraphQLString)})
        )

        document = parse(
            """
            { foo }

            type Query { bar: String }
            """
        )

        result = execute_sync(schema, document)
        assert result == ({"foo": None}, None)

    def uses_a_custom_field_resolver():
        schema = GraphQLSchema(
            GraphQLObjectType("Query", {"foo": GraphQLField(GraphQLString)})
        )
        document = parse("{ foo }")

        def field_resolver(_source, info):
            # For the purposes of test, just return the name of the field!
            return info.field_name

        result = execute_sync(schema, document, field_resolver=field_resolver)
        assert result == ({"foo": "foo"}, None)

    def uses_a_custom_type_resolver():
        document = parse("{ foo { bar } }")

        foo_interface = GraphQLInterfaceType(
            "FooInterface", {"bar": GraphQLField(GraphQLString)}
        )

        foo_object = GraphQLObjectType(
            "FooObject", {"bar": GraphQLField(GraphQLString)}, [foo_interface]
        )

        schema = GraphQLSchema(
            GraphQLObjectType("Query", {"foo": GraphQLField(foo_interface)}),
            types=[foo_object],
        )

        possible_types = None

        def type_resolver(_source, info, abstract_type):
            # Resolver should be able to figure out all possible types on its own
            nonlocal possible_types
            possible_types = info.schema.get_possible_types(abstract_type)
            return "FooObject"

        root_value = {"foo": {"bar": "bar"}}
        result = execute_sync(schema, document, root_value, type_resolver=type_resolver)

        assert result == ({"foo": {"bar": "bar"}}, None)
        assert possible_types == [foo_object]