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
|
import json
import urllib.parse
import uuid
import pytest
import responses
import globus_sdk
from globus_sdk._missing import filter_missing
from globus_sdk.testing import get_last_request, load_response
from tests.common import register_api_route_fixture_file
@pytest.fixture
def search_client():
client = globus_sdk.SearchClient()
with client.retry_config.tune(max_retries=0):
yield client
def test_search_query_simple(search_client):
meta = load_response(search_client.search).metadata
res = search_client.search(meta["index_id"], q="foo")
assert res.http_status == 200
data = res.data
assert isinstance(data, dict)
assert data["gmeta"][0]["entries"][0]["content"]["foo"] == "bar"
req = get_last_request()
assert req.body is None
parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
assert parsed_qs == {"q": ["foo"]}
@pytest.mark.parametrize("query_doc", [{"q": "foo"}, {"q": "foo", "limit": 10}])
def test_search_post_query_simple(search_client, query_doc):
meta = load_response(search_client.post_search).metadata
res = search_client.post_search(meta["index_id"], query_doc)
assert res.http_status == 200
data = res.data
assert isinstance(data, dict)
assert data["gmeta"][0]["entries"][0]["content"]["foo"] == "bar"
req = get_last_request()
assert req.body is not None
req_body = json.loads(req.body)
assert req_body == dict(query_doc)
def test_search_post_query_simple_with_v1_helper(search_client):
query_doc = globus_sdk.SearchQueryV1(q="foo")
meta = load_response(search_client.post_search).metadata
res = search_client.post_search(meta["index_id"], query_doc)
assert res.http_status == 200
data = res.data
assert isinstance(data, dict)
assert data["gmeta"][0]["entries"][0]["content"]["foo"] == "bar"
req = get_last_request()
assert req.body is not None
req_body = json.loads(req.body)
assert req_body == {"@version": "query#1.0.0", "q": "foo"}
@pytest.mark.parametrize("doc_type", ("dict", "helper"))
def test_search_post_query_arg_overrides(search_client, doc_type):
meta = load_response(search_client.post_search).metadata
if doc_type == "dict":
query_doc = {"q": "foo", "limit": 10, "offset": 0}
elif doc_type == "helper":
query_doc = globus_sdk.SearchQueryV1(q="foo", limit=10, offset=0)
else:
raise NotImplementedError(doc_type)
res = search_client.post_search(meta["index_id"], query_doc, limit=100, offset=150)
assert res.http_status == 200
data = res.data
assert isinstance(data, dict)
assert data["gmeta"][0]["entries"][0]["content"]["foo"] == "bar"
req = get_last_request()
assert req.body is not None
req_body = json.loads(req.body)
assert req_body != dict(query_doc)
assert req_body["q"] == query_doc["q"]
assert req_body["limit"] == 100
assert req_body["offset"] == 150
# important! these should be unchanged (no side-effects)
assert query_doc["limit"] == 10
assert query_doc["offset"] == 0
@pytest.mark.parametrize(
"query_doc",
[
{"q": "foo"},
globus_sdk.SearchScrollQuery("foo"),
],
)
def test_search_paginated_scroll_query(search_client, query_doc):
index_id = str(uuid.uuid1())
register_api_route_fixture_file(
"search",
f"/v1/index/{index_id}/scroll",
"scroll_result_1.json",
method="POST",
match=[responses.matchers.json_params_matcher({"q": "foo"})],
)
register_api_route_fixture_file(
"search",
f"/v1/index/{index_id}/scroll",
"scroll_result_2.json",
method="POST",
match=[
responses.matchers.json_params_matcher(
{"q": "foo", "marker": "3d34900e3e4211ebb0a806b2af333354"}
)
],
)
data = list(search_client.paginated.scroll(index_id, query_doc).items())
assert len(responses.calls) == 2
assert len(data) == 2
assert isinstance(data[0], dict)
assert data[0]["entries"][0]["content"]["foo"] == "bar"
assert isinstance(data[1], dict)
assert data[1]["entries"][0]["content"]["foo"] == "baz"
# confirm that pagination was not side-effecting
assert "marker" not in filter_missing(query_doc)
|
