tests: abstracted pagination testing out into a more generalized pagination_tester function

Author: theodorehreuter

tests/conftest.py

@@ -1,18 +1,12 @@
 from collections.abc import Iterator
-from datetime import datetime, timedelta, timezone
 from typing import Any, Callable
 from urllib.parse import urljoin
-from uuid import uuid4
 
 import pytest
-from fastapi import FastAPI
+from fastapi import FastAPI, status
 from fastapi.testclient import TestClient
-from geojson_pydantic import Point
-from geojson_pydantic.types import Position2D
+from httpx import Response
 
-from stapi_fastapi.models.opportunity import (
-    Opportunity,
-)
 from stapi_fastapi.models.product import (
     Product,
     Provider,
@@ -161,35 +155,58 @@ def mock_provider() -> Provider:
     )
 
 
-@pytest.fixture
-def mock_test_spotlight_opportunities() -> list[Opportunity]:
-    """Fixture to create mock data for Opportunities for `test-spotlight-1`."""
-    now = datetime.now(timezone.utc)  # Use timezone-aware datetime
-    start = now
-    end = start + timedelta(days=5)
-
-    # Create a list of mock opportunities for the given product
-    return [
-        Opportunity(
-            id=str(uuid4()),
-            type="Feature",
-            geometry=Point(
-                type="Point",
-                coordinates=Position2D(longitude=0.0, latitude=0.0),
-            ),
-            properties=MyOpportunityProperties(
-                product_id="xyz123",
-                datetime=(start, end),
-                off_nadir={"minimum": 20, "maximum": 22},
-                vehicle_id=[1],
-                platform="platform_id",
-            ),
-        ),
-    ]
-
-
-@pytest.fixture
-def mock_test_pagination_opportunities(
-    mock_test_spotlight_opportunities,
-) -> list[Opportunity]:
-    return [opp for opp in mock_test_spotlight_opportunities for __ in range(0, 3)]
+def pagination_tester(
+    stapi_client: TestClient,
+    endpoint: str,
+    method: str,
+    limit: int,
+    target: str,
+    expected_total_returns: int,
+    body: dict | None = None,
+) -> None:
+    retrieved = []
+
+    res = make_request(stapi_client, endpoint, method, body, None, limit)
+    assert res.status_code == status.HTTP_200_OK
+    resp_body = res.json()
+
+    assert len(resp_body[target]) <= limit
+    retrieved.extend(resp_body[target])
+    next_url = next((d["href"] for d in resp_body["links"] if d["rel"] == "next"), None)
+
+    while next_url:
+        res = make_request(stapi_client, next_url, method, body, next_url, limit)
+        assert res.status_code == status.HTTP_200_OK
+        assert len(resp_body[target]) <= limit
+        resp_body = res.json()
+        retrieved.extend(resp_body[target])
+
+        if resp_body["links"]:
+            next_url = next(
+                (d["href"] for d in resp_body["links"] if d["rel"] == "next"), None
+            )
+        else:
+            next_url = None
+
+    assert len(retrieved) == expected_total_returns
+
+
+def make_request(
+    stapi_client: TestClient,
+    endpoint: str,
+    method: str,
+    body: dict | None,
+    next_token: str | None,
+    limit: int,
+) -> Response:
+    """request wrapper for pagination tests"""
+    if next_token and "next=" in next_token:
+        next_token = next_token.split("next=")[1]
+    params = {"next": next_token, "limit": limit}
+
+    if method == "GET":
+        res = stapi_client.get(endpoint, params=params)
+    if method == "POST":
+        res = stapi_client.post(endpoint, json=body, params=params)
+
+    return res

tests/test_opportunity.py

@@ -1,16 +1,54 @@
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime, timedelta, timezone
 from typing import List
+from uuid import uuid4
 
 import pytest
 from fastapi import status
 from fastapi.testclient import TestClient
+from geojson_pydantic import Point
+from geojson_pydantic.types import Position2D
 
 from stapi_fastapi.models.opportunity import Opportunity, OpportunityCollection
+from tests.application import MyOpportunityProperties
 
 from .backends import MockProductBackend
 from .test_datetime_interval import rfc3339_strftime
 
 
