permissions: conditioanlly disable can_include_directlly

invenio_communities/permissions.py

@@ -107,14 +107,22 @@ class CommunityPermissionPolicy(BasePermissionPolicy):
     ]
 
     # who can include a record directly, without a review
-    can_include_directly = [
+    _can_include_directly = [
         ReviewPolicy(
             closed_=[Disable()],
             open_=[CommunityCurators()],
             members_=[CommunityMembers()],
         )
     ]
 
+    can_include_directly = [
+        IfConfig(
+            "RDM_COMMUNITY_REQUIRED_TO_PUBLISH",
+            then_=[Disable()],
+            else_=_can_include_directly,
+        )
+    ]
+
     can_members_add = [
         CommunityManagersForRole(),
         AllowedMemberTypes("group"),

tests/communities/test_community_ui_serializer.py

@@ -7,11 +7,17 @@
 
 """Resources serializers tests."""
 
+from collections import namedtuple
+from functools import partial
+
 from flask import g
+from flask_principal import Identity, RoleNeed
+from invenio_access.permissions import system_identity
 
 from invenio_communities.communities.resources.serializer import (
     UICommunityJSONSerializer,
 )
+from invenio_communities.permissions import CommunityPermissionPolicy
 
 
 def test_ui_serializer(app, community, users, any_user):
@@ -75,3 +81,227 @@ def test_ui_serializer(app, community, users, any_user):
         serialized_record["ui"]["permissions"]
         == require_review_expected_data["permissions"]
     )
+
+
+def test_can_include_directly_closed(
+    app,
+    db,
+    community,
+    anon_identity,
+    any_user,
+    superuser_identity,
+    owner,
+    community_service,
+):
+    """Test 'include_directly' permission with 'closed' review policy."""
+    # Set the community's review policy to 'closed'
+    community_data = community_service.read(system_identity, community.id).data
+    community_data["access"]["review_policy"] = "closed"
+    community_service.update(system_identity, community.id, community_data)
+    community_record = community_service.read(system_identity, community.id)
+
+    # Identities
+    anonymous_identity = anon_identity
+    authenticated_identity = any_user.identity
+    owner_identity = owner.identity
+    superuser_id = superuser_identity
+    system_process_identity = system_identity
+
+    # Community Creator
+    community_creator_identity = Identity(any_user.id)
+    community_creator_identity.provides.update(authenticated_identity.provides)
+    community_creator_identity.provides.add(RoleNeed("community-creator"))
+
+    # Community Member (Owner)
+    member_identity = Identity(owner.id)
+    member_identity.provides.update(owner_identity.provides)
+    # Assume owner is a member
+
+    _Need = namedtuple("Need", ["method", "value", "role"])
+    CommunityRoleNeed = partial(_Need, "community")
+    community_id = str(community.id)
+    member_identity.provides.add(CommunityRoleNeed(community_id, "member"))
+
+    # Community Curator
+    curator_identity = Identity(any_user.id)
+    curator_identity.provides.update(authenticated_identity.provides)
+    curator_identity.provides.add(CommunityRoleNeed(community_id, "curator"))
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to True
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = True
+
+    policy = CommunityPermissionPolicy
+
+    # Assertions: No one can include directly
+    identities = [
+        anonymous_identity,
+        authenticated_identity,
+        community_creator_identity,
+        member_identity,
+        curator_identity,
+        owner_identity,
+        superuser_id,
+        system_process_identity,
+    ]
+
+    for identity in identities:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )
+
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = False
+
+    # Assertions remain the same: No one can include directly
+    for identity in identities:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )
+
+
+def test_can_include_directly_open(
+    app,
+    db,
+    community,
+    anon_identity,
+    any_user,
+    superuser_identity,
+    owner,
+    community_service,
+):
+    """Test 'include_directly' permission with 'open' review policy."""
+    # Set the community's review policy to 'open'
+    community_data = community_service.read(system_identity, community.id).data
+    community_data["access"]["review_policy"] = "open"
+    community_service.update(system_identity, community.id, community_data)
+    community_record = community_service.read(system_identity, community.id)
+
+    # Identities
+    anonymous_identity = anon_identity
+    authenticated_identity = any_user.identity
+    owner_identity = owner.identity
+    system_process_identity = system_identity
+
+    # Create identities with roles
+    from flask_principal import Identity, RoleNeed
+
+    # Community Creator
+    community_creator_identity = Identity(any_user.id)
+    community_creator_identity.provides.update(authenticated_identity.provides)
+    community_creator_identity.provides.add(RoleNeed("community-creator"))
+
+    # Community Curator
+    from collections import namedtuple
+    from functools import partial
+
+    _Need = namedtuple("Need", ["method", "value", "role"])
+    CommunityRoleNeed = partial(_Need, "community")
+    community_id = str(community.id)
+    curator_identity = Identity(any_user.id)
+    curator_identity.provides.update(authenticated_identity.provides)
+    curator_identity.provides.add(CommunityRoleNeed(community_id, "curator"))
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to False
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = False
+
+    policy = CommunityPermissionPolicy
+
+    # Assertions when RDM_COMMUNITY_REQUIRED_TO_PUBLISH is False
+    # Only community curators can include directly
+    # Superuser should not be allowed
+    assert not policy(action="include_directly", record=community_record).allows(
+        anonymous_identity
+    )
+    assert not policy(action="include_directly", record=community_record).allows(
+        authenticated_identity
+    )
+    assert not policy(action="include_directly", record=community_record).allows(
+        community_creator_identity
+    )
+
+    # Community curator can include directly
+    assert policy(action="include_directly", record=community_record).allows(
+        curator_identity
+    )
+
+
+def test_can_include_directly_members(
+    app,
+    db,
+    community,
+    anon_identity,
+    any_user,
+    superuser_identity,
+    owner,
+    community_service,
+):
+    """Test 'include_directly' permission with 'members' review policy."""
+    # Set the community's review policy to 'members'
+    community_data = community_service.read(system_identity, community.id).data
+    community_data["access"]["review_policy"] = "members"
+    community_service.update(system_identity, community.id, community_data)
+    community_record = community_service.read(system_identity, community.id)
+
+    # Identities
+    anonymous_identity = anon_identity
+    authenticated_identity = any_user.identity
+    owner_identity = owner.identity
+    superuser_id = superuser_identity
+    system_process_identity = system_identity
+
+    # Create identities with roles
+    from flask_principal import Identity, RoleNeed
+
+    # Community Creator
+    community_creator_identity = Identity(any_user.id)
+    community_creator_identity.provides.update(authenticated_identity.provides)
+    community_creator_identity.provides.add(RoleNeed("community-creator"))
+
+    # Community Member
+    member_identity = Identity(any_user.id)
+    member_identity.provides.update(authenticated_identity.provides)
+    from collections import namedtuple
+    from functools import partial
+
+    _Need = namedtuple("Need", ["method", "value", "role"])
+    CommunityRoleNeed = partial(_Need, "community")
+    community_id = str(community.id)
+    member_identity.provides.add(CommunityRoleNeed(community_id, "member"))
+
+    # Assume owner is a member
+    owner_member_identity = Identity(owner.id)
+    owner_member_identity.provides.update(owner_identity.provides)
+    owner_member_identity.provides.add(CommunityRoleNeed(community_id, "member"))
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to True
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = True
+
+    policy = CommunityPermissionPolicy
+
+    # Assertions when RDM_COMMUNITY_REQUIRED_TO_PUBLISH is True
+    identities = [
+        anonymous_identity,
+        authenticated_identity,
+        community_creator_identity,
+        member_identity,
+    ]
+
+    # All users are denied
+    for identity in identities:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )
+
+    # Set RDM_COMMUNITY_REQUIRED_TO_PUBLISH to False
+    app.config["RDM_COMMUNITY_REQUIRED_TO_PUBLISH"] = False
+
+    # Assertions when RDM_COMMUNITY_REQUIRED_TO_PUBLISH is False
+    # Only community members can include directly
+    for identity in [
+        anonymous_identity,
+        authenticated_identity,
+        community_creator_identity,
+        system_process_identity,
+    ]:
+        assert not policy(action="include_directly", record=community_record).allows(
+            identity
+        )