File: test_mutation_async.py

package info (click to toggle)
python-graphene 3.4.3-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,120 kB
  • sloc: python: 8,935; makefile: 214; sh: 18
file content (90 lines) | stat: -rw-r--r-- 2,262 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
from pytest import mark

from graphene.types import ID, Field, ObjectType, Schema
from graphene.types.scalars import String
from graphene.relay.mutation import ClientIDMutation
from graphene.test import Client


class SharedFields(object):
    shared = String()


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


class SaySomethingAsync(ClientIDMutation):
    class Input:
        what = String()

    phrase = String()

    @staticmethod
    async def mutate_and_get_payload(self, info, what, client_mutation_id=None):
        return SaySomethingAsync(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_promise = SaySomethingAsync.Field()
    other = OtherMutation.Field()


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


@mark.asyncio
async def test_node_query_promise():
    executed = await client.execute_async(
        'mutation a { sayPromise(input: {what:"hello", clientMutationId:"1"}) { phrase } }'
    )
    assert isinstance(executed, dict)
    assert "errors" not in executed
    assert executed["data"] == {"sayPromise": {"phrase": "hello"}}


@mark.asyncio
async def test_edge_query():
    executed = await client.execute_async(
        'mutation a { other(input: {clientMutationId:"1"}) { clientMutationId, myNodeEdge { cursor node { name }} } }'
    )
    assert isinstance(executed, dict)
    assert "errors" not in executed
    assert executed["data"] == {
        "other": {
            "clientMutationId": "1",
            "myNodeEdge": {"cursor": "1", "node": {"name": "name"}},
        }
    }