Skip to content

Commit

Permalink
Issue #678/#682 more tests for load_stac spatial_extent handling
Browse files Browse the repository at this point in the history
  • Loading branch information
@soxofaan
soxofaan committed Jan 17, 2025
1 parent 4a45519 commit dda8bcb
Showing 1 changed file with 217 additions and 40 deletions.
257 changes: 217 additions & 40 deletions tests/rest/test_connection.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,44 @@

BASIC_ENDPOINTS = [{"path": "/credentials/basic", "methods": ["GET"]}]


GEOJSON_POINT_01 = {"type": "Point", "coordinates": [3, 52]}
GEOJSON_LINESTRING_01 = {"type": "LineString", "coordinates": [[3, 50], [4, 51], [5, 53]]}
GEOJSON_POLYGON_01 = {
"type": "Polygon",
"coordinates": [[[3, 51], [4, 51], [4, 52], [3, 52], [3, 51]]],
}
GEOJSON_MULTIPOLYGON_01 = {
"type": "MultiPolygon",
"coordinates": [[[[3, 51], [4, 51], [4, 52], [3, 52], [3, 51]]]],
}
GEOJSON_FEATURE_01 = {
"type": "Feature",
"properties": {},
"geometry": GEOJSON_POLYGON_01,
}
GEOJSON_FEATURE_02 = {
"type": "Feature",
"properties": {},
"geometry": GEOJSON_MULTIPOLYGON_01,
}
GEOJSON_FEATURECOLLECTION_01 = {
"type": "FeatureCollection",
"features": [
GEOJSON_FEATURE_01,
GEOJSON_FEATURE_02,
],
}
GEOJSON_GEOMETRYCOLLECTION_01 = {
"type": "GeometryCollection",
"geometries": [
GEOJSON_POINT_01,
GEOJSON_POLYGON_01,
],
}

# Trick to avoid linting/auto-formatting tools to complain about or fix unused imports of these pytest fixtures
# TODO: use proper way to reuse fixtures instead of this hack
auth_config = auth_config
refresh_token_store = refresh_token_store

Expand Down Expand Up @@ -2427,42 +2464,6 @@ def test_load_collection_spatial_extent_bbox(self, dummy_backend):
"temporal_extent": None,
}

# TODO: make this more reusable
GEOJSON_POINT_01 = {"type": "Point", "coordinates": [3, 52]}
GEOJSON_LINESTRING_01 = {"type": "LineString", "coordinates": [[3, 50], [4, 51], [5, 53]]}
GEOJSON_POLYGON_01 = {
"type": "Polygon",
"coordinates": [[[3, 51], [4, 51], [4, 52], [3, 52], [3, 51]]],
}
GEOJSON_MULTIPOLYGON_01 = {
"type": "MultiPolygon",
"coordinates": [[[[3, 51], [4, 51], [4, 52], [3, 52], [3, 51]]]],
}
GEOJSON_FEATURE_01 = {
"type": "Feature",
"properties": {},
"geometry": GEOJSON_POLYGON_01,
}
GEOJSON_FEATURE_02 = {
"type": "Feature",
"properties": {},
"geometry": GEOJSON_MULTIPOLYGON_01,
}
GEOJSON_FEATURECOLLECTION_01 = {
"type": "FeatureCollection",
"features": [
GEOJSON_FEATURE_01,
GEOJSON_FEATURE_02,
],
}
GEOJSON_GEOMETRYCOLLECTION_01 = {
"type": "GeometryCollection",
"geometries": [
GEOJSON_POINT_01,
GEOJSON_POLYGON_01,
],
}

@pytest.mark.parametrize(
"spatial_extent",
[
Expand All @@ -2485,9 +2486,9 @@ def test_load_collection_spatial_extent_geojson(self, dummy_backend, spatial_ext
"spatial_extent",
[GEOJSON_POINT_01, GEOJSON_LINESTRING_01, GEOJSON_GEOMETRYCOLLECTION_01],
)
def test_load_collection_spatial_extent_geojson_wrong_type(self, con120, spatial_extent):
def test_load_collection_spatial_extent_geojson_wrong_type(self, dummy_backend, spatial_extent):
with pytest.raises(OpenEoClientException, match="Invalid geometry type"):
_ = con120.load_collection("S2", spatial_extent=spatial_extent)
_ = dummy_backend.connection.load_collection("S2", spatial_extent=spatial_extent)

@pytest.mark.parametrize(
"geojson",
Expand Down Expand Up @@ -2544,11 +2545,18 @@ def test_load_collection_spatial_extent_url(self, dummy_backend):
assert dummy_backend.get_sync_pg() == {
"loadurl1": {
"process_id": "load_url",
"arguments": {"url": "https://geo.test/geometry.json", "format": "GeoJSON"},
"arguments": {
"url": "https://geo.test/geometry.json",
"format": "GeoJSON",
},
},
"loadcollection1": {
"process_id": "load_collection",
"arguments": {"id": "S2", "spatial_extent": {"from_node": "loadurl1"}, "temporal_extent": None},
"arguments": {
"id": "S2",
"spatial_extent": {"from_node": "loadurl1"},
"temporal_extent": None,
},
"result": True,
},
}
Expand Down Expand Up @@ -2921,6 +2929,175 @@ def test_bands_parameterized(self, con120):
}
}

def test_load_stac_spatial_extent_bbox(self, dummy_backend):
spatial_extent = {"west": 1, "south": 2, "east": 3, "north": 4}
# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
cube = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=spatial_extent)
cube.execute()
assert dummy_backend.get_sync_pg()["loadstac1"]["arguments"] == {
"spatial_extent": {"east": 3, "north": 4, "south": 2, "west": 1},
"url": "https://stac.test/data",
}

@pytest.mark.parametrize(
"spatial_extent",
[
GEOJSON_POLYGON_01,
GEOJSON_MULTIPOLYGON_01,
GEOJSON_FEATURE_01,
GEOJSON_FEATURECOLLECTION_01,
],
)
def test_load_stac_spatial_extent_geojson(self, dummy_backend, spatial_extent):
# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
cube = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=spatial_extent)
cube.execute()
assert dummy_backend.get_sync_pg()["loadstac1"]["arguments"] == {
"spatial_extent": spatial_extent,
"url": "https://stac.test/data",
}

@pytest.mark.parametrize(
"spatial_extent",
[
GEOJSON_POINT_01,
GEOJSON_LINESTRING_01,
GEOJSON_GEOMETRYCOLLECTION_01,
],
)
def test_load_stac_spatial_extent_geojson_wrong_type(self, dummy_backend, spatial_extent):
# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
with pytest.raises(OpenEoClientException, match="Invalid geometry type"):
_ = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=spatial_extent)

@pytest.mark.parametrize(
"geojson",
[
GEOJSON_POLYGON_01,
GEOJSON_MULTIPOLYGON_01,
],
)
def test_load_stac_spatial_extent_shapely(self, dummy_backend, geojson):
spatial_extent = shapely.geometry.shape(geojson)
# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
cube = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=spatial_extent)
cube.execute()
assert dummy_backend.get_sync_pg()["loadstac1"]["arguments"] == {
"url": "https://stac.test/data",
"spatial_extent": geojson,
}

@pytest.mark.parametrize(
"geojson",
[
GEOJSON_POINT_01,
GEOJSON_GEOMETRYCOLLECTION_01,
],
)
def test_load_stac_spatial_extent_shapely_wront_type(self, dummy_backend, geojson):
spatial_extent = shapely.geometry.shape(geojson)
# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
with pytest.raises(OpenEoClientException, match="Invalid geometry type"):
_ = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=spatial_extent)

@pytest.mark.parametrize(
"geojson",
[
GEOJSON_MULTIPOLYGON_01,
GEOJSON_FEATURECOLLECTION_01,
],
)
@pytest.mark.parametrize("path_factory", [str, Path])
def test_load_stac_spatial_extent_path(self, geojson, dummy_backend, tmp_path, path_factory):
path = tmp_path / "geometry.json"
with path.open("w") as f:
json.dump(geojson, f)

# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
cube = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=path_factory(path))
cube.execute()
assert dummy_backend.get_sync_pg()["loadstac1"]["arguments"] == {
"url": "https://stac.test/data",
"spatial_extent": geojson,
}

def test_load_stac_spatial_extent_url(self, dummy_backend):
# TODO #694 how to avoid request to dummy STAC URL (without mocking, which is overkill for this test)
cube = dummy_backend.connection.load_stac(
"https://stac.test/data", spatial_extent="https://geo.test/geometry.json"
)
cube.execute()
assert dummy_backend.get_sync_pg() == {
"loadurl1": {
"process_id": "load_url",
"arguments": {
"url": "https://geo.test/geometry.json",
"format": "GeoJSON",
},
},
"loadstac1": {
"process_id": "load_stac",
"arguments": {
"url": "https://stac.test/data",
"spatial_extent": {"from_node": "loadurl1"},
},
"result": True,
},
}

@pytest.mark.parametrize(
"parameter",
[
Parameter("zpatial_extent"),
Parameter.spatial_extent("zpatial_extent"),
Parameter.geojson("zpatial_extent"),
],
)
def test_load_stac_spatial_extent_parameter(self, dummy_backend, parameter, recwarn):
cube = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=parameter)
assert len(recwarn) == 0

cube.execute()
assert dummy_backend.get_sync_pg()["loadstac1"]["arguments"] == {
"url": "https://stac.test/data",
"spatial_extent": {"from_parameter": "zpatial_extent"},
}

def test_load_stac_spatial_extent_parameter_schema_mismatch(self, dummy_backend, recwarn):
cube = dummy_backend.connection.load_stac(
"https://stac.test/data", spatial_extent=Parameter.number("zpatial_extent", description="foo")
)
assert [str(w.message) for w in recwarn] == [
"Schema mismatch with parameter given to `spatial_extent` in `load_stac`: expected a schema compatible with type 'object' but got {'type': 'number'}."
]

cube.execute()
assert dummy_backend.get_sync_pg()["loadstac1"]["arguments"] == {
"url": "https://stac.test/data",
"spatial_extent": {"from_parameter": "zpatial_extent"},
}

def test_load_stac_spatial_extent_vector_cube(self, dummy_backend):
vector_cube = VectorCube.load_url(
connection=dummy_backend.connection, url="https://geo.test/geometry.json", format="GeoJSON"
)
cube = dummy_backend.connection.load_stac("https://stac.test/data", spatial_extent=vector_cube)
cube.execute()
assert dummy_backend.get_sync_pg() == {
"loadurl1": {
"process_id": "load_url",
"arguments": {"format": "GeoJSON", "url": "https://geo.test/geometry.json"},
},
"loadstac1": {
"process_id": "load_stac",
"arguments": {
"url": "https://stac.test/data",
"spatial_extent": {"from_node": "loadurl1"},
},
"result": True,
},
}


@pytest.mark.parametrize(
"data",
Expand Down

0 comments on commit dda8bcb

Please sign in to comment.