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
|
import json
import os
import tempfile
import unittest
from pathlib import Path
import pytest
import pystac
from pystac.stac_io import DefaultStacIO, DuplicateKeyReportingMixin, StacIO
from tests.utils import TestCases
def test_read_write_collection() -> None:
collection = pystac.read_file(
TestCases.get_path("data-files/collections/multi-extent.json")
)
with tempfile.TemporaryDirectory() as tmp_dir:
dest_href = os.path.join(tmp_dir, "collection.json")
pystac.write_file(collection, dest_href=dest_href)
assert os.path.exists(dest_href), "File was not written."
def test_read_write_collection_with_file_protocol() -> None:
collection = pystac.read_file(
"file://" + TestCases.get_path("data-files/collections/multi-extent.json")
)
with tempfile.TemporaryDirectory() as tmp_dir:
dest_href = os.path.join(tmp_dir, "collection.json")
pystac.write_file(collection, dest_href="file://" + dest_href)
assert os.path.exists(dest_href), "File was not written."
def test_read_item() -> None:
item = pystac.read_file(TestCases.get_path("data-files/item/sample-item.json"))
with tempfile.TemporaryDirectory() as tmp_dir:
dest_href = os.path.join(tmp_dir, "item.json")
pystac.write_file(item, dest_href=dest_href)
assert os.path.exists(dest_href), "File was not written."
def test_read_write_catalog() -> None:
catalog = pystac.read_file(
TestCases.get_path("data-files/catalogs/test-case-1/catalog.json")
)
with tempfile.TemporaryDirectory() as tmp_dir:
dest_href = os.path.join(tmp_dir, "catalog.json")
pystac.write_file(catalog, dest_href=dest_href)
assert os.path.exists(dest_href), "File was not written."
def test_read_item_collection_raises_exception() -> None:
with pytest.raises(pystac.STACTypeError):
_ = pystac.read_file(
TestCases.get_path("data-files/item-collection/sample-item-collection.json")
)
def test_read_item_dict() -> None:
stac_io = StacIO.default()
item_dict = stac_io.read_json(
TestCases.get_path("data-files/item/sample-item.json")
)
item = pystac.read_dict(item_dict)
assert isinstance(item, pystac.Item)
def test_read_collection_dict() -> None:
stac_io = StacIO.default()
collection_dict = stac_io.read_json(
TestCases.get_path("data-files/collections/multi-extent.json")
)
collection = pystac.read_dict(collection_dict)
assert isinstance(collection, pystac.Collection)
def test_read_catalog_dict() -> None:
stac_io = StacIO.default()
catalog_dict = stac_io.read_json(
TestCases.get_path("data-files/catalogs/test-case-1/catalog.json")
)
catalog = pystac.read_dict(catalog_dict)
assert isinstance(catalog, pystac.Catalog)
def test_read_from_stac_object() -> None:
catalog = pystac.STACObject.from_file(
TestCases.get_path("data-files/catalogs/test-case-1/catalog.json")
)
assert isinstance(catalog, pystac.Catalog)
def test_report_duplicate_keys() -> None:
# Directly from dict
class ReportingStacIO(DefaultStacIO, DuplicateKeyReportingMixin):
pass
stac_io = ReportingStacIO()
test_json = """{
"key": "value_1",
"key": "value_2"
}"""
with pytest.raises(pystac.DuplicateObjectKeyError) as excinfo:
stac_io.json_loads(test_json)
assert str(excinfo.value) == 'Found duplicate object name "key"'
# From file
with tempfile.TemporaryDirectory() as tmp_dir:
src_href = os.path.join(tmp_dir, "test.json")
with open(src_href, "w") as dst:
dst.write(test_json)
with pytest.raises(pystac.DuplicateObjectKeyError) as excinfo:
stac_io.read_json(src_href)
assert str(excinfo.value), f'Found duplicate object name "key" in {src_href}'
@unittest.mock.patch("pystac.stac_io.urlopen")
def test_headers_stac_io(urlopen_mock: unittest.mock.MagicMock) -> None:
stac_io = DefaultStacIO(headers={"Authorization": "api-key fake-api-key-value"})
catalog = pystac.Catalog("an-id", "a description").to_dict()
# required until https://github.com/stac-utils/pystac/pull/896 is merged
catalog["links"] = []
urlopen_mock.return_value.__enter__.return_value.read.return_value = json.dumps(
catalog
).encode("utf-8")
pystac.Catalog.from_file("https://example.com/catalog.json", stac_io=stac_io)
request_obj = urlopen_mock.call_args[0][0]
assert request_obj.headers == stac_io.headers
@pytest.mark.vcr()
def test_retry_stac_io() -> None:
# This doesn't test any retry behavior, but it does make sure that we can
# still read objects.
_ = pytest.importorskip("urllib3")
from pystac.stac_io import RetryStacIO
stac_io = RetryStacIO()
_ = stac_io.read_stac_object("https://planetarycomputer.microsoft.com/api/stac/v1")
@pytest.mark.vcr()
def test_retry_stac_io_404() -> None:
# This doesn't test any retry behavior, but it does make sure that we can
# error when an object doesn't exist.
_ = pytest.importorskip("urllib3")
from pystac.stac_io import RetryStacIO
stac_io = RetryStacIO()
with pytest.raises(Exception):
_ = stac_io.read_stac_object(
"https://planetarycomputer.microsoft.com"
"/api/stac/v1/collections/not-a-collection-id"
)
def test_save_http_href_errors(tmp_path: Path) -> None:
catalog = pystac.Catalog(id="test-catalog", description="")
catalog.set_self_href("http://pystac.test/catalog.json")
with pytest.raises(NotImplementedError):
catalog.save_object()
|
