-
Notifications
You must be signed in to change notification settings - Fork 35
/
test_vectorcube.py
189 lines (161 loc) · 6.07 KB
/
test_vectorcube.py
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
from pathlib import Path
from typing import List
import pytest
from openeo.internal.graph_building import PGNode
from openeo.rest.vectorcube import VectorCube
@pytest.fixture
def vector_cube(con100) -> VectorCube:
pgnode = PGNode(process_id="create_vector_cube")
return VectorCube(graph=pgnode, connection=con100)
class DownloadSpy:
"""
Test helper to track download requests and optionally override next response to return.
"""
__slots__ = ["requests", "next_response"]
def __init__(self):
self.requests: List[dict] = []
self.next_response: bytes = b"Spy data"
@property
def only_request(self) -> dict:
"""Get progress graph of only request done"""
assert len(self.requests) == 1
return self.requests[-1]
@property
def last_request(self) -> dict:
"""Get last progress graph"""
assert len(self.requests) > 0
return self.requests[-1]
@pytest.fixture
def download_spy(requests_mock, con100) -> DownloadSpy:
"""Test fixture to spy on (and mock) `POST /result` (download) requests."""
spy = DownloadSpy()
def post_result(request, context):
pg = request.json()["process"]["process_graph"]
spy.requests.append(pg)
return spy.next_response
requests_mock.post(con100.build_url("/result"), content=post_result)
yield spy
def test_raster_to_vector(con100):
img = con100.load_collection("S2")
vector_cube = img.raster_to_vector()
vector_cube_tranformed = vector_cube.run_udf(udf="python source code", runtime="Python")
assert vector_cube_tranformed.flat_graph() == {
'loadcollection1': {
'arguments': {
'id': 'S2',
'spatial_extent': None,
'temporal_extent': None
},
'process_id': 'load_collection'
},
'rastertovector1': {
'arguments': {
'data': {'from_node': 'loadcollection1'}
},
'process_id': 'raster_to_vector'
},
'runudf1': {
'arguments': {
'data': {'from_node': 'rastertovector1'},
'runtime': 'Python',
'udf': 'python source code'
},
'process_id': 'run_udf',
'result': True}
}
@pytest.mark.parametrize(
["filename", "expected_format"],
[
("result.json", "JSON"), # TODO possible to allow "GeoJSON" with ".json" extension?
("result.geojson", "GeoJSON"),
("result.nc", "netCDF"),
],
)
@pytest.mark.parametrize("path_class", [str, Path])
def test_download_auto_save_result_only_file(
vector_cube, download_spy, tmp_path, filename, expected_format, path_class
):
output_path = tmp_path / filename
vector_cube.download(path_class(output_path))
assert download_spy.only_request == {
"createvectorcube1": {"process_id": "create_vector_cube", "arguments": {}},
"saveresult1": {
"process_id": "save_result",
"arguments": {
"data": {"from_node": "createvectorcube1"},
"format": expected_format,
"options": {},
},
"result": True,
},
}
assert output_path.read_bytes() == b"Spy data"
@pytest.mark.parametrize(
["filename", "format", "expected_format"],
[
("result.json", "JSON", "JSON"),
("result.geojson", "GeoJSON", "GeoJSON"),
("result.nc", "netCDF", "netCDF"),
("result.nc", "NETcDf", "NETcDf"), # TODO #401 normalize format
("result.nc", "inV6l1d!!!", "inV6l1d!!!"), # TODO #401 this should fail?
("result.json", None, "JSON"),
("result.geojson", None, "GeoJSON"),
("result.nc", None, "netCDF"),
# TODO #449 more formats to autodetect?
],
)
def test_download_auto_save_result_with_format(vector_cube, download_spy, tmp_path, filename, format, expected_format):
output_path = tmp_path / filename
vector_cube.download(output_path, format=format)
assert download_spy.only_request == {
"createvectorcube1": {"process_id": "create_vector_cube", "arguments": {}},
"saveresult1": {
"process_id": "save_result",
"arguments": {
"data": {"from_node": "createvectorcube1"},
"format": expected_format,
"options": {},
},
"result": True,
},
}
assert output_path.read_bytes() == b"Spy data"
def test_download_auto_save_result_with_options(vector_cube, download_spy, tmp_path):
output_path = tmp_path / "result.json"
vector_cube.download(output_path, format="GeoJSON", options={"precision": 7})
assert download_spy.only_request == {
"createvectorcube1": {"process_id": "create_vector_cube", "arguments": {}},
"saveresult1": {
"process_id": "save_result",
"arguments": {
"data": {"from_node": "createvectorcube1"},
"format": "GeoJSON",
"options": {"precision": 7},
},
"result": True,
},
}
assert output_path.read_bytes() == b"Spy data"
@pytest.mark.parametrize(
["output_file", "format", "expected_format"],
[
("result.geojson", None, "GeoJSON"),
("result.geojson", "GeoJSON", "GeoJSON"),
("result.json", "JSON", "JSON"),
("result.nc", "netCDF", "netCDF"),
],
)
def test_save_result_and_download(vector_cube, download_spy, tmp_path, output_file, format, expected_format):
"""e.g. https://github.com/Open-EO/openeo-geopyspark-driver/issues/477"""
vector_cube = vector_cube.save_result(format=format)
output_path = tmp_path / output_file
vector_cube.download(output_path)
assert download_spy.only_request == {
"createvectorcube1": {"process_id": "create_vector_cube", "arguments": {}},
"saveresult1": {
"process_id": "save_result",
"arguments": {"data": {"from_node": "createvectorcube1"}, "format": expected_format, "options": {}},
"result": True,
},
}
assert output_path.read_bytes() == b"Spy data"