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
|
import urllib.parse
import pytest
from globus_sdk import MISSING
from globus_sdk.testing import get_last_request, load_response
@pytest.mark.parametrize("filter_fulltext", [MISSING, "foo"])
@pytest.mark.parametrize("filter_roles", [MISSING, "bar"])
@pytest.mark.parametrize("orderby", [MISSING, "created_at ASC"])
def test_list_flows_simple(flows_client, filter_fulltext, filter_roles, orderby):
meta = load_response(flows_client.list_flows).metadata
add_kwargs = {}
if filter_fulltext:
add_kwargs["filter_fulltext"] = filter_fulltext
if filter_roles:
add_kwargs["filter_roles"] = filter_roles
if orderby:
add_kwargs["orderby"] = orderby
res = flows_client.list_flows(**add_kwargs)
assert res.http_status == 200
# dict-like indexing
assert meta["first_flow_id"] == res["flows"][0]["id"]
# list conversion (using __iter__) and indexing
assert meta["first_flow_id"] == list(res)[0]["id"]
req = get_last_request()
assert req.body is None
parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
expect_query_params = {
k: [v]
for k, v in (
("filter_fulltext", filter_fulltext),
("filter_roles", filter_roles),
("orderby", orderby),
)
if v is not MISSING
}
assert parsed_qs == expect_query_params
@pytest.mark.parametrize("by_pages", [True, False])
def test_list_flows_paginated(flows_client, by_pages):
meta = load_response(flows_client.list_flows, case="paginated").metadata
total_items = meta["total_items"]
num_pages = meta["num_pages"]
expect_markers = meta["expect_markers"]
res = flows_client.paginated.list_flows()
if by_pages:
pages = list(res)
assert len(pages) == num_pages
for i, page in enumerate(pages):
assert page["marker"] == expect_markers[i]
if i < num_pages - 1:
assert page["has_next_page"] is True
else:
assert page["has_next_page"] is False
else:
items = list(res.items())
assert len(items) == total_items
@pytest.mark.parametrize(
"orderby_style, orderby_value",
[
# single str -> single param
("str", "created_at ASC"),
# sequence of strs (tuple, list) -> multi param
("seq", ["created_at ASC", "updated_at DESC"]),
("seq", ("created_at ASC", "updated_at DESC")),
# more complex cases to handle within the test: generators and sets
("generator", ("created_at ASC", "updated_at DESC")),
("set", ("created_at ASC", "updated_at DESC")),
],
)
def test_list_flows_orderby_multi(flows_client, orderby_style, orderby_value):
meta = load_response(flows_client.list_flows).metadata
if orderby_style == "str":
orderby = orderby_value
expected_orderby_value = [orderby_value]
elif orderby_style == "seq":
orderby = orderby_value
expected_orderby_value = list(orderby_value)
elif orderby_style == "generator":
orderby = (x for x in orderby_value)
expected_orderby_value = list(orderby_value)
elif orderby_style == "set":
orderby = set(orderby_value)
expected_orderby_value = set(orderby_value)
else:
raise NotImplementedError
res = flows_client.list_flows(orderby=orderby)
assert res.http_status == 200
# check result correctness
assert meta["first_flow_id"] == res["flows"][0]["id"]
req = get_last_request()
assert req.body is None
parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
# for the set case, handle arbitrary ordering by converting the value seen
# back to a set
if isinstance(expected_orderby_value, set):
assert set(parsed_qs["orderby"]) == expected_orderby_value
else:
assert parsed_qs["orderby"] == expected_orderby_value
@pytest.mark.parametrize(
"filter_roles, expected_filter_roles",
[
# empty list/tuple, list/tuple with empty string, and None do not send the param
((), None),
([], None),
([""], None),
(("",), None),
(MISSING, None),
# single role as string
("foo", ["foo"]),
# single role as list/tuple
(["foo"], ["foo"]),
(("foo",), ["foo"]),
# multiple roles as comma-separated string
("foo,bar", ["foo,bar"]),
# multiple roles as list/tuple
(["foo", "bar"], ["foo,bar"]),
(("foo", "bar"), ["foo,bar"]),
],
)
def test_list_flows_with_filter_roles_parameter(
flows_client, filter_roles, expected_filter_roles
):
load_response(flows_client.list_flows).metadata
res = flows_client.list_flows(filter_roles=filter_roles)
assert res.http_status == 200
req = get_last_request()
assert req.body is None
parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
# Only check filter_roles in parsed_qs if we expect a non-empty value
if expected_filter_roles:
assert parsed_qs["filter_roles"] == expected_filter_roles
else:
assert "filter_roles" not in parsed_qs
|