+@pytest.fixture
+def mock_test_spotlight_opportunities() -> list[Opportunity]:
+    """Fixture to create mock data for Opportunities for `test-spotlight-1`."""
+    now = datetime.now(timezone.utc)  # Use timezone-aware datetime
+    start = now
+    end = start + timedelta(days=5)
+
+    # Create a list of mock opportunities for the given product
+    return [
+        Opportunity(
+            id=str(uuid4()),
+            type="Feature",
+            geometry=Point(
+                type="Point",
+                coordinates=Position2D(longitude=0.0, latitude=0.0),
+            ),
+            properties=MyOpportunityProperties(
+                product_id="xyz123",
+                datetime=(start, end),
+                off_nadir={"minimum": 20, "maximum": 22},
+                vehicle_id=[1],
+                platform="platform_id",
+            ),
+        ),
+    ]
+
+
+@pytest.fixture
+def mock_test_pagination_opportunities(
+    mock_test_spotlight_opportunities,
+) -> list[Opportunity]:
+    return [opp for opp in mock_test_spotlight_opportunities for __ in range(0, 3)]
+
+
 @pytest.mark.parametrize("product_id", ["test-spotlight"])
 def test_search_opportunities_response(
     product_id: str,

tests/test_order.py

@@ -8,6 +8,7 @@
 from httpx import Response
 
 from stapi_fastapi.models.order import OrderPayload, OrderStatus, OrderStatusCode
+from tests.conftest import pagination_tester
 
 from .application import InMemoryOrderDB, MyOrderParameters
 from .backends import MockProductBackend
@@ -161,26 +162,15 @@ def setup_orders_pagination(stapi_client: TestClient, create_order_payloads) ->
         assert res.headers["Content-Type"] == "application/geo+json"
 
 
-@pytest.mark.parametrize("limit", [2])
-def test_order_pagination(
-    stapi_client: TestClient, setup_orders_pagination, limit
-) -> None:
-    res = stapi_client.get("/orders", params={"next": None, "limit": limit})
-    assert res.status_code == status.HTTP_200_OK
-    body = res.json()
-    assert len(body["features"]) == limit
-    next = body["links"][0]["href"]
-
-    while next:
-        res = stapi_client.get(next)
-        assert res.status_code == status.HTTP_200_OK
-        body = res.json()
-        assert body["features"] != []
-        if body["links"]:
-            assert len(body["features"]) == limit
-            next = body["links"][0]["href"]
-        else:
-            break
+def test_order_pagination(setup_orders_pagination, stapi_client: TestClient) -> None:
+    pagination_tester(
+        stapi_client=stapi_client,
+        endpoint="/orders",
+        method="GET",
+        limit=2,
+        target="features",
+        expected_total_returns=3,
+    )
 
 
 def test_token_not_found(stapi_client: TestClient) -> None:
@@ -223,30 +213,15 @@ def test_order_status_pagination(
     order_db._statuses = order_statuses
 
     order_id = "test_order_id"
-    res = stapi_client.get(f"/orders/{order_id}/statuses")
-    assert res.status_code == status.HTTP_200_OK
 
-    order_id = "test_order_id"
-    res = stapi_client.get(
-        f"/orders/{order_id}/statuses", params={"next": None, "limit": limit}
+    pagination_tester(
+        stapi_client=stapi_client,
+        endpoint=f"/orders/{order_id}/statuses",
+        method="GET",
+        limit=2,
+        target="statuses",
+        expected_total_returns=3,
     )
-    body = res.json()
-    links = body["links"]
-    for link in body["links"]:
-        if ("rel", "next") in link.items():
-            assert len(body["statuses"]) == limit
-            next = link["href"]
-
-    while len(links) > 1:
-        res = stapi_client.get(next)
-        assert res.status_code == status.HTTP_200_OK
-        body = res.json()
-        assert body["statuses"] != []
-        links = body["links"]
-        for link in body["links"]:
-            if ("rel", "next") in link.items():
-                assert len(body["statuses"]) == limit
-                next = body["links"][0]["href"]
 
 
 def test_get_order_statuses_bad_token(