Test alternate geometries in pgstac and sqlalchemy (#264)

* add point, line string tests to sqlalchemy

* add point, line string tests to pgstac

* run pre-commit

* change varianle name fromo poly to geom in sqlalchemy core

* run pre-commit

* lint

Co-authored-by: Rob Emanuele <[email protected]>
Co-authored-by: Nathan Zimmerman <[email protected]>

Author: 3 people

stac_fastapi/pgstac/tests/api/test_api.py

@@ -229,3 +229,47 @@ async def test_app_search_response(load_test_data, app_client, load_test_collect
     # stac_version and stac_extensions were removed in v1.0.0-beta.3
     assert resp_json.get("stac_version") is None
     assert resp_json.get("stac_extensions") is None
+
+
+@pytest.mark.asyncio
+async def test_search_point_intersects(
+    load_test_data, app_client, load_test_collection
+):
+    coll = load_test_collection
+    item = load_test_data("test_item.json")
+    resp = await app_client.post(f"/collections/{coll.id}/items", json=item)
+    assert resp.status_code == 200
+
+    point = [150.04, -33.14]
+    intersects = {"type": "Point", "coordinates": point}
+
+    params = {
+        "intersects": intersects,
+        "collections": [item["collection"]],
+    }
+    resp = await app_client.post("/search", json=params)
+    assert resp.status_code == 200
+    resp_json = resp.json()
+    assert len(resp_json["features"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_search_line_string_intersects(
+    load_test_data, app_client, load_test_collection
+):
+    coll = load_test_collection
+    item = load_test_data("test_item.json")
+    resp = await app_client.post(f"/collections/{coll.id}/items", json=item)
+    assert resp.status_code == 200
+
+    line = [[150.04, -33.14], [150.22, -33.89]]
+    intersects = {"type": "LineString", "coordinates": line}
+
+    params = {
+        "intersects": intersects,
+        "collections": [item["collection"]],
+    }
+    resp = await app_client.post("/search", json=params)
+    assert resp.status_code == 200
+    resp_json = resp.json()
+    assert len(resp_json["features"]) == 1

stac_fastapi/sqlalchemy/stac_fastapi/sqlalchemy/core.py

@@ -312,12 +312,12 @@ def post_search(
 
             else:
                 # Spatial query
-                poly = None
+                geom = None
                 if search_request.intersects is not None:
-                    poly = shape(search_request.intersects)
+                    geom = shape(search_request.intersects)
                 elif search_request.bbox:
                     if len(search_request.bbox) == 4:
-                        poly = ShapelyPolygon.from_bounds(*search_request.bbox)
+                        geom = ShapelyPolygon.from_bounds(*search_request.bbox)
                     elif len(search_request.bbox) == 6:
                         """Shapely doesn't support 3d bounding boxes we'll just use the 2d portion"""
                         bbox_2d = [
@@ -326,10 +326,10 @@ def post_search(
                             search_request.bbox[3],
                             search_request.bbox[4],
                         ]
-                        poly = ShapelyPolygon.from_bounds(*bbox_2d)
+                        geom = ShapelyPolygon.from_bounds(*bbox_2d)
 
-                if poly:
-                    filter_geom = ga.shape.from_shape(poly, srid=4326)
+                if geom:
+                    filter_geom = ga.shape.from_shape(geom, srid=4326)
                     query = query.filter(
                         ga.func.ST_Intersects(self.item_table.geometry, filter_geom)
                     )

stac_fastapi/sqlalchemy/tests/api/test_api.py

@@ -174,6 +174,23 @@ def test_search_invalid_date(load_test_data, app_client, postgres_transactions):
     assert resp.status_code == 400
 
 
+def test_search_point_intersects(load_test_data, app_client, postgres_transactions):
+    item = load_test_data("test_item.json")
+    postgres_transactions.create_item(item, request=MockStarletteRequest)
+
+    point = [150.04, -33.14]
+    intersects = {"type": "Point", "coordinates": point}
+
+    params = {
+        "intersects": intersects,
+        "collections": [item["collection"]],
+    }
+    resp = app_client.post("/search", json=params)
+    assert resp.status_code == 200
+    resp_json = resp.json()
+    assert len(resp_json["features"]) == 1
+
+
 def test_datetime_non_interval(load_test_data, app_client, postgres_transactions):
     item = load_test_data("test_item.json")
     postgres_transactions.create_item(item, request=MockStarletteRequest)
@@ -207,6 +224,24 @@ def test_bbox_3d(load_test_data, app_client, postgres_transactions):
     }
     resp = app_client.post("/search", json=params)
     assert resp.status_code == 200
+    resp_json = resp.json()
+    assert len(resp_json["features"]) == 1
+
+
+def test_search_line_string_intersects(
+    load_test_data, app_client, postgres_transactions
+):
+    item = load_test_data("test_item.json")
+    postgres_transactions.create_item(item, request=MockStarletteRequest)
 
+    line = [[150.04, -33.14], [150.22, -33.89]]
+    intersects = {"type": "LineString", "coordinates": line}
+
+    params = {
+        "intersects": intersects,
+        "collections": [item["collection"]],
+    }
+    resp = app_client.post("/search", json=params)
+    assert resp.status_code == 200
     resp_json = resp.json()
     assert len(resp_json["features"]) == 1