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
|
import json
import pytest
from responses import matchers
from globus_sdk import MISSING, FlowsAPIError
from globus_sdk.testing import get_last_request, load_response
from globus_sdk.testing.models import RegisteredResponse
@pytest.mark.parametrize("subscription_id", [MISSING, None, "dummy_subscription_id"])
def test_create_flow(flows_client, subscription_id):
metadata = load_response(flows_client.create_flow).metadata
resp = flows_client.create_flow(
**metadata["params"], subscription_id=subscription_id
)
assert resp.data["title"] == "Multi Step Transfer"
last_req = get_last_request()
req_body = json.loads(last_req.body)
if subscription_id is not MISSING:
assert req_body["subscription_id"] == subscription_id
else:
assert "subscription_id" not in req_body
@pytest.mark.parametrize("value", [MISSING, [], ["dummy_value"]])
@pytest.mark.parametrize("key", ["run_managers", "run_monitors"])
def test_create_flow_run_role_serialization(flows_client, key, value):
request_body = {
"title": "Multi Step Transfer",
"definition": {
"StartAt": "Transfer1",
"States": {
"Transfer1": {"Type": "Pass", "End": True},
},
},
"input_schema": {},
}
if value is not MISSING:
request_body[key] = value
load_response(
RegisteredResponse(
service="flows",
path="/flows",
method="POST",
json=request_body,
match=[
matchers.json_params_matcher(
params={
"title": request_body["title"],
"definition": request_body["definition"],
"input_schema": request_body["input_schema"],
},
strict_match=False,
)
],
)
)
flows_client.create_flow(**request_body)
last_req = get_last_request()
req_body = json.loads(last_req.body)
if value is MISSING:
assert key not in req_body
else:
assert req_body[key] == value
def test_create_flow_error_parsing(flows_client):
metadata = load_response(
flows_client.create_flow, case="bad_admin_principal_error"
).metadata
with pytest.raises(FlowsAPIError) as excinfo:
flows_client.create_flow(**metadata["params"])
err = excinfo.value
assert err.code == "UNPROCESSABLE_ENTITY"
assert err.messages == [
(
'Unrecognized principal string: "ae341a98-d274-11e5-b888-dbae3a8ba545". '
'Allowed principal types in role "FlowAdministrator": '
"[<AuthGroupUrn>, <AuthIdentityUrn>]"
),
(
'Unrecognized principal string: "4fab4345-6d20-43a0-9a25-16b2e3bbe765". '
'Allowed principal types in role "FlowAdministrator": '
"[<AuthGroupUrn>, <AuthIdentityUrn>]"
),
]
def test_get_flow(flows_client):
meta = load_response(flows_client.get_flow).metadata
resp = flows_client.get_flow(meta["flow_id"])
assert resp.data["title"] == meta["title"]
def test_update_flow(flows_client):
meta = load_response(flows_client.update_flow).metadata
resp = flows_client.update_flow(meta["flow_id"], **meta["params"])
for k, v in meta["params"].items():
assert k in resp
assert resp[k] == v
@pytest.mark.parametrize("value", [MISSING, [], ["dummy_value"]])
@pytest.mark.parametrize("key", ["run_managers", "run_monitors"])
def test_update_flow_run_role_serialization(flows_client, key, value):
metadata = load_response(flows_client.update_flow).metadata
params = {**metadata["params"], key: value}
flows_client.update_flow(metadata["flow_id"], **params)
last_req = get_last_request()
req_body = json.loads(last_req.body)
if value is MISSING:
assert key not in req_body
else:
assert req_body[key] == value
def test_delete_flow(flows_client):
metadata = load_response(flows_client.delete_flow).metadata
resp = flows_client.delete_flow(metadata["flow_id"])
assert resp.data["title"] == "Multi Step Transfer"
assert resp.data["DELETED"] is True
|
