[Fixes #10251] Review geonode management command set_layers_permissio…

…ns (#10252) (#10253)

* [Fixes #10251] Review geonode management command set_layers_permissions

* [Fixes #10251] Review geonode management command set_layers_permissions

* [Fixes #10251] Review geonode management command set_layers_permissions

* [Fixes #10251] Review geonode management command set_layers_permissions

* [Fixes #10251] Review geonode management command set_layers_permissions

* [Fixes #10251] Review geonode management command set_layers_permissions

* [Fixes #10251] Review geonode management command set_layers_permissions

Co-authored-by: mattiagiupponi <51856725+mattiagiupponi@users.noreply.github.com>

geonode/base/admin.py

@@ -99,7 +99,7 @@ def set_user_and_group_dataset_permission(modeladmin, request, queryset):
     }
 
     form = UserAndGroupPermissionsForm({
-        'permission_type': ('r', ),
+        'permission_type': 'read',
         'mode': 'set',
         'ids': ids,
     })

geonode/base/forms.py

@@ -638,13 +638,13 @@ def __init__(self, *args, **kwargs):
     layers = forms.ModelMultipleChoiceField(
         queryset=Dataset.objects.all(),
         required=False)
-    permission_type = forms.MultipleChoiceField(
+    permission_type = forms.ChoiceField(
         required=True,
-        widget=forms.CheckboxSelectMultiple,
+        widget=forms.RadioSelect,
         choices=(
-            ('r', 'Read'),
-            ('w', 'Write'),
-            ('d', 'Download'),
+            ('read', 'Read'),
+            ('edit', 'Edit'),
+            ('download', 'Download')
         ),
     )
     mode = forms.ChoiceField(

geonode/base/templates/base/user_and_group_permissions.html

@@ -11,7 +11,7 @@
 <div class="page-header">
   <h2 class="page-title">{% trans "Batch Edit" %}</h2>
 </div>
-<form action="/{{ model|lower }}/layer/permission/" method="post">
+<form action="/{{ model|lower }}/dataset/permission/" method="post">
   {% csrf_token %}
   {{ form|as_bootstrap }}
   <div>

geonode/base/views.py

@@ -99,7 +99,7 @@ def user_and_group_permission(request, model):
         form = UserAndGroupPermissionsForm(request.POST)
         ids = ids.split(",")
         if form.is_valid():
-            resources_names = [layer.name for layer in form.cleaned_data.get('layers')]
+            resources_names = [layer.title for layer in form.cleaned_data.get('layers')]
             users_usernames = [user.username for user in model_class.objects.filter(
                 id__in=ids)] if model == 'profile' else None
             groups_names = [group_profile.group.name for group_profile in model_class.objects.filter(
@@ -121,13 +121,13 @@ def user_and_group_permission(request, model):
 
             if permissions_names:
                 set_permissions.apply_async(
-                    (permissions_names, resources_names, users_usernames, groups_names, delete_flag))
+                    ([permissions_names], resources_names, users_usernames, groups_names, delete_flag))
 
         return HttpResponseRedirect(
             get_url_for_app_model(model, model_class))
 
     form = UserAndGroupPermissionsForm({
-        'permission_type': ('r', ),
+        'permission_type': 'read',
         'mode': 'set',
     })
     return render(

geonode/layers/management/commands/set_layers_permissions.py

@@ -17,10 +17,15 @@
 #
 #########################################################################
 
-from django.core.management.base import BaseCommand
+import logging
 from argparse import RawTextHelpFormatter
+
+from django.core.management.base import BaseCommand
+from geonode.layers.models import Dataset
 from geonode.layers.utils import set_datasets_permissions
 
+logger = logging.getLogger("geonode.layers.management.set_layers_permissions")
+
 
 class Command(BaseCommand):
 
@@ -48,67 +53,82 @@ def create_parser(self, *args, **kwargs):
 
     def add_arguments(self, parser):
         parser.add_argument(
-            '-r',
-            '--resources',
-            dest='resources',
-            nargs='*',
+            "-r",
+            "--resources",
+            dest="resources",
+            nargs="*",
             type=str,
-            default=None,
-            help='Resources names for which permissions will be assigned to. '
-                 'Default value: None (all the layers will be considered). '
-                 'Multiple choices can be typed with white space separator.'
-                 'A Note: names with white spaces must be typed inside quotation marks.'
+            default=[],
+            help="Resources IDs for which permissions will be assigned to. Default value: [] (all the layers will be considered). "
         )
         parser.add_argument(
-            '-p',
-            '--permission',
-            dest='permission',
+            "-p",
+            "--permission",
+            dest="permission",
             type=str,
             default=None,
-            help='Permissions to be assigned. '
-                 'Allowed values are: read (r), write (w), download (d) and owner (o).'
+            help="Permissions to be assigned. " "Allowed values are: view, download, edit and manage.",
         )
         parser.add_argument(
-            '-u',
-            '--users',
-            dest='users',
-            nargs='*',
+            "-u",
+            "--users",
+            dest="users",
+            nargs="*",
             type=str,
-            default=None,
-            help='Users for which permissions will be assigned to. '
-                 'Multiple choices can be typed with white space separator.'
+            default=[],
+            help="Users for which permissions will be assigned to. "
+            "Multiple choices can be typed with comma separator.",
         )
         parser.add_argument(
-            '-g',
-            '--groups',
-            dest='groups',
-            nargs='*',
+            "-g",
+            "--groups",
+            dest="groups",
+            nargs="*",
             type=str,
-            default=None,
-            help='Groups for which permissions will be assigned to. '
-                 'Multiple choices can be typed with white space separator.'
+            default=[],
+            help="Groups for which permissions will be assigned to. "
+            "Multiple choices can be typed with comma separator.",
         )
         parser.add_argument(
-            '-d',
-            '--delete',
-            dest='delete_flag',
-            action='store_true',
+            "-d",
+            "--delete",
+            dest="delete_flag",
+            action="store_true",
             default=False,
-            help='Delete permission if it exists.'
+            help="Delete permission if it exists.",
         )
 
     def handle(self, *args, **options):
         # Retrieving the arguments
-        resources_names = options.get('resources')
-        permissions_name = options.get('permission')
-        users_usernames = options.get('users')
-        groups_names = options.get('groups')
-        delete_flag = options.get('delete_flag')
+        permissions_name = options.get("permission").replace(" ", "")
+
+        resources_pk = [x.replace(" ", "") for x in options.get("resources", [])]
+        if resources_pk:
+            resources_pk = resources_pk[0].split(",")
+        else:
+            resources_pk = [x for x in Dataset.objects.values_list('pk', flat=True)]
+
+        users_usernames = [x.replace(" ", "") for x in options.get("users", [])]
+        if users_usernames:
+            users_usernames = users_usernames[0].split(",")
+
+        groups_names = [x.replace(" ", "") for x in options.get("groups", [])]
+        if groups_names:
+            groups_names = groups_names[0].split(",")
+
+        delete_flag = options.get("delete_flag")
+
+        if isinstance(permissions_name, list):
+            # it accept one kind of permission per request
+            raise Exception("Only one permission name must be specified")
+
+        if not users_usernames and not groups_names:
+            raise Exception("Groups or Usernames must be specified")
+
         set_datasets_permissions(
             permissions_name,
-            resources_names,
-            users_usernames,
-            groups_names,
-            delete_flag,
-            verbose=True
+            resources_names=resources_pk,
+            users_usernames=users_usernames,
+            groups_names=groups_names,
+            delete_flag=delete_flag
         )

geonode/layers/tests.py

@@ -18,6 +18,7 @@
 #########################################################################
 
 import io
+import itertools
 import os
 import shutil
 import gisdata
@@ -35,6 +36,7 @@
 from django.test.client import RequestFactory
 from django.contrib.contenttypes.models import ContentType
 from django.core.files.uploadedfile import SimpleUploadedFile
+from django.core.management import call_command
 from django.contrib.auth.models import Group
 from django.contrib.gis.geos import Polygon
 from django.db.models import Count
@@ -902,7 +904,7 @@ def test_assign_remove_permissions(self):
         perm_spec = layer.get_all_level_info()
         self.assertNotIn(get_user_model().objects.get(username="norman"), perm_spec["users"])
 
-        utils.set_datasets_permissions("write", resources_names=[layer.name], users_usernames=["norman"], delete_flag=False, verbose=True)
+        utils.set_datasets_permissions("edit", resources_names=[layer.name], users_usernames=["norman"], delete_flag=False, verbose=True)
         perm_spec = layer.get_all_level_info()
         _c = 0
         if "users" in perm_spec:
@@ -2017,3 +2019,114 @@ def test_dataset_time_form_should_raise_error_if_invalid_payload(self):
         self.assertFalse(form.is_valid())
         self.assertTrue('presentation' in form.errors)
         self.assertEqual("Select a valid choice. INVALID_PRESENTATION_VALUE is not one of the available choices.", form.errors['presentation'][0])
+
+
+class SetLayersPermissionsCommand(GeoNodeBaseTestSupport):
+    '''
+    Unittest to ensure that the management command "set_layers_permissions"
+    behaves as expected
+    '''
+
+    def test_user_get_the_download_permissions_for_the_selected_dataset(self):
+        '''
+        Given a user, the compact perms and the resource id, it shoul set the
+        permissions for the selected resource
+        '''
+        try:
+            expected_perms = {'view_resourcebase', 'download_resourcebase'}
+
+            dataset, args, username, opts = self._create_arguments(perms_type='download')
+            call_command('set_layers_permissions', *args, **opts)
+
+            self._assert_perms(expected_perms, dataset, username)
+        finally:
+            if dataset:
+                dataset.delete()
+
+    def test_user_get_the_view_permissions_for_the_selected_dataset(self):
+        '''
+        Given a user, the compact perms and the resource id, it shoul set the
+        permissions for the selected resource
+        '''
+        try:
+            expected_perms = {'view_resourcebase'}
+            dataset, args, username, opts = self._create_arguments(perms_type='view')
+
+            call_command('set_layers_permissions', *args, **opts)
+
+            self._assert_perms(expected_perms, dataset, username)
+
+        finally:
+            if dataset:
+                dataset.delete()
+
+    def test_user_get_the_edit_permissions_for_the_selected_dataset(self):
+        '''
+        Given a user, the compact perms and the resource id, it shoul set the
+        permissions for the selected resource
+        '''
+        try:
+            expected_perms = {
+                'view_resourcebase',
+                'change_dataset_style',
+                'download_resourcebase',
+                'change_resourcebase_metadata',
+                'change_dataset_data',
+                'change_resourcebase'
+            }
+
+            dataset, args, username, opts = self._create_arguments(perms_type='edit')
+
+            call_command('set_layers_permissions', *args, **opts)
+
+            self._assert_perms(expected_perms, dataset, username)
+        finally:
+            if dataset:
+                dataset.delete()
+
+    def test_user_get_the_manage_permissions_for_the_selected_dataset(self):
+        '''
+        Given a user, the compact perms and the resource id, it shoul set the
+        permissions for the selected resource
+        '''
+        try:
+            expected_perms = {
+                'delete_resourcebase',
+                'change_resourcebase',
+                'view_resourcebase',
+                'change_resourcebase_permissions',
+                'change_dataset_style',
+                'change_resourcebase_metadata',
+                'publish_resourcebase',
+                'change_dataset_data',
+                'download_resourcebase'
+            }
+
+            dataset, args, username, opts = self._create_arguments(perms_type='manage')
+
+            call_command('set_layers_permissions', *args, **opts)
+
+            self._assert_perms(expected_perms, dataset, username)
+        finally:
+            if dataset:
+                dataset.delete()
+
+    def _create_arguments(self, perms_type):
+        dataset = create_single_dataset('dataset_for_management_command')
+        args = []
+        username = get_user_model().objects.exclude(username='admin').exclude(username='AnonymousUser').first().username
+        opts = {
+                "permission": perms_type,
+                "users": [username],
+                "resources": str(dataset.id)
+            }
+
+        return dataset, args, username, opts
+
+    def _assert_perms(self, expected_perms, dataset, username):
+        dataset.refresh_from_db()
+
+        perms = dataset.get_all_level_info()
+        self.assertTrue(username in [user.username for user in perms['users']])
+        actual = set(itertools.chain.from_iterable([perms for user, perms in perms['users'].items() if user.username == username]))
+        self.assertSetEqual(expected_perms, actual)