Add search params tests (#11)

Author: ungarj

pyproject.toml

@@ -28,5 +28,6 @@ dev = [
     "pytest-cov>=7.0.0",
     "requests>=2.32.5",
     "ruff>=0.12.12",
+    "shapely>=2.1.2",
     "tqdm>=4.67.1",
 ]

src/stac_fastapi_pgstac_pair_search/models.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 import re
@@ -187,16 +188,25 @@ def end_date(self) -> Optional[dt]:
     @model_validator(mode="before")
     def validate_spatial(cls, values: Dict[str, Any]) -> Dict[str, Any]:
         for prefix in ["first", "second"]:
-            if (
-                values.get(f"{prefix}-intersects")
-                and values.get(f"{prefix}-bbox") is not None
-            ):
+            bbox = values.get(f"{prefix}-bbox")
+            intersects = values.get(f"{prefix}-intersects")
+            collections = values.get(f"{prefix}-collections")
+            ids = values.get(f"{prefix}-ids")
+            if bbox and intersects:
                 raise ValueError(
                     f"{prefix}-intersects and {prefix}-bbox parameters are mutually exclusive"
                 )
-            collections = values.get(f"{prefix}-collections")
+            if bbox:
+                if isinstance(bbox, str):
+                    values[f"{prefix}-bbox"] = list(map(float, bbox.split(",")))
+            if intersects:
+                if isinstance(intersects, str):
+                    values[f"{prefix}-intersects"] = json.loads(intersects)
             if isinstance(collections, str):
                 values[f"{prefix}-collections"] = collections.split(",")
+            if ids:
+                if isinstance(ids, str):
+                    values[f"{prefix}-ids"] = [ids]
         filter_expr = values.get("filter_expr")
         if filter_expr:
             try:

tests/requirements.txt

@@ -4,4 +4,5 @@ pytest>=8.4.2
 pytest-cov>=7.0.0
 requests>=2.32.5
 ruff>=0.12.12
+shapely>=2.1.2
 tqdm>=4.67.1

tests/test_app.py

@@ -1,6 +1,7 @@
 import json
 
 import pytest
+from shapely.geometry import box, mapping
 
 
 def test_app(client):
@@ -181,10 +182,213 @@ def test_pair_search_response(client, method: str, response_type: str):
             assert first != second
 
 
-# https://pair-search-demo.eox.at/catalogue/pair-search?
-#   response-type=pair&
-#   first-collection=ASA_IMS_1P&
-#   second-collection=ASA_IMS_1P&
-#   first-bbox=12.6,41.8,12.7,41.9&
-#   first-datetime=2003-01-01T00:00Z/2004-01-01T00:00Z&
-#   filter=((N_DIFF(second.oads:baseline_perpendicular_offset, first.oads:baseline_perpendicular_offset) BETWEEN-500 AND 500) AND (T_DIFF(T_START(first.datetime),T_START(seconds.datetime)) BETWEEN TimeDelta('0D') ANDTimeDelta('356D') AND (first.sat:orbit_state = second.sat:orbit_state)AND (first.sar:beam_id = second.sar:beam_id) AND(first.sar:polarization = second.sar:polarization) AND(first.oads:wrs_longitude_grid = second.oads:wrs_longitude_grid) AND(first.oads:wrs_latitude_grid = second.oads:wrs_latitude_grid) AND(first.oads:mission_phase = second.oads:mission_phase))
+# first-bbox, second-bbox
+@pytest.mark.parametrize("method", ["get", "post"])
+@pytest.mark.parametrize("first_bbox", [None, [6.21203, 43.36210, 6.35085, 43.48009]])
+@pytest.mark.parametrize("second_bbox", [None, [6.21203, 43.36210, 6.35085, 43.48009]])
+def test_bbox(client, method: str, first_bbox, second_bbox):
+    # bbox only intersects with product ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000
+    intersecting_id = "ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000"
+    params = {
+        "first-collections": ["ENVISAT.ASA.IMS_1P"],
+        "second-collections": ["ENVISAT.ASA.IMS_1P"],
+    }
+    if first_bbox:
+        params["first-bbox"] = ",".join(map(str, first_bbox))
+    if second_bbox:
+        params["second-bbox"] = ",".join(map(str, second_bbox))
+
+    response = client.request(
+        method=method,
+        url="/pair-search",
+        params=params if method == "get" else None,
+        content=json.dumps(params) if method == "post" else None,
+    )
+    if response.status_code != 200:
+        raise ValueError(f"Response: {response.status_code}, {response.text}")
+
+    response_json = response.json()
+
+    assert "numberPairsReturned" in response_json
+    assert "numberPairsMatched" in response_json
+
+    # special case: if both bboxes are set, no pairs are returned, because there is
+    # only one product
+    if first_bbox and second_bbox:
+        assert response_json["numberPairsReturned"] == 0
+        assert response_json["numberPairsMatched"] == 0
+        assert not response_json["featurePairs"]
+    else:
+        assert response_json["numberPairsReturned"] > 0
+        assert response_json["numberPairsMatched"] > 0
+        assert response_json["featurePairs"]
+        for first, second in response_json["featurePairs"]:
+            if first_bbox:
+                assert first == intersecting_id
+            if second_bbox:
+                assert second == intersecting_id
+
+
+# first-intersects, second-intersects
+@pytest.mark.parametrize("method", ["get", "post"])
+@pytest.mark.parametrize(
+    "first_intersects", [None, mapping(box(6.21203, 43.36210, 6.35085, 43.48009))]
+)
+@pytest.mark.parametrize(
+    "second_intersects", [None, mapping(box(6.21203, 43.36210, 6.35085, 43.48009))]
+)
+def test_intersects(client, method: str, first_intersects, second_intersects):
+    # bbox only intersects with product ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000
+    intersecting_id = "ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000"
+    params = {
+        "first-collections": ["ENVISAT.ASA.IMS_1P"],
+        "second-collections": ["ENVISAT.ASA.IMS_1P"],
+    }
+    if first_intersects:
+        params["first-intersects"] = json.dumps(first_intersects, separators=(",", ":"))
+    if second_intersects:
+        params["second-intersects"] = json.dumps(
+            second_intersects, separators=(",", ":")
+        )
+
+    response = client.request(
+        method=method,
+        url="/pair-search",
+        params=params if method == "get" else None,
+        content=json.dumps(params) if method == "post" else None,
+    )
+    if response.status_code != 200:
+        raise ValueError(f"Response: {response.status_code}, {response.text}")
+
+    response_json = response.json()
+
+    assert "numberPairsReturned" in response_json
+    assert "numberPairsMatched" in response_json
+
+    # special case: if both bboxes are set, no pairs are returned, because there is
+    # only one product
+    if first_intersects and second_intersects:
+        assert response_json["numberPairsReturned"] == 0
+        assert response_json["numberPairsMatched"] == 0
+        assert not response_json["featurePairs"]
+    else:
+        assert response_json["numberPairsReturned"] > 0
+        assert response_json["numberPairsMatched"] > 0
+        assert response_json["featurePairs"]
+        for first, second in response_json["featurePairs"]:
+            if first_intersects:
+                assert first == intersecting_id
+            if second_intersects:
+                assert second == intersecting_id
+
+
+# first-ids, second-ids
+@pytest.mark.parametrize("method", ["get", "post"])
+@pytest.mark.parametrize(
+    "first_ids",
+    [
+        None,
+        "ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000",
+        ["ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000"],
+    ],
+)
+@pytest.mark.parametrize(
+    "second_ids",
+    [
+        None,
+        "ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000",
+        ["ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000"],
+    ],
+)
+def test_ids(client, method: str, first_ids, second_ids):
+    intersecting_id = "ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000"
+    params = {
+        "first-collections": ["ENVISAT.ASA.IMS_1P"],
+        "second-collections": ["ENVISAT.ASA.IMS_1P"],
+    }
+    if first_ids:
+        params["first-ids"] = first_ids
+    if second_ids:
+        params["second-ids"] = second_ids
+
+    response = client.request(
+        method=method,
+        url="/pair-search",
+        params=params if method == "get" else None,
+        content=json.dumps(params) if method == "post" else None,
+    )
+    if response.status_code != 200:
+        raise ValueError(f"Response: {response.status_code}, {response.text}")
+
+    response_json = response.json()
+
+    assert "numberPairsReturned" in response_json
+    assert "numberPairsMatched" in response_json
+
+    # special case: if both bboxes are set, no pairs are returned, because there is
+    # only one product
+    if first_ids and second_ids:
+        assert response_json["numberPairsReturned"] == 0
+        assert response_json["numberPairsMatched"] == 0
+        assert not response_json["featurePairs"]
+    else:
+        assert response_json["numberPairsReturned"] > 0
+        assert response_json["numberPairsMatched"] > 0
+        assert response_json["featurePairs"]
+        for first, second in response_json["featurePairs"]:
+            if first_ids:
+                assert first == intersecting_id
+            if second_ids:
+                assert second == intersecting_id
+
+
+# first-datetime, second-datetime
+@pytest.mark.parametrize("method", ["get", "post"])
+@pytest.mark.parametrize(
+    "first_datetime",
+    [None, "2010-06-02T09:49:53.149000Z", "2010-06-02T00:00:00Z/2010-06-02T23:59:59Z"],
+)
+@pytest.mark.parametrize(
+    "second_datetime",
+    [None, "2010-06-02T09:49:53.149000Z", "2010-06-02T00:00:00Z/2010-06-02T23:59:59Z"],
+)
+def test_datetime(client, method: str, first_datetime, second_datetime):
+    # bbox only intersects with product ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000
+    intersecting_id = "ASA_IMS_1PNESA20100602_094953_000000152090_00022_43162_0000"
+    params = {
+        "first-collections": ["ENVISAT.ASA.IMS_1P"],
+        "second-collections": ["ENVISAT.ASA.IMS_1P"],
+    }
+    if first_datetime:
+        params["first-datetime"] = first_datetime
+    if second_datetime:
+        params["second-datetime"] = second_datetime
+
+    response = client.request(
+        method=method,
+        url="/pair-search",
+        params=params if method == "get" else None,
+        content=json.dumps(params) if method == "post" else None,
+    )
+    if response.status_code != 200:
+        raise ValueError(f"Response: {response.status_code}, {response.text}")
+
+    response_json = response.json()
+    assert "numberPairsReturned" in response_json
+    assert "numberPairsMatched" in response_json
+
+    # special case: if both bboxes are set, no pairs are returned, because there is
+    # only one product
+    if first_datetime and second_datetime:
+        assert response_json["numberPairsReturned"] == 0
+        assert response_json["numberPairsMatched"] == 0
+        assert not response_json["featurePairs"]
+    else:
+        assert response_json["numberPairsReturned"] > 0
+        assert response_json["numberPairsMatched"] > 0
+        assert response_json["featurePairs"]
+        for first, second in response_json["featurePairs"]:
+            if first_datetime:
+                assert first == intersecting_id
+            if second_datetime:
+                assert second == intersecting_id