permissions and tests refactor

oarepo_communities/permissions/presets.py

@@ -6,15 +6,24 @@
 )
 from invenio_records_permissions.policies.base import BasePermissionPolicy
 
-from .record import RecordCommunitiesGenerator
+from .record import CommunityRolePermittedInCF
 
 
 class CommunityPermissionPolicy(RecordPermissionPolicy):
     can_search = [SystemProcess(), AnyUser()]
-    can_read = [SystemProcess(), RecordCommunitiesGenerator("can_read")]
+    can_read = [
+        SystemProcess(),
+        CommunityRolePermittedInCF(community_permission_name="can_read"),
+    ]
     can_create = [SystemProcess(), AuthenticatedUser()]
-    can_update = [SystemProcess(), RecordCommunitiesGenerator("can_update")]
-    can_delete = [SystemProcess(), RecordCommunitiesGenerator("can_delete")]
+    can_update = [
+        SystemProcess(),
+        CommunityRolePermittedInCF(community_permission_name="can_update"),
+    ]
+    can_delete = [
+        SystemProcess(),
+        CommunityRolePermittedInCF(community_permission_name="can_delete"),
+    ]
     can_manage = [SystemProcess()]
 
     can_create_files = [SystemProcess()]
@@ -31,7 +40,10 @@ class CommunityPermissionPolicy(RecordPermissionPolicy):
     can_read_draft = [SystemProcess()]
     can_update_draft = [SystemProcess()]
     can_delete_draft = [SystemProcess()]
-    can_publish = [SystemProcess(), RecordCommunitiesGenerator("can_publish")]
+    can_publish = [
+        SystemProcess(),
+        CommunityRolePermittedInCF(community_permission_name="can_publish_directly"),
+    ]
     can_draft_create_files = [SystemProcess()]
     can_draft_set_content_files = [SystemProcess()]
     can_draft_get_content_files = [SystemProcess()]
@@ -49,13 +61,13 @@ class CommunitiesEveryonePermissionPolicy(BasePermissionPolicy):
 class CommunitiesFromCFPermissionPolicy(BasePermissionPolicy):
     can_add_community = [
         SystemProcess(),
-        RecordCommunitiesGenerator("can_add_community"),
+        CommunityRolePermittedInCF(community_permission_name="can_add_community"),
     ]
     can_remove_community = [
         SystemProcess(),
-        RecordCommunitiesGenerator("can_remove_community"),
+        CommunityRolePermittedInCF(community_permission_name="can_remove_community"),
     ]
     can_remove_record = [
         SystemProcess(),
-        RecordCommunitiesGenerator("can_remove_record"),
+        CommunityRolePermittedInCF(community_permission_name="can_remove_record"),
     ]

oarepo_communities/permissions/record.py

@@ -7,11 +7,11 @@
 from ..proxies import current_communities_permissions
 
 
-class RecordCommunitiesGenerator(Generator):
+class CommunityRolePermittedInCF(Generator):
     """Allows system_process role."""
 
-    def __init__(self, action):
-        self.action = action
+    def __init__(self, community_permission_name):
+        self.community_permission_name = community_permission_name
 
     def needs(self, **kwargs):
         if "record" in kwargs and hasattr(kwargs["record"], "parent"):
@@ -20,15 +20,17 @@ def needs(self, **kwargs):
                 community_ids = record.parent["communities"]["ids"]
             except KeyError:
                 return []
-            return needs_from_community_ids(community_ids, self.action)
+            return needs_from_community_ids(
+                community_ids, self.community_permission_name
+            )
         return []
 
 
-def needs_from_community_ids(community_ids, action):
+def needs_from_community_ids(community_ids, community_permission_name):
     _needs = set()
     by_community_permission = record_community_permissions(frozenset(community_ids))
-    if action in by_community_permission:
-        community2role_list = by_community_permission[action]
+    if community_permission_name in by_community_permission:
+        community2role_list = by_community_permission[community_permission_name]
         for community_id, roles in community2role_list.items():
             for role in roles:
                 _needs.add(CommunityRoleNeed(community_id, role))

tests/test_permissions/conftest.py

@@ -176,24 +176,28 @@ def community_permissions_cf():
             "permissions": {
                 "owner": {
                     "can_publish": True,
+                    "can_publish_directly": True,
                     "can_read": True,
                     "can_update": True,
                     "can_delete": True,
                 },
                 "manager": {
                     "can_publish": True,
+                    "can_publish_directly": True,
                     "can_read": False,
                     "can_update": False,
                     "can_delete": False,
                 },
                 "curator": {
                     "can_publish": True,
+                    "can_publish_directly": True,
                     "can_read": True,
                     "can_update": True,
                     "can_delete": False,
                 },
                 "reader": {
                     "can_publish": False,
+                    "can_publish_directly": True,
                     "can_read": True,
                     "can_update": False,
                     "can_delete": False,
@@ -254,7 +258,6 @@ def community(app, community_owner_helper, minimal_community):
 @pytest.fixture()
 def community_with_permissions_cf(community, community_permissions_cf, vocab_cf):
     data = current_communities.service.read(system_identity, community.id).data
-    # data = _resp_to_input(community.data)
     data |= community_permissions_cf
     community = current_communities.service.update(system_identity, data["id"], data)
     Community.index.refresh()

tests/test_permissions/test_permissions_api.py

@@ -1,13 +1,11 @@
-from invenio_communities import current_communities
-from invenio_communities.communities.records.api import Community
 from thesis.resources.record_communities.config import (
     ThesisRecordCommunitiesResourceConfig,
 )
 
 RECORD_COMMUNITIES_BASE_URL = ThesisRecordCommunitiesResourceConfig.url_prefix
 
 
-def _create_and_publish(client, input_data, community, publish_authorized):
+def _create_and_publish(client, input_data, community):
     """Create a draft and publish it."""
     # Create the draft
     response = client.post(RECORD_COMMUNITIES_BASE_URL, json=input_data)
@@ -23,117 +21,67 @@ def _create_and_publish(client, input_data, community, publish_authorized):
             ]
         },
     )
-
     # Publish it
     response = client.post(
         f"{RECORD_COMMUNITIES_BASE_URL}{recid}/draft/actions/publish"
     )
-    if publish_authorized:
-        assert response.status_code == 202
-    else:
-        assert response.status_code == 403
-    return recid
-
-
-def _resp_to_input(resp):
-    return {
-        "slug": resp["slug"],
-        "metadata": resp["metadata"],
-        "access": resp["access"],
-        "id": resp["id"],
-    }
-
-
-def _community_with_permissions_cf(community, identity, community_permissions_cf):
-    data = _resp_to_input(community.data)
-    data |= community_permissions_cf
-    community = current_communities.service.update(identity, data["id"], data)
-    Community.index.refresh()
-    return community
-
-
-def _recid_with_community(
-    owner_client,
-    input_data,
-    community,
-    community_owner,
-    community_permissions_cf,
-    publish_authorized=True,
-):
-    comm = _community_with_permissions_cf(
-        community, community_owner.identity, community_permissions_cf
-    )
-    recid = _create_and_publish(owner_client, input_data, comm, publish_authorized)
-    return recid
+    return response
 
 
 def test_owner(
     client,
     community_owner,
-    rando_user,
-    community,
-    community_permissions_cf,
+    community_with_permissions_cf,
     input_data,
-    vocab_cf,
     search_clear,
 ):
     owner_client = community_owner.login(client)
-    recid = _recid_with_community(
-        owner_client, input_data, community, community_owner, community_permissions_cf
+    record_resp = _create_and_publish(
+        owner_client, input_data, community_with_permissions_cf
     )
+    assert record_resp.status_code == 202
+    recid = record_resp.json["id"]
 
     response_read = owner_client.get(f"{RECORD_COMMUNITIES_BASE_URL}{recid}")
     assert response_read.status_code == 200
     response_delete = owner_client.delete(f"{RECORD_COMMUNITIES_BASE_URL}{recid}")
     assert response_delete.status_code == 204
     response_read = owner_client.get(f"{RECORD_COMMUNITIES_BASE_URL}{recid}")
     assert response_read.status_code == 410
-    """
-    jsn = response_read.json["metadata"]
-    jsn["title"] = "updated title"
-    response_update = owner_client.put(f"{RECORD_COMMUNITIES_BASE_URL}{recid}", json=jsn)
-    response_read = owner_client.get(
-        f"{RECORD_COMMUNITIES_BASE_URL}{recid}"
-    )
-    print()
-    """
 
 
 def test_cf(
     client,
     community_owner,
-    community,
-    community_permissions_cf,
+    community_with_permissions_cf,
     input_data,
-    vocab_cf,
     search_clear,
 ):
     community_owner.login(client)
-    recid = _recid_with_community(
-        client, input_data, community, community_owner, community_permissions_cf
-    )
+    record_resp = _create_and_publish(client, input_data, community_with_permissions_cf)
+    assert record_resp.status_code == 202
+    recid = record_resp.json["id"]
     # sleep(5)
     response = client.get(f"{RECORD_COMMUNITIES_BASE_URL}{recid}/communities")
     assert (
         response.json["hits"]["hits"][0]["custom_fields"]
-        == community_permissions_cf["custom_fields"]
+        == community_with_permissions_cf["custom_fields"]
     )
 
 
 def test_reader(
     client,
-    community_owner,
     community_reader,
-    community,
-    community_permissions_cf,
+    community_with_permissions_cf,
     input_data,
-    vocab_cf,
     search_clear,
 ):
     reader_client = community_reader.login(client)
-    recid = _recid_with_community(
-        reader_client, input_data, community, community_owner, community_permissions_cf
+    record_resp = _create_and_publish(
+        reader_client, input_data, community_with_permissions_cf
     )
+    assert record_resp.status_code == 202
+    recid = record_resp.json["id"]
 
     response_read = reader_client.get(f"{RECORD_COMMUNITIES_BASE_URL}{recid}")
     assert response_read.status_code == 200
@@ -143,20 +91,13 @@ def test_reader(
 
 def test_rando(
     client,
-    community_owner,
     rando_user,
-    community,
-    community_permissions_cf,
+    community_with_permissions_cf,
     input_data,
-    vocab_cf,
     search_clear,
 ):
     rando_client = rando_user.login(client)
-    _recid_with_community(
-        rando_client,
-        input_data,
-        community,
-        community_owner,
-        community_permissions_cf,
-        publish_authorized=False,
+    record_resp = _create_and_publish(
+        rando_client, input_data, community_with_permissions_cf
     )
+    assert record_resp.status_code == 403