From d213661bd86185ad32127410837e6526f12474ad Mon Sep 17 00:00:00 2001 From: Tim DiLauro Date: Fri, 24 May 2024 10:19:51 -0400 Subject: [PATCH] Admin must have role for a library to generate inventory report. (PP-1293) (#1860) * Admin needs rights on library to generate inventory report. --- .../manager/api/admin/controller/report.py | 66 +++++++------ src/palace/manager/api/admin/routes.py | 4 +- .../api/admin/controller/test_report.py | 93 ++++++++++++++++++- 3 files changed, 129 insertions(+), 34 deletions(-) diff --git a/src/palace/manager/api/admin/controller/report.py b/src/palace/manager/api/admin/controller/report.py index fc223d7a7f..f911340046 100644 --- a/src/palace/manager/api/admin/controller/report.py +++ b/src/palace/manager/api/admin/controller/report.py @@ -2,7 +2,7 @@ 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 ( @@ -10,6 +10,7 @@ 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, + ) diff --git a/src/palace/manager/api/admin/routes.py b/src/palace/manager/api/admin/routes.py index c9cfd50278..cd847640b6 100644 --- a/src/palace/manager/api/admin/routes.py +++ b/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"], ) diff --git a/tests/manager/api/admin/controller/test_report.py b/tests/manager/api/admin/controller/test_report.py index 28cd0071ac..c7909eea66 100644 --- a/tests/manager/api/admin/controller/test_report.py +++ b/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 = "sysadmin@example.org" + librarian_email = "librarian@example.org" + + 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",