Add test for the metadata edit notification

tests/test_misc.py

@@ -10,11 +10,17 @@
 from logging.handlers import SMTPHandler
 
 from flask import g
+from invenio_access.permissions import Identity, system_identity
 from invenio_db import db
+from invenio_rdm_records.proxies import current_rdm_records_service as service
+from invenio_records_resources.services.uow import UnitOfWork
 
 import invenio_config_tuw
 from invenio_config_tuw.startup import register_smtp_error_handler
-from invenio_config_tuw.tasks import send_publication_notification
+from invenio_config_tuw.tasks import (
+    send_metadata_edit_notification,
+    send_publication_notification,
+)
 from invenio_config_tuw.users.utils import current_user_as_creator
 
 
@@ -50,6 +56,53 @@ def test_send_publication_notification_email(example_record, mocker):
         )
 
 
+def test_send_edit_notification_email(example_record, users, mocker):
+    """Test if published record edits send a notification to the record owner."""
+    recid = example_record.pid.pid_value
+    old_check_permission = service.check_permission
+    service.check_permission = (lambda *args, **kwargs: True).__get__(
+        service, type(service)
+    )
+
+    def _update_draft(title, identity):
+        """Update the draft with the given title."""
+        draft = service.edit(identity, recid)
+        new_data = draft.data.copy()
+        new_data["metadata"]["title"] = title
+        service.update_draft(identity, recid, new_data)
+
+        uow = UnitOfWork()
+        service.publish(identity, recid, uow=uow)
+        uow.commit()
+        return uow
+
+    # the system identity *should not* trigger the notification
+    uow = _update_draft("Different title", system_identity)
+    assert not [
+        op
+        for op in uow._operations
+        if getattr(op, "_celery_task", None) == send_metadata_edit_notification
+    ]
+
+    # the owner *should not* trigger the notification
+    uow = _update_draft("Yet another title", Identity(users[0].id))
+    assert not [
+        op
+        for op in uow._operations
+        if getattr(op, "_celery_task", None) == send_metadata_edit_notification
+    ]
+
+    # another user *should* trigger the notification
+    uow = _update_draft("Last new title", Identity(users[1].id))
+    assert [
+        op
+        for op in uow._operations
+        if getattr(op, "_celery_task", None) == send_metadata_edit_notification
+    ]
+
+    service.check_permission = old_check_permission
+
+
 def test_record_metadata_current_user_as_creator(client_with_login):
     """Test the auto-generation of a "creator" entry for the current user."""
     user = client_with_login._user