Add-GET-/books/{user_id}/reservations-(JWT-protected) (#16)

### Role-based Access Control (RBAC)
- Added `require_admin` decorator for admin-only routes
- Updated `login_user` route and DB logic to include a `role` field
- Default role set to `user`; fixtures and sample data updated
- Added tests for all RBAC cases (happy path, invalid JWT, non-admin access, etc.)

### Reservations Endpoint
- Implemented `GET /books/{book_id}/reservations` (JWT-protected, admin-only)
- Added tests for valid/invalid book IDs, missing/no auth headers, and non-admin roles
- Log warning and skip reservations referencing non-existent users
- Refactored response structure (`status → state`, nested `message` field, unified body construction)

### Fixtures & Test Infrastructure
- Consolidated `users_db_setup` → `mongo_setup` to reset all collections
- Added fixtures for seeding users/admins with JWTs, books, and reservations
- Cleaned up/reorganized fixtures; fixed imports and test DB usage
- Expanded coverage to 100% with edge case tests

### Refactors & Fixes
- Fixed token return value, forbidden return handling, and test failures
- Standardized response formatting across endpoints
- Spelling/typo corrections, linter/formatter run

Author: codesungrape

app/routes/auth_routes.py

@@ -59,6 +59,7 @@ def register_user():
             "email": email,
             # The hash is stored as a string in the DB
             "password": hashed_password,
+            "role": "user",  # all users assigned a default 'user' role
         }
     ).inserted_id
 
