diff --git a/rero_ils/modules/documents/serializers.py b/rero_ils/modules/documents/serializers.py index bb49ecc6b4..c797129ce7 100644 --- a/rero_ils/modules/documents/serializers.py +++ b/rero_ils/modules/documents/serializers.py @@ -61,11 +61,9 @@ def preprocess_record(self, pid, record, links_factory=None, **kwargs): person = Person.get_record_by_mef_pid(pid_value) if person: authors[idx] = person.dumps_for_document() - data = super(JSONSerializer, self).preprocess_record( + return super(JSONSerializer, self).preprocess_record( pid=pid, record=rec, links_factory=links_factory, kwargs=kwargs) - return JSONSerializer.add_item_links_and_permissions(record, data, pid) - def post_process_serialize_search(self, results, pid_fetcher): """Post process the search results.""" # Item filters. diff --git a/rero_ils/modules/permissions.py b/rero_ils/modules/permissions.py index e063928b9c..a40a4b7c68 100644 --- a/rero_ils/modules/permissions.py +++ b/rero_ils/modules/permissions.py @@ -20,44 +20,50 @@ from flask import jsonify -from .utils import get_record_class_update_permission_from_route +from .utils import get_record_class_permissions_factories_from_route -def jsonify_permission_api_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_update_delete_permissions(record_pid=None, route_name=None): +def record_permissions(record_pid=None, route_name=None): """Return record permissions.""" try: - rec_class, update_permission, delete_permission = \ - get_record_class_update_permission_from_route(route_name) - record = rec_class.get_record_by_pid(record_pid) - - if not record: - return jsonify({'status': 'error: Record not found.'}), 404 - - # We have two behavior for 'can_delete'. Either the record has linked - # resource and so children resources should be deleted before ; either - # the `delete_permissions_factory` for this record should be called. If - # this call send 'False' then the reason_not_to_delete should be - # "permission denied" - can_delete = record.can_delete and delete_permission(record).can() - reasons = record.reasons_not_to_delete() - if not can_delete and not reasons: - # in this case, it's because config delete factory return `False` - # So the reason is 'Permission denied' - reasons = {'others': {'permission': 'permission denied'}} - - return jsonify_permission_api_response( - can_update=update_permission(record).can(), - can_delete=can_delete, - reasons=reasons - ) - except Exception as error: + rec_class, create_permission, update_permission, delete_permission = \ + get_record_class_permissions_factories_from_route(route_name) + + # To check create permission, we don't need to check if the record_pid + # exists. Just call the create permission (if exists) with `None` value + # as record. + permissions = { + 'create': {'can': True} + } + if create_permission: + permissions['create']['can'] = create_permission(record=None).can() + + # If record_pid is not None, we can check about 'delete' and 'update' + # permissions. + if record_pid: + record = rec_class.get_record_by_pid(record_pid) + if not record: + return jsonify({'status': 'error: Record not found.'}), 404 + + # To check if the record could be update, just call the update + # permission factory to get the answer + permissions['update'] = {'can': update_permission(record).can()} + + # We have two behaviors for 'can_delete'. Either the record has + # linked resources and so children resources should be deleted + # before ; either the `delete_permissions_factory` for this record + # should be called. If this call send 'False' then the + # reason_not_to_delete should be "permission denied" + permissions['delete'] = { + 'can': record.can_delete and delete_permission(record).can() + } + reasons = record.reasons_not_to_delete() + if not permissions['delete']['can'] and not reasons: + # in this case, it's because config delete factory return + # `False`, so the reason is 'Permission denied' + reasons = {'others': {'permission': 'permission denied'}} + permissions['delete']['reasons'] = reasons + + return jsonify(permissions) + except Exception: return jsonify({'status': 'error: Bad request'}), 400 diff --git a/rero_ils/modules/serializers.py b/rero_ils/modules/serializers.py index 25a8570fe8..90707ce903 100644 --- a/rero_ils/modules/serializers.py +++ b/rero_ils/modules/serializers.py @@ -17,16 +17,13 @@ """Record serialization.""" -from flask import current_app, json, request, url_for -from invenio_pidstore.errors import PIDDoesNotExistError -from invenio_records_rest import current_records_rest +from flask import json, request, url_for from invenio_records_rest.schemas import \ RecordSchemaJSONV1 as _RecordSchemaJSONV1 from invenio_records_rest.serializers.json import \ JSONSerializer as _JSONSerializer from invenio_records_rest.serializers.response import record_responsify, \ search_responsify -from invenio_records_rest.utils import obj_or_import_string from marshmallow import fields @@ -48,117 +45,36 @@ def preprocess_record(self, pid, record, links_factory=None, **kwargs): rec = record if request and request.args.get('resolve') == '1': rec = record.replace_refs() - data = super(JSONSerializer, self).preprocess_record( + return super(JSONSerializer, self).preprocess_record( pid=pid, record=rec, links_factory=links_factory, kwargs=kwargs) - return JSONSerializer.add_item_links_and_permissions(record, data, pid) - @staticmethod def preprocess_search_hit(pid, record_hit, links_factory=None, **kwargs): """Prepare a record hit from Elasticsearch for serialization.""" - from invenio_records.api import Record - from invenio_pidstore.models import PersistentIdentifier - data = super(JSONSerializer, JSONSerializer).preprocess_search_hit( - pid=pid, record_hit=record_hit, - links_factory=links_factory, kwargs=kwargs) - record_class = obj_or_import_string( - current_app.config - .get('RECORDS_REST_ENDPOINTS') - .get(pid.pid_type).get('record_class', Record)) - try: - persistent_identifier = PersistentIdentifier.get( - pid.pid_type, pid.pid_value) - record = record_class.get_record( - persistent_identifier.object_uuid - ) - json = JSONSerializer.add_item_links_and_permissions( - record, data, pid - ) - permissions = json.get('permissions') - except PIDDoesNotExistError: - permissions = { - 'cannot_update': {'permisson': 'permission denied'}, - 'cannot_delete': {'permisson': 'permission denied'} - } + super(JSONSerializer, JSONSerializer).preprocess_search_hit( + pid=pid, + record_hit=record_hit, + links_factory=links_factory, + kwargs=kwargs + ) search_hit = dict( pid=pid, metadata=record_hit['_source'], links=links_factory(pid, record_hit=record_hit, **kwargs), - revision=record_hit['_version'], - permissions=permissions + revision=record_hit['_version'] ) if record_hit.get('_explanation'): search_hit['explanation'] = record_hit.get('_explanation') return search_hit - @staticmethod - def add_item_links_and_permissions(record, data, pid): - """Update the record with action links and 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 - def get_permission(action, pid_type): - """Get the permission given an action.""" - default_action = getattr( - current_records_rest, - '{action}_permission_factory'.format(action=action)) - permission = obj_or_import_string( - current_app.config - .get('RECORDS_REST_ENDPOINTS') - .get(pid_type) - .get( - '{action}_permission_factory_imp'.format(action=action), - default_action)) - return permission - def post_process_serialize_search(self, results, pid_fetcher): """Post process the search results.""" pid_type = pid_fetcher('foo', dict(pid='1')).pid_type - - # add permissions and links actions - permission = self.get_permission('create', pid_type) - permissions = {} - links = {} - if permission: - can = permission(record=None).can() - if can: - links['create'] = url_for( - 'invenio_records_rest.{pid_type}_list'.format( - pid_type=pid_type), _external=True) - else: - permissions['cannot_create'] = { - 'permission': "permission denied"} - results['permissions'] = permissions - results['links'].update(links) + results['links'].update({ + 'create': url_for('invenio_records_rest.{pid_type}_list'.format( + pid_type=pid_type), _external=True + ) + }) return results def serialize_search(self, pid_fetcher, search_result, links=None, diff --git a/rero_ils/modules/utils.py b/rero_ils/modules/utils.py index 8d255bcb5b..fc874dc9ab 100644 --- a/rero_ils/modules/utils.py +++ b/rero_ils/modules/utils.py @@ -109,19 +109,22 @@ def read_json_record(json_file, buf_size=1024, decoder=JSONDecoder()): buffer = buffer[1:].lstrip() -def get_record_class_update_permission_from_route(route_name): - """Return the record class for a given record route name.""" +def get_record_class_permissions_factories_from_route(route_name): + """Get record class and permission factories for a 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')) + create_permission = obj_or_import_string( + record.get('create_permission_factory_imp')) update_permission = obj_or_import_string( record.get('update_permission_factory_imp')) delete_permission = obj_or_import_string( record.get('delete_permission_factory_imp')) - return record_class, update_permission, delete_permission + return record_class, create_permission, \ + update_permission, delete_permission def get_endpoint_configuration(module): diff --git a/rero_ils/modules/views.py b/rero_ils/modules/views.py index f4cf051a2d..0867700a47 100644 --- a/rero_ils/modules/views.py +++ b/rero_ils/modules/views.py @@ -24,7 +24,7 @@ from flask import Blueprint, jsonify from flask_login import current_user -from .permissions import record_update_delete_permissions +from .permissions import record_permissions from ..permissions import librarian_permission api_blueprint = Blueprint( @@ -47,13 +47,15 @@ def decorated_view(*args, **kwargs): return decorated_view -@api_blueprint.route( - '/permissions//', methods=['GET']) +@api_blueprint.route('/permissions/', methods=['GET']) +@api_blueprint.route('/permissions//', methods=['GET']) @check_authentication -def permissions(route_name, record_pid): +def permissions(route_name, record_pid=None): """HTTP GET request for record permissions. - Required parameters: route_name, record_pid + :param route_name : the list route of the resource + :param record_pid : the record pid + :return a JSON object with create/update/delete permissions for this + record/resource """ - return record_update_delete_permissions( - record_pid=record_pid, route_name=route_name) + return record_permissions(record_pid=record_pid, route_name=route_name) diff --git a/tests/api/test_record_permissions.py b/tests/api/test_record_permissions.py index fa3e774d70..a8d4d8e120 100644 --- a/tests/api/test_record_permissions.py +++ b/tests/api/test_record_permissions.py @@ -16,11 +16,9 @@ # along with this program. If not, see . """Test record permissions API.""" - - from flask import url_for from invenio_accounts.testutils import login_user_via_session -from utils import get_json +from utils import get_json, login_user def test_document_permissions( @@ -59,29 +57,13 @@ def test_document_permissions( # 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) + data = call_api_permissions(client, 'documents', document.pid) 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) + data = call_api_permissions(client, 'documents', ebook_1.pid) assert 'update' in data assert 'delete' in data @@ -96,15 +78,105 @@ def test_document_permissions( assert res.status_code == 400 # failed: permission denied - res = client.get( + data = call_api_permissions(client, 'circ_policies', + circ_policy_short_martigny.pid) + assert data.get('delete', {}).get('can') is False + reasons = data.get('delete', {}).get('reasons', {}) + assert 'others' in reasons and 'permission' in reasons['others'] + + +def test_patrons_permissions( + client, + patron_martigny_no_email, + librarian_martigny_no_email, + librarian2_martigny_no_email, + librarian_saxon_no_email, + system_librarian_martigny_no_email, + system_librarian_sion_no_email, + librarian_sion_no_email +): + """Test serializers for patrons.""" + + # simple librarian ----------------------------------------------- + login_user(client, librarian_martigny_no_email) + # 1) should update and delete a librarian of the same library + data = call_api_permissions(client, 'patrons', + librarian2_martigny_no_email.pid) + assert data['delete']['can'] + assert data['update']['can'] + # 2) should not update and delete a librarian of an other library + data = call_api_permissions(client, 'patrons', + librarian_saxon_no_email.pid) + assert not data['delete']['can'] + assert not data['update']['can'] + # 3) should not update and delete a system librarian + data = call_api_permissions(client, 'patrons', + system_librarian_martigny_no_email.pid) + assert not data['delete']['can'] + assert not data['update']['can'] + + # system librarian ---------------------------------------------- + login_user(client, system_librarian_martigny_no_email) + # should update and delete a librarian of the same library + data = call_api_permissions(client, 'patrons', + librarian2_martigny_no_email.pid) + assert data['delete']['can'] + assert data['update']['can'] + + # should update and delete a librarian of an other library + data = call_api_permissions(client, 'patrons', + librarian_saxon_no_email.pid) + assert data['delete']['can'] + assert data['update']['can'] + + # should update and delete a system librarian of the same organisation + data = call_api_permissions(client, 'patrons', + system_librarian_martigny_no_email.pid) + assert data['delete']['can'] + assert data['update']['can'] + + # should not update and delete a system librarian of an other organisation + data = call_api_permissions(client, 'patrons', + system_librarian_sion_no_email.pid) + assert not data['delete']['can'] + assert not data['update']['can'] + + +def test_items_permissions( + client, + item_lib_martigny, # on shelf + item_lib_fully, # on loan + librarian_martigny_no_email +): + """Test record retrieval.""" + login_user(client, librarian_martigny_no_email) + + data = call_api_permissions(client, 'items', item_lib_fully.pid) + assert not data['delete']['can'] + assert not data['update']['can'] + + data = call_api_permissions(client, 'items', item_lib_martigny.pid) + assert data['delete']['can'] + assert data['update']['can'] + + response = client.get( url_for( 'api_blueprint.permissions', - route_name='circ_policies', - record_pid=circ_policy_short_martigny.pid + route_name='items', + record_pid='dummy_item_pid' ) ) - data = get_json(res) - assert res.status_code == 200 - assert data.get('delete', {}).get('can') is False - reasons = data.get('delete', {}).get('reasons', {}) - assert 'others' in reasons and 'permission' in reasons['others'] + assert response.status_code == 404 + + +def call_api_permissions(client, route_name, pid): + """Get permissions from permissions API.""" + response = client.get( + url_for( + 'api_blueprint.permissions', + route_name=route_name, + record_pid=pid + ) + ) + assert response.status_code == 200 + return get_json(response) diff --git a/tests/api/test_serializers.py b/tests/api/test_serializers.py index 5f6aff8d79..1b2f78cc14 100644 --- a/tests/api/test_serializers.py +++ b/tests/api/test_serializers.py @@ -25,94 +25,15 @@ def test_patrons_serializers( client, json_header, - patron_martigny_no_email, librarian_martigny_no_email, - librarian2_martigny_no_email, - librarian_saxon_no_email, - system_librarian_martigny_no_email, - system_librarian_sion_no_email, - librarian_sion_no_email + librarian2_martigny_no_email ): """Test serializers for patrons.""" - - # simple librarian login_user(client, librarian_martigny_no_email) - # should update and delete a librarian of the same library - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=librarian2_martigny_no_email.pid) - - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - data = get_json(response) - assert 'cannot_delete' not in data['permissions'] - assert 'cannot_update' not in data['permissions'] - assert data['links']['update'] - assert data['links']['delete'] - - # should not update and delete a librarian of an other library - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=librarian_saxon_no_email.pid) - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - assert 'cannot_delete' in get_json(response)['permissions'] - assert 'cannot_update' in get_json(response)['permissions'] - - # should not update and delete a system librarian - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=system_librarian_martigny_no_email.pid) - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - assert 'cannot_delete' in get_json(response)['permissions'] - assert 'cannot_update' in get_json(response)['permissions'] - - # simple librarian - login_user(client, system_librarian_martigny_no_email) - # should update and delete a librarian of the same library - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=librarian2_martigny_no_email.pid) - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - assert 'cannot_delete' not in get_json(response)['permissions'] - assert 'cannot_update' not in get_json(response)['permissions'] - - # should update and delete a librarian of an other library - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=librarian_saxon_no_email.pid) - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - assert 'cannot_delete' not in get_json(response)['permissions'] - assert 'cannot_update' not in get_json(response)['permissions'] - - # should update and delete a system librarian of the same organistion - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=system_librarian_martigny_no_email.pid) - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - assert 'cannot_delete' not in get_json(response)['permissions'] - assert 'cannot_update' not in get_json(response)['permissions'] - - # should not update and delete a system librarian of an other organisation - item_url = url_for( - 'invenio_records_rest.ptrn_item', - pid_value=system_librarian_martigny_no_email.pid) - response = client.get(item_url, headers=json_header) - assert response.status_code == 200 - assert 'cannot_delete' not in get_json(response)['permissions'] - assert 'cannot_update' not in get_json(response)['permissions'] - - list_url = url_for( - 'invenio_records_rest.ptrn_list') - + list_url = url_for('invenio_records_rest.ptrn_list') response = client.get(list_url, headers=json_header) assert response.status_code == 200 - data = get_json(response) def test_items_serializers( @@ -133,8 +54,6 @@ def test_items_serializers( response = client.get(item_url, headers=json_header) assert response.status_code == 200 data = get_json(response) - assert 'cannot_delete' in data['permissions'] - assert 'cannot_update' in data['permissions'] assert data['metadata'].get('item_type').get('$ref') item_url = url_for( @@ -142,8 +61,6 @@ def test_items_serializers( response = client.get(item_url, headers=json_header) assert response.status_code == 200 data = get_json(response) - assert 'cannot_delete' not in data['permissions'] - assert 'cannot_update' not in data['permissions'] assert data['metadata'].get('item_type').get('$ref') item_url = url_for( diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index cc1f9588dc..b430a79c4c 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -24,7 +24,7 @@ from rero_ils.modules.patrons.api import Patron from rero_ils.modules.utils import add_years, extracted_data_from_ref, \ get_endpoint_configuration, get_schema_for_resource, read_json_record -from rero_ils.utils import unique_list +from rero_ils.utils import get_current_language, unique_list def test_unique_list(): @@ -92,3 +92,9 @@ def test_extract_data_from_ref(app, patron_sion_data, assert extracted_data_from_ref('dummy_data', data='record_class') is None assert extracted_data_from_ref('dummy_data', data='record') is None assert extracted_data_from_ref(ptty, data='dummy') is None + + +def test_current_language(app): + """Test current language.""" + # Just test this function return otherwise than None + assert get_current_language()