Continue test buildout

Author: alukach

tests/conftest.py

@@ -3,8 +3,8 @@
 import json
 import os
 import threading
-from typing import Any
-from unittest.mock import MagicMock, patch
+from typing import Any, Generator
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 import uvicorn
@@ -140,3 +140,13 @@ def mock_env():
     """Clear environment variables to avoid poluting configs from runtime env."""
     with patch.dict(os.environ, clear=True):
         yield
+
+
+@pytest.fixture
+def mock_upstream() -> Generator[MagicMock, None, None]:
+    """Mock the HTTPX send method. Useful when we want to inspect the request is sent to upstream API."""
+    with patch(
+        "stac_auth_proxy.handlers.reverse_proxy.httpx.AsyncClient.send",
+        new_callable=AsyncMock,
+    ) as mock_send_method:
+        yield mock_send_method

tests/test_filters_jinja2.py

@@ -1,52 +1,26 @@
 """Tests for Jinja2 CQL2 filter."""
 
-from dataclasses import dataclass
-from typing import Generator
-from unittest.mock import AsyncMock, MagicMock, patch
 from urllib.parse import parse_qs
 
 import httpx
 import pytest
 from fastapi.testclient import TestClient
 from utils import AppFactory
 
+from tests.utils import single_chunk_async_stream_response
+
 app_factory = AppFactory(
     oidc_discovery_url="https://example-stac-api.com/.well-known/openid-configuration",
     default_public=False,
 )
 
 
-@pytest.fixture
-def mock_send() -> Generator[MagicMock, None, None]:
-    """Mock the HTTPX send method. Useful when we want to inspect the request is sent to upstream API."""
-    with patch(
-        "stac_auth_proxy.handlers.reverse_proxy.httpx.AsyncClient.send",
-        new_callable=AsyncMock,
-    ) as mock_send_method:
-        yield mock_send_method
-
-
-@dataclass
-class SingleChunkAsyncStream(httpx.AsyncByteStream):
-    """Mock async stream that returns a single chunk of data."""
-
-    body: bytes
-
-    async def __aiter__(self):
-        """Return a single chunk of data."""
-        yield self.body
-
-
 def test_collections_filter_contained_by_token(
-    mock_send, source_api_server, token_builder
+    mock_upstream, source_api_server, token_builder
 ):
     """Test that the collections filter is applied correctly."""
     # Mock response from upstream API
-    mock_send.return_value = httpx.Response(
-        200,
-        stream=SingleChunkAsyncStream(b"{}"),
-        headers={"content-type": "application/json"},
-    )
+    mock_upstream.return_value = single_chunk_async_stream_response(b"{}")
 
     app = app_factory(
         upstream_url=source_api_server,
@@ -59,15 +33,49 @@ def test_collections_filter_contained_by_token(
     )
 
     auth_token = token_builder({"collections": ["foo", "bar"]})
-    client = TestClient(
-        app,
-        headers={"Authorization": f"Bearer {auth_token}"},
-    )
-
+    client = TestClient(app, headers={"Authorization": f"Bearer {auth_token}"})
     response = client.get("/collections")
+
     assert response.status_code == 200
-    assert mock_send.call_count == 1
-    [r] = mock_send.call_args[0]
+    assert mock_upstream.call_count == 1
+    [r] = mock_upstream.call_args[0]
     assert parse_qs(r.url.query.decode()) == {
         "filter": ["a_containedby(id, ('foo', 'bar'))"]
     }
+
+
+@pytest.mark.parametrize(
+    "authenticated, expected_filter",
+    [
+        (True, "true"),
+        (False, "(private = false)"),
+    ],
+)
+def test_collections_filter_private_and_public(
+    mock_upstream, source_api_server, token_builder, authenticated, expected_filter
+):
+    """Test that filter can be used for private/public collections."""
+    # Mock response from upstream API
+    mock_upstream.return_value = single_chunk_async_stream_response(b"{}")
+
+    app = app_factory(
+        upstream_url=source_api_server,
+        collections_filter={
+            "cls": "stac_auth_proxy.filters.Template",
+            "args": ["{{ '(private = false)' if token is none else true }}"],
+        },
+        default_public=True,
+    )
+
+    client = TestClient(
+        app,
+        headers=(
+            {"Authorization": f"Bearer {token_builder({})}"} if authenticated else {}
+        ),
+    )
+    response = client.get("/collections")
+
+    assert response.status_code == 200
+    assert mock_upstream.call_count == 1
+    [r] = mock_upstream.call_args[0]
+    assert parse_qs(r.url.query.decode()) == {"filter": [expected_filter]}

tests/utils.py

@@ -1,7 +1,10 @@
 """Utilities for testing."""
 
+from dataclasses import dataclass
 from typing import Callable
 
+import httpx
+
 from stac_auth_proxy import Settings, create_app
 
 
@@ -23,3 +26,25 @@ def __call__(self, *, upstream_url, **overrides) -> Callable:
                 },
             )
         )
+
+
+@dataclass
+class SingleChunkAsyncStream(httpx.AsyncByteStream):
+    """Mock async stream that returns a single chunk of data."""
+
+    body: bytes
+
+    async def __aiter__(self):
+        """Return a single chunk of data."""
+        yield self.body
+
+
+def single_chunk_async_stream_response(
+    body: bytes, status_code=200, headers={"content-type": "application/json"}
+):
+    """Create a response with a single chunk of data."""
+    return httpx.Response(
+        stream=SingleChunkAsyncStream(body),
+        status_code=status_code,
+        headers=headers,
+    )