Admin must have role for a library to generate inventory report. (PP-…

…1293) (#1860)

* Admin needs rights on library to generate inventory report.

src/palace/manager/api/admin/controller/report.py

@@ -2,14 +2,15 @@
 from http import HTTPStatus
 
 import flask
-from flask import Response
+from flask import Request, Response
 from sqlalchemy.orm import Session
 
 from palace.manager.api.admin.model.inventory_report import (
     InventoryReportCollectionInfo,
     InventoryReportInfo,
 )
 from palace.manager.api.admin.problem_details import ADMIN_NOT_AUTHORIZED
+from palace.manager.api.problem_details import LIBRARY_NOT_FOUND
 from palace.manager.celery.tasks.generate_inventory_and_hold_reports import (
     generate_inventory_and_hold_reports,
     library_report_integrations,
@@ -22,6 +23,26 @@
 from palace.manager.util.problem_detail import ProblemDetail, ProblemDetailException
 
 
+def _authorize_from_request(
+    request: Request,
+) -> tuple[Admin, Library]:
+    """Authorize the admin for the library specified in the request.
+
+    :param request: A Flask Request object.
+    :return: A 2-tuple of admin and library, if the admin is authorized for the library.
+    :raise: ProblemDetailException, if no library or if admin not authorized for the library.
+    """
+    library: Library | None = getattr(request, "library")
+    if library is None:
+        raise ProblemDetailException(LIBRARY_NOT_FOUND)
+
+    admin: Admin = getattr(request, "admin")
+    if not admin.is_librarian(library):
+        raise ProblemDetailException(ADMIN_NOT_AUTHORIZED)
+
+    return admin, library
+
+
 class ReportController(LoggerMixin):
     def __init__(self, db: Session):
         self._db = db
@@ -31,17 +52,8 @@ def inventory_report_info(self) -> Response:
 
         returns: Inventory report info response, if the library exists and
             the admin is authorized.
-            Otherwise, return a 404 response if the library does not exist
-            or raise an ADMIN_NOT_AUTHORIZED ProblemDetailException, if the
-            admin is not authorized.
         """
-        library: Library | None = getattr(flask.request, "library")
-        if library is None:
-            return Response(status=404)
-
-        admin: Admin = getattr(flask.request, "admin")
-        if not admin.is_librarian(library):
-            raise ProblemDetailException(ADMIN_NOT_AUTHORIZED)
+        admin, library = _authorize_from_request(flask.request)
 
         collections = [
             integration.collection
@@ -64,31 +76,25 @@ def inventory_report_info(self) -> Response:
         )
 
     def generate_inventory_report(self) -> Response | ProblemDetail:
-        library: Library = getattr(flask.request, "library")
-        admin: Admin = getattr(flask.request, "admin")
-        try:
-            # these values should never be None
-            assert admin.email
-            assert admin.id
-            assert library.id
+        admin, library = _authorize_from_request(flask.request)
 
+        try:
             task = generate_inventory_and_hold_reports.delay(
                 email_address=admin.email, library_id=library.id
             )
-
-            msg = (
-                f"An inventory and hold report request was received. Report processing can take a few minutes to "
-                f"finish depending on current server load. The completed reports will be sent to {admin.email}."
-            )
-
-            self.log.info(msg + f"(Task Request Id: {task.id})")
-            return Response(
-                json.dumps(dict(message=msg)),
-                HTTPStatus.ACCEPTED,
-                mimetype=MediaTypes.APPLICATION_JSON_MEDIA_TYPE,
-            )
         except Exception as e:
             msg = f"failed to generate inventory report request: {e}"
             self.log.error(msg=msg, exc_info=e)
             self._db.rollback()
             return INTERNAL_SERVER_ERROR.detailed(detail=msg)
+
+        msg = (
+            f"An inventory and hold report request was received. Report processing can take a few minutes to "
+            f"finish depending on current server load. The completed reports will be sent to {admin.email}."
+        )
+        self.log.info(f"({msg} Task Request Id: {task.id})")
+        return Response(
+            json.dumps(dict(message=msg)),
+            HTTPStatus.ACCEPTED,
+            mimetype=MediaTypes.APPLICATION_JSON_MEDIA_TYPE,
+        )

src/palace/manager/api/admin/routes.py

@@ -714,7 +714,9 @@ def diagnostics():
 )
 @api_spec.validate(
     resp=SpecResponse(
-        HTTP_200=InventoryReportInfo, HTTP_403=ProblemDetailModel, HTTP_404=None
+        HTTP_200=InventoryReportInfo,
+        HTTP_403=ProblemDetailModel,
+        HTTP_404=ProblemDetailModel,
     ),
     tags=["admin.inventory"],
 )

tests/manager/api/admin/controller/test_report.py

@@ -1,4 +1,6 @@
+import logging
 from http import HTTPStatus
+from types import SimpleNamespace
 from unittest.mock import patch
 
 import pytest
@@ -11,10 +13,11 @@
 )
 from palace.manager.api.admin.problem_details import ADMIN_NOT_AUTHORIZED
 from palace.manager.api.overdrive import OverdriveAPI
+from palace.manager.api.problem_details import LIBRARY_NOT_FOUND
 from palace.manager.core.opds_import import OPDSAPI
 from palace.manager.sqlalchemy.model.admin import Admin, AdminRole
 from palace.manager.sqlalchemy.util import create
-from palace.manager.util.problem_detail import ProblemDetailException
+from palace.manager.util.problem_detail import ProblemDetail, ProblemDetailException
 from tests.fixtures.api_admin import AdminControllerFixture
 from tests.fixtures.api_controller import ControllerFixture
 from tests.fixtures.database import DatabaseTransactionFixture
@@ -64,6 +67,89 @@ def test_generate_inventory_and_hold_reports(
             email_address=email_address, library_id=library_id
         )
 
+    @patch(
+        "palace.manager.api.admin.controller.report.generate_inventory_and_hold_reports"
+    )
+    def test_generate_report_authorization(
+        self,
+        mock_generate_reports,
+        db: DatabaseTransactionFixture,
+        flask_app_fixture: FlaskAppFixture,
+        caplog: pytest.LogCaptureFixture,
+    ):
+        caplog.set_level(
+            logging.INFO,
+            "palace.manager.api.admin.controller.report",
+        )
+
+        task_id = 7
+        mock_generate_reports.delay.return_value = SimpleNamespace(id=task_id)
+        log_message_suffix = f"Task Request Id: {task_id})"
+
+        controller = ReportController(db.session)
+        method = controller.generate_inventory_report
+
+        library1 = db.library()
+        library2 = db.library()
+
+        sysadmin_email = "[email protected]"
+        librarian_email = "[email protected]"
+
+        sysadmin = flask_app_fixture.admin_user(
+            email=sysadmin_email, role=AdminRole.SYSTEM_ADMIN
+        )
+        librarian1 = flask_app_fixture.admin_user(
+            email=librarian_email, role=AdminRole.LIBRARIAN, library=library1
+        )
+
+        collection = db.collection(
+            protocol=OPDSAPI.label(),
+            settings={"data_source": "test", "external_account_id": "http://url"},
+        )
+        collection.libraries = [library1, library2]
+
+        def assert_and_clear_caplog(
+            response: Response | ProblemDetail, email: str
+        ) -> None:
+            assert isinstance(response, Response)
+            assert response.status_code == 202
+            assert "The completed reports will be sent to" in response.get_json().get(
+                "message"
+            )
+            assert email in response.get_json().get("message")
+            assert log_message_suffix in caplog.text
+            caplog.clear()
+
+        # Sysadmin can get info for any library.
+        with flask_app_fixture.test_request_context(
+            "/", admin=sysadmin, library=library1
+        ):
+            assert_and_clear_caplog(method(), sysadmin_email)
+
+        with flask_app_fixture.test_request_context(
+            "/", admin=sysadmin, library=library2
+        ):
+            assert_and_clear_caplog(method(), sysadmin_email)
+
+        # The librarian for library 1 can get info only for that library...
+        with flask_app_fixture.test_request_context(
+            "/", admin=librarian1, library=library1
+        ):
+            assert_and_clear_caplog(method(), librarian_email)
+        # ... since it does not have an admin role for library2.
+        with flask_app_fixture.test_request_context(
+            "/", admin=librarian1, library=library2
+        ):
+            with pytest.raises(ProblemDetailException) as exc:
+                method()
+        assert exc.value.problem_detail == ADMIN_NOT_AUTHORIZED
+
+        # A library must be provided.
+        with flask_app_fixture.test_request_context("/", admin=sysadmin, library=None):
+            with pytest.raises(ProblemDetailException) as exc:
+                method()
+        assert exc.value.problem_detail == LIBRARY_NOT_FOUND
+
     def test_inventory_report_info(
         self, db: DatabaseTransactionFixture, flask_app_fixture: FlaskAppFixture
     ):
@@ -125,8 +211,9 @@ def test_inventory_report_info(
 
         # A library must be provided.
         with flask_app_fixture.test_request_context("/", admin=sysadmin, library=None):
-            admin_response_none = controller.inventory_report_info()
-        assert admin_response_none.status_code == 404
+            with pytest.raises(ProblemDetailException) as exc:
+                controller.inventory_report_info()
+        assert exc.value.problem_detail == LIBRARY_NOT_FOUND
 
     @pytest.mark.parametrize(
         "protocol, settings, expect_collection",