Bound limit and offset on search and typeahead

app/routers/search.py

@@ -15,6 +15,9 @@
 
 router = APIRouter(tags=["Search"])
 
+# LIMIT and OFFSET are bigint in Postgres; larger values fail when the query runs.
+MAX_BIGINT = 2**63 - 1
+
 
 def parse_query_params(
     # String filters
@@ -61,9 +64,9 @@ def parse_query_params(
         default=ResponseFormat.JSON, description="Response format (json or csv)"
     ),
     limit: Optional[int] = Query(
-        None, description="Maximum number of records to return"
+        None, ge=1, le=100000, description="Maximum number of records to return"
     ),
-    offset: Optional[int] = Query(0, description="Number of records to skip"),
+    offset: Optional[int] = Query(0, ge=0, le=MAX_BIGINT, description="Number of records to skip"),
     ordering: Optional[str] = Query(
         None,
         description="Column to sort by. Prefix with '-' for descending order. "
@@ -326,10 +329,10 @@ def parse_typeahead_params(
         description="Search term for typeahead suggestions (minimum 3 characters)"
     ),
     limit: Optional[int] = Query(
-        20, description="Maximum number of records to return"
+        20, ge=1, le=100, description="Maximum number of records to return"
     ),
     offset: Optional[int] = Query(
-        0, description="Number of records to skip"
+        0, ge=0, le=MAX_BIGINT, description="Number of records to skip"
     ),
     # Search context filters
     accession: Optional[List[str]] = Query(

tests/conftest.py

@@ -1,5 +1,29 @@
 import os
+from unittest.mock import AsyncMock
+
+import pytest
 
 # Settings require DB credentials at import time; tests mock the DB calls.
 for key in ("CLEAN_DB_HOST", "CLEAN_DB_USER", "CLEAN_DB_PASSWORD", "CLEAN_DB_NAME"):
     os.environ.setdefault(key, "test")
+
+
+@pytest.fixture
+def db():
+    db = AsyncMock()
+    db.fetch.return_value = []
+    db.fetchval.return_value = 0
+    return db
+
+
+@pytest.fixture
+def client(db):
+    # Imported here so the environment above is set before settings load.
+    from fastapi.testclient import TestClient
+
+    from app.db.database import get_db
+    from app.main import app
+
+    app.dependency_overrides[get_db] = lambda: db
+    yield TestClient(app)
+    app.dependency_overrides.clear()

tests/test_pagination_bounds.py

@@ -0,0 +1,45 @@
+import pytest
+
+SEARCH = "/api/v1/search?accession=P51451"
+TYPEAHEAD = "/api/v1/typeahead?field_name=organism&search=homo"
+TOO_BIG = 2**63  # one past Postgres bigint
+
+# Out-of-range values used to reach the SQL: negative or oversized ones made
+# Postgres fail (a 500), and limit=0 gave /search a `next` link to the same page.
+REJECTED = [
+    (SEARCH, "limit", -1),
+    (SEARCH, "limit", 0),
+    (SEARCH, "limit", 100001),
+    (SEARCH, "offset", -5),
+    (SEARCH, "offset", TOO_BIG),
+    (TYPEAHEAD, "limit", -1),
+    (TYPEAHEAD, "limit", 0),
+    (TYPEAHEAD, "limit", 101),
+    (TYPEAHEAD, "offset", -5),
+    (TYPEAHEAD, "offset", TOO_BIG),
+]
+
+ACCEPTED = [
+    (SEARCH, "limit", 100000),
+    (SEARCH, "offset", TOO_BIG - 1),
+    (TYPEAHEAD, "limit", 100),
+    (TYPEAHEAD, "offset", TOO_BIG - 1),
+]
+
+
+@pytest.mark.parametrize("url, param, value", REJECTED)
+def test_out_of_range_pagination_is_rejected(client, db, url, param, value):
+    response = client.get(f"{url}&{param}={value}")
+
+    assert response.status_code == 422, response.text
+    assert [e["loc"] for e in response.json()["detail"]] == [["query", param]]
+    db.fetch.assert_not_called()
+    db.fetchval.assert_not_called()
+
+
+@pytest.mark.parametrize("url, param, value", ACCEPTED)
+def test_pagination_bounds_are_inclusive(client, db, url, param, value):
+    response = client.get(f"{url}&{param}={value}")
+
+    assert response.status_code == 200, response.text
+    assert f"{param.upper()} {value}" in db.fetch.call_args.args[0]

tests/test_search_csv.py

@@ -2,24 +2,12 @@
 from io import StringIO
 from unittest.mock import AsyncMock, patch
 
-import pytest
-from fastapi.testclient import TestClient
-
 import app.routers.search as search
-from app.db.database import get_db
-from app.main import app
 from app.models.clean_data import CLEANColumn
 
 HEADER = [c.value for c in CLEANColumn]
 
 
-@pytest.fixture
-def client():
-    app.dependency_overrides[get_db] = lambda: None
-    yield TestClient(app)
-    app.dependency_overrides.clear()
-
-
 def get_csv(client, rows):
     with (
         patch.object(search, "get_total_count", AsyncMock(return_value=len(rows))),