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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
|
import contextlib
import json
from io import BytesIO
import pytest
from urllib3 import encode_multipart_formdata
from tests.views.schema import schema
from .clients.base import HttpClient
@pytest.fixture
def http_client(http_client_class: type[HttpClient]) -> HttpClient:
with contextlib.suppress(ImportError):
from .clients.chalice import ChaliceHttpClient
if http_client_class is ChaliceHttpClient:
pytest.xfail(reason="Chalice does not support uploads")
return http_client_class(schema)
@pytest.fixture
def enabled_http_client(http_client_class: type[HttpClient]) -> HttpClient:
with contextlib.suppress(ImportError):
from .clients.chalice import ChaliceHttpClient
if http_client_class is ChaliceHttpClient:
pytest.xfail(reason="Chalice does not support uploads")
return http_client_class(schema, multipart_uploads_enabled=True)
async def test_multipart_uploads_are_disabled_by_default(http_client: HttpClient):
f = BytesIO(b"strawberry")
query = """
mutation($textFile: Upload!) {
readText(textFile: $textFile)
}
"""
response = await http_client.query(
query,
variables={"textFile": None},
files={"textFile": f},
)
assert response.status_code == 400
assert response.data == b"Unsupported content type"
async def test_upload(enabled_http_client: HttpClient):
f = BytesIO(b"strawberry")
query = """
mutation($textFile: Upload!) {
readText(textFile: $textFile)
}
"""
response = await enabled_http_client.query(
query,
variables={"textFile": None},
files={"textFile": f},
)
assert response.json.get("errors") is None
assert response.json["data"] == {"readText": "strawberry"}
async def test_file_list_upload(enabled_http_client: HttpClient):
query = "mutation($files: [Upload!]!) { readFiles(files: $files) }"
file1 = BytesIO(b"strawberry1")
file2 = BytesIO(b"strawberry2")
response = await enabled_http_client.query(
query=query,
variables={"files": [None, None]},
files={"file1": file1, "file2": file2},
)
data = response.json["data"]
assert isinstance(data, dict)
assert len(data["readFiles"]) == 2
assert data["readFiles"][0] == "strawberry1"
assert data["readFiles"][1] == "strawberry2"
async def test_nested_file_list(enabled_http_client: HttpClient):
query = "mutation($folder: FolderInput!) { readFolder(folder: $folder) }"
file1 = BytesIO(b"strawberry1")
file2 = BytesIO(b"strawberry2")
response = await enabled_http_client.query(
query=query,
variables={"folder": {"files": [None, None]}},
files={"file1": file1, "file2": file2},
)
data = response.json["data"]
assert isinstance(data, dict)
assert len(data["readFolder"]) == 2
assert data["readFolder"][0] == "strawberry1"
assert data["readFolder"][1] == "strawberry2"
async def test_upload_single_and_list_file_together(enabled_http_client: HttpClient):
query = """
mutation($files: [Upload!]!, $textFile: Upload!) {
readFiles(files: $files)
readText(textFile: $textFile)
}
"""
file1 = BytesIO(b"strawberry1")
file2 = BytesIO(b"strawberry2")
file3 = BytesIO(b"strawberry3")
response = await enabled_http_client.query(
query=query,
variables={"files": [None, None], "textFile": None},
files={"file1": file1, "file2": file2, "textFile": file3},
)
data = response.json["data"]
assert isinstance(data, dict)
assert len(data["readFiles"]) == 2
assert data["readFiles"][0] == "strawberry1"
assert data["readFiles"][1] == "strawberry2"
assert data["readText"] == "strawberry3"
async def test_upload_invalid_query(enabled_http_client: HttpClient):
f = BytesIO(b"strawberry")
query = """
mutation($textFile: Upload!) {
readT
"""
response = await enabled_http_client.query(
query,
variables={"textFile": None},
files={"textFile": f},
)
assert response.status_code == 200
assert response.json["data"] is None
assert response.json["errors"] == [
{
"locations": [{"column": 5, "line": 4}],
"message": "Syntax Error: Expected Name, found <EOF>.",
}
]
async def test_upload_missing_file(enabled_http_client: HttpClient):
f = BytesIO(b"strawberry")
query = """
mutation($textFile: Upload!) {
readText(textFile: $textFile)
}
"""
response = await enabled_http_client.query(
query,
variables={"textFile": None},
# using the wrong name to simulate a missing file
# this is to make it easier to run tests with our client
files={"a": f},
)
assert response.status_code == 400
assert "File(s) missing in form data" in response.text
class FakeWriter:
def __init__(self):
self.buffer = BytesIO()
async def write(self, data: bytes):
self.buffer.write(data)
@property
def value(self) -> bytes:
return self.buffer.getvalue()
async def test_extra_form_data_fields_are_ignored(enabled_http_client: HttpClient):
query = """mutation($textFile: Upload!) {
readText(textFile: $textFile)
}"""
f = BytesIO(b"strawberry")
operations = json.dumps({"query": query, "variables": {"textFile": None}})
file_map = json.dumps({"textFile": ["variables.textFile"]})
extra_field_data = json.dumps({})
f = BytesIO(b"strawberry")
fields = {
"operations": operations,
"map": file_map,
"extra_field": extra_field_data,
"textFile": ("textFile.txt", f.read(), "text/plain"),
}
data, header = encode_multipart_formdata(fields)
response = await enabled_http_client.post(
url="/graphql",
data=data,
headers={
"content-type": header,
"content-length": f"{len(data)}",
},
)
assert response.status_code == 200
assert response.json["data"] == {"readText": "strawberry"}
async def test_sending_invalid_form_data(enabled_http_client: HttpClient):
headers = {"content-type": "multipart/form-data; boundary=----fake"}
response = await enabled_http_client.post("/graphql", headers=headers)
assert response.status_code == 400
# TODO: consolidate this, it seems only AIOHTTP returns the second error
# due to validating the boundary
assert (
"No GraphQL query found in the request" in response.text
or "Unable to parse the multipart body" in response.text
)
@pytest.mark.aiohttp
async def test_sending_invalid_json_body(enabled_http_client: HttpClient):
f = BytesIO(b"strawberry")
operations = "}"
file_map = json.dumps({"textFile": ["variables.textFile"]})
fields = {
"operations": operations,
"map": file_map,
"textFile": ("textFile.txt", f.read(), "text/plain"),
}
data, header = encode_multipart_formdata(fields)
response = await enabled_http_client.post(
"/graphql",
data=data,
headers={
"content-type": header,
"content-length": f"{len(data)}",
},
)
assert response.status_code == 400
assert (
"Unable to parse the multipart body" in response.text
or "Unable to parse request body as JSON" in response.text
)
|