@@ -97,6 +98,7 @@ def login_user():
     # 4. Generate the JWT payload
     payload = {
         "sub": str(user["_id"]),  # sub (subject)- standard claim for user ID
+        "role": user.get("role", "user"),  # embed the role in token
         "iat": datetime.datetime.now(
             datetime.UTC
         ),  # iat (issued at)- when token was created

app/routes/reservation_routes.py

@@ -1,13 +1,14 @@
 """Routes for /books/<id>/reservations endpoint"""
 
 import datetime
+import logging
 
 from bson import ObjectId
 from bson.errors import InvalidId
 from flask import Blueprint, g, jsonify, url_for
 
 from app.extensions import mongo
-from app.utils.decorators import require_jwt
+from app.utils.decorators import require_admin, require_jwt
 
 reservations_bp = Blueprint(
     "reservations_bp", __name__, url_prefix="/books/<book_id_str>"
@@ -19,8 +20,8 @@
 def create_reservation(book_id_str):
     """
     This POST endpoint lets an authenticated user reserve a book by its ID.
-    It validates the ID, checks book availability, 
-    prevents duplicate reservations, 
+    It validates the ID, checks book availability,
+    prevents duplicate reservations,
     creates the reservation, and returns its details.
     """
 
@@ -76,3 +77,65 @@ def create_reservation(book_id_str):
     }
 
     return jsonify(response_data), 201
+
+
+@reservations_bp.route("/reservations", methods=["GET"])
+@require_admin
+def get_reservations_for_book_id(book_id_str):
+    """
+    Retrieves all reservations for a specific book.
+    Accessible only by users with the 'admin' role.
+    """
+    # Validate the book_id format
+    try:
+        oid = ObjectId(book_id_str)
+    except (InvalidId, TypeError):
+        return jsonify({"error": "Invalid Book ID"}), 400
+
+    # Check if book exists
+    book = mongo.db.books.find_one({"_id": oid})
+    if not book:
+        return "book not found", 404, {"Content-Type": "text/plain"}
+
+    # Fetch reservations for the given book_id
+    reservations_cursor = mongo.db.reservations.find({"book_id": oid})
+
+    items = []
+    for r in reservations_cursor:
+        # For each reservation, fetch the associated user's details
+        user = mongo.db.users.find_one({"_id": r["user_id"]})
+
+        # Skip if the user for a reservation is not found, to avoid errors
+        if not user:
+            logging.warning(
+                "Data integrity issue: Reservation '%s' references a non-existent user_id '%s'. Skipping.",  # pylint: disable=line-too-long
+                r["_id"],
+                r["user_id"],
+            )
+            continue
+
+        # Build the item object according to spec
+        reservation_item = {
+            "id": str(r["_id"]),
+            "state": r.get("state", "UNKNOWN"),  # Use .get() for safety
+            "user": {
+                "forenames": user.get("forenames"),
+                "surname": user.get("surname"),
+            },
+            "book_id": str(r["book_id"]),
+            "links": {
+                "self": url_for(
+                    ".get_reservations_for_book_id",
+                    reservation_id=str(r["_id"]),
+                    _external=True,
+                ),
+                "book": url_for("get_book", book_id=str(r["book_id"]), _external=True),
+            },
+        }
+
+        items.append(reservation_item)
+
+    # Construct the final response body
+    response_body = {"total_count": len(items), "items": items}
+
+    return jsonify(response_body), 200

app/utils/decorators.py

@@ -7,7 +7,7 @@
 import jwt
 from bson.errors import InvalidId
 from bson.objectid import ObjectId
-from flask import current_app, g, jsonify, request
+from flask import abort, current_app, g, jsonify, request
 
 from app.extensions import mongo
 
@@ -68,3 +68,19 @@ def decorated_function(*args, **kwargs):
         return f(*args, **kwargs)
 
     return decorated_function
+
+
+def require_admin(f):
+    """A decorator that requires a user to be admin"""
+
+    @functools.wraps(f)
+    @require_jwt  # ensure the user is logged in with a valid JWT
+    def decorated_function(*args, **kwargs):
+        user_role = g.current_user.get("role")
+
+        if user_role != "admin":
+            abort(403, description="Admin privileges required.")
+
+        return f(*args, **kwargs)
+
+    return decorated_function

scripts/seed_users.py

@@ -26,6 +26,7 @@ def seed_users(users_to_seed: list) -> str:
 
     for user_data in users_to_seed:
         email = user_data["email"]
+        role = user_data["role"]
 
         # Check if data already exists
         if mongo.db.users.find_one({"email": email}):
@@ -39,7 +40,7 @@ def seed_users(users_to_seed: list) -> str:
 
         # insert to new user
         mongo.db.users.insert_one(
-            {"email": email, "password": hashed_password.decode("utf-8")}
+            {"email": email, "password": hashed_password.decode("utf-8"), "role": role}
         )
         count += 1
         print(f"Created user: {email}")

scripts/test_data/sample_user_data.json

@@ -1,5 +1,5 @@
 [
-    {"email": "[email protected]", "password": "AdminPassword123"},
-    {"email": "[email protected]", "password": "UserPassword456"},
-    {"email": "[email protected]", "password": "DAYSend1991"}
+    {"email": "[email protected]", "password": "AdminPassword123", "role": "admin"},
+    {"email": "[email protected]", "password": "UserPassword456", "role": "user"},
+    {"email": "[email protected]", "password": "DAYSend1991", "role": "user"}
 ]

tests/conftest.py

@@ -125,20 +125,25 @@ def db_setup(test_app):  # pylint: disable=redefined-outer-name
 
 # Fixture for tests/test_auth.py
 @pytest.fixture(scope="function")
-def users_db_setup(test_app):  # pylint: disable=redefined-outer-name
+def mongo_setup(test_app):  # pylint: disable=redefined-outer-name
     """
-    Sets up and tears down the 'users' collection for a test.
+    Updated to a more robust setup fixture.
+    Ensures all relevant collections ('users', 'books', 'reservations')
+    are clean before each test function runs.
     """
     with test_app.app_context():
-        # Now, the 'mongo' variable is defined and linked to the test_app
-        users_collection = mongo.db.users
-        users_collection.delete_many({})
+        # Clear all collections before using in tests
+        mongo.db.users.delete_many({})
+        mongo.db.books.delete_many({})
+        mongo.db.reservations.delete_many({})
 
     yield
 
     with test_app.app_context():
-        users_collection = mongo.db.users
-        users_collection.delete_many({})
+        # Clear after tests have run
+        mongo.db.users.delete_many({})
+        mongo.db.books.delete_many({})
+        mongo.db.reservations.delete_many({})
 
 
 TEST_USER_ID = "6154b3a3e4a5b6c7d8e9f0a1"
@@ -155,29 +160,82 @@ def mock_user_data():
         "_id": ObjectId(TEST_USER_ID),
         "email": "[email protected]",
         "password": hashed_password,
+        "role": "user",
+    }
+
+
+@pytest.fixture(scope="session")
+def mock_admin_data():
+    """
+    PROVIDES a dictionary of a test admin's data,
+    WITH a hashed password
+    """
+    hashed_password = bcrypt.generate_password_hash("admin-password").decode("utf-8")
+
+    return {
+        "email": "[email protected]",
+        "password": hashed_password,
+        "role": "admin",  # Role explicitly set to 'admin'
     }
 
 
 @pytest.fixture
 def seeded_user_in_db(
-    test_app, mock_user_data, users_db_setup
+    test_app, mock_user_data, mongo_setup
 ):  # pylint: disable=redefined-outer-name
     """
     Ensures the test database is clean and contains exactly one predefined user.
     Depends on:
     - test_app: To get the application context and correct mongo.db object
     - mock_user_data: To get the user data to insert.
-    - users_db_Setup: To ensure the users collection is empty before seeding.
+    - mongo_setup: To ensure the users collection is empty before seeding.
     """
-    _ = users_db_setup
+    _ = mongo_setup
 
     with test_app.app_context():
         mongo.db.users.insert_one(mock_user_data)
 
-    # When yielding the mock data back to the test,
-    # we must convert it back the ObjectId back to the string,
-    # because that's what 'auth_token' fixture expects to put into the JWT 'sub' claim
     yield_data = mock_user_data.copy()
     yield_data["_id"] = str(yield_data["_id"])
 
     yield yield_data
+
+
+@pytest.fixture
+def seeded_admin_in_db(
+    test_app, mock_admin_data, mongo_setup
+):  # pylint: disable=redefined-outer-name
+    """
+    Ensures the test database is clean and
+    Contains exactly one predefined admin
+    """
+    _ = mongo_setup
+
+    with test_app.app_context():
+        result = mongo.db.users.insert_one(mock_admin_data)
+
+    yield_data = mock_admin_data.copy()
+    yield_data["_id"] = str(result.inserted_id)
+    yield yield_data
+
+
+@pytest.fixture
+def user_token(client, seeded_user_in_db):  # pylint: disable=redefined-outer-name
+    """Logs in the seeded user and returns a valid JWT"""
+    _ = seeded_user_in_db
+    login_payload = {"email": "[email protected]", "password": PLAIN_PASSWORD}
+
+    response = client.post("/auth/login", json=login_payload)
+    assert response.status_code == 200, "Failed to log in test user"
+    return response.get_json()["token"]
+
+
+@pytest.fixture
+def admin_token(client, seeded_admin_in_db):  # pylint: disable=redefined-outer-name
+    """Logs in the seeded admin and returns a valid JWT."""
+    _ = seeded_admin_in_db
+    login_payload = {"email": "[email protected]", "password": "admin-password"}
+
+    response = client.post("/auth/login", json=login_payload)
+    assert response.status_code == 200, "Failed to log in test admin"
+    return response.get_json()["token"]

tests/scripts/test_seed_users.py

@@ -14,13 +14,22 @@ def test_seed_users_successfully(test_app):
     """
     GIVEN an empty database and a list of user data
     WHEN the seed_users function is called
-    THEN the users should be created in the database with hashed passwords.
+    THEN the users should be created in the database with hashed passwords
+    AND the correct roles.
     """
     # Arrange
     # define the user data we want to seed the database with
     sample_users = [
-        {"email": "[email protected]", "password": "AdminPassword123"},
-        {"email": "[email protected]", "password": "UserPassword456"},
+        {
+            "email": "[email protected]",
+            "password": "AdminPassword123",
+            "role": "admin",
+        },
+        {
+            "email": "[email protected]",
+            "password": "UserPassword456",
+            "role": "user",
+        },
     ]
 
     # Enter application context and
@@ -41,6 +50,11 @@ def test_seed_users_successfully(test_app):
         assert bcrypt.checkpw(
             b"AdminPassword123", admin_user["password"].encode("utf-8")
         )
+
+        # verify roles
+        assert admin_user["role"] == "admin"
+        test_user = mongo.db.users.find_one({"email": "[email protected]"})
+        assert test_user["role"] == "user"
     assert "Successfully seeded 2 users" in result_message
 
 
@@ -52,8 +66,12 @@ def test_seed_users_skips_if_user_already_exists(test_app, capsys):
     """
     # Arrange
     users_to_attempt_seeding = [
-        {"email": "[email protected]", "password": "Password123"},
-        {"email": "[email protected]", "password": "Password456"},
+        {
+            "email": "[email protected]",
+            "password": "Password123",
+            "role": "user",
+        },
+        {"email": "[email protected]", "password": "Password456", "role": "user"},
     ]
 
     with test_app.app_context():