permissions: evaluate per transition #26

Added support for custom permission factory per transition.

invenio_circulation/config.py

@@ -8,6 +8,8 @@
 
 """Invenio module for the circulation of bibliographic items."""
 
+from invenio_records_rest.utils import allow_all
+
 from .api import Loan
 from .links import loan_links_factory
 from .transitions.transitions import CreatedToItemOnLoan, CreatedToPending, \
@@ -33,6 +35,8 @@
 _CIRCULATION_LOAN_LINKS_FACTORY = loan_links_factory
 """."""
 
+CIRCULATION_PERMISSION_FACTORY = allow_all
+"""."""
 
 CIRCULATION_STATES_ITEM_AVAILABLE = ['ITEM_RETURNED']
 """."""

invenio_circulation/errors.py

@@ -37,5 +37,9 @@ class LoanActionError(CirculationException):
     """."""
 
 
+class InvalidCirculationPermission(CirculationException):
+    """Raised when permissions are not satisfied for transition."""
+
+
 class TransitionConstraintsViolation(CirculationException):
     """Exception raised when constraints for the transition failed."""

invenio_circulation/transitions/base.py

@@ -13,8 +13,9 @@
 from flask import current_app
 
 from ..api import is_item_available
-from ..errors import InvalidState, ItemNotAvailable, \
-    TransitionConditionsFailed, TransitionConstraintsViolation
+from ..errors import InvalidCirculationPermission, InvalidState, \
+    ItemNotAvailable, TransitionConditionsFailed, \
+    TransitionConstraintsViolation
 from ..signals import loan_state_changed
 from ..utils import parse_date
 
@@ -86,7 +87,8 @@ def __init__(self, src, dest, trigger='next', permission_factory=None,
         self.src = src
         self.dest = dest
         self.trigger = trigger
-        self.permission_factory = permission_factory
+        self.permission_factory = permission_factory or \
+            current_app.config['CIRCULATION_PERMISSION_FACTORY']
         # validate states
         self.validate_transition_states()
 
@@ -111,6 +113,10 @@ def validate_transition_states(self):
     @check_trigger
     def before(self, loan, **kwargs):
         """Validate input, evaluate conditions and raise if failed."""
+        if self.permission_factory and not self.permission_factory(loan).can():
+            msg = 'Invalid circulation permission'
+            raise InvalidCirculationPermission(msg=msg)
+
         kwargs.setdefault('transaction_date', datetime.now())
         kwargs['transaction_date'] = parse_date(kwargs['transaction_date'])
         loan.update(kwargs)

invenio_circulation/views.py

@@ -19,8 +19,8 @@
 from invenio_rest import ContentNegotiatedMethodView
 from invenio_rest.views import create_api_errorhandler
 
-from invenio_circulation.errors import ItemNotAvailable, LoanActionError, \
-    NoValidTransitionAvailable
+from invenio_circulation.errors import InvalidCirculationPermission, \
+    ItemNotAvailable, LoanActionError, NoValidTransitionAvailable
 from invenio_circulation.proxies import current_circulation
 
 HTTP_CODES = {
@@ -125,7 +125,11 @@ def post(self, pid, record, action, **kwargs):
                 record, **dict(params, trigger=action)
             )
             db.session.commit()
-        except (ItemNotAvailable, NoValidTransitionAvailable) as ex:
+        except (
+            ItemNotAvailable,
+            InvalidCirculationPermission,
+            NoValidTransitionAvailable
+        ) as ex:
             current_app.logger.exception(ex.msg)
             raise LoanActionError(ex)
 

tests/test_permissions.py

@@ -0,0 +1,36 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2018 CERN.
+# Copyright (C) 2018 RERO.
+#
+# Invenio-Circulation is free software; you can redistribute it and/or modify
+# it under the terms of the MIT License; see LICENSE file for more details.
+
+
+"""Test circulation permissions on transitions."""
+
+import pytest
+from invenio_records_rest.utils import allow_all, deny_all
+
+from invenio_circulation.api import Loan
+from invenio_circulation.errors import InvalidCirculationPermission
+from invenio_circulation.transitions.transitions import CreatedToPending
+
+
+def test_valid_permission(loan_created, params):
+    """Test transition with valid permission."""
+    transition = CreatedToPending(
+        'CREATED', 'PENDING', trigger='next', permission_factory=allow_all
+    )
+    transition.execute(loan_created, **params)
+    assert loan_created['state'] == 'PENDING'
+
+
+def test_invalid_permission(loan_created, params):
+    """Test transition without permission."""
+    transition = CreatedToPending(
+        'CREATED', 'PENDING', trigger='next', permission_factory=deny_all
+    )
+    with pytest.raises(InvalidCirculationPermission):
+        transition.execute(loan_created, **params)
+    assert loan_created['state'] == 'CREATED'