graphene/graphene/relay/tests/test_mutation.py

import pytest
from promise import Promise

from ...types import (ID, Argument, Field, InputField, InputObjectType,
                      NonNull, ObjectType, Schema)
from ...types.scalars import String
from ..mutation import ClientIDMutation


class SharedFields(object):
    shared = String()


class MyNode(ObjectType):
    # class Meta:
    #     interfaces = (Node, )
    id = ID()
    name = String()


class SaySomething(ClientIDMutation):

    class Input:
        what = String()

    phrase = String()

    @staticmethod
    def mutate_and_get_payload(self, info, what, client_mutation_id=None):
        return SaySomething(phrase=str(what))


class FixedSaySomething(object):
    __slots__ = 'phrase',

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


class SaySomethingFixed(ClientIDMutation):

    class Input:
        what = String()

    phrase = String()

    @staticmethod
    def mutate_and_get_payload(self, info, what, client_mutation_id=None):
        return FixedSaySomething(phrase=str(what))


class SaySomethingPromise(ClientIDMutation):

    class Input:
        what = String()

    phrase = String()

    @staticmethod
    def mutate_and_get_payload(self, info, what, client_mutation_id=None):
        return Promise.resolve(SaySomething(phrase=str(what)))


# MyEdge = MyNode.Connection.Edge
class MyEdge(ObjectType):
    node = Field(MyNode)
    cursor = String()


class OtherMutation(ClientIDMutation):

    class Input(SharedFields):
        additional_field = String()

    name = String()
    my_node_edge = Field(MyEdge)

    @staticmethod
    def mutate_and_get_payload(self, info, shared='', additional_field='', client_mutation_id=None):
        edge_type = MyEdge
        return OtherMutation(
            name=shared + additional_field,
            my_node_edge=edge_type(cursor='1', node=MyNode(name='name')))


class RootQuery(ObjectType):
    something = String()


class Mutation(ObjectType):
    say = SaySomething.Field()
    say_fixed = SaySomethingFixed.Field()
    say_promise = SaySomethingPromise.Field()
    other = OtherMutation.Field()


schema = Schema(query=RootQuery, mutation=Mutation)


def test_no_mutate_and_get_payload():
    with pytest.raises(AssertionError) as excinfo:

        class MyMutation(ClientIDMutation):
            pass

    assert "MyMutation.mutate_and_get_payload method is required in a ClientIDMutation." == str(
        excinfo.value)


def test_mutation():
    fields = SaySomething._meta.fields
    assert list(fields.keys()) == ['phrase', 'client_mutation_id']
    assert SaySomething._meta.name == "SaySomethingPayload"
    assert isinstance(fields['phrase'], Field)
    field = SaySomething.Field()
    assert field.type == SaySomething
    assert list(field.args.keys()) == ['input']
    assert isinstance(field.args['input'], Argument)
    assert isinstance(field.args['input'].type, NonNull)
    assert field.args['input'].type.of_type == SaySomething.Input
    assert isinstance(fields['client_mutation_id'], Field)
    assert fields['client_mutation_id'].name == 'clientMutationId'
    assert fields['client_mutation_id'].type == String


def test_mutation_input():
    Input = SaySomething.Input
    assert issubclass(Input, InputObjectType)
    fields = Input._meta.fields
    assert list(fields.keys()) == ['what', 'client_mutation_id']
    assert isinstance(fields['what'], InputField)
    assert fields['what'].type == String
    assert isinstance(fields['client_mutation_id'], InputField)
    assert fields['client_mutation_id'].type == String


def test_subclassed_mutation():
    fields = OtherMutation._meta.fields
    assert list(fields.keys()) == [
        'name', 'my_node_edge', 'client_mutation_id'
    ]
    assert isinstance(fields['name'], Field)
    field = OtherMutation.Field()
    assert field.type == OtherMutation
    assert list(field.args.keys()) == ['input']
    assert isinstance(field.args['input'], Argument)
    assert isinstance(field.args['input'].type, NonNull)
    assert field.args['input'].type.of_type == OtherMutation.Input


def test_subclassed_mutation_input():
    Input = OtherMutation.Input
    assert issubclass(Input, InputObjectType)
    fields = Input._meta.fields
    assert list(fields.keys()) == [
        'shared', 'additional_field', 'client_mutation_id'
    ]
    assert isinstance(fields['shared'], InputField)
    assert fields['shared'].type == String
    assert isinstance(fields['additional_field'], InputField)
    assert fields['additional_field'].type == String
    assert isinstance(fields['client_mutation_id'], InputField)
    assert fields['client_mutation_id'].type == String


def test_node_query():
    executed = schema.execute(
        'mutation a { say(input: {what:"hello", clientMutationId:"1"}) { phrase } }'
    )
    assert not executed.errors
    assert executed.data == {'say': {'phrase': 'hello'}}


def test_node_query_fixed():
    executed = schema.execute(
        'mutation a { sayFixed(input: {what:"hello", clientMutationId:"1"}) { phrase } }'
    )
    assert "Cannot set client_mutation_id in the payload object" in str(executed.errors[0])


def test_node_query_promise():
    executed = schema.execute(
        'mutation a { sayPromise(input: {what:"hello", clientMutationId:"1"}) { phrase } }'
    )
    assert not executed.errors
    assert executed.data == {'sayPromise': {'phrase': 'hello'}}


def test_edge_query():
    executed = schema.execute(
        'mutation a { other(input: {clientMutationId:"1"}) { clientMutationId, myNodeEdge { cursor node { name }} } }'
    )
    assert not executed.errors
    assert dict(executed.data) == {
        'other': {
            'clientMutationId': '1',
            'myNodeEdge': {
                'cursor': '1',
                'node': {
                    'name': 'name'
                }
            }
        }
    }