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
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
|
import uuid
import pytest
import globus_sdk
from globus_sdk import GlobusApp, GlobusAppConfig, UserApp
from globus_sdk.testing import load_response
from globus_sdk.token_storage import MemoryTokenStorage
@pytest.fixture
def app() -> GlobusApp:
config = GlobusAppConfig(token_storage=MemoryTokenStorage())
return UserApp("test-app", client_id="client_id", config=config)
def test_client_inherits_environment_from_globus_app():
config = GlobusAppConfig(token_storage=MemoryTokenStorage(), environment="sandbox")
app = UserApp("test-app", client_id="client_id", config=config)
client = globus_sdk.AuthClient(app=app)
assert client.environment == "sandbox"
def test_client_environment_does_not_match_the_globus_app_environment():
config = GlobusAppConfig(token_storage=MemoryTokenStorage(), environment="sandbox")
app = UserApp("test-app", client_id="client_id", config=config)
with pytest.raises(globus_sdk.GlobusSDKUsageError) as exc:
globus_sdk.AuthClient(app=app, environment="preview")
expected = "[Environment Mismatch] AuthClient's environment (preview) does not match the GlobusApp's configured environment (sandbox)." # noqa: E501
assert str(exc.value) == expected
def test_transfer_client_default_scopes(app):
globus_sdk.TransferClient(app=app)
assert [str(s) for s in app.scope_requirements["transfer.api.globus.org"]] == [
"urn:globus:auth:scope:transfer.api.globus.org:all"
]
def test_transfer_client_add_app_data_access_scope(app):
client = globus_sdk.TransferClient(app=app)
collection_id = str(uuid.UUID(int=0))
client.add_app_data_access_scope(collection_id)
str_list = [str(s) for s in app.scope_requirements["transfer.api.globus.org"]]
expected = f"urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id}/data_access]" # noqa: E501
assert expected in str_list
def test_timers_client_add_app_data_access_scope(app):
client = globus_sdk.TimersClient(app=app)
collection_id = str(uuid.UUID(int=0))
client.add_app_transfer_data_access_scope(collection_id)
str_list = [
str(s) for s in app.scope_requirements[globus_sdk.TimersClient.resource_server]
]
expected = f"{globus_sdk.TimersClient.scopes.timer}[urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id}/data_access]]" # noqa: E501
assert expected in str_list
def test_timers_client_add_app_flow_user_scope(app):
client = globus_sdk.TimersClient(app=app)
flow_id = str(uuid.UUID(int=0))
client.add_app_flow_user_scope(flow_id)
str_list = [
str(s) for s in app.scope_requirements[globus_sdk.TimersClient.resource_server]
]
flow_client = globus_sdk.SpecificFlowClient(flow_id, app=app)
expected = f"{globus_sdk.TimersClient.scopes.timer}[{flow_client.scopes.user}]" # noqa: E501
assert expected in str_list
def test_specific_flow_client_add_app_data_access_scope(app):
flow_id = str(uuid.UUID(int=1))
client = globus_sdk.SpecificFlowClient(flow_id, app=app)
collection_id = str(uuid.UUID(int=0))
client.add_app_transfer_data_access_scope(collection_id)
str_list = [str(s) for s in app.scope_requirements[client.resource_server]]
expected = f"{client.scopes.user}[*urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id}/data_access]]" # noqa: E501
assert expected in str_list
def test_transfer_client_add_app_data_access_scope_chaining(app):
collection_id_1 = str(uuid.UUID(int=1))
collection_id_2 = str(uuid.UUID(int=2))
(
globus_sdk.TransferClient(app=app)
.add_app_data_access_scope(collection_id_1)
.add_app_data_access_scope(collection_id_2)
)
str_list = [str(s) for s in app.scope_requirements["transfer.api.globus.org"]]
expected_1 = f"urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id_1}/data_access]" # noqa: E501
expected_2 = f"urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id_2}/data_access]" # noqa: E501
assert expected_1 in str_list
assert expected_2 in str_list
def test_timers_client_add_app_flow_user_scope_chaining(app):
flow_id_1 = str(uuid.UUID(int=1))
flow_id_2 = str(uuid.UUID(int=2))
(
globus_sdk.TimersClient(app=app)
.add_app_flow_user_scope(flow_id_1)
.add_app_flow_user_scope(flow_id_2)
)
str_list = [
str(s) for s in app.scope_requirements[globus_sdk.TimersClient.resource_server]
]
flow_client_1 = globus_sdk.SpecificFlowClient(flow_id_1, app=app)
expected_1 = f"{globus_sdk.TimersClient.scopes.timer}[{flow_client_1.scopes.user}]" # noqa: E501
flow_client_2 = globus_sdk.SpecificFlowClient(flow_id_2, app=app)
expected_2 = f"{globus_sdk.TimersClient.scopes.timer}[{flow_client_2.scopes.user}]" # noqa: E501
assert expected_1 in str_list
assert expected_2 in str_list
def test_transfer_client_add_app_data_access_scope_in_iterable(app):
collection_id_1 = str(uuid.UUID(int=1))
collection_id_2 = str(uuid.UUID(int=2))
globus_sdk.TransferClient(app=app).add_app_data_access_scope(
(collection_id_1, collection_id_2)
)
expected_1 = f"https://auth.globus.org/scopes/{collection_id_1}/data_access"
expected_2 = f"https://auth.globus.org/scopes/{collection_id_2}/data_access"
transfer_dependencies = []
for scope in app.scope_requirements["transfer.api.globus.org"]:
if scope.scope_string != str(globus_sdk.TransferClient.scopes.all):
continue
for dep in scope.dependencies:
transfer_dependencies.append((dep.scope_string, dep.optional))
assert (expected_1, True) in transfer_dependencies
assert (expected_2, True) in transfer_dependencies
def test_timers_client_add_app_data_access_scope_in_iterable(app):
collection_id_1 = str(uuid.UUID(int=1))
collection_id_2 = str(uuid.UUID(int=2))
globus_sdk.TimersClient(app=app).add_app_transfer_data_access_scope(
(collection_id_1, collection_id_2)
)
expected_1 = f"https://auth.globus.org/scopes/{collection_id_1}/data_access"
expected_2 = f"https://auth.globus.org/scopes/{collection_id_2}/data_access"
transfer_dependencies = []
for scope in app.scope_requirements[globus_sdk.TimersClient.resource_server]:
if scope.scope_string != str(globus_sdk.TimersClient.scopes.timer):
continue
for dep in scope.dependencies:
if dep.scope_string != str(globus_sdk.TransferClient.scopes.all):
continue
for subdep in dep.dependencies:
transfer_dependencies.append((subdep.scope_string, subdep.optional))
assert (expected_1, True) in transfer_dependencies
assert (expected_2, True) in transfer_dependencies
def test_timers_client_add_app_flow_user_scope_in_iterable(app):
flow_id_1 = str(uuid.UUID(int=1))
flow_id_2 = str(uuid.UUID(int=2))
globus_sdk.TimersClient(app=app).add_app_flow_user_scope((flow_id_1, flow_id_2))
timer_dependencies = [
scope_dep.scope_string
for scope in app.scope_requirements[globus_sdk.TimersClient.resource_server]
for scope_dep in scope.dependencies
]
flow_client_1 = globus_sdk.SpecificFlowClient(flow_id_1, app=app)
flow_client_2 = globus_sdk.SpecificFlowClient(flow_id_2, app=app)
assert flow_client_1.scopes.user.scope_string in timer_dependencies
assert flow_client_2.scopes.user.scope_string in timer_dependencies
def test_transfer_client_add_app_data_access_scope_catches_bad_uuid(app):
with pytest.raises(ValueError, match="'collection_ids' must be a valid UUID"):
globus_sdk.TransferClient(app=app).add_app_data_access_scope("foo")
def test_transfer_client_add_app_data_access_scope_catches_bad_uuid_in_iterable(app):
collection_id_1 = str(uuid.UUID(int=1))
with pytest.raises(ValueError, match=r"'collection_ids\[1\]' must be a valid UUID"):
globus_sdk.TransferClient(app=app).add_app_data_access_scope(
[collection_id_1, "foo"]
)
def test_timers_client_add_app_data_access_scope_catches_bad_uuid(app):
with pytest.raises(ValueError, match="'collection_ids' must be a valid UUID"):
globus_sdk.TimersClient(app=app).add_app_transfer_data_access_scope("foo")
def test_timers_client_add_app_flow_user_scope_catches_bad_uuid(app):
with pytest.raises(ValueError, match="'flow_ids' must be a valid UUID"):
globus_sdk.TimersClient(app=app).add_app_flow_user_scope("foo")
def test_timers_client_add_app_data_access_scope_catches_bad_uuid_in_iterable(app):
collection_id_1 = str(uuid.UUID(int=1))
with pytest.raises(ValueError, match=r"'collection_ids\[1\]' must be a valid UUID"):
globus_sdk.TimersClient(app=app).add_app_transfer_data_access_scope(
[collection_id_1, "foo"]
)
def test_timers_client_add_app_flow_user_scope_catches_bad_uuid_in_iterable(app):
flow_id = str(uuid.UUID(int=1))
with pytest.raises(ValueError, match=r"'flow_ids\[1\]' must be a valid UUID"):
globus_sdk.TimersClient(app=app).add_app_flow_user_scope([flow_id, "foo"])
def test_auth_client_default_scopes(app):
globus_sdk.AuthClient(app=app)
str_list = [str(s) for s in app.scope_requirements["auth.globus.org"]]
assert "openid" in str_list
assert "profile" in str_list
assert "email" in str_list
def test_groups_client_default_scopes(app):
globus_sdk.GroupsClient(app=app)
assert [str(s) for s in app.scope_requirements["groups.api.globus.org"]] == [
"urn:globus:auth:scope:groups.api.globus.org:view_my_groups_and_memberships"
]
def test_search_client_default_scopes(app):
globus_sdk.SearchClient(app=app)
assert [str(s) for s in app.scope_requirements["search.api.globus.org"]] == [
"urn:globus:auth:scope:search.api.globus.org:search"
]
def test_timer_client_default_scopes(app):
globus_sdk.TimersClient(app=app)
timer_client_id = "524230d7-ea86-4a52-8312-86065a9e0417"
str_list = [str(s) for s in app.scope_requirements[timer_client_id]]
assert str_list == [f"https://auth.globus.org/scopes/{timer_client_id}/timer"]
def test_flows_client_default_scopes(app):
globus_sdk.FlowsClient(app=app)
flows_client_id = "eec9b274-0c81-4334-bdc2-54e90e689b9a"
str_list = [str(s) for s in app.scope_requirements["flows.globus.org"]]
assert len(str_list) == 1
assert str_list == [f"https://auth.globus.org/scopes/{flows_client_id}/all"]
def test_specific_flow_client_default_scopes(app):
globus_sdk.SpecificFlowClient("flow_id", app=app)
assert [str(s) for s in app.scope_requirements["flow_id"]] == [
"https://auth.globus.org/scopes/flow_id/flow_flow_id_user"
]
def test_gcs_client_default_scopes(app):
meta = load_response(globus_sdk.GCSClient.get_gcs_info).metadata
endpoint_client_id = meta["endpoint_client_id"]
domain_name = meta["domain_name"]
globus_sdk.GCSClient(domain_name, app=app)
assert [str(s) for s in app.scope_requirements[endpoint_client_id]] == [
f"urn:globus:auth:scope:{endpoint_client_id}:manage_collections"
]
|
