feat: rework test backend to make code smoother feat: use default feat: use include_query_params instead of parsing query param string ourselves

Author: theodorehreuter

src/stapi_fastapi/backends/root_backend.py

@@ -14,7 +14,7 @@
 
 class RootBackend[T: OrderStatusPayload, U: OrderStatus](Protocol):  # pragma: nocover
     async def get_orders(
-        self, request: Request, next: str | None, limit: int | None
+        self, request: Request, next: str | None, limit: int
     ) -> ResultE[tuple[OrderCollection, str]]:
         """
         Return a list of existing orders and pagination token if applicable

src/stapi_fastapi/routers/root_router.py

@@ -160,7 +160,7 @@ def get_products(self, request: Request) -> ProductsCollection:
         )
 
     async def get_orders(
-        self, request: Request, next: str | None = None, limit: int | None = None
+        self, request: Request, next: str | None = None, limit: int = 10
     ) -> OrderCollection:
         match await self.backend.get_orders(request, next, limit):
             case Success((collections, token)):
@@ -177,13 +177,7 @@ async def get_orders(
                         )
                     )
                 if token:  # pagination link if backend returns token
-                    query = request.url.components.query
-                    params = {
-                        param.split("=")[0]: param.split("=")[1]
-                        for param in query.split("&")
-                    }
-                    params["next"] = token
-                    updated_url = request.url.replace_query_params(**params)
+                    updated_url = request.url.include_query_params(next=token)
                     collections.links.append(
                         Link(href=str(updated_url), rel="next", type=TYPE_JSON)
                     )

tests/application.py

@@ -44,33 +44,35 @@ def __init__(self, orders: InMemoryOrderDB) -> None:
         self._orders_db: InMemoryOrderDB = orders
 
     async def get_orders(
-        self, request: Request, next: str | None = None, limit: int | None = None
+        self, request: Request, next: str | None, limit: int
     ) -> ResultE[tuple[OrderCollection, str]]:
         """
         Return orders from backend.  Handle pagination/limit if applicable
         """
-        features = list(self._orders_db._orders.values())
-        if limit and not next:
-            return Success((OrderCollection(features=features[:limit]), ""))
-        if next:
-
-            def token_processor(next: str) -> tuple[str, int]:
-                """process token to return new token and start loc"""
-                return "new_token", 0
-
-            token, start = token_processor(next)
-            if limit:
-                if start + limit > len(features):
-                    features = features[start:]
-                else:
-                    features = features[start:limit]
-                return Success((OrderCollection(features=features), token))
-            else:  # token and no limit
-                return Success((OrderCollection(features=features[start:]), ""))
-        else:
-            return Success(
-                (OrderCollection(features=list(self._orders_db._orders.values())), "")
-            )
+        try:
+            start = 0
+            order_ids = sorted(self._orders_db._orders.keys())
+            if not order_ids:  # not data in db return empty
+                return Success(
+                    (
+                        OrderCollection(
+                            features=list(self._orders_db._orders.values())
+                        ),
+                        "",
+                    )
+                )
+
+            if next:
+                start = [i for i, x in enumerate(order_ids) if x == next][0]
+            end = start + limit
+            ids = order_ids[start:end]
+
+            feats = [self._orders_db._orders[order_id] for order_id in ids]
+            next = self._orders_db._orders[order_ids[end]].id
+
+            return Success((OrderCollection(features=feats), next))
+        except Exception as e:
+            return Failure(e)
 
     async def get_order(self, order_id: str, request: Request) -> ResultE[Maybe[Order]]:
         """

tests/test_order.py

@@ -187,7 +187,7 @@ def create_second_order_allowed_payloads() -> list[OrderPayload]:
     ]
 
 
-@pytest.mark.parametrize("product_id", ["test-spotlight"])
+@pytest.mark.parametrize("product_id", [pytest.param("test-spotlight", id="base test")])
 def test_order_pagination(
     product_id: str,
     product_backend: MockProductBackend,
@@ -196,7 +196,6 @@ def test_order_pagination(
     create_second_order_allowed_payloads: list[OrderPayload],
 ) -> None:
     product_backend._allowed_payloads = create_order_allowed_payloads
-    OLD_TOKEN = "a_token"
 
     # check empty
     res = stapi_client.get("/orders")
@@ -220,9 +219,9 @@ def test_order_pagination(
         f"products/{product_id}/orders",
         json=create_second_order_allowed_payloads[0].model_dump(),
     )
-
     # call all orders
-    res = stapi_client.get("/orders", params={"next": OLD_TOKEN, "limit": 1})
+    next = res.json()["id"]
+    res = stapi_client.get("/orders", params={"next": next, "limit": 1})
     checker = res.json()
 
     assert res.status_code == status.HTTP_200_OK
@@ -233,6 +232,6 @@ def test_order_pagination(
     assert checker["links"] != []
 
     # check to make sure new token in link
-    assert OLD_TOKEN not in checker["links"][0]["href"]
+    assert next not in checker["links"][0]["href"]
 
     assert len(checker["features"]) == 1