Lint

Author: alukach

src/stac_auth_proxy/handlers/reverse_proxy.py

@@ -4,7 +4,7 @@
 import logging
 import time
 from dataclasses import dataclass
-from typing import Annotated, Any, Callable, Optional
+from typing import Annotated, Callable, Optional
 
 import httpx
 from cql2 import Expr
@@ -62,6 +62,7 @@ def _get_filter(self, path: str) -> Optional[Callable[..., Expr]]:
         for check, builder in endpoint_filters:
             if check(path):
                 return builder
+        return None
 
     async def proxy_request(self, request: Request, *, stream=False) -> httpx.Response:
         """Proxy a request to the upstream STAC API."""

tests/fixtures/__init__.py

@@ -5,12 +5,10 @@
 
 import cql2
 import pytest
-from httpx import Request
 from fastapi.testclient import TestClient
+from httpx import Request
 from utils import AppFactory
 
-from fixtures.demo_searches import SEARCHES
-
 app_factory = AppFactory(
     oidc_discovery_url="https://example-stac-api.com/.well-known/openid-configuration",
     default_public=False,
@@ -198,21 +196,33 @@
             "true",
             """{"op": "=", "args": [{"property": "private"}, true]}""",
         ],
-        # Not sure what is demonstrated in this...
-        # [
-        #     """{
-        #         "op": "and",
-        #         "args": [
-        #             { "op": "=", "args": [{ "property": "id" }, "LC08_L1TP_060247_20180905_20180912_01_T1_L1TP" ] },
-        #             { "op": "=", "args": [{ "property": "collection" }, "landsat8_l1tp"] }
-        #         ]
-        #     }"""
-        # ]
-        # * 3,
     ],
 )
 @pytest.mark.parametrize("is_authenticated", [True, False])
-@pytest.mark.parametrize("input_query", SEARCHES)
+@pytest.mark.parametrize(
+    "input_query",
+    [
+        # Not using filter
+        {
+            "collections": ["example-collection"],
+            "bbox": [-120.5, 35.7, -120.0, 36.0],
+            "datetime": "2021-06-01T00:00:00Z/2021-06-30T23:59:59Z",
+        },
+        # Using filter
+        {
+            "filter-lang": "cql2-json",
+            "filter": {
+                "op": "and",
+                "args": [
+                    {"op": "=", "args": [{"property": "collection"}, "landsat-8-l1"]},
+                    {"op": "<=", "args": [{"property": "eo:cloud_cover"}, 20]},
+                    {"op": "=", "args": [{"property": "platform"}, "landsat-8"]},
+                ],
+            },
+            "limit": 5,
+        },
+    ],
+)
 def test_search_post(
     mock_upstream,
     source_api_server,
@@ -223,6 +233,7 @@ def test_search_post(
     input_query,
     token_builder,
 ):
+    """Test filter is applied to search with full-featured filtering."""
     # Setup app
     app = app_factory(
         upstream_url=source_api_server,
@@ -246,19 +257,17 @@ def test_search_post(
     output_query = json.loads(r.read().decode())
 
     # Parse query from upstream
-    # input_filter_lang = input_query.get("filter-lang")
     input_filter = input_query.get("filter")
     expected_filter = auth_filter if is_authenticated else anon_filter
     expected_filter_exprs = [
         cql2.Expr(expr).to_text()
         for expr in [input_filter, expected_filter.strip()]
         if expr
     ]
-    expected_filter_out = cql2.Expr(" AND ".join(expected_filter_exprs)).to_json()
 
     expected_output_query = {
         **input_query,
-        "filter": expected_filter_out,
+        "filter": cql2.Expr(" AND ".join(expected_filter_exprs)).to_json(),
     }
 
     assert (