File: test_transport_batch.py

package info (click to toggle)
python-gql 4.0.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 1,900 kB
  • sloc: python: 21,677; makefile: 54
file content (153 lines) | stat: -rw-r--r-- 3,797 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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import os

import pytest

from gql import Client, gql

# We serve https://github.com/graphql-python/swapi-graphene locally:
URL = "http://127.0.0.1:8000/graphql"

# Marking all tests in this file with the requests marker
pytestmark = pytest.mark.requests


def use_cassette(name):
    import vcr

    query_vcr = vcr.VCR(
        cassette_library_dir=os.path.join(
            os.path.dirname(__file__), "fixtures", "vcr_cassettes"
        ),
        record_mode="new_episodes",
        match_on=["uri", "method", "body"],
    )

    return query_vcr.use_cassette(name + ".yaml")


@pytest.fixture
def client():
    import requests

    from gql.transport.requests import RequestsHTTPTransport

    with use_cassette("client"):
        response = requests.get(
            URL, headers={"Host": "swapi.graphene-python.org", "Accept": "text/html"}
        )
        response.raise_for_status()
        csrf = response.cookies["csrftoken"]

        return Client(
            transport=RequestsHTTPTransport(
                url=URL, cookies={"csrftoken": csrf}, headers={"x-csrftoken": csrf}
            ),
            fetch_schema_from_transport=True,
            introspection_args={
                "input_value_deprecation": False,
            },
        )


def test_hero_name_query(client):
    query = gql(
        """
        {
          myFavoriteFilm: film(id:"RmlsbToz") {
            id
            title
            episodeId
            characters(first:5) {
              edges {
                node {
                  name
                }
              }
            }
          }
        }
        """
    )
    expected = [
        {
            "myFavoriteFilm": {
                "id": "RmlsbToz",
                "title": "Return of the Jedi",
                "episodeId": 6,
                "characters": {
                    "edges": [
                        {"node": {"name": "Luke Skywalker"}},
                        {"node": {"name": "C-3PO"}},
                        {"node": {"name": "R2-D2"}},
                        {"node": {"name": "Darth Vader"}},
                        {"node": {"name": "Leia Organa"}},
                    ]
                },
            }
        }
    ]
    with use_cassette("queries_batch"):
        results = client.execute_batch([query])
    assert results == expected


def test_query_with_variable(client):
    query = gql(
        """
        query Planet($id: ID!) {
          planet(id: $id) {
            id
            name
          }
        }
        """
    )
    query.variable_values = {"id": "UGxhbmV0OjEw"}
    expected = [{"planet": {"id": "UGxhbmV0OjEw", "name": "Kamino"}}]
    with use_cassette("queries_batch"):
        results = client.execute_batch([query])
    assert results == expected


def test_named_query(client):
    query = gql(
        """
        query Planet1 {
          planet(id: "UGxhbmV0OjEw") {
            id
            name
          }
        }
        query Planet2 {
          planet(id: "UGxhbmV0OjEx") {
            id
            name
          }
        }
        """
    )
    query.operation_name = "Planet2"
    expected = [{"planet": {"id": "UGxhbmV0OjEx", "name": "Geonosis"}}]
    with use_cassette("queries_batch"):
        results = client.execute_batch([query])
    assert results == expected


def test_header_query(client):
    query = gql(
        """
        query Planet($id: ID!) {
          planet(id: $id) {
            id
            name
          }
        }
        """
    )
    expected = [{"planet": {"id": "UGxhbmV0OjEx", "name": "Geonosis"}}]
    with use_cassette("queries_batch"):
        results = client.execute_batch(
            [query],
            extra_args={"headers": {"authorization": "xxx-123"}},
        )
    assert results == expected