feat: adding pagination to GET /products endpoint in root_router.py tests: adding tests for GET /products pagination

Author: theodorehreuter

src/stapi_fastapi/routers/root_router.py

@@ -140,17 +140,47 @@ def get_root(self, request: Request) -> RootResponse:
     def get_conformance(self, request: Request) -> Conformance:
         return Conformance(conforms_to=self.conformances)
 
-    def get_products(self, request: Request) -> ProductsCollection:
-        return ProductsCollection(
-            products=[pr.get_product(request) for pr in self.product_routers.values()],
-            links=[
+    def get_products(
+        self, request: Request, next: str | None = None, limit: int = 10
+    ) -> ProductsCollection:
+        try:
+            start = 0
+            product_ids = [*self.product_routers.keys()]
+            if next:
+                start = product_ids.index(next)
+            if not product_ids and not next:
+                ProductsCollection(
+                    products=[],
+                    links=[
+                        Link(
+                            href=str(request.url_for(f"{self.name}:list-products")),
+                            rel="self",
+                            type=TYPE_JSON,
+                        )
+                    ],
+                )
+            end = start + limit
+            ids = product_ids[start:end]
+            products = [
+                self.product_routers[product_id].get_product(request)
+                for product_id in ids
+            ]
+            links = [
                 Link(
                     href=str(request.url_for(f"{self.name}:list-products")),
                     rel="self",
                     type=TYPE_JSON,
-                )
-            ],
-        )
+                ),
+            ]
+            next = ""
+            if end < len(product_ids):
+                next = self.product_routers[product_ids[end]].product.id
+                updated_url = request.url.include_query_params(next=next)
+                links.append(Link(href=str(updated_url), rel="next", type=TYPE_JSON))
+            return ProductsCollection(products=products, links=links)
+        except ValueError as e:
+            logging.exception(f"An error occurred while retrieving orders: {e}")
+            raise NotFoundException(detail="Error finding pagination token")
 
     async def get_orders(
         self, request: Request, next: str | None = None, limit: int = 10

tests/conftest.py

@@ -71,12 +71,46 @@ def mock_product_test_spotlight(
     )
 
 
+@pytest.fixture
+def mock_product_test_wolf_cola(
+    product_backend: MockProductBackend, mock_provider: Provider
+) -> Product:
+    """Fixture for a mock Wolf Cola product."""
+    return Product(
+        id="test-wolf-cola",
+        title="Test Wolf Cola Product",
+        description="Test product for Wolf Cola for testing GET /product pagination",
+        license="CC-BY-4.0",
+        keywords=["test", "satellite", "wolf-cola"],
+        providers=[mock_provider],
+        links=[],
+        constraints=MyProductConstraints,
+        opportunity_properties=MyOpportunityProperties,
+        order_parameters=MyOrderParameters,
+        backend=product_backend,
+    )
+
+
 @pytest.fixture
 def stapi_client(
-    root_backend, mock_product_test_spotlight, base_url: str
+    root_backend,
+    mock_product_test_spotlight,
+    mock_product_test_wolf_cola,
+    base_url: str,
 ) -> Iterator[TestClient]:
     root_router = RootRouter(root_backend)
     root_router.add_product(mock_product_test_spotlight)
+    root_router.add_product(mock_product_test_wolf_cola)
+    app = FastAPI()
+    app.include_router(root_router, prefix="")
+
+    with TestClient(app, base_url=f"{base_url}") as client:
+        yield client
+
+
+@pytest.fixture
+def empty_stapi_client(root_backend, base_url: str) -> Iterator[TestClient]:
+    root_router = RootRouter(root_backend)
     app = FastAPI()
     app.include_router(root_router, prefix="")
 

tests/test_order.py

@@ -166,6 +166,7 @@ def test_order_pagination(stapi_client: TestClient, setup_pagination, limit) ->
     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:
@@ -179,7 +180,6 @@ def test_order_pagination(stapi_client: TestClient, setup_pagination, limit) ->
             break
 
 
-def test_token_not_found(stapi_client: TestClient):
+def test_token_not_found(stapi_client: TestClient) -> None:
     res = stapi_client.get("/orders", params={"next": "a_token"})
-    # should return 404 as a result of bad token
     assert res.status_code == status.HTTP_404_NOT_FOUND

tests/test_product.py

@@ -63,3 +63,36 @@ def test_product_order_parameters_response(
     json_schema = res.json()
     assert "properties" in json_schema
     assert "s3_path" in json_schema["properties"]
+
+
+@pytest.mark.parametrize("limit", [1])
+def test_product_pagination(stapi_client: TestClient, limit: int):
+    res = stapi_client.get("/products", params={"next": None, "limit": limit})
+    assert res.status_code == status.HTTP_200_OK
+    body = res.json()
+    assert len(body["products"]) == limit
+    for d in body["links"]:
+        if ("rel", "next") in d.items():
+            next = d["href"]
+
+    while len(body["links"]) > 1:
+        res = stapi_client.get(next)
+        assert res.status_code == status.HTTP_200_OK
+        body = res.json()
+        assert len(body["products"]) == limit
+        for d in body["links"]:
+            if ("rel", "next") in d.items():
+                next = body["links"][0]["href"]
+
+
+def test_token_not_found(stapi_client: TestClient) -> None:
+    res = stapi_client.get("/products", params={"next": "a_token"})
+    assert res.status_code == status.HTTP_404_NOT_FOUND
+
+
+def test_no_products(empty_stapi_client: TestClient):
+    res = empty_stapi_client.get("/products")
+    body = res.json()
+    print("hold")
+    assert res.status_code == status.HTTP_200_OK
+    assert len(body["products"]) == 0