File: test_cli.py

package info (click to toggle)
pystac-client 0.9.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 47,416 kB
  • sloc: python: 4,652; sh: 74; makefile: 60
file content (320 lines) | stat: -rw-r--r-- 10,249 bytes parent folder | download
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
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
import json
import tempfile

import pytest
from pytest_console_scripts import ScriptRunner

import pystac_client.cli
from tests.helpers import STAC_URLS, TEST_DATA

FILTER_JSON = {"op": "lte", "args": [{"property": "eo:cloud_cover"}, 40]}


# We want to ensure that the CLI is handling these warnings properly
@pytest.mark.filterwarnings("ignore::pystac_client.warnings.PystacClientWarning")
class TestCLISearch:
    @pytest.mark.vcr
    def test_item_search(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["PLANETARY-COMPUTER"],
            "-c",
            "naip",
            "--max-items",
            "20",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success

    @pytest.mark.vcr
    def test_filter(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["EARTH-SEARCH"],
            "--filter",
            json.dumps(FILTER_JSON),
            "--max-items",
            "20",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success, result.stderr
        assert result.stdout

    @pytest.mark.vcr
    @pytest.mark.parametrize("filename", ["netherlands_aoi.json", "sample-item.json"])
    def test_intersects(self, script_runner: ScriptRunner, filename: str) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["PLANETARY-COMPUTER"],
            "--collections",
            "landsat-8-c2-l2",
            "--intersects",
            str(TEST_DATA / filename),
            "--max-items",
            "10",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success, result.stderr
        assert result.stdout

    @pytest.mark.vcr
    def test_intersects_despite_warning(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["PLANETARY-COMPUTER"],
            "--collections",
            "landsat-8-c2-l2",
            "--intersects",
            str(TEST_DATA / "sample-item-collection.json"),
            "--max-items",
            "10",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success
        assert result.stdout
        assert "input to intersects is a FeatureCollection" in result.stderr

    @pytest.mark.parametrize(
        "headers,good_header_count",
        [
            (["kick=flip", "home=run"], 2),
            (["mad=pow"], 1),
            (["=no-var"], 0),
            (["no-val="], 0),
            (["good=header", "bad-header"], 1),
            (["header=value-with-three-=-signs-=", "plain=jane"], 2),
        ],
    )
    def test_headers(self, headers: list[str], good_header_count: int) -> None:
        args = [
            "search",
            STAC_URLS["PLANETARY-COMPUTER"],
            "-c",
            "naip",
            "--max-items",
            "20",
            "--headers",
        ] + headers
        pargs = pystac_client.cli.parse_args(args)
        assert len(pargs["headers"]) == good_header_count

    def test_no_arguments(self, script_runner: ScriptRunner) -> None:
        args = ["stac-client"]
        result = script_runner.run(*args, print_result=False)
        assert not result.success
        assert result.returncode == 1

    @pytest.mark.vcr
    def test_non_conformant_raises_by_default(
        self, script_runner: ScriptRunner
    ) -> None:
        args = [
            "stac-client",
            "search",
            "https://earth-search.aws.element84.com/v0",
            "-c",
            "sentinel-s2-l2a-cogs",
            "--matched",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success is False
        assert "Server does not conform to ITEM_SEARCH" in result.stderr
        assert result.returncode == 1

    @pytest.mark.vcr
    @pytest.mark.parametrize("warning_flag", ["--error", "--error=no-conforms-to"])
    def test_non_conformant_raises_if_warning_set_to_error(
        self, script_runner: ScriptRunner, warning_flag: str
    ) -> None:
        args = [
            "stac-client",
            "search",
            "https://earth-search.aws.element84.com/v0",
            "-c",
            "sentinel-s2-l2a-cogs",
            warning_flag,
            "--matched",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success is False
        assert "Server does not advertise any conformance classes" in result.stderr
        assert result.returncode == 1

    @pytest.mark.vcr
    def test_non_conformant_can_be_fixed(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            "https://earth-search.aws.element84.com/v0",
            "-c",
            "sentinel-s2-l2a-cogs",
            "--add-conforms-to=ITEM_SEARCH",
            "--matched",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success

    @pytest.mark.vcr
    @pytest.mark.parametrize("warning_flag", ["--ignore", "--ignore=no-conforms-to"])
    def test_non_conformant_can_be_ignored(
        self, script_runner: ScriptRunner, warning_flag: str
    ) -> None:
        args = [
            "stac-client",
            "search",
            "https://earth-search.aws.element84.com/v0",
            "-c",
            "sentinel-s2-l2a-cogs",
            warning_flag,
            "--add-conforms-to=ITEM_SEARCH",
            "--matched",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success

    @pytest.mark.vcr
    @pytest.mark.parametrize(
        "conforms_to_flag", ["--clear-conforms-to", "--remove-conforms-to=ITEM_SEARCH"]
    )
    def test_altering_conforms_to(
        self, script_runner: ScriptRunner, conforms_to_flag: str
    ) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["EARTH-SEARCH"],
            conforms_to_flag,
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success is False
        assert "Server does not conform to ITEM_SEARCH" in result.stderr
        assert result.returncode == 1

    @pytest.mark.vcr
    @pytest.mark.filterwarnings("ignore::Warning")
    def test_matched_not_available(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["PLANETARY-COMPUTER"],
            "-c",
            "naip",
            "--matched",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success is False
        assert "'matched' is not supported for this catalog" in result.stderr
        assert result.returncode == 1

    @pytest.mark.vcr
    def test_matched(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["EARTH-SEARCH"],
            "-c",
            "cop-dem-glo-30",
            "--max-items",
            "1",
            "--matched",
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success
        assert result.stdout[0].isdigit(), "Output does not start with a number"

    @pytest.mark.vcr
    def test_fields(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "search",
            STAC_URLS["EARTH-SEARCH"],
            "-c",
            "cop-dem-glo-30",
            "--max-items",
            "1",
            "--fields=-geometry,-assets",
        ]
        result = script_runner.run(args, print_result=True)
        assert result.success
        assert "assets" not in result.stdout
        assert "geometry" not in result.stdout

    @pytest.mark.vcr
    def test_save(self, script_runner: ScriptRunner) -> None:
        with tempfile.NamedTemporaryFile() as fp:
            path = fp.name
        args = [
            "stac-client",
            "search",
            STAC_URLS["EARTH-SEARCH"],
            "-c",
            "cop-dem-glo-30",
            "--max-items",
            "1",
            "--save",
            path,
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success

        with open(path) as f:
            output = json.loads(f.read())

        assert "features" in output
        assert len(output["features"]) == 1


@pytest.mark.filterwarnings("ignore::pystac_client.warnings.PystacClientWarning")
class TestCLICollections:
    @pytest.mark.vcr
    def test_collections(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "collections",
            STAC_URLS["EARTH-SEARCH"],
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success
        assert result.stdout.startswith('[{"type": "Collection"')

    @pytest.mark.vcr
    def test_collection_search(self, script_runner: ScriptRunner) -> None:
        args = [
            "stac-client",
            "collections",
            STAC_URLS["EARTH-SEARCH"],
            "--q",
            "sentinel",
        ]
        with pytest.warns(UserWarning, match="COLLECTION_SEARCH"):
            result = script_runner.run(args, print_result=False)

        assert result.success

        collections = json.loads(result.stdout)
        assert len(collections) == 5

    @pytest.mark.vcr
    def test_save(self, script_runner: ScriptRunner) -> None:
        with tempfile.NamedTemporaryFile() as fp:
            path = fp.name
        args = [
            "stac-client",
            "collections",
            STAC_URLS["EARTH-SEARCH"],
            "--save",
            path,
        ]
        result = script_runner.run(args, print_result=False)
        assert result.success

        with open(path) as f:
            output = json.loads(f.read())

        assert isinstance(output, list)
        assert len(output) == 9, "earth-search does not have 9 collections"
        assert all(c["type"] == "Collection" for c in output)