Move pagination to DB level in get_assertions_by_badge endpoint

conftest.py

@@ -0,0 +1,29 @@
+# Copyright (c) 2026 Red Hat, Inc.
+
+"""Pytest configuration and shared fixtures."""
+
+import pytest
+
+from tahrir.app import create_app
+
+
+@pytest.fixture(scope="module")
+def app():
+    app = create_app(
+        {
+            "TESTING": True,
+            "SQLALCHEMY_DATABASE_URI": "sqlite:///:memory:",
+            "TAHRIR_PNGS_PATH": "/tmp",
+            "TAHRIR_ADMIN_GROUPS": ["admins"],
+            "TAHRIR_TITLE": "Test Badges",
+            "OIDC_ENABLED": False,
+            "WTF_CSRF_ENABLED": False,
+            "CACHE": {"backend": "dogpile.cache.null"},
+        }
+    )
+    return app
+
+
+@pytest.fixture(scope="module")
+def client(app):
+    return app.test_client()

tahrir/endpoints/admin/authorizations.py

@@ -31,11 +31,17 @@ def add_authorization():
     if not result:
         return abort(400, "Failed to add authorization")
 
-    return jsonify(
-        {
-            "message": f"Badge {data.get(required_fields[0])!r} authorized to {data.get(required_fields[1])!r}"
-        }
-    ), 201
+    return (
+        jsonify(
+            {
+                "message": (
+                    f"Badge {data.get(required_fields[0])!r}"
+                    f" authorized to {data.get(required_fields[1])!r}"
+                )
+            }
+        ),
+        201,
+    )
 
 
 @bp.route("/api/admin/authorization", methods=["DELETE"])
@@ -64,8 +70,14 @@ def remove_authorization():
     if not result:
         return abort(404, "Authorization not found or failed to remove")
 
-    return jsonify(
-        {
-            "message": f"Badge {data.get(required_fields[0])!r} authorization revoked from {data.get(required_fields[1])!r}"
-        }
-    ), 200
+    return (
+        jsonify(
+            {
+                "message": (
+                    f"Badge {data.get(required_fields[0])!r}"
+                    f" authorization revoked from {data.get(required_fields[1])!r}"
+                )
+            }
+        ),
+        200,
+    )

tahrir/endpoints/assertions.py

@@ -30,23 +30,19 @@ def get_recent_assertions():
 def get_assertions_by_badge(badge_id: str):
     """Endpoint to fetch all assertions for a specific badge."""
 
-    assertions = g.tahrirdb.get_assertions_by_badge(badge_id)
+    begin = request.args.get("begin", 0, type=int)
+    limit = request.args.get("limit", 100, type=int)
+
+    assertions = g.tahrirdb.get_assertions_by_badge(badge_id, limit=min(limit, 100), offset=begin)
 
     if assertions is False:
         return abort(404, f"No such badge {badge_id!r}")
 
     if not assertions:
         return jsonify([])
 
-    assertions = sorted(assertions, key=lambda assertion: assertion.issued_on)
-
-    # This is a very unoptimised implementation for achieving pagination.
-    # The implementation should have been there in the upstream `tahrir-api` at database level.
-    begin = request.args.get("begin", 0, type=int)
-    limit = request.args.get("limit", 100, type=int)
-
     result = []
-    for assertion in assertions[begin : begin + (limit if limit < 100 else 100)]:
+    for assertion in assertions:
         assertion_data = assertion.as_dict()
         assertion_data.pop("badge", None)  # Remove unwanted fields
         assertion_data["person"] = assertion.person.as_dict()  # Add user info

tests/test_assertions.py

@@ -0,0 +1,65 @@
+# Copyright (c) 2026 Red Hat, Inc.
+
+"""Test tahrir.endpoints.assertions."""
+
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+
+def make_mock_assertion(badge_id="test-badge"):
+    """Create a mock assertion object."""
+    assertion = MagicMock()
+    assertion.as_dict.return_value = {"id": "assertion-1", "badge": {}}
+    assertion.person.as_dict.return_value = {"username": "testuser"}
+    assertion.person.avatar = "https://example.com/avatar.png"
+    assertion.issued_on = datetime(2024, 1, 1)
+    assertion.badge.id = badge_id
+    return assertion
+
+
+def test_get_assertions_by_badge_returns_results(client):
+    """Paginated assertions are returned for a valid badge."""
+    mock_db = MagicMock()
+    mock_db.get_assertions_by_badge.return_value = [
+        make_mock_assertion("test-badge") for _ in range(3)
+    ]
+
+    with patch("tahrir.database.TahrirDatabaseExtension.get_db", return_value=mock_db):
+        response = client.get("/api/assertions/test-badge?begin=0&limit=3")
+
+    assert response.status_code == 200
+    assert len(response.get_json()) == 3
+
+
+def test_get_assertions_by_badge_limit_capped_at_100(client):
+    """Limit is capped at 100 even if a higher value is requested."""
+    mock_db = MagicMock()
+    mock_db.get_assertions_by_badge.return_value = []
+
+    with patch("tahrir.database.TahrirDatabaseExtension.get_db", return_value=mock_db):
+        client.get("/api/assertions/test-badge?begin=0&limit=999")
+        _, kwargs = mock_db.get_assertions_by_badge.call_args
+        assert kwargs["limit"] <= 100
+
+
+def test_get_assertions_by_badge_not_found(client):
+    """A non-existent badge returns 404."""
+    mock_db = MagicMock()
+    mock_db.get_assertions_by_badge.return_value = False
+
+    with patch("tahrir.database.TahrirDatabaseExtension.get_db", return_value=mock_db):
+        response = client.get("/api/assertions/nonexistent-badge")
+
+    assert response.status_code == 404
+
+
+def test_get_assertions_by_badge_empty(client):
+    """A badge with no assertions returns an empty list."""
+    mock_db = MagicMock()
+    mock_db.get_assertions_by_badge.return_value = []
+
+    with patch("tahrir.database.TahrirDatabaseExtension.get_db", return_value=mock_db):
+        response = client.get("/api/assertions/empty-badge")
+
+    assert response.status_code == 200
+    assert response.get_json() == []