diff --git a/tests/rest/test_connection.py b/tests/rest/test_connection.py index 767b6bfe1..408d79bd0 100644 --- a/tests/rest/test_connection.py +++ b/tests/rest/test_connection.py @@ -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 @@ -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", [ @@ -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", @@ -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, }, } @@ -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",