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
288
289
290
291
292
293
294
295
296
297
298
299
300
|
import pytest
from globus_sdk import MISSING, DeleteData, TransferData
from globus_sdk.services.transfer.client import _format_filter
from tests.common import GO_EP1_ID, GO_EP2_ID
def test_transfer_init_no_params():
"""
Creates a TransferData object without optional parameters and
verifies field initialization.
"""
# default init
tdata = TransferData(GO_EP1_ID, GO_EP2_ID)
assert tdata["DATA_TYPE"] == "transfer"
assert tdata["source_endpoint"] == GO_EP1_ID
assert tdata["destination_endpoint"] == GO_EP2_ID
assert tdata["submission_id"] is MISSING
assert "DATA" in tdata
assert len(tdata["DATA"]) == 0
def test_transfer_init_w_params():
# init with params
label = "label"
params = {"param1": "value1", "param2": "value2"}
tdata = TransferData(
GO_EP1_ID,
GO_EP2_ID,
label=label,
sync_level="exists",
additional_fields=params,
)
assert tdata["label"] == label
assert tdata["submission_id"] is MISSING
# sync_level of "exists" should be converted to 0
assert tdata["sync_level"] == 0
for par in params:
assert tdata[par] == params[par]
def test_transfer_add_item():
"""
Adds items to TransferData, verifies results
"""
tdata = TransferData(GO_EP1_ID, GO_EP2_ID)
# add item
source_path = "source/path/"
dest_path = "dest/path/"
tdata.add_item(source_path, dest_path)
# verify results
assert len(tdata["DATA"]) == 1
data = tdata["DATA"][0]
assert data["DATA_TYPE"] == "transfer_item"
assert data["source_path"] == source_path
assert data["destination_path"] == dest_path
assert data["recursive"] == MISSING
assert data["external_checksum"] == MISSING
assert data["checksum_algorithm"] == MISSING
# add recursive item
tdata.add_item(source_path, dest_path, recursive=True)
# verify results
assert len(tdata["DATA"]) == 2
r_data = tdata["DATA"][1]
assert r_data["DATA_TYPE"] == "transfer_item"
assert r_data["source_path"] == source_path
assert r_data["destination_path"] == dest_path
assert r_data["recursive"]
assert data["external_checksum"] == MISSING
assert data["checksum_algorithm"] == MISSING
# item with checksum
checksum = "d577273ff885c3f84dadb8578bb41399"
algorithm = "MD5"
tdata.add_item(
source_path, dest_path, external_checksum=checksum, checksum_algorithm=algorithm
)
assert len(tdata["DATA"]) == 3
c_data = tdata["DATA"][2]
assert c_data["DATA_TYPE"] == "transfer_item"
assert c_data["source_path"] == source_path
assert c_data["destination_path"] == dest_path
assert c_data["recursive"] == MISSING
assert c_data["external_checksum"] == checksum
assert c_data["checksum_algorithm"] == algorithm
# add an item which uses `additional_fields`
addfields = {"foo": "bar", "bar": "baz"}
tdata.add_item(source_path, dest_path, additional_fields=addfields)
assert len(tdata["DATA"]) == 4
fields_data = tdata["DATA"][3]
assert fields_data["DATA_TYPE"] == "transfer_item"
assert fields_data["source_path"] == source_path
assert fields_data["destination_path"] == dest_path
assert fields_data["recursive"] == MISSING
assert all(fields_data[k] == v for k, v in addfields.items())
@pytest.mark.parametrize(
"args, kwargs",
(
((GO_EP1_ID,), {}),
((), {"endpoint": GO_EP1_ID}),
),
)
def test_delete_data_noparams_init(args, kwargs):
# the minimal, required argument is the endpoint ID -- less than that results
# in a TypeError because the signature is not obeyed
ddata = DeleteData(*args, **kwargs)
assert ddata["DATA_TYPE"] == "delete"
assert ddata["endpoint"] == GO_EP1_ID
assert ddata["submission_id"] is MISSING
assert "DATA" in ddata
assert len(ddata["DATA"]) == 0
@pytest.mark.parametrize(
"add_kwargs",
(
{"recursive": True},
{"ignore_missing": True},
{"interpret_globs": True},
{"label": "somelabel"},
),
)
def test_delete_init_with_supported_parameters(add_kwargs):
ddata = DeleteData(GO_EP1_ID, **add_kwargs)
for k, v in add_kwargs.items():
assert ddata[k] == v
def test_delete_init_with_additional_fields():
params = {"param1": "value1", "param2": "value2"}
ddata = DeleteData(GO_EP1_ID, additional_fields=params)
assert ddata["param1"] == "value1"
assert ddata["param2"] == "value2"
def test_delete_add_item():
"""
Adds items to DeleteData, verifies results
"""
ddata = DeleteData(GO_EP1_ID)
# add normal item
path = "source/path/"
ddata.add_item(path)
assert len(ddata["DATA"]) == 1
data = ddata["DATA"][0]
assert data["DATA_TYPE"] == "delete_item"
assert data["path"] == path
# add an item which uses `additional_fields`
addfields = {"foo": "bar", "bar": "baz"}
ddata.add_item(path, additional_fields=addfields)
assert len(ddata["DATA"]) == 2
fields_data = ddata["DATA"][1]
assert fields_data["DATA_TYPE"] == "delete_item"
assert fields_data["path"] == path
assert all(fields_data[k] == v for k, v in addfields.items())
def test_delete_iter_items():
ddata = DeleteData(GO_EP1_ID)
# add item
ddata.add_item("abc/")
ddata.add_item("def/")
# order preserved, well-formed
as_list = list(ddata.iter_items())
assert len(as_list) == 2
def check_item(x, path):
assert isinstance(x, dict)
assert x.get("DATA_TYPE") == "delete_item"
assert x.get("path") == path
check_item(as_list[0], "abc/")
check_item(as_list[1], "def/")
def test_transfer_iter_items():
tdata = TransferData(GO_EP1_ID, GO_EP2_ID)
tdata.add_item("source/abc.txt", "dest/abc.txt")
tdata.add_item("source/def/", "dest/def/", recursive=True)
# order preserved, well-formed
as_list = list(tdata.iter_items())
assert len(as_list) == 2
def check_item(x, src, dst, params=None):
params = params or {}
assert isinstance(x, dict)
assert x.get("DATA_TYPE") == "transfer_item"
assert x.get("source_path") == src
assert x.get("destination_path") == dst
for k, v in params.items():
assert x.get(k) == v
check_item(as_list[0], "source/abc.txt", "dest/abc.txt")
check_item(as_list[1], "source/def/", "dest/def/", {"recursive": True})
@pytest.mark.parametrize("n_succeeded", [None, True, False])
@pytest.mark.parametrize("n_failed", [None, True, False])
@pytest.mark.parametrize("n_inactive", [None, True, False])
def test_notification_options(n_succeeded, n_failed, n_inactive):
notify_kwargs = {}
if n_succeeded is not None:
notify_kwargs["notify_on_succeeded"] = n_succeeded
if n_failed is not None:
notify_kwargs["notify_on_failed"] = n_failed
if n_inactive is not None:
notify_kwargs["notify_on_inactive"] = n_inactive
ddata = DeleteData(GO_EP1_ID, **notify_kwargs)
tdata = TransferData(GO_EP1_ID, GO_EP2_ID, **notify_kwargs)
def _default(x):
return x if x is not None else True
expect = {
"notify_on_succeeded": _default(n_succeeded),
"notify_on_failed": _default(n_failed),
"notify_on_inactive": _default(n_inactive),
}
for k, v in expect.items():
if k in notify_kwargs:
assert tdata[k] is v
assert ddata[k] is v
else:
assert tdata[k] is MISSING
assert ddata[k] is MISSING
@pytest.mark.parametrize(
"sync_level, result",
[
("exists", 0),
(0, 0),
("size", 1),
(1, 1),
("mtime", 2),
(2, 2),
("checksum", 3),
(3, 3),
("EXISTS", ValueError),
("hash", ValueError),
(100, 100),
],
)
def test_transfer_sync_levels_result(sync_level, result):
if isinstance(result, type) and issubclass(result, Exception):
with pytest.raises(result):
TransferData(GO_EP1_ID, GO_EP2_ID, sync_level=sync_level)
else:
tdata = TransferData(GO_EP1_ID, GO_EP2_ID, sync_level=sync_level)
assert tdata["sync_level"] == result
def test_add_filter_rule():
tdata = TransferData(GO_EP1_ID, GO_EP2_ID)
assert "filter_rules" not in tdata
tdata.add_filter_rule("*.tgz", type="file")
assert "filter_rules" in tdata
assert isinstance(tdata["filter_rules"], list)
assert len(tdata["filter_rules"]) == 1
assert tdata["filter_rules"][0]["DATA_TYPE"] == "filter_rule"
assert tdata["filter_rules"][0]["method"] == "exclude"
assert tdata["filter_rules"][0]["name"] == "*.tgz"
assert tdata["filter_rules"][0]["type"] == "file"
tdata.add_filter_rule("tmp")
assert len(tdata["filter_rules"]) == 2
assert tdata["filter_rules"][1]["DATA_TYPE"] == "filter_rule"
assert tdata["filter_rules"][1]["method"] == "exclude"
assert tdata["filter_rules"][1]["name"] == "tmp"
assert tdata["filter_rules"][1]["type"] == MISSING
@pytest.mark.parametrize(
"filter,expected",
[
("foo", "foo"),
({"foo": "bar", "a": ["b", "c"]}, "foo:bar/a:b,c"),
(
[
"foo",
{
"a": ["b", "c"],
},
],
["foo", "a:b,c"],
),
],
)
def test_ls_format_filter(filter, expected):
assert _format_filter(filter) == expected
|
