Add report controller and tests.

api/admin/controller/__init__.py

@@ -3,6 +3,7 @@
 from typing import TYPE_CHECKING
 
 from api.admin.controller.quicksight import QuickSightController
+from api.admin.controller.report import ReportController
 
 if TYPE_CHECKING:
     from api.circulation_manager import CirculationManager
@@ -63,3 +64,4 @@ def setup_admin_controllers(manager: CirculationManager):
     manager.admin_announcement_service = AnnouncementSettings(manager._db)
     manager.admin_search_controller = AdminSearchController(manager)
     manager.admin_quicksight_controller = QuickSightController(manager)
+    manager.admin_report_controller = ReportController(manager)

api/admin/controller/report.py

@@ -0,0 +1,34 @@
+import logging
+from http import HTTPStatus
+
+import flask
+from flask import Response
+
+from api.controller.circulation_manager import CirculationManagerController
+from core.model.admin import Admin
+from core.model.asynctask import AsyncTaskType, queue_task
+from core.util.problem_detail import ProblemDetail, ProblemDetailException
+
+
+class ReportController(CirculationManagerController):
+    def generate_inventory_report(self) -> Response | ProblemDetail:
+        log = logging.getLogger(self.__class__.__name__)
+        admin: Admin = getattr(flask.request, "admin")
+        try:
+            email = admin.email
+            data = dict(admin_email=email, admin_id=admin.id)
+            task, is_new = queue_task(
+                self._db, task_type=AsyncTaskType.INVENTORY_REPORT, data=data
+            )
+            self._db.commit()
+
+            msg = (
+                f"An inventory report request was {'already' if not is_new else ''} received at {task.created}. "
+                f"When processing is complete, the report will be sent to {email}."
+            )
+            http_status = HTTPStatus.ACCEPTED if is_new else HTTPStatus.CONFLICT
+
+            return Response(dict(message=msg), http_status)
+        except ProblemDetailException as e:
+            self._db.rollback()
+            return e.problem_detail

api/admin/routes.py

@@ -702,6 +702,13 @@ def diagnostics():
     return app.manager.timestamps_controller.diagnostics()
 
 
+@app.route("/admin/reports/generate_inventory_report", methods=["POST"])
+@returns_json_or_response_or_problem_detail
+@requires_admin
+def generate_inventory_report():
+    return app.manager.admin_report_controller.generate_inventory_report()
+
+
 @app.route("/admin/sign_in_again")
 def admin_sign_in_again():
     """Allows an  admin with expired credentials to sign back in

api/circulation_manager.py

@@ -65,6 +65,7 @@
     from api.admin.controller.patron import PatronController
     from api.admin.controller.patron_auth_services import PatronAuthServicesController
     from api.admin.controller.quicksight import QuickSightController
+    from api.admin.controller.report import ReportController
     from api.admin.controller.reset_password import ResetPasswordController
     from api.admin.controller.sign_in import SignInController
     from api.admin.controller.timestamps import TimestampsController
@@ -110,6 +111,7 @@ class CirculationManager(LoggerMixin):
     admin_search_controller: AdminSearchController
     admin_view_controller: ViewController
     admin_quicksight_controller: QuickSightController
+    admin_report_controller: ReportController
 
     @inject
     def __init__(

tests/api/admin/controller/test_report.py

@@ -0,0 +1,49 @@
+from http import HTTPStatus
+
+import pytest
+
+from core.model import create
+from core.model.admin import Admin, AdminRole
+from tests.fixtures.api_admin import AdminControllerFixture
+from tests.fixtures.api_controller import ControllerFixture
+
+
+class ReportControllerFixture(AdminControllerFixture):
+    def __init__(self, controller_fixture: ControllerFixture):
+        super().__init__(controller_fixture)
+
+
+@pytest.fixture
+def report_fixture(
+    controller_fixture: ControllerFixture,
+) -> ReportControllerFixture:
+    return ReportControllerFixture(controller_fixture)
+
+
+class TestReportController:
+    def test_generate_inventory_report(self, report_fixture: ReportControllerFixture):
+        ctrl = report_fixture.manager.admin_report_controller
+        db = report_fixture.ctrl.db
+
+        system_admin, _ = create(db.session, Admin, email="[email protected]")
+        system_admin.add_role(AdminRole.SYSTEM_ADMIN)
+        default = db.default_library()
+        library1 = db.library()
+        with report_fixture.request_context_with_admin(
+            f"/",
+            admin=system_admin,
+        ) as ctx:
+            response = ctrl.generate_inventory_report()
+            assert response.status_code == HTTPStatus.ACCEPTED
+            assert response.response["message"].__contains__("[email protected]")
+            assert not response.response["message"].__contains__("already")
+
+        # check that when generating a duplicate request a 409 is returned.
+        with report_fixture.request_context_with_admin(
+            f"/",
+            admin=system_admin,
+        ) as ctx:
+            response = ctrl.generate_inventory_report()
+            assert response.status_code == HTTPStatus.CONFLICT
+            assert response.response["message"].__contains__("[email protected]")
+            assert response.response["message"].__contains__("already")