feat(permissions): add RDM_ALLOW_OWNERS_REMOVE_COMMUNITY_FROM_RECORD

* add ``RDM_ALLOW_OWNERS_REMOVE_COMMUNITY_FROM_RECORD`` configuration
  variable to control whether record owners can remove communities from records.
  When set to False, only community curators, managers, and owners can remove
  communities.
* closes CERNDocumentServer/cds-rdm#630

Author: jrcastro2

invenio_rdm_records/config.py

@@ -273,6 +273,16 @@ def always_valid(identifier):
 """Enforces at least one community per record."""
 RDM_COMMUNITY_INCLUSION_REQUEST_CLS = CommunityInclusion
 """Request type for record inclusion requests."""
+RDM_ALLOW_OWNERS_REMOVE_COMMUNITY_FROM_RECORD = True
+"""Allow record owners to remove communities from records.
+
+When set to False, only community curators, managers, and owners can remove
+communities from records. This applies to published records.
+Record owners will still be able to add communities, but removal is restricted
+to community curators, managers, and owners.
+
+Default: True (backwards compatible - owners can remove communities)
+"""
 
 #
 # Search configuration

invenio_rdm_records/services/communities/service.py

@@ -217,7 +217,10 @@ def _remove(self, identity, community, record, valid_data, errors, uow):
 
         try:
             self.require_permission(
-                identity, "remove_community", record=record, community_id=community_id
+                identity,
+                "remove_community_from_record",
+                record=record,
+                community_id=community_id,
             )
             # By default, admin/superuser has permission to do everything,
             # so PermissionDeniedError won't be raised for admin in any case
@@ -233,7 +236,7 @@ def _remove(self, identity, community, record, valid_data, errors, uow):
 
         # Run components
         self.run_components(
-            "remove_community",
+            "remove_community_from_record",
             identity,
             record=record,
             community=community,

invenio_rdm_records/services/permissions.py

@@ -251,21 +251,23 @@ class RDMRecordPermissionPolicy(RecordPermissionPolicy):
     # Who can add record to a community
     can_add_community = can_manage
     # Who can remove a community from a record
-    can_remove_community_ = [
-        RecordOwners(),
-        CommunityCurators(),
-        SystemProcess(),
+    can_remove_community_from_record_ = [
+        IfConfig(
+            "RDM_ALLOW_OWNERS_REMOVE_COMMUNITY_FROM_RECORD",
+            then_=[RecordOwners(), CommunityCurators(), SystemProcess()],
+            else_=[CommunityCurators(), SystemProcess()],
+        ),
     ]
-    can_remove_community = [
+    can_remove_community_from_record = [
         IfConfig(
             "RDM_COMMUNITY_REQUIRED_TO_PUBLISH",
             then_=[
                 IfOneCommunity(
                     then_=[Administration(), SystemProcess()],
-                    else_=can_remove_community_,
+                    else_=can_remove_community_from_record_,
                 ),
             ],
-            else_=can_remove_community_,
+            else_=can_remove_community_from_record_,
         ),
     ]
     # Who can remove records from a community

tests/resources/test_resources_communities.py

@@ -569,7 +569,7 @@ def test_invalid_record_or_draft(
     assert response.status_code == 404
 
 
-def test_remove_community(
+def test_remove_community_from_record(
     client, uploader, curator, record_community, headers, community
 ):
     """Test removal of a community from the record."""
@@ -1110,3 +1110,60 @@ def test_remove_record_last_community_with_multiple_communities(
     record_saved = client.get(f"/records/{record_pid}", headers=headers)
     # check that only one community ID is associated with the record
     assert len(record_saved.json["parent"]["communities"]["ids"]) == 1
+
+
+def test_remove_community_from_record_restricted_to_curators(
+    db, client, uploader, record_community, headers, community, monkeypatch
+):
+    """Test that record owners cannot remove communities when config is disabled."""
+    from flask import current_app
+
+    # Create a record with a community as uploader (record owner)
+    client_uploader = uploader.login(client)
+    record = record_community.create_record()
+
+    # Verify the community is added
+    record_data = client_uploader.get(
+        f"/records/{record.pid.pid_value}", headers=headers
+    )
+    assert community.id in record_data.json["parent"]["communities"]["ids"]
+
+    # Try to remove community as owner - should succeed with default config
+    data = {"communities": [{"id": community.id}]}
+    response = client_uploader.delete(
+        f"/records/{record.pid.pid_value}/communities",
+        headers=headers,
+        json=data,
+    )
+    assert response.status_code == 200
+    assert not response.json.get("errors")
+
+    # Re-add the community for next test
+    record = _add_to_community(db, record, community)
+
+    # Now disable the config - owners should NOT be able to remove
+    monkeypatch.setitem(
+        current_app.config, "RDM_ALLOW_OWNERS_REMOVE_COMMUNITY_FROM_RECORD", False
+    )
+
+    # Try to remove community as owner - should fail
+    response = client_uploader.delete(
+        f"/records/{record.pid.pid_value}/communities",
+        headers=headers,
+        json=data,
+    )
+
+    # Should get errors (permission denied for owner)
+    assert response.status_code == 400
+    assert len(response.json.get("errors", [])) > 0
+
+    # Verify community is still there
+    record_data = client_uploader.get(
+        f"/records/{record.pid.pid_value}", headers=headers
+    )
+    assert community.id in record_data.json["parent"]["communities"]["ids"]
+
+    # Reset config
+    monkeypatch.setitem(
+        current_app.config, "RDM_ALLOW_OWNERS_REMOVE_COMMUNITY_FROM_RECORD", True
+    )

tests/resources/test_resources_community_records.py

@@ -18,7 +18,9 @@ def service():
     return current_community_records_service
 
 
-def test_remove_community(client, curator, record_community, headers, community):
+def test_remove_community_from_record(
+    client, curator, record_community, headers, community
+):
     """Remove a record from the community."""
     client = curator.login(client)
     record = record_community.create_record()

tests/services/test_service_record_communities.py

@@ -98,14 +98,14 @@ def add_community(self, identity, record, communities, **kwargs):
     assert "dummy_id" not in [c["community_id"] for c in requests]
 
 
-def test_remove_community_component_called(
+def test_remove_community_from_record_component_called(
     db, community, community2, uploader, record_factory, set_app_config_fn_scoped
 ):
-    """The component `remove_community` method is called and blocks the removal."""
+    """The component `remove_community_from_record` method is called and blocks the removal."""
     test_community_id = community.id
 
     class MockRemoveComponent(ServiceComponent):
-        def remove_community(
+        def remove_community_from_record(
             self, identity, record, community, valid_data, errors, **kwargs
         ):
             if community["id"] == str(community2.id):