permissions: update and delete permissions api for records

* Removes update and delete permissions from document serializer.
* Removes unused codes.
* Increases test code coverage.

Co-Authored-by: Aly Badr <aly.badr@rero.ch>

rero_ils/config.py

@@ -388,7 +388,8 @@ def _(x):
         max_result_window=5000000,
         search_factory_imp='rero_ils.query:document_search_factory',
         read_permission_factory_imp=allow_all,
-        list_permission_factory_imp=allow_all
+        list_permission_factory_imp=allow_all,
+        update_permission_factory_imp=librarian_update_permission_factory
     ),
     item=dict(
         pid_type='item',

rero_ils/modules/permissions.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+#
+# RERO ILS
+# Copyright (C) 2019 RERO
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""Permissions for all modules."""
+
+
+from flask import jsonify
+
+from .utils import get_record_class_update_permission_from_route
+
+
+def jsonify_response(
+        can_update=False, can_delete=False, reasons={}):
+    """Jsonify api response."""
+    return jsonify({
+        'update': {'can': can_update},
+        'delete': {'can': can_delete, 'reasons': reasons}
+    })
+
+
+def record_permissions(record_pid=None, route_name=None):
+    """Return record permissions."""
+    try:
+        rec_class, update_permission = \
+            get_record_class_update_permission_from_route(route_name)
+        record = rec_class.get_record_by_pid(record_pid)
+        can_update = update_permission(record).can()
+        if not record:
+            return jsonify_response(reasons='Record not found.')
+
+        if record.can_delete:
+            return jsonify_response(
+                can_update=can_update,
+                can_delete=True)
+        else:
+            return jsonify_response(
+                reasons=record.reasons_not_to_delete(),
+                can_update=can_update)
+    except Exception as error:
+        return jsonify({'status': 'error: Bad request'}), 400

rero_ils/modules/serializers.py

@@ -94,32 +94,35 @@ def preprocess_search_hit(pid, record_hit, links_factory=None, **kwargs):
     @staticmethod
     def add_item_links_and_permissions(record, data, pid):
         """Update the record with action links and permissions."""
-        actions = [
-            'update',
-            'delete'
-        ]
-        permissions = {}
-        action_links = {}
-        for action in actions:
-            permission = JSONSerializer.get_permission(action, pid.pid_type)
-            if permission:
-                can = permission(record, credentials_only=True).can()
-                if can:
-                    action_links[action] = url_for(
-                        'invenio_records_rest.{pid_type}_item'.format(
-                            pid_type=pid.pid_type),
-                        pid_value=pid.pid_value, _external=True)
-                else:
-                    action_key = 'cannot_{action}'.format(action=action)
-                    permissions[action_key] = {
-                        'permission': "permission denied"}
-        if not record.can_delete:
-            permissions.setdefault(
-                'cannot_delete',
-                {}
-            ).update(record.reasons_not_to_delete())
-        data['links'].update(action_links)
-        data['permissions'] = permissions
+        # TODO: remove this function and use the permission api
+        if pid.pid_type != 'doc':
+            actions = [
+                'update',
+                'delete'
+            ]
+            permissions = {}
+            action_links = {}
+            for action in actions:
+                permission = JSONSerializer.get_permission(
+                    action, pid.pid_type)
+                if permission:
+                    can = permission(record, credentials_only=True).can()
+                    if can:
+                        action_links[action] = url_for(
+                            'invenio_records_rest.{pid_type}_item'.format(
+                                pid_type=pid.pid_type),
+                            pid_value=pid.pid_value, _external=True)
+                    else:
+                        action_key = 'cannot_{action}'.format(action=action)
+                        permissions[action_key] = {
+                            'permission': "permission denied"}
+            if not record.can_delete:
+                permissions.setdefault(
+                    'cannot_delete',
+                    {}
+                ).update(record.reasons_not_to_delete())
+            data['links'].update(action_links)
+            data['permissions'] = permissions
         return data
 
     @staticmethod

rero_ils/modules/utils.py

@@ -25,6 +25,7 @@
 import pytz
 from dateutil import parser
 from flask import current_app
+from invenio_records_rest.utils import obj_or_import_string
 
 from .api import IlsRecordIndexer
 
@@ -106,3 +107,16 @@ def read_json_record(json_file, buf_size=1024, decoder=JSONDecoder()):
                 if buffer.startswith(','):
                     # delete records deliminators
                     buffer = buffer[1:].lstrip()
+
+
+def get_record_class_update_permission_from_route(route_name):
+    """Return the record class for a given record route name."""
+    endpoints = current_app.config.get('RECORDS_REST_ENDPOINTS')
+    for endpoint in endpoints.items():
+        record = endpoint[1]
+        list_route = record.get('list_route').replace('/', '')
+        if list_route == route_name:
+            record_class = obj_or_import_string(record.get('record_class'))
+            update_permission = obj_or_import_string(
+                record.get('update_permission_factory_imp'))
+            return record_class, update_permission

rero_ils/modules/views.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+#
+# RERO ILS
+# Copyright (C) 2019 RERO
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""Blueprint used for loading templates for all modules."""
+
+from __future__ import absolute_import, print_function
+
+from functools import wraps
+
+from flask import Blueprint, jsonify
+from flask_login import current_user
+
+from .permissions import record_permissions
+from ..permissions import librarian_permission
+
+api_blueprint = Blueprint(
+    'api_blueprint',
+    __name__,
+    url_prefix=''
+)
+
+
+def check_authentication(func):
+    """Decorator to check authentication for permissions HTTP API."""
+    @wraps(func)
+    def decorated_view(*args, **kwargs):
+        if not current_user.is_authenticated:
+            return jsonify({'status': 'error: Unauthorized'}), 401
+        if not librarian_permission.require().can():
+            return jsonify({'status': 'error: Forbidden'}), 403
+        return func(*args, **kwargs)
+
+    return decorated_view
+
+
+@api_blueprint.route(
+    '/permissions/<route_name>/<record_pid>', methods=['GET'])
+@check_authentication
+def permissions(route_name, record_pid):
+    """HTTP GET request for record permissions.
+
+    Required parameters: route_name, record_pid
+    """
+    return record_permissions(
+        record_pid=record_pid, route_name=route_name)

rero_ils/query.py

@@ -99,19 +99,6 @@ def organisation_search_factory(self, search, query_parser=None):
     return (search, urlkwargs)
 
 
-def library_search_factory(self, search, query_parser=None):
-    """Search factory."""
-    search, urlkwargs = search_factory(self, search)
-    if current_patron:
-        if current_patron.is_system_librarian:
-            search = search.filter(
-                'term', organisation__pid=current_patron.organisation_pid)
-        elif current_patron.is_librarian:
-            search = search.filter(
-                'term', library__pid=current_patron.library_pid)
-    return (search, urlkwargs)
-
-
 def loans_search_factory(self, search, query_parser=None):
     """Loan search factory.
 

setup.py

@@ -95,6 +95,7 @@ def run(self):
             'holdings = rero_ils.modules.holdings.views:blueprint',
         ],
         'invenio_base.api_blueprints': [
+            'rero_ils = rero_ils.modules.views:api_blueprint',
             'circ_policies = rero_ils.modules.circ_policies.views:blueprint',
             'item_types = rero_ils.modules.item_types.views:blueprint',
             'patron_types = rero_ils.modules.patron_types.views:blueprint',

tests/api/test_record_permissions.py

@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+#
+# RERO ILS
+# Copyright (C) 2019 RERO
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""Test record permissions API."""
+
+
+from flask import url_for
+from invenio_accounts.testutils import login_user_via_session
+from utils import get_json
+
+
+def test_document_permissions(
+        client, document, librarian_martigny_no_email,
+        patron_martigny_no_email, ebook_1):
+    """Test document permissions."""
+    # failed: invlaid document pid is given
+    res = client.get(
+        url_for(
+            'api_blueprint.permissions',
+            route_name='documents',
+            record_pid='no_pid'
+        )
+    )
+    assert res.status_code == 401
+    # failed: no logged user
+    res = client.get(
+        url_for(
+            'api_blueprint.permissions',
+            route_name='documents',
+            record_pid=document.pid
+        )
+    )
+    assert res.status_code == 401
+
+    # failed: logged patron and a valid document pid is given
+    login_user_via_session(client, patron_martigny_no_email.user)
+    res = client.get(
+        url_for(
+            'api_blueprint.permissions',
+            route_name='documents',
+            record_pid=document.pid
+        )
+    )
+    assert res.status_code == 403
+
+    # success: logged user and a valid document pid is given
+    login_user_via_session(client, librarian_martigny_no_email.user)
+    res = client.get(
+        url_for(
+            'api_blueprint.permissions',
+            route_name='documents',
+            record_pid=document.pid
+        )
+    )
+    assert res.status_code == 200
+    data = get_json(res)
+    assert 'update' in data
+    assert 'delete' in data
+
+    # success: logged user and a valid document pid is given
+    login_user_via_session(client, librarian_martigny_no_email.user)
+    res = client.get(
+        url_for(
+            'api_blueprint.permissions',
+            route_name='documents',
+            record_pid=ebook_1.pid
+        )
+    )
+    assert res.status_code == 200
+    data = get_json(res)
+    assert 'update' in data
+    assert 'delete' in data
+
+    # failed: invlaid route name
+    res = client.get(
+        url_for(
+            'api_blueprint.permissions',
+            route_name='no_route',
+            record_pid=document.pid
+        )
+    )
+    assert res.status_code == 400