documents: add permissions for documents with registred URN

 * prevent deleting documents with registered URN
 * prevent deleting pdf files of documents with registered URN

Co-Authored-by: Valeria Granata <valeria@chaw.com>

sonar/modules/documents/api.py

@@ -259,7 +259,8 @@ def create_fulltext_file(self, file):
                 'Error during fulltext extraction of {file} of record '
                 '{record}: {error}'.format(file=file.key,
                                            error=exception,
-                                           record=self['identifiedBy']))
+                                           record=self.get(
+                                               'identifiedBy', self['pid'])))
 
     def create_thumbnail(self, file):
         """Create a thumbnail for record.

sonar/modules/documents/permissions.py

@@ -17,8 +17,10 @@
 
 """Permissions for documents."""
 
-from flask import request
-from invenio_files_rest.models import Bucket
+from flask import current_app, request
+from invenio_files_rest.models import Bucket, ObjectVersion
+from invenio_pidstore.errors import PIDDoesNotExistError
+from invenio_pidstore.models import PersistentIdentifier
 
 from sonar.modules.documents.api import DocumentRecord
 from sonar.modules.organisations.api import current_organisation
@@ -119,8 +121,36 @@ def delete(cls, user, record):
         if not user or not user.is_admin:
             return False
 
-        # Same rules as update
-        return cls.update(user, record)
+        # Check delete conditions and consider same rules as update
+        return cls.can_delete(user, record) and cls.update(user, record)
+
+    @classmethod
+    def can_delete(cls, user, record):
+        """Delete permission conditions.
+
+        :param user: Current user record.
+        :param record: Record to check.
+        :returns: True is action can be done.
+        """
+        # Delete only documents with no registred URN
+        try:
+            document = DocumentRecord.get_record_by_pid(record['pid'])
+            if document:
+                urn_identifier = PersistentIdentifier\
+                    .get_by_object('urn', 'rec', document.id)
+                urn_config = current_app.config.get("SONAR_APP_DOCUMENT_URN")
+                org_pid = document.replace_refs()\
+                    .get("organisation", [{}])[0].get("pid")
+
+                if config := urn_config.get("organisations", {}).get(org_pid):
+                    if record.get("documentType") in config.get("types")\
+                        and urn_identifier\
+                            and urn_identifier.is_registered():
+                        return False
+        except PIDDoesNotExistError:
+            pass
+
+        return True
 
 class DocumentFilesPermission(FilesPermission):
     """Documents files permissions.
@@ -195,4 +225,10 @@ def delete(cls, user, record, pid, parent_record):
         :param parent_record: the record related to the bucket.
         :returns: True is action can be done.
         """
+        document = cls.get_document(parent_record)
+        if isinstance(record, ObjectVersion):
+            file_type = document.files[record.key]['type']
+            if file_type == 'file' and record.mimetype == 'application/pdf':
+                return DocumentPermission.can_delete(user, parent_record)\
+                    and cls.update(user, record, pid, parent_record)
         return cls.update(user, record, pid, parent_record)

sonar/modules/documents/urn.py

@@ -98,13 +98,17 @@ def create_urn(cls, record):
                         object_uuid=record.id,
                         status=PIDStatus.NEW,
                     )
-                    record["identifiedBy"].append(
-                        {"type": "bf:Urn", "value": urn_code}
-                    )
+                    if "identifiedBy" in record:
+                        record["identifiedBy"].append(
+                            {"type": "bf:Urn", "value": urn_code}
+                        )
+                    else:
+                        record["identifiedBy"] = \
+                            [{"type": "bf:Urn", "value": urn_code}]
                 except PIDAlreadyExists:
                     current_app.logger.error(
                         'generated urn already exist for document: '
-                            + record.get('pid'))
+                        + record.get('pid'))
 
     @classmethod
     def urn_query(cls, status=None):
@@ -114,9 +118,8 @@ def urn_query(cls, status=None):
         :returns: Base query.
         """
         return PersistentIdentifier.query\
-                .filter_by(pid_type=cls.urn_pid_type)\
-                .filter_by(status=status)
-
+            .filter_by(pid_type=cls.urn_pid_type)\
+            .filter_by(status=status)
 
     @classmethod
     def get_urn_pids(cls, status=PIDStatus.NEW, days=None):
@@ -138,7 +141,6 @@ def get_urn_pids(cls, status=PIDStatus.NEW, days=None):
             count = query.count()
         return count
 
-
     @classmethod
     def get_unregistered_urns(cls):
         """Get list of unregistered URNs.
@@ -161,8 +163,8 @@ def register_urn_code_from_document(cls, record):
         if DnbUrnService.register_document(record):
             urn_code = DocumentRecord.get_rero_urn_code(record)
             pid = PersistentIdentifier.query\
-                    .filter_by(pid_type=cls.urn_pid_type)\
-                    .filter_by(pid_value=urn_code).first()
+                .filter_by(pid_type=cls.urn_pid_type)\
+                .filter_by(pid_value=urn_code).first()
             if pid and pid.status == PIDStatus.NEW:
                 pid.status = PIDStatus.REGISTERED
                 db.session.commit()
@@ -177,7 +179,6 @@ def register_urn_pid(cls, urn=None):
             pid.status = PIDStatus.REGISTERED
             db.session.commit()
 
-
     @classmethod
     def get_documents_to_generate_urns(cls):
         """Get documents that need a URN code..

tests/api/documents/test_documents_files_permissions.py

@@ -17,21 +17,25 @@
 
 """Test documents files permissions."""
 
+from io import BytesIO
 
 from flask import url_for
 from flask_security import url_for_security
 from invenio_accounts.testutils import login_user_via_session
 
+from sonar.modules.utils import wait_empty_tasks
+
 
 def test_update_delete(client, superuser, admin, moderator,
-              submitter, user, document, pdf_file):
+                       submitter, user, document, pdf_file):
     """Test permissions for uploading and deleting files."""
     file_name = 'test.pdf'
     users = [superuser, admin, moderator, submitter, user, None]
 
     # upload the file
     url_file_content = url_for(
-        'invenio_records_files.doc_object_api', pid_value=document.get('pid'), key=file_name)
+        'invenio_records_files.doc_object_api',
+        pid_value=document.get('pid'), key=file_name)
     for u, status in zip(users, [200, 200, 200, 404, 404, 404]):
         if u:
             login_user_via_session(client, email=u['email'])
@@ -46,9 +50,8 @@ def test_update_delete(client, superuser, admin, moderator,
             assert res.status_code == status
 
 
-
 def test_read_metadata(client, superuser, admin, moderator,
-              submitter, user, document_with_file):
+                       submitter, user, document_with_file):
     """Test read files permissions."""
 
     users = [superuser, admin, moderator, submitter, user, None]
@@ -74,8 +77,9 @@ def test_read_metadata(client, superuser, admin, moderator,
         res = client.get(url_files)
         assert res.status_code == status
 
+
 def test_read_content(client, superuser, admin, moderator,
-              submitter, user, user_without_org, document_with_file):
+                      submitter, user, user_without_org, document_with_file):
     """Test read documents permissions."""
 
     file_name = 'test1.pdf'
@@ -84,7 +88,7 @@ def test_read_content(client, superuser, admin, moderator,
     url_file_content = url_for(
         'invenio_records_files.doc_object_api',
         pid_value=document_with_file.get('pid'),
-    key=file_name)
+        key=file_name)
     open_access_code = "coar:c_abf2"
     document_with_file.files[file_name]['access'] = open_access_code
     document_with_file.commit()
@@ -111,7 +115,8 @@ def test_read_content(client, superuser, admin, moderator,
     # restricted files
     restricted_code = "coar:c_16ec"
     document_with_file.files[file_name]['access'] = restricted_code
-    document_with_file.files[file_name]['restricted_outside_organisation'] = True
+    document_with_file.files[file_name]['restricted_outside_organisation'] \
+        = True
     document_with_file.commit()
     for u, status in zip(users, [200, 200, 200, 200, 200, 404, 404]):
         if u:
@@ -120,3 +125,36 @@ def test_read_content(client, superuser, admin, moderator,
             client.get(url_for_security('logout'))
         res = client.get(url_file_content)
         assert res.status_code == status
+
+
+def test_file_of_document_with_urn_delete(client, superuser,
+                                          minimal_thesis_document):
+    """Test delete file of document with registered URN identifier."""
+    # Logged as superuser
+    login_user_via_session(client, email=superuser['email'])
+
+    # Add pdf file to document
+    minimal_thesis_document.files['test.pdf'] = BytesIO(b'File content')
+    minimal_thesis_document.files['test.pdf']['type'] = 'file'
+    minimal_thesis_document.commit()
+
+    wait_empty_tasks(delay=3, verbose=True)
+
+    url_file_content = url_for(
+        'invenio_records_files.doc_object_api',
+        pid_value=minimal_thesis_document['pid'], key='test.pdf')
+    res = client.delete(url_file_content)
+    assert res.status_code == 403
+
+    # Add png file to document
+    minimal_thesis_document.files['test.png'] = BytesIO(b'File content')
+    minimal_thesis_document.files['test.png']['type'] = 'file'
+    minimal_thesis_document.commit()
+
+    wait_empty_tasks(delay=3, verbose=True)
+
+    url_file_content = url_for(
+        'invenio_records_files.doc_object_api',
+        pid_value=minimal_thesis_document['pid'], key='test.png')
+    res = client.delete(url_file_content)
+    assert res.status_code == 204

tests/api/documents/test_documents_permissions.py

@@ -18,12 +18,15 @@
 """Test documents permissions."""
 
 import json
+from io import BytesIO
 
 import mock
 from flask import url_for
 from invenio_accounts.testutils import login_user_via_session
 from invenio_pidstore.models import PersistentIdentifier, PIDStatus
 
+from sonar.modules.utils import wait_empty_tasks
+
 
 def test_list(app, client, make_document, superuser, admin, moderator,
               submitter, user):
@@ -157,7 +160,8 @@ def test_read(client, document, make_user, superuser, admin, moderator,
         magic_mock
     ):
         res = client.get(
-            url_for('invenio_records_rest.doc_item', pid_value=document['pid']))
+            url_for('invenio_records_rest.doc_item',
+                    pid_value=document['pid']))
         assert res.status_code == 401
 
     # Logged as user
@@ -184,7 +188,7 @@ def test_read(client, document, make_user, superuser, admin, moderator,
     }
     assert res.json['metadata']['partOf'][0][
         'text'] == 'Journal du dimanche / Renato, Ferrari ; Albano, Mesta. ' \
-            '- John Doe Publications inc.. - 2020, vol. 6, no. 12, p. 135-139'
+        '- John Doe Publications inc.. - 2020, vol. 6, no. 12, p. 135-139'
     assert res.json['metadata']['provisionActivity'][0]['text'] == {
         'default':
         'Bienne : Impr. Weber, [2006] ; Lausanne ; Rippone : Impr. Coustaud'
@@ -390,3 +394,20 @@ def test_delete(client, document, make_document, make_user, superuser, admin,
     assert res.status_code == 204
     assert PersistentIdentifier.get('ark', ark_id).status == \
         PIDStatus.DELETED
+
+
+def test_document_with_urn_delete(client, superuser, minimal_thesis_document):
+    """Test delete document with registered URN identifier."""
+    # Add file to document
+    minimal_thesis_document.files['test.pdf'] = BytesIO(b'File content')
+    minimal_thesis_document.files['test.pdf']['type'] = 'file'
+    minimal_thesis_document.commit()
+
+    wait_empty_tasks(delay=3, verbose=True)
+
+    # Logged as superuser
+    login_user_via_session(client, email=superuser['email'])
+    pid = minimal_thesis_document['pid']
+    res = client.delete(url_for('invenio_records_rest.doc_item',
+                                pid_value=pid))
+    assert res.status_code == 403