Re-create UnforgivingExecutionContext compatible with graphql-core 3.1.5+

graphene/types/schema.py

@@ -11,6 +11,7 @@ from graphql import (
     print_schema,
     subscribe,
     validate,
+    ExecutionContext,
     ExecutionResult,
     GraphQLArgument,
     GraphQLBoolean,
@@ -54,6 +55,22 @@ from .utils import get_field_as
 introspection_query = get_introspection_query()
 IntrospectionSchema = introspection_types["__Schema"]
 
+class UnforgivingExecutionContext(ExecutionContext):
+    """An execution context which doesn't swallow exceptions.
+    Instead it re-raises the original error, except for
+    GraphQLError, which is handled by graphql-core
+    """
+
+    def handle_field_error(
+        self,
+        error: GraphQLError,
+        return_type,
+    ) -> None:
+        if type(error.original_error) is GraphQLError:
+            super().handle_field_error(error, return_type)
+        else:
+            raise error.original_error
+
 
 def assert_valid_root_type(type_):
     if type_ is None:

graphene/types/tests/test_schema.py

@@ -1,12 +1,12 @@
 from graphql.type import GraphQLObjectType, GraphQLSchema
-from pytest import raises
-
+from graphql import GraphQLError
+from pytest import fixture, mark, raises
 from graphene.tests.utils import dedent
 
 from ..field import Field
 from ..objecttype import ObjectType
 from ..scalars import String
-from ..schema import Schema
+from ..schema import Schema, UnforgivingExecutionContext
 
 
 class MyOtherType(ObjectType):
@@ -68,3 +68,115 @@ def test_schema_requires_query_type():
     assert len(result.errors) == 1
     error = result.errors[0]
     assert error.message == "Query root type must be provided."
+
+
+class TestUnforgivingExecutionContext:
+    @fixture
+    def schema(self):
+        class ErrorFieldsMixin:
+            sanity_field = String()
+            expected_error_field = String()
+            unexpected_value_error_field = String()
+            unexpected_type_error_field = String()
+            unexpected_attribute_error_field = String()
+            unexpected_key_error_field = String()
+
+            @staticmethod
+            def resolve_sanity_field(obj, info):
+                return "not an error"
+
+            @staticmethod
+            def resolve_expected_error_field(obj, info):
+                raise GraphQLError("expected error")
+
+            @staticmethod
+            def resolve_unexpected_value_error_field(obj, info):
+                raise ValueError("unexpected error")
+
+            @staticmethod
+            def resolve_unexpected_type_error_field(obj, info):
+                raise TypeError("unexpected error")
+
+            @staticmethod
+            def resolve_unexpected_attribute_error_field(obj, info):
+                raise AttributeError("unexpected error")
+
+            @staticmethod
+            def resolve_unexpected_key_error_field(obj, info):
+                return {}["fails"]
+
+        class NestedObject(ErrorFieldsMixin, ObjectType):
+            pass
+
+        class MyQuery(ErrorFieldsMixin, ObjectType):
+            nested_object = Field(NestedObject)
+            nested_object_error = Field(NestedObject)
+
+            @staticmethod
+            def resolve_nested_object(obj, info):
+                return object()
+
+            @staticmethod
+            def resolve_nested_object_error(obj, info):
+                raise TypeError()
+
+        schema = Schema(query=MyQuery)
+        return schema
+
+    def test_sanity_check(self, schema):
+        # this should pass with no errors (sanity check)
+        result = schema.execute(
+            "query { sanityField }",
+            execution_context_class=UnforgivingExecutionContext,
+        )
+        assert not result.errors
+        assert result.data == {"sanityField": "not an error"}
+
+    def test_nested_sanity_check(self, schema):
+        # this should pass with no errors (sanity check)
+        result = schema.execute(
+            r"query { nestedObject { sanityField } }",
+            execution_context_class=UnforgivingExecutionContext,
+        )
+        assert not result.errors
+        assert result.data == {"nestedObject": {"sanityField": "not an error"}}
+
+    def test_graphql_error(self, schema):
+        result = schema.execute(
+            "query { expectedErrorField }",
+            execution_context_class=UnforgivingExecutionContext,
+        )
+        assert len(result.errors) == 1
+        assert result.errors[0].message == "expected error"
+        assert result.data == {"expectedErrorField": None}
+
+    def test_nested_graphql_error(self, schema):
+        result = schema.execute(
+            r"query { nestedObject { expectedErrorField } }",
+            execution_context_class=UnforgivingExecutionContext,
+        )
+        assert len(result.errors) == 1
+        assert result.errors[0].message == "expected error"
+        assert result.data == {"nestedObject": {"expectedErrorField": None}}
+
+    @mark.parametrize(
+        "field,exception",
+        [
+            ("unexpectedValueErrorField", ValueError),
+            ("unexpectedTypeErrorField", TypeError),
+            ("unexpectedAttributeErrorField", AttributeError),
+            ("unexpectedKeyErrorField", KeyError),
+            ("nestedObject { unexpectedValueErrorField }", ValueError),
+            ("nestedObject { unexpectedTypeErrorField }", TypeError),
+            ("nestedObject { unexpectedAttributeErrorField }", AttributeError),
+            ("nestedObject { unexpectedKeyErrorField }", KeyError),
+            ("nestedObjectError { __typename }", TypeError),
+        ],
+    )
+    def test_unexpected_error(self, field, exception, schema):
+        with raises(exception):
+            # no result, but the exception should be propagated
+            schema.execute(
+                f"query {{ {field} }}",
+                execution_context_class=UnforgivingExecutionContext,
+            )