Python graphene 模块，Schema() 实例源码

def query_route(request):
    """
    Query database based on Facebook's GraphQL Library.
    Parameters must be coded into a "q" GET parameter, e.g.
    `curl "localhost:4284/api/v2/query?q=query\{statements\{uid,isStartpoint\}\}"`

    :return: JSON containing queried data
    """
    q = request.params.get("q")
    if q:
        schema = graphene.Schema(query=Query)
        result = schema.execute(q, context_value={'session': DBDiscussionSession})
        if result.errors:
            return {"errors": {"message": "Not all requested parameters could be queried. Some fields are not "
                                          "allowed, e.g. the password.",
                               "exception": str(result.errors)}}
        return result.data
    return {"errors": {"message": "No valid query provided."}}

def test_should_query_only_fields():
    with raises(Exception):
        class ReporterType(DjangoObjectType):

            class Meta:
                model = Reporter
                only_fields = ('articles', )

        schema = graphene.Schema(query=ReporterType)
        query = '''
            query ReporterQuery {
              articles
            }
        '''
        result = schema.execute(query)
        assert not result.errors

def testQuery_excludedField(self):
        Article(headline="h1", summary="s1").put()

        class ArticleType(NdbObjectType):
            class Meta:
                model = Article
                exclude_fields = ['summary']

        class QueryType(graphene.ObjectType):
            articles = graphene.List(ArticleType)

            def resolve_articles(self, info):
                return Article.query()

        schema = graphene.Schema(query=QueryType)
        query = '''
            query ArticlesQuery {
              articles { headline, summary }
            }
        '''

        result = schema.execute(query)

        self.assertIsNotNone(result.errors)
        self.assertTrue('Cannot query field "summary"' in result.errors[0].message)

def schema():
    class Query(graphene.ObjectType):
        test_string = graphene.String()

        def resolve_test_string(self, args, context, info):
            return 'works'

    # TODO: Implement case conversion for arg names
    class Subscription(graphene.ObjectType):
        test_subscription = graphene.String()
        test_context = graphene.String()
        test_filter = graphene.String(filterBoolean=graphene.Boolean())
        test_filter_multi = graphene.String(
            filterBoolean=graphene.Boolean(),
            a=graphene.String(),
            b=graphene.Int())
        test_channel_options = graphene.String()

        def resolve_test_subscription(self, args, context, info):
            return self

        def resolve_test_context(self, args, context, info):
            return context

        def resolve_test_filter(self, args, context, info):
            return 'good_filter' if args.get('filterBoolean') else 'bad_filter'

        def resolve_test_filter_multi(self, args, context, info):
            return 'good_filter' if args.get('filterBoolean') else 'bad_filter'

        def resolve_test_channel_options(self, args, context, info):
            return self

    return graphene.Schema(query=Query, subscription=Subscription)

def validation_schema():
    class Query(graphene.ObjectType):
        placeholder = graphene.String()

    class Subscription(graphene.ObjectType):
        test_1 = graphene.String()
        test_2 = graphene.String()

    return graphene.Schema(query=Query, subscription=Subscription)

