Method _native_graphql_query

edb/testbase/http.py:379–423

(
        self,
        query,
        *,
        # Can/should we support operation_name somehow...
        variables=None,
        globals=None,
    )

        raise ex

    async def _native_graphql_query(
        self,
        query,
        *,
        # Can/should we support operation_name somehow...
        variables=None,
        globals=None,
    ):
        # The graphql tests are all synchronous, and our gel
        # connections need to be async... so we spin up a new
        # connection and asyncio.run the coro.
        con = await self.connect()
        try:
            # Ahhhhhh. We don't support with_globals on testbase
            # connections, so....
            if globals:
                glob_defs = {
                    obj.name: obj
                    for obj in
                    await con.query('''
                        select schema::Global {
                            name, required, tname := .target.name
                        }
                    ''')
                }

                for k, v in globals.items():
                    glob = glob_defs[k]
                    # Why do we allow this for the HTTP proto??
                    # We don't for binary proto stuff.
                    if v is None and glob.required:
                        continue
                    mod = 'required ' if glob.required else ''
                    await con.execute(
                        f'set global {k} := <{mod}{glob.tname}><json>$0',
                        json.dumps(v),
                    )

            async with server.RollbackChanges(con):
                return json.loads(await con.query_graphql_json(
                    query,
                    **(variables or {}),
                ))
        finally:
            await con.aclose()

    def assert_graphql_query_result(
        self,