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
|
import urllib.parse
import pytest
import globus_sdk
from globus_sdk import MISSING
from globus_sdk.testing import get_last_request, load_response
@pytest.mark.parametrize(
"include_param",
[MISSING, "private_policies", "private_policies,foo", ("private_policies", "foo")],
)
def test_get_storage_gateway_list(client, include_param):
meta = load_response(client.get_storage_gateway_list).metadata
expect_ids = meta["ids"]
res = client.get_storage_gateway_list(include=include_param)
assert res.http_status == 200
# confirm iterable and sanity check some fields
assert len(list(res)) > 0
for sg in res:
assert sg["DATA_TYPE"] == "storage_gateway#1.0.0"
assert "id" in sg
assert "display_name" in sg
assert [sg["id"] for sg in res] == expect_ids
req = get_last_request()
assert req.body is None
parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
if include_param is MISSING:
assert parsed_qs == {}
elif isinstance(include_param, str):
assert parsed_qs == {"include": [include_param]}
else:
assert parsed_qs == {"include": [",".join(include_param)]}
def test_create_storage_gateway(client):
meta = load_response(client.create_storage_gateway).metadata
# the SDK does not validate the create document, so an empty document is fine, if
# unrealistic, with a mocked response
res = client.create_storage_gateway({})
assert res.http_status == 200
# confirm top level access to storage gateway data
assert res["id"] == meta["id"]
assert res["display_name"] == meta["display_name"]
def test_create_storage_gateway_validation_error(client):
meta = load_response(
client.create_storage_gateway, case="validation_error"
).metadata
with pytest.raises(globus_sdk.GCSAPIError) as excinfo:
client.create_storage_gateway({})
error = excinfo.value
assert error.http_status == meta["http_status"]
assert error.code == meta["code"]
assert error.message == meta["message"]
@pytest.mark.parametrize(
"include_param",
[MISSING, "private_policies", "private_policies,foo", ("private_policies", "foo")],
)
def test_get_storage_gateway(client, include_param):
meta = load_response(client.get_storage_gateway).metadata
res = client.get_storage_gateway(meta["id"], include=include_param)
assert res.http_status == 200
# confirm top level access to storage gateway data
assert res["id"] == meta["id"]
assert res["display_name"] == meta["display_name"]
req = get_last_request()
assert req.body is None
parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
if include_param is MISSING:
assert parsed_qs == {}
elif isinstance(include_param, str):
assert parsed_qs == {"include": [include_param]}
else:
assert parsed_qs == {"include": [",".join(include_param)]}
def test_update_storage_gateway(client):
meta = load_response(client.update_storage_gateway).metadata
# as in the create test, an empty update document is not very realistic
# but because there's no request validation, this is fine
res = client.update_storage_gateway(meta["id"], {})
assert res.http_status == 200
# confirm top level access to response data
assert res["code"] == "success"
assert res["message"] == "Operation successful"
def test_delete_storage_gateway(client):
meta = load_response(client.delete_storage_gateway).metadata
res = client.delete_storage_gateway(meta["id"])
assert res.http_status == 200
# confirm top level access to response data
assert res["code"] == "success"
assert res["message"] == "Operation successful"
def test_delete_storage_gateway_permission_denied(client):
meta = load_response(
client.delete_storage_gateway, case="permission_denied_error"
).metadata
with pytest.raises(globus_sdk.GCSAPIError) as excinfo:
client.delete_storage_gateway(meta["id"])
error = excinfo.value
assert error.http_status == meta["http_status"]
assert error.code == meta["code"]
assert error.message == meta["message"]
|