def schema(data):
    class UserType(graphene.ObjectType):
        id = graphene.String()
        name = graphene.String()

    class Query(graphene.ObjectType):
        test_string = graphene.String()

    class Subscription(graphene.ObjectType):
        user = graphene.Field(UserType, id=graphene.String())
        user_filtered = graphene.Field(UserType, id=graphene.String())
        context = graphene.String()
        error = graphene.String()

        def resolve_user(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_user_filtered(self, args, context, info):
            id = args['id']
            name = data[args['id']]['name']
            return UserType(id=id, name=name)

        def resolve_context(self, args, context, info):
            return context

        def resolve_error(self, args, context, info):
            raise Exception('E1')

    return graphene.Schema(query=Query, subscription=Subscription)

def test_root_scan_should_fastforward_on_after():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        articles = PynamoConnectionField(ArticleNode)

        def resolve_articles(self, *args, **kwargs):
            return [
                Article(1, headline='One'),
                Article(2, headline='Two'),
                Article(3, headline='Three'),
                Article(4, headline='Four')
            ]

    query = '''
        query ArticlesQuery {
          articles(after: "QXJ0aWNsZU5vZGU6Mq==", first: 1) {
            edges {
              node {
                id
                headline
              }
            }
          }
        }
    '''
    expected = [{
        'node': {
            'headline': 'Three',
            'id': 'QXJ0aWNsZU5vZGU6Mw=='
        }
    }]

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['articles']['edges'] == expected

def schema_builder():
    def build(schemas, mutations=None):
        mutations = [] if not mutations else mutations

        attrs = {schema[0].model.__name__.lower(): schema[1] for schema in schemas}
        Query = type('Query', (graphene.ObjectType,), attrs)

        attrs = {'create_' + m.model.__name__.lower(): m.mutate for m in mutations}
        Mutation = type('Mutation', (graphene.ObjectType,), attrs)

        return graphene.Schema(query=Query, mutation=Mutation)
    return build

def graphql_schema(self):
        """
        :return: Graphql Schema 
        :rtype: graphene.Schema
        """
        return self._graphql_schema

def init_graphql_schema():
        """
        Go through all sub apps defined Queries, Types and mutations
        defined in (graphql.py) and register them in one global schema
        """

        # Import all (graphql) submodules defined in package 'graphql'
        # After importing we'll have
        # All Queries under ::  BaseQuery.__subclasses__()
        # All Types under :: SQLAlchemyObjectType.__subclasses__()
        # All Mutations under :: BaseMutation.__subclasses__()
        CRM._load_modules(module_type='types')
        CRM._load_modules(module_type='queries')
        CRM._load_modules(module_type='mutations')

        schema = graphene.Schema(

            # Make dynamic Query class that inherits all defined queries
            query=type(
                'Query',
                tuple(BaseQuery.__subclasses__()),
                {}
            ),

            types=list(SQLAlchemyObjectType.__subclasses__()),

            # Make dynamic Mutations class that inherits all defined mutations
            mutation=type(
                'Mutations',
                tuple(BaseMutation.__subclasses__()),
                {}
            )
        )

        return schema

def test_should_query_well():
    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter

    class Query(graphene.ObjectType):
        reporter = graphene.Field(ReporterType)

        def resolve_reporter(self, info):
            return Reporter(first_name='ABA', last_name='X')

    query = '''
        query ReporterQuery {
          reporter {
            firstName,
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'firstName': 'ABA',
            'lastName': 'X',
            'email': ''
        }
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data == expected

def test_should_query_promise_connectionfields():
    from promise import Promise

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )

    class Query(graphene.ObjectType):
        all_reporters = DjangoConnectionField(ReporterType)

        def resolve_all_reporters(self, info, **args):
            return Promise.resolve([Reporter(id=1)])

    schema = graphene.Schema(query=Query)
    query = '''
        query ReporterPromiseConnectionQuery {
            allReporters(first: 1) {
                edges {
                    node {
                        id
                    }
                }
            }
        }
    '''

    expected = {
        'allReporters': {
            'edges': [{
                'node': {
                    'id': 'UmVwb3J0ZXJUeXBlOjE='
                }
            }]
        }
    }

    result = schema.execute(query)
    assert not result.errors
    assert result.data == expected

def test_should_handle_inherited_choices():
    class BaseModel(models.Model):
        choice_field = models.IntegerField(choices=((0, 'zero'), (1, 'one')))

    class ChildModel(BaseModel):
        class Meta:
            proxy = True

    class BaseType(DjangoObjectType):
        class Meta:
            model = BaseModel

    class ChildType(DjangoObjectType):
        class Meta:
            model = ChildModel

    class Query(graphene.ObjectType):
        base = graphene.Field(BaseType)
        child = graphene.Field(ChildType)

    schema = graphene.Schema(query=Query)
    query = '''
        query {
          child {
            choiceField
          }
        }
    '''
    result = schema.execute(query)
    assert not result.errors

def __init__(self, app=None, blueprint_options=None):
        self._schema = graphene.Schema()
        self._webpack = None
        if blueprint_options is None:
            blueprint_options = {}
        self._blueprint_options = blueprint_options
        self._blueprint = self._create_blueprint()
        self._add_url_rules()
        self.jwt = None
        if app is not None:
            self.init_app(app)

def test_noSchema_returns500(self):
        graphql_application = webapp2.WSGIApplication([
            ('/graphql', GraphQLHandler)
        ])

        app = webtest.TestApp(graphql_application)
        for method in (app.get, app.post):
            response = method('/graphql', expect_errors=True)
            self.assertEqual(response.status_int, 500)
            self.assertEqual(response.json_body['errors'][0]['message'], 'GraphQL Schema is missing.')

def testQuery_onlyFields(self):
        Article(headline="h1", summary="s1").put()

        class ArticleType(NdbObjectType):
            class Meta:
                model = Article
                only_fields = ['headline']

        class QueryType(graphene.ObjectType):
            articles = graphene.List(ArticleType)

            def resolve_articles(self, info):
                return Article.query()

        schema = graphene.Schema(query=QueryType)
        query = '''
                    query ArticlesQuery {
                      articles { headline }
                    }
                '''

        result = schema.execute(query)

        self.assertIsNotNone(result.data)
        self.assertEqual(result.data['articles'][0]['headline'], 'h1')

        query = '''
                    query ArticlesQuery {
                      articles { headline, summary }
                    }
                '''
        result = schema.execute(query)

        self.assertIsNotNone(result.errors)
        self.assertTrue('Cannot query field "summary"' in result.errors[0].message)

def router(app):
    schema = graphene.Schema(query=RootQuery)
    router = GrapheneRouter.from_schema(schema).serving_on(app.app)
    return router

def __init__(self, schema: graphene.Schema):
        self.schema = schema

def from_schema(cls, schema: graphene.Schema) -> "GrapheneRouter":
        """Create a router with a Graphene schema.

        Args:
            schema: A :class:`graphene.Schema` instance.

        Returns:
            A :class:`.GrapheneRouter` set up with the given schema.
        """
        router = GrapheneRouter()
        return router.with_schema(schema)

def test_should_query_well():
    class ReporterType(PynamoObjectType):
        class Meta:
            model = Reporter

    class Query(graphene.ObjectType):
        reporter = graphene.Field(ReporterType)
        reporters = graphene.List(ReporterType)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(1)

        def resolve_reporters(self, *args, **kwargs):
            return list(Reporter.scan())

    query = '''
        query ReporterQuery {
          reporter {
            firstName,
            lastName,
            email,
            customMap,
            awards
          }
          reporters {
            firstName
          }
        }
    '''
    expected = {
        'reporter': {
            'email': None,
            'firstName': 'ABA',
            'lastName': 'X',
            'customMap': {"key1": "value1", "key2": "value2"},
            'awards': ['pulizer']
        },
        'reporters': [{
            'firstName': 'ABO',
        }, {
            'firstName': 'ABA',
        }]
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    result.data['reporter']["customMap"] = json.loads(result.data['reporter']["customMap"])
    assert dict(result.data['reporter']) == expected['reporter']
    assert all(item in result.data['reporters'] for item in expected['reporters'])

def test_should_custom_identifier():
    class EditorNode(PynamoObjectType):
        class Meta:
            model = Editor
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        all_editors = PynamoConnectionField(EditorNode)

    query = '''
        query EditorQuery {
          allEditors {
            edges {
                node {
                    id,
                    name
                }
            }
          },
          node(id: "RWRpdG9yTm9kZTox") {
            ...on EditorNode {
              name
            }
          }
        }
    '''
    expected = {
        'allEditors': {
            'edges': [{
                'node': {
                    'id': 'RWRpdG9yTm9kZTox',
                    'name': 'John'
                }
            }]
        },
        'node': {
            'name': 'John'
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['allEditors'] == expected['allEditors']

def test_should_return_empty_cursors_on_empty():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class ReporterNode(PynamoObjectType):
        class Meta:
            model = Reporter
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        reporter = graphene.Field(ReporterNode)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(2)

    query = '''
        query ReporterQuery {
          reporter {
            id,
            firstName,
            articles(first: 1) {
              edges {
                node {
                  id
                  headline
                }
              }
              pageInfo {
                hasNextPage
                hasPreviousPage
                startCursor
                endCursor
              }
            }
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'articles': {
                'edges': [],
                'pageInfo': {
                    'hasNextPage': False,
                    'hasPreviousPage': False,
                    'startCursor': '',
                    'endCursor': ''
                }
            }
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']
    assert result.data['reporter']['articles']['pageInfo'] == expected['reporter']['articles']['pageInfo']

def test_should_support_last():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class ReporterNode(PynamoObjectType):
        class Meta:
            model = Reporter
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        reporter = graphene.Field(ReporterNode)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(1)

    query = '''
        query ReporterQuery {
          reporter {
            id,
            firstName,
            articles(last: 1) {
              edges {
                node {
                  id
                  headline
                }
              }
            }
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'articles': {
                'edges': [{
                    'node': {
                        'id': 'QXJ0aWNsZU5vZGU6Mw==',
                        'headline': 'My Article'
                    }
                }]
            }
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']

def test_should_support_after():
    class ArticleNode(PynamoObjectType):
        class Meta:
            model = Article
            interfaces = (Node,)

    class ReporterNode(PynamoObjectType):
        class Meta:
            model = Reporter
            interfaces = (Node,)

    class Query(graphene.ObjectType):
        node = Node.Field()
        reporter = graphene.Field(ReporterNode)

        def resolve_reporter(self, *args, **kwargs):
            return Reporter.get(1)

    query = '''
        query ReporterQuery {
          reporter {
            id,
            firstName,
            articles(after: "QXJ0aWNsZU5vZGU6MQ==") {
              edges {
                node {
                  id
                  headline
                }
              }
            }
            lastName,
            email
          }
        }
    '''
    expected = {
        'reporter': {
            'articles': {
                'edges': [{
                    'node': {
                        'id': 'QXJ0aWNsZU5vZGU6Mw==',
                        'headline': 'My Article'
                    }
                }]
            }
        }
    }

    schema = graphene.Schema(query=Query)
    result = schema.execute(query)
    assert not result.errors
    assert result.data['reporter']['articles']['edges'] == expected['reporter']['articles']['edges']

def test_should_query_field():
    r1 = Reporter(last_name='ABA')
    r1.save()
    r2 = Reporter(last_name='Griffin')
    r2.save()

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )

    class Query(graphene.ObjectType):
        reporter = graphene.Field(ReporterType)
        debug = graphene.Field(DjangoDebug, name='__debug')

        def resolve_reporter(self, info, **args):
            return Reporter.objects.first()

    query = '''
        query ReporterQuery {
          reporter {
            lastName
          }
          __debug {
            sql {
              rawSql
            }
          }
        }
    '''
    expected = {
        'reporter': {
            'lastName': 'ABA',
        },
        '__debug': {
            'sql': [{
                'rawSql': str(Reporter.objects.order_by('pk')[:1].query)
            }]
        }
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()])
    assert not result.errors
    assert result.data == expected

def test_should_query_list():
    r1 = Reporter(last_name='ABA')
    r1.save()
    r2 = Reporter(last_name='Griffin')
    r2.save()

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )

    class Query(graphene.ObjectType):
        all_reporters = graphene.List(ReporterType)
        debug = graphene.Field(DjangoDebug, name='__debug')

        def resolve_all_reporters(self, info, **args):
            return Reporter.objects.all()

    query = '''
        query ReporterQuery {
          allReporters {
            lastName
          }
          __debug {
            sql {
              rawSql
            }
          }
        }
    '''
    expected = {
        'allReporters': [{
            'lastName': 'ABA',
        }, {
            'lastName': 'Griffin',
        }],
        '__debug': {
            'sql': [{
                'rawSql': str(Reporter.objects.all().query)
            }]
        }
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()])
    assert not result.errors
    assert result.data == expected

def test_should_query_connection():
    r1 = Reporter(last_name='ABA')
    r1.save()
    r2 = Reporter(last_name='Griffin')
    r2.save()

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )

    class Query(graphene.ObjectType):
        all_reporters = DjangoConnectionField(ReporterType)
        debug = graphene.Field(DjangoDebug, name='__debug')

        def resolve_all_reporters(self, info, **args):
            return Reporter.objects.all()

    query = '''
        query ReporterQuery {
          allReporters(first:1) {
            edges {
              node {
                lastName
              }
            }
          }
          __debug {
            sql {
              rawSql
            }
          }
        }
    '''
    expected = {
        'allReporters': {
            'edges': [{
                'node': {
                    'lastName': 'ABA',
                }
            }]
        },
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()])
    assert not result.errors
    assert result.data['allReporters'] == expected['allReporters']
    assert 'COUNT' in result.data['__debug']['sql'][0]['rawSql']
    query = str(Reporter.objects.all()[:1].query)
    assert result.data['__debug']['sql'][1]['rawSql'] == query

def test_should_query_connectionfilter():
    from ...filter import DjangoFilterConnectionField

    r1 = Reporter(last_name='ABA')
    r1.save()
    r2 = Reporter(last_name='Griffin')
    r2.save()

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )

    class Query(graphene.ObjectType):
        all_reporters = DjangoFilterConnectionField(ReporterType, fields=['last_name'])
        s = graphene.String(resolver=lambda *_: "S")
        debug = graphene.Field(DjangoDebug, name='__debug')

        def resolve_all_reporters(self, info, **args):
            return Reporter.objects.all()

    query = '''
        query ReporterQuery {
          allReporters(first:1) {
            edges {
              node {
                lastName
              }
            }
          }
          __debug {
            sql {
              rawSql
            }
          }
        }
    '''
    expected = {
        'allReporters': {
            'edges': [{
                'node': {
                    'lastName': 'ABA',
                }
            }]
        },
    }
    schema = graphene.Schema(query=Query)
    result = schema.execute(query, context_value=context(), middleware=[DjangoDebugMiddleware()])
    assert not result.errors
    assert result.data['allReporters'] == expected['allReporters']
    assert 'COUNT' in result.data['__debug']['sql'][0]['rawSql']
    query = str(Reporter.objects.all()[:1].query)
    assert result.data['__debug']['sql'][1]['rawSql'] == query

def test_annotation_is_perserved():
    class ReporterType(DjangoObjectType):
        full_name = String()

        def resolve_full_name(instance, info, **args):
            return instance.full_name

        class Meta:
            model = Reporter
            interfaces = (Node, )
            filter_fields = ()

    class Query(ObjectType):
        all_reporters = DjangoFilterConnectionField(ReporterType)

        def resolve_all_reporters(self, info, **args):
            return Reporter.objects.annotate(
                full_name=Concat('first_name', Value(' '), 'last_name', output_field=TextField())
            )

    Reporter.objects.create(
        first_name='John',
        last_name='Doe',
    )

    schema = Schema(query=Query)

    query = '''
        query NodeFilteringQuery {
            allReporters(first: 1) {
                edges {
                    node {
                        fullName
                    }
                }
            }
        }
    '''
    expected = {
        'allReporters': {
            'edges': [{
                'node': {
                    'fullName': 'John Doe',
                }
            }]
        }
    }

    result = schema.execute(query)

    assert not result.errors
    assert result.data == expected

def test_should_query_postgres_fields():
    from django.contrib.postgres.fields import IntegerRangeField, ArrayField, JSONField, HStoreField

    class Event(models.Model):
        ages = IntegerRangeField(help_text='The age ranges')
        data = JSONField(help_text='Data')
        store = HStoreField()
        tags = ArrayField(models.CharField(max_length=50))

    class EventType(DjangoObjectType):

        class Meta:
            model = Event

    class Query(graphene.ObjectType):
        event = graphene.Field(EventType)

        def resolve_event(self, info):
            return Event(
                ages=(0, 10),
                data={'angry_babies': True},
                store={'h': 'store'},
                tags=['child', 'angry', 'babies']
            )

    schema = graphene.Schema(query=Query)
    query = '''
        query myQuery {
          event {
            ages
            tags
            data
            store
          }
        }
    '''
    expected = {
        'event': {
            'ages': [0, 10],
            'tags': ['child', 'angry', 'babies'],
            'data': '{"angry_babies": true}',
            'store': '{"h": "store"}',
        },
    }
    result = schema.execute(query)
    assert not result.errors
    assert result.data == expected

def test_should_query_connectionfields():
    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )
            only_fields = ('articles', )

    class Query(graphene.ObjectType):
        all_reporters = DjangoConnectionField(ReporterType)

        def resolve_all_reporters(self, info, **args):
            return [Reporter(id=1)]

    schema = graphene.Schema(query=Query)
    query = '''
        query ReporterConnectionQuery {
          allReporters {
            pageInfo {
              hasNextPage
            }
            edges {
              node {
                id
              }
            }
          }
        }
    '''
    result = schema.execute(query)
    assert not result.errors
    assert result.data == {
        'allReporters': {
            'pageInfo': {
                'hasNextPage': False,
            },
            'edges': [{
                'node': {
                    'id': 'UmVwb3J0ZXJUeXBlOjE='
                }
            }]
        }
    }

def test_should_keep_annotations():
    from django.db.models import (
        Count,
        Avg,
    )

    class ReporterType(DjangoObjectType):

        class Meta:
            model = Reporter
            interfaces = (Node, )
            only_fields = ('articles', )

    class ArticleType(DjangoObjectType):

        class Meta:
            model = Article
            interfaces = (Node, )
            filter_fields = ('lang', )

    class Query(graphene.ObjectType):
        all_reporters = DjangoConnectionField(ReporterType)
        all_articles = DjangoConnectionField(ArticleType)

        def resolve_all_reporters(self, info, **args):
            return Reporter.objects.annotate(articles_c=Count('articles')).order_by('articles_c')

        def resolve_all_articles(self, info, **args):
            return Article.objects.annotate(import_avg=Avg('importance')).order_by('import_avg')

    schema = graphene.Schema(query=Query)
    query = '''
        query ReporterConnectionQuery {
          allReporters {
            pageInfo {
              hasNextPage
            }
            edges {
              node {
                id
              }
            }
          }
          allArticles {
            pageInfo {
              hasNextPage
            }
            edges {
              node {
                id
              }
            }
          }
        }
    '''
    result = schema.execute(query)
    assert not result.errors

def testNdbObjectType_should_raise_if_model_is_invalid(self):
        with self.assertRaises(Exception) as context:
            class Character2(NdbObjectType):
                class Meta:
                    model = 1

        assert 'not an NDB model' in str(context.exception.message)

    # def testNdbObjectType_keyProperty_kindDoesntExist_raisesException(self):
    #     with self.assertRaises(Exception) as context:
    #         class ArticleType(NdbObjectType):
    #             class Meta:
    #                 model = Article
    #                 only_fields = ('prop',)
    #
    #             prop = NdbKeyReferenceField('foo', 'bar')
    #
    #         class QueryType(graphene.ObjectType):
    #             articles = graphene.List(ArticleType)
    #
    #             @graphene.resolve_only_args
    #             def resolve_articles(self):
    #                 return Article.query()
    #
    #         schema = graphene.Schema(query=QueryType)
    #         schema.execute('query test {  articles { prop } }')
    #
    #     self.assertIn("Model 'bar' is not accessible by the schema.", str(context.exception.message))

    # def testNdbObjectType_keyProperty_stringRepresentation_kindDoesntExist_raisesException(self):
    #     with self.assertRaises(Exception) as context:
    #         class ArticleType(NdbObjectType):
    #             class Meta:
    #                 model = Article
    #                 only_fields = ('prop',)
    #
    #             prop = NdbKeyStringField('foo', 'bar')
    #
    #         class QueryType(graphene.ObjectType):
    #             articles = graphene.List(ArticleType)
    #
    #             @graphene.resolve_only_args
    #             def resolve_articles(self):
    #                 return Article.query()
    #
    #         schema = graphene.Schema(query=QueryType)
    #         schema.execute('query test {  articles { prop } }')
    #
    #     self.assertIn("Model 'bar' is not accessible by the schema.", str(context.exception.message))