diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/migrations/0002_populate.py b/SAS/TMSS/backend/src/tmss/tmssapp/migrations/0002_populate.py index 92baffd4c15a8c025d234eeffed61ae9f443fabf..0fece500a4fdfb63d13d81b325dd60bc7c955b7b 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/migrations/0002_populate.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/migrations/0002_populate.py @@ -21,4 +21,4 @@ class Migration(migrations.Migration): migrations.RunPython(populate_misc), migrations.RunPython(populate_resources), migrations.RunPython(populate_cycles), - migrations.RunPython(populate_projects) ] + migrations.RunPython(populate_projects)] diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/models/CMakeLists.txt b/SAS/TMSS/backend/src/tmss/tmssapp/models/CMakeLists.txt index 64b472d0532f38dcee337722ff16d68881b85099..3496efd57358ab186b665fe2dc3bd40264d4deaa 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/models/CMakeLists.txt +++ b/SAS/TMSS/backend/src/tmss/tmssapp/models/CMakeLists.txt @@ -3,6 +3,7 @@ include(PythonInstall) set(_py_files __init__.py + permissions.py specification.py scheduling.py common.py diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/models/permissions.py b/SAS/TMSS/backend/src/tmss/tmssapp/models/permissions.py index 9ff0dd5a806e427b40cf485bf700181440d16b75..8cdbf52b01eb27e1c3e8467cbba34e4494c37b3d 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/models/permissions.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/models/permissions.py @@ -9,6 +9,9 @@ from .common import NamedCommonPK, AbstractChoice from django.db.models import ManyToManyField from enum import Enum +from rest_framework.permissions import DjangoModelPermissions + + # # Project Permissions # @@ -31,3 +34,20 @@ class ProjectPermission(NamedCommonPK): POST = ManyToManyField('ProjectRole', related_name='can_POST', blank=True) PATCH = ManyToManyField('ProjectRole', related_name='can_PATCH', blank=True) DELETE = ManyToManyField('ProjectRole', related_name='can_DELETE', blank=True) + + +# todo: move to viewsets / merge with TMSSDjangoModelPermissions class +class TMSSBasePermissions(DjangoModelPermissions): + # This enforces permissions as "deny any" by default. + view_permissions = ['%(app_label)s.view_%(model_name)s'] + + perms_map = { + 'GET': view_permissions, + 'OPTIONS': view_permissions, + 'HEAD': view_permissions, + 'POST': DjangoModelPermissions.perms_map['POST'], + 'PUT': DjangoModelPermissions.perms_map['PUT'], + 'PATCH': DjangoModelPermissions.perms_map['PATCH'], + 'DELETE': DjangoModelPermissions.perms_map['DELETE'], + } + diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/models/specification.py b/SAS/TMSS/backend/src/tmss/tmssapp/models/specification.py index aee721d29894c4271786bddd2c3a9241a25199b8..4c00aa3da2312e0849ca93f5deb0d3a67bee7a9a 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/models/specification.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/models/specification.py @@ -19,6 +19,24 @@ from collections import Counter from django.utils.functional import cached_property +# +# Mixins +# + +class ProjectPropertyMixin: + @cached_property + def project(self): # -> Project: + '''return the related project of this task + ''' + if not hasattr(self, 'path_to_project'): + return TMSSException("Please define a 'path_to_project' attribute on the object for the ProjectPropertyMixin to function.") + obj = self + for attr in self.path_to_project.split('__'): + obj = getattr(obj, attr) + if attr == 'project': + return obj + + # # I/O # @@ -270,6 +288,7 @@ class Project(NamedCommonPK): project_category = ForeignKey('ProjectCategory', null=True, on_delete=PROTECT, help_text='Project category.') period_category = ForeignKey('PeriodCategory', null=True, on_delete=PROTECT, help_text='Period category.') auto_pin = BooleanField(default=False, help_text='True if the output_pinned flag of tasks in this project should be set True on creation.') + path_to_project = "project" @cached_property def duration(self) -> datetime.timedelta: @@ -277,6 +296,10 @@ class Project(NamedCommonPK): ''' return self.relative_stop_time - self.relative_start_time + @cached_property + def project(self): + return self + class ProjectQuota(Model): project = ForeignKey('Project', related_name="quota", on_delete=PROTECT, help_text='Project to wich this quota belongs.') # protected to avoid accidents diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/populate.py b/SAS/TMSS/backend/src/tmss/tmssapp/populate.py index 5d06a24e9c529a8b1bc5e4b51773f2f47967632d..b0f83fb47480960bf27e542b4bcbdb2398f445a0 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/populate.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/populate.py @@ -17,16 +17,18 @@ class Migration(migrations.Migration): import logging logger = logging.getLogger(__name__) -import json -import os +import inspect +import re from datetime import datetime, timezone from lofar.sas.tmss.tmss.tmssapp import models +from lofar.sas.tmss.tmss.tmssapp import viewsets from lofar.sas.tmss.tmss.tmssapp.models.specification import * from lofar.sas.tmss.tmss.tmssapp.models.scheduling import * from lofar.sas.tmss.tmss.tmssapp.models.permissions import * from lofar.common import isTestEnvironment, isDevelopmentEnvironment from concurrent.futures import ThreadPoolExecutor -from django.contrib.auth.models import Group, Permission +from django.contrib.auth.models import User, Group, Permission +from django.contrib.contenttypes.models import ContentType working_dir = os.path.dirname(os.path.abspath(__file__)) @@ -245,6 +247,8 @@ def populate_connectors(): def populate_permissions(): + logger.info('Populating permissions') + perm = ProjectPermission.objects.create(name='taskdraft') perm.GET.set([ProjectRole.objects.get(value='shared_support_user')]) perm.POST.set([ProjectRole.objects.get(value='shared_support_user')]) @@ -254,7 +258,175 @@ def populate_permissions(): perm.GET.set([ProjectRole.objects.get(value='shared_support_user')]) perm.save() + perm = ProjectPermission.objects.create(name='project') + perm.GET.set([ProjectRole.objects.get(value='shared_support_user')]) + perm.POST.set([ProjectRole.objects.get(value='shared_support_user')]) + perm.save() + populate_system_permissions() + populate_system_roles() + populate_system_test_users() + + + +def populate_system_permissions(): + # For each viewset create custom permissions for extra actions. + for name, obj in inspect.getmembers(viewsets): + if inspect.isclass(obj): + try: + ct = ContentType.objects.get_for_model(obj.serializer_class.Meta.model) + extra_actions = obj.get_extra_actions() + if extra_actions: + for action in extra_actions: + codename = ("%s_%s" % (action.__name__, obj.serializer_class.Meta.model.__name__)).lower() + name = f'Can {action.__name__} {obj.serializer_class.Meta.model.__name__.lower()}' + Permission.objects.create(codename=codename, name=name, content_type=ct) + except: + pass + + +def populate_system_roles(): + to_observer_group = Group.objects.create(name='TO observer') + sdco_support_group = Group.objects.create(name='SDCO support') + tmss_maintainer_group = Group.objects.create(name='TMSS Maintainer') + tmss_admin_group = Group.objects.create(name='TMSS Admin') + to_maintenance_group = Group.objects.create(name='TO maintenance') + to_user_group = Group.objects.create(name='TO user') scientist_group = Group.objects.create(name='Scientist') + e_scientist_group = Group.objects.create(name='Scientist (Expert)') + guest_group = Group.objects.create(name='Guest') + lta_user_group = Group.objects.create(name='LTA User') + + assign_system_permissions() + + +def assign_system_permissions(): + ''' + Assign system permission to each system role, accordingly. + ''' + + # Get system roles + to_observer_group = Group.objects.get(name='TO observer') + sdco_support_group = Group.objects.get(name='SDCO support') + tmss_maintainer_group = Group.objects.get(name='TMSS Maintainer') + tmss_admin_group = Group.objects.get(name='TMSS Admin') + to_maintenance_group = Group.objects.get(name='TO maintenance') + to_user_group = Group.objects.get(name='TO user') + scientist_group = Group.objects.get(name='Scientist') + e_scientist_group = Group.objects.get(name='Scientist (Expert)') + guest_group = Group.objects.get(name='Guest') + lta_user_group = Group.objects.get(name='LTA User') + + # Existing tests rely on this # todo: adapt tests to match actual permissions scientist_group.permissions.add(Permission.objects.get(codename='view_cycle')) scientist_group.permissions.add(Permission.objects.get(codename='view_project')) + + # Subtask model permissions + ct = ContentType.objects.get(model='subtask') + perm = Permission.objects.get(codename='view_subtask') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + to_observer_group.permissions.add(Permission.objects.get(codename='schedule_subtask')) + sdco_support_group.permissions.add(Permission.objects.get(codename='schedule_subtask')) + tmss_maintainer_group.permissions.add(Permission.objects.get(codename='schedule_subtask')) + tmss_admin_group.permissions.add(Permission.objects.get(codename='schedule_subtask')) + # TODO: This is needed for testing atm. The table does not specify which roles should have this permission for now. + for subtask_perm in ['get_progress_subtask', 'input_dataproducts_subtask', 'output_dataproducts_subtask', + 'parset_subtask', 'predecessors_subtask', 'process_feedback_and_set_to_finished_if_complete_subtask', + 'reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask', 'state_log_subtask', + 'successors_subtask', 'task_log_subtask', 'transformed_output_dataproduct_subtask', 'unschedule_subtask' ]: + to_observer_group.permissions.add(Permission.objects.get(codename=subtask_perm)) + + # Template models permissions + for name, obj in inspect.getmembers(models): + if inspect.isclass(obj) and issubclass(obj, Template) and obj is not Template: + ct = ContentType.objects.get(model=name.lower()) + perm_name = ''.join('_'.join(re.findall(r'[A-Z](?:[a-z]+|[A-Z]*(?=[A-Z]|$))', name)).lower()) + perm = Permission.objects.get(codename=f'add_{name.lower()}') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + e_scientist_group.permissions.add(perm) + perm = Permission.objects.get(codename=f'view_{name.lower()}') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + e_scientist_group.permissions.add(perm) + perm = Permission.objects.get(codename=f'change_{name.lower()}') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + e_scientist_group.permissions.add(perm) + + # SchedulingUnit models permissions + for template in ['schedulingunitdraft', 'schedulingunitblueprint']: + ct = ContentType.objects.get(model=template) + perm = Permission.objects.get(codename=f'add_{template}') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + to_user_group.permissions.add(perm) + e_scientist_group.permissions.add(perm) if template is 'schedulingunitdraft' else None + perm = Permission.objects.get(codename=f'view_{template}') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + to_user_group.permissions.add(perm) + + # Project model permissions + ct = ContentType.objects.get(model='project') + perm = Permission.objects.get(codename='add_project') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + perm = Permission.objects.get(codename='view_project') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + to_user_group.permissions.add(perm) + perm = Permission.objects.get(codename='change_project') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + + # User model permissions + ct = ContentType.objects.get(model='user') + perm = Permission.objects.get(codename='add_user') + to_observer_group.permissions.add(perm) + sdco_support_group.permissions.add(perm) + tmss_maintainer_group.permissions.add(perm) + tmss_admin_group.permissions.add(perm) + + +def populate_system_test_users(): + # TODO: Set proper credentials (passwords at least). + to_observer_user = User.objects.create(username='to_observer', password='to_observer') + to_observer_user.groups.add(Group.objects.get(name='TO observer')) + sdco_support_user = User.objects.create(username='sdco_support', password='sdco_support') + sdco_support_user.groups.add(Group.objects.get(name='SDCO support')) + tmss_maintainer_user = User.objects.create(username='tmss_maintainer', password='tmss_maintainer') + tmss_maintainer_user.groups.add(Group.objects.get(name='TMSS Maintainer')) + tmss_admin_user = User.objects.create(username='tmss_admin', password='tmss_admin') + tmss_admin_user.groups.add(Group.objects.get(name='TMSS Admin')) + to_maintenance_user = User.objects.create(username='to_maintenance', password='to_maintenance') + to_maintenance_user.groups.add(Group.objects.get(name='TO maintenance')) + to_user = User.objects.create(username='to_user', password='to_user') + to_user.groups.add(Group.objects.get(name='TO user')) + scientist_user = User.objects.create(username='scientist', password='scientist') + scientist_user.groups.add(Group.objects.get(name='Scientist')) + e_scientist_user = User.objects.create(username='e_scientist', password='e_scientist') + e_scientist_user.groups.add(Group.objects.get(name='Scientist (Expert)')) + guest_user = User.objects.create(username='guest', password='guest') + guest_user.groups.add(Group.objects.get(name='Guest')) + lta_user = User.objects.create(username='lta_user', password='lta_user') + lta_user.groups.add(Group.objects.get(name='LTA User')) diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/CMakeLists.txt b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/CMakeLists.txt index 8f21d3c956f637dd42e50f3676699db3b5bd8139..186d29924f2c1706f57804848474f1a74bfeebb8 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/CMakeLists.txt +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/CMakeLists.txt @@ -7,6 +7,7 @@ set(_py_files specification.py scheduling.py permissions.py + project_permissions.py ) python_install(${_py_files} diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/__init__.py b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/__init__.py index 767fdee003df98d3837ee05217cf3e8de1ab9846..0f7980fabfd9022b1389bf2ac72a975f9d2fb1e8 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/__init__.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/__init__.py @@ -1,3 +1,4 @@ from .specification import * from .scheduling import * -from .permissions import * \ No newline at end of file +from .permissions import * +from .project_permissions import * \ No newline at end of file diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/lofar_viewset.py b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/lofar_viewset.py index b30fe6463f3e2d652f1b53db5c818c999aa3bfee..fa74b76b1a3291235d865b4b9914b15324f36871 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/lofar_viewset.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/lofar_viewset.py @@ -17,12 +17,61 @@ from django.http import JsonResponse from django.urls import reverse as revese_url from rest_framework.decorators import action from lofar.common import json_utils +from lofar.sas.tmss.tmss.tmssapp.viewsets.permissions import TMSSPermissions, IsProjectMemberFilterBackend +from lofar.sas.tmss.tmss.tmssapp.models import permissions +from django_filters.rest_framework import DjangoFilterBackend +from rest_framework.filters import OrderingFilter + +#class TMSSPermissionsMixin: + + # def __init__(self, *args, **kwargs): + # self.permission_classes = (TMSSPermissions,) + # self.filter_backends = (IsProjectMemberFilterBackend,) + # self.extra_action_permission_classes = self._create_extra_action_permission_classes() + # super(TMSSPermissionsMixin, self).__init__(*args, **kwargs) + # + # # TODO: Cache this method to avoid redundancy and overhead. + # def _create_extra_action_permission_classes(self): + # extra_action_permission_classes = [] + # extra_actions = [a.__name__ for a in self.get_extra_actions()] + # for ea in extra_actions: # Create permission classes + # permission_name = f'{ea}_{self.serializer_class.Meta.model.__name__.lower()}' + # permission_class_name = f'Can {ea} {self.serializer_class.Meta.model.__name__.lower()}' + # new_permission_class = type(f'{permission_class_name}', (permissions.TMSSBasePermissions,), { + # # TODO: Is it necessary to have both permissions and object permissions? + # # TODO: Find a way to use the "%(app_label)s." syntax. + # 'permission_name': permission_name, + # 'has_permission': lambda self, request, view: request.user.has_perm(f'tmssapp.{self.permission_name}'), + # 'has_object_permission': lambda self, request, view, obj: request.user.has_perm(f'tmssapp.{self.permission_name}'), + # }) + # new_permission_class.__setattr__(self, 'permission_name', permission_name) + # extra_action_permission_classes.append({ea: new_permission_class},) + # return extra_action_permission_classes + # + # # TODO: Refactoring. + # def get_model_permissions(self): + # extra_actions = [a.__name__ for a in self.get_extra_actions()] + # if self.action in extra_actions: + # for ea_permission_class in self.extra_action_permission_classes: + # if ea_permission_class.get(self.action): + # return [permissions.TMSSBasePermissions, ea_permission_class.get(self.action),] + # else: + # return [permissions.TMSSBasePermissions,] + # else: + # return [permissions.TMSSBasePermissions, ] + + #def get_permissions(self): + # self.get_extra_action_permission_classes() + # return super(TMSSPermissionsMixin, self).get_permissions() + class LOFARViewSet(viewsets.ModelViewSet): """ If you're using format suffixes, make sure to also include the `format=None` keyword argument for each action. """ + permission_classes = (TMSSPermissions,) + filter_backends = (DjangoFilterBackend, OrderingFilter, IsProjectMemberFilterBackend,) @swagger_auto_schema(responses={403: 'forbidden'}) def list(self, request, **kwargs): @@ -52,7 +101,7 @@ class LOFARNestedViewSet(mixins.CreateModelMixin, mixins.ListModelMixin, #mixins.RetrieveModelMixin, viewsets.GenericViewSet): - + @swagger_auto_schema(responses={403: 'forbidden'}) def list(self, request, **kwargs): return super(LOFARNestedViewSet, self).list(request, **kwargs) @@ -70,6 +119,7 @@ class LOFARCopyViewSet(mixins.CreateModelMixin, viewsets.GenericViewSet): def list(self, request, **kwargs): return super(LOFARCopyViewSet, self).list(request, **kwargs) """ + @swagger_auto_schema(responses={400: 'invalid specification', 403: 'forbidden'}) def create(self, request, **kwargs): return super(LOFARCopyViewSet, self).create(request, **kwargs) diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/permissions.py b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/permissions.py index 3e68aeba65f43de96852b093d3c05f289fb4329f..66124b5c3ba77f70eecd7533369037b9d1f5d88e 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/permissions.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/permissions.py @@ -4,8 +4,6 @@ This file contains permissions and filters that are used in the viewsets from rest_framework import permissions as drf_permissions, filters as drf_filters from .. import models -from .. import serializers -from .lofar_viewset import LOFARViewSet from lofar.sas.tmss.tmss.exceptions import * from django.core.exceptions import ObjectDoesNotExist import logging @@ -13,18 +11,6 @@ logger = logging.getLogger(__name__) logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO) from django.urls import resolve import urllib.parse -# -# ViewSets -# - -class ProjectRoleViewSet(LOFARViewSet): - queryset = models.ProjectRole.objects.all() - serializer_class = serializers.ProjectRoleSerializer - - -class ProjectPermissionViewSet(LOFARViewSet): - queryset = models.ProjectPermission.objects.all() - serializer_class = serializers.ProjectPermissionSerializer # @@ -50,8 +36,8 @@ def get_project_roles_for_user(user): try: if user == models.User.objects.get(username='paulus'): return ({'project': 'test_user_is_shared_support', 'role': 'shared_support_user'}, - {'project': 'test_user_is_contact', 'role': 'contact_author'}, - {'project': 'high', 'role': 'shared_support_user'}) + {'project': 'test_user_is_contact', 'role': 'contact_author'}) + #{'project': 'high', 'role': 'shared_support_user'}) except: pass @@ -66,8 +52,8 @@ def get_project_roles_with_permission(permission_name, method='GET'): # ...determine what project roles are allowed to perform the requested action return getattr(project_permission, method).all() except ObjectDoesNotExist: - raise TMSSException("This action was configured to enforce project permissions, but no project permission with name '%s' has been defined." % permission_name) - + logger.error("This action was configured to enforce project permissions, but no project permission with name '%s' has been defined." % permission_name) + return [] class IsProjectMember(drf_permissions.DjangoObjectPermissions): """ @@ -109,13 +95,25 @@ class IsProjectMember(drf_permissions.DjangoObjectPermissions): # check whether the related project of this object is one that the user has permission to see for project_role in user_project_roles: - if project_role['project'] == obj.project.name and \ - models.ProjectRole.objects.get(value=project_role['role']) in permitted_project_roles: - logger.info('user=%s is permitted to access object=%s' % (request.user, obj)) - logger.info('### IsProjectMember.has_object_permission %s %s True' % (request._request, request.method)) - return True - - logger.info('User=%s is not permitted to access object=%s with related project=%s since it requires one of project_roles=%s' % (request.user, obj, obj.project, permitted_project_roles)) + if hasattr(obj, 'project'): + related_project = obj.project + if project_role['project'] == obj.project.name and \ + models.ProjectRole.objects.get(value=project_role['role']) in permitted_project_roles: + logger.info('user=%s is permitted to access object=%s' % (request.user, obj)) + logger.info('### IsProjectMember.has_object_permission %s %s True' % (request._request, request.method)) + return True + else: + related_project = None + logger.error("No project property on object %s, so cannot check project permission." % obj) + # todo: how to deal with objects that do not have a unique project associated to them? + # Do need users need the required role in all of them? Or just one? + # Also, we need to support lists in 'path_to_project' if we want to handle this. + if issubclass(type(obj), models.Template) and view.action == 'default': + # todo: review this. Why is the default action called in a GET of an object that has a template reference? + logger.warning("'%s' is a Template and action is '%s' so granting object access nonetheless." % (obj, view.action)) + return True + + logger.info('User=%s is not permitted to access object=%s with related project=%s since it requires one of project_roles=%s' % (request.user, obj, related_project, permitted_project_roles)) logger.info('### IsProjectMember.has_object_permission %s False' % (request._request)) return False @@ -124,15 +122,17 @@ class IsProjectMember(drf_permissions.DjangoObjectPermissions): # We need to check the project name before the object is created, but project is an object property. # Turning the path to the project into a (static) model attribute, allows us to use it both in the object # property as well as here (where we did not create an object yet). - if request.method == 'POST' and request.data: + if view.action == 'create' and request.data: obj = None + if view.serializer_class.Meta.model == models.Project: + return False # project creation solely depends on system role for attr in view.serializer_class.Meta.model.path_to_project.split('__'): if not obj: # on first iteration, the referenced object needs to be resolved from POSTed FQDN obj_ref = request.data[attr] path = urllib.parse.urlparse(obj_ref).path resolved_func, _, resolved_kwargs = resolve(path) - obj = resolved_func.cls().get_queryset().get(id=resolved_kwargs['pk']) + obj = resolved_func.cls().get_queryset().get(pk=resolved_kwargs['pk']) else: if attr == 'project': # has_object_permission checks the project from obj, so we can just check project permission on @@ -141,14 +141,15 @@ class IsProjectMember(drf_permissions.DjangoObjectPermissions): logger.info('### IsProjectMember.has_permission %s %s' % (request._request, p)) return p obj = getattr(obj, attr) - if view.action == 'list': - logger.info('Allowing list action %s %s' % (request.method, view.action)) - logger.info('### IsProjectMember.has_permission %s %s True' % (request._request, request.method)) - return True - logger.info('Allowing empty POST or non-POST request %s %s' % (request.method, view.action)) - logger.info('### IsProjectMember.has_permission %s %s True' % (request._request, request.method)) - return True + pk = view.kwargs.get('pk', None) + if pk: + obj = view.serializer_class.Meta.model.objects.get(pk=pk) + p = self.has_object_permission(request, view, obj) + else: + p = super().has_permission(request, view) + logger.info('### IsProjectMember.has_permission %s %s' % (request._request, p)) + return p class IsProjectMemberOrReadOnly(IsProjectMember): @@ -184,7 +185,13 @@ class TMSSDjangoModelPermissions(drf_permissions.DjangoModelPermissions): } def has_permission(self, request, view): - p = super().has_permission(request, view) + extra_actions = [a.__name__ for a in view.get_extra_actions()] + if view.action in extra_actions: + permission_name = f'{view.action}_{view.serializer_class.Meta.model.__name__.lower()}' + logger.info('### TMSSDjangoModelPermissions checking extra permission %s %s' % (request._request, permission_name)) + p = request.user.has_perm(f'tmssapp.{permission_name}') + else: + p = super().has_permission(request, view) logger.info('### TMSSDjangoModelPermissions.has_permission %s %s' % (request._request, p)) return p @@ -194,15 +201,16 @@ class TMSSPermissions(drf_permissions.DjangoObjectPermissions): Create custom permission class Note: required because the composition using & and | in the permission_classes does not seem to work as it should. """ - model_permissions = TMSSDjangoModelPermissions() project_permissions = IsProjectMember() + model_permissions = TMSSDjangoModelPermissions() def has_permission(self, request, view): return (self.model_permissions.has_permission(request, view) or self.project_permissions.has_permission(request, view)) and request.user.is_authenticated def has_object_permission(self, request, view, obj): - return IsProjectMember().has_object_permission(request, view, obj) and request.user.is_authenticated + return (self.model_permissions.has_permission(request, view) or + self.project_permissions.has_object_permission(request, view, obj)) and request.user.is_authenticated # @@ -256,9 +264,10 @@ class IsProjectMemberFilterBackend(drf_filters.BaseFilterBackend): permitted_fetched_objects = [] not_permitted = [o for o in queryset if o not in permitted_fetched_objects] - logger.info('User=%s is not permitted to access objects=%s with related projects=%s' % (request.user, not_permitted, [o.project for o in not_permitted])) + logger.info('### User=%s is not permitted to access objects=%s with related projects=%s' % (request.user, not_permitted, [o.project for o in not_permitted if hasattr(o, 'project')])) + logger.info('### User=%s is permitted to access objects=%s with related projects=%s' % (request.user, permitted_fetched_objects, [o.project for o in permitted_fetched_objects if hasattr(o, 'project')])) # we could return the list of objects, which seems to work if you don't touch the get_queryset. # But are supposed to return a queryset instead, so we make a new one, even though we fetched already. # I don't know, there must be a better way... - return queryset.filter(id__in=[o.id for o in permitted_fetched_objects]) + return queryset.filter(pk__in=[o.pk for o in permitted_fetched_objects]) diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/project_permissions.py b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/project_permissions.py new file mode 100644 index 0000000000000000000000000000000000000000..cf7fc57258e0c4fcf45e419e2107998a9969b10a --- /dev/null +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/project_permissions.py @@ -0,0 +1,17 @@ +from .. import models +from .. import serializers +from .lofar_viewset import LOFARViewSet + + +# +# Project Permission ViewSets +# + +class ProjectRoleViewSet(LOFARViewSet): + queryset = models.ProjectRole.objects.all() + serializer_class = serializers.ProjectRoleSerializer + + +class ProjectPermissionViewSet(LOFARViewSet): + queryset = models.ProjectPermission.objects.all() + serializer_class = serializers.ProjectPermissionSerializer diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/scheduling.py b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/scheduling.py index 767c7a835c1ebb74b5ae3a8cfff1a3411a8d1321..8bda4797baab86e450e9d4bec910d9132503287f 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/scheduling.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/scheduling.py @@ -17,6 +17,7 @@ from drf_yasg.inspectors import SwaggerAutoSchema from drf_yasg.openapi import Parameter from rest_framework.decorators import action +from rest_framework.decorators import permission_classes from django.http import HttpResponse, JsonResponse, HttpResponseRedirect, HttpResponseNotFound from rest_framework.response import Response as RestResponse @@ -138,7 +139,6 @@ class SubTaskFilter(filters.FilterSet): class SubtaskViewSet(LOFARViewSet): queryset = models.Subtask.objects.all() serializer_class = serializers.SubtaskSerializer - filter_backends = (filters.DjangoFilterBackend, OrderingFilter,) filter_class = SubTaskFilter ordering = ('start_time',) @@ -343,7 +343,6 @@ class SubtaskViewSet(LOFARViewSet): class SubtaskNestedViewSet(LOFARNestedViewSet): queryset = models.Subtask.objects.all() serializer_class = serializers.SubtaskSerializer - filter_backends = (filters.DjangoFilterBackend,) filter_class = SubTaskFilter ordering = ('start_time',) diff --git a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/specification.py b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/specification.py index 12fa2504a1131c98730c5519ec06a4d997ff5ae4..9605ead221a9ae4a18596d0c6d887b4ad2791bc2 100644 --- a/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/specification.py +++ b/SAS/TMSS/backend/src/tmss/tmssapp/viewsets/specification.py @@ -20,7 +20,6 @@ from drf_yasg.utils import swagger_auto_schema from drf_yasg.openapi import Parameter from lofar.sas.tmss.tmss.tmssapp.viewsets.lofar_viewset import LOFARViewSet, LOFARNestedViewSet, AbstractTemplateViewSet, LOFARCopyViewSet -from lofar.sas.tmss.tmss.tmssapp.viewsets.permissions import IsProjectMemberFilterBackend, IsProjectMember, IsProjectMemberOrReadOnly, TMSSDjangoModelPermissions, TMSSPermissions from lofar.sas.tmss.tmss.tmssapp import models from lofar.sas.tmss.tmss.tmssapp import serializers from django.http import JsonResponse @@ -30,8 +29,11 @@ from lofar.common.json_utils import get_default_json_object_for_schema from lofar.common.datetimeutils import formatDatetime from lofar.sas.tmss.tmss.tmssapp.tasks import * from lofar.sas.tmss.tmss.tmssapp.subtasks import * +from lofar.sas.tmss.tmss.tmssapp.viewsets.permissions import TMSSDjangoModelPermissions from django.urls import resolve, get_script_prefix,Resolver404 +from django_filters.rest_framework import DjangoFilterBackend +from rest_framework.filters import OrderingFilter import json import logging @@ -242,8 +244,9 @@ class TaskConnectorTypeViewSet(LOFARViewSet): serializer_class = serializers.TaskConnectorTypeSerializer -@permission_classes((TMSSDjangoModelPermissions,)) class CycleViewSet(LOFARViewSet): + permission_classes = (TMSSDjangoModelPermissions,) # override default project permission + filter_backends = (DjangoFilterBackend, OrderingFilter) # override default project permission queryset = models.Cycle.objects.all() serializer_class = serializers.CycleSerializer ordering = ['start'] @@ -264,7 +267,6 @@ class CycleQuotaViewSet(LOFARViewSet): return queryset -@permission_classes((TMSSDjangoModelPermissions,)) class ProjectViewSet(LOFARViewSet): queryset = models.Project.objects.all() serializer_class = serializers.ProjectSerializer @@ -281,7 +283,6 @@ class ProjectViewSet(LOFARViewSet): return queryset -@permission_classes((TMSSDjangoModelPermissions,)) class ProjectNestedViewSet(LOFARNestedViewSet): queryset = models.Project.objects.all() serializer_class = serializers.ProjectSerializer @@ -750,8 +751,6 @@ class SchedulingUnitBlueprintNestedViewSet(LOFARNestedViewSet): class TaskDraftViewSet(LOFARViewSet): queryset = models.TaskDraft.objects.all() serializer_class = serializers.TaskDraftSerializer - permission_classes = (TMSSPermissions,) # todo: move to LOFARViewSet eventually? # Note: this should work the same, but something funny is going on: [IsProjectMember | TMSSDjangoModelPermissions] - filter_backends = (IsProjectMemberFilterBackend,) # todo: move to LOFARViewSet eventually? # prefetch all reverse related references from other models on their related_name to avoid a ton of duplicate queries queryset = queryset.prefetch_related('first_scheduling_relation') \ @@ -769,18 +768,20 @@ class TaskDraftViewSet(LOFARViewSet): queryset = queryset.select_related('copies') \ .select_related('copy_reason') - # do not permit listing if the queryset contains objects that the user has not permission for. - # Note: this is not required if we apply correct filtering, but it's a nice check that we do filter correctly. - # Do not check on other actions, as the queryset might contain items that will be filtered later. - # todo: see if there is something like a get_filtered_queryset that always only includes what will be returned - # to the user, so we can always check object permissions on everything. - def get_queryset(self): - qs = super().get_queryset() - if self.action == 'list': - qs = self.filter_queryset(qs) - for obj in qs: - self.check_object_permissions(self.request, obj) - return qs + # # do not permit listing if the queryset contains objects that the user has not permission for. + # # Note: this is not required if we apply correct filtering, but it's a nice check that we do filter correctly. + # # Do not check on other actions, as the queryset might contain items that will be filtered later. + # # todo: see if there is something like a get_filtered_queryset that always only includes what will be returned + # # to the user, so we can always check object permissions on everything. + # # todo: this causes a recursion error if has_permission is called from has_object_permission in TMSSPermissions + # # Probably a non-issue since we do the filtering anyway. + # def get_queryset(self): + # qs = super().get_queryset() + # if self.action == 'list': + # qs = self.filter_queryset(qs) + # for obj in qs: + # self.check_object_permissions(self.request, obj) + # return qs @swagger_auto_schema(responses={201: 'The created task blueprint, see Location in Response header', 403: 'forbidden'}, @@ -788,10 +789,6 @@ class TaskDraftViewSet(LOFARViewSet): @action(methods=['get'], detail=True, url_name="create_task_blueprint", name="Create TaskBlueprint") # todo: I think these actions should be 'post'-only, since they alter the DB ?! def create_task_blueprint(self, request, pk=None): task_draft = get_object_or_404(models.TaskDraft, pk=pk) - # todo: it seems there is no way to pass IsProjectMember as a permission class to actions without explicitly - # registering them in the router for some reason. but explicitly doing a check_object_permission works as well. - # We may want to extend get_object_or_404 to also perform a permission check before returning an object...? - self.check_object_permissions(self.request, task_draft) # or request.user.has_perm('create_task_blueprint') task_blueprint = create_task_blueprint_from_task_draft(task_draft) # url path magic to construct the new task_blueprint_path url diff --git a/SAS/TMSS/backend/test/CMakeLists.txt b/SAS/TMSS/backend/test/CMakeLists.txt index 9f0f6d3f3333c923241195bd75664e2eb385b70d..18d00552af28dc70f428041fd2c028057a64b021 100644 --- a/SAS/TMSS/backend/test/CMakeLists.txt +++ b/SAS/TMSS/backend/test/CMakeLists.txt @@ -34,6 +34,7 @@ if(BUILD_TESTING) lofar_add_test(t_scheduling) lofar_add_test(t_conversions) lofar_add_test(t_permissions) + lofar_add_test(t_permissions_system_roles) lofar_add_test(t_complex_serializers) set_tests_properties(t_scheduling PROPERTIES TIMEOUT 300) diff --git a/SAS/TMSS/backend/test/t_permissions.py b/SAS/TMSS/backend/test/t_permissions.py index 9b112ba4992374b682b722cc2f32eb8315bb8500..e79c126f907c493a146acd0118344435702df1c6 100755 --- a/SAS/TMSS/backend/test/t_permissions.py +++ b/SAS/TMSS/backend/test/t_permissions.py @@ -51,7 +51,13 @@ from lofar.sas.tmss.test.tmss_test_data_rest import TMSSRESTTestDataCreator from django.test import TestCase +from django.contrib.auth.models import User, Group, Permission + class ProjectPermissionTestCase(TestCase): + # This tests that the project permissions are enforced in light of the project roles that are externally provided + # for the user through the user admin. This test does not rely on the project permissions as defined in the system, + # but project permissions are created specifically for this test. + @classmethod def setUpClass(cls) -> None: diff --git a/SAS/TMSS/backend/test/t_permissions_system_roles.py b/SAS/TMSS/backend/test/t_permissions_system_roles.py new file mode 100755 index 0000000000000000000000000000000000000000..5d05682bec00597c71fc3ae94f46eaf6cc35a0d9 --- /dev/null +++ b/SAS/TMSS/backend/test/t_permissions_system_roles.py @@ -0,0 +1,674 @@ +#!/usr/bin/env python3 + +# Copyright (C) 2018 ASTRON (Netherlands Institute for Radio Astronomy) +# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands +# +# This file is part of the LOFAR software suite. +# The LOFAR software suite is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as published +# by the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# The LOFAR software suite is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>. + +# $Id: $ + + +# This functional test talks to the API like a regular user would. +# It is supposed to cover all REST http methods for all ViewSets. +# I am still a bit under the impression that we re-test Django functionality that we can expect to just work +# with some of these tests. On the other hand a lot of these provide us a nice basis for differentiating out +# behavior in a controlled way. +# We should probably also fully test behavior wrt mandatory and nullable fields. + +import unittest +import logging + +logger = logging.getLogger(__name__) +logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO) + +from lofar.common.test_utils import skip_integration_tests +if skip_integration_tests(): + exit(3) + +# Do Mandatory setup step: +# use setup/teardown magic for tmss test database, ldap server and django server +# (ignore pycharm unused import statement, python unittests does use at RunTime the tmss_test_environment_unittest_setup module) +from lofar.sas.tmss.test.tmss_test_environment_unittest_setup import * +# --- +from lofar.sas.tmss.test.test_utils import TMSSTestEnvironment +tmss_test_env = TMSSTestEnvironment(populate_schemas=True, populate_test_data=False, start_ra_test_environment=True, + start_postgres_listener=False, start_subtask_scheduler=False, start_dynamic_scheduler=False, + enable_viewflow=False) + +try: + tmss_test_env.start() +except: + tmss_test_env.stop() + exit(1) + +# tell unittest to stop (and automagically cleanup) the test database once all testing is done. +def tearDownModule(): + tmss_test_env.stop() +# --- + +from lofar.sas.tmss.test.tmss_test_data_django_models import * + +# import and setup test data creator +from lofar.sas.tmss.test.tmss_test_data_rest import TMSSRESTTestDataCreator +test_data_creator = TMSSRESTTestDataCreator(BASE_URL, AUTH) + +from lofar.sas.tmss.tmss.tmssapp.viewsets.permissions import TMSSPermissions +from lofar.sas.tmss.tmss.tmssapp.viewsets.scheduling import SubtaskViewSet +from django.contrib.auth.models import User, Group + + +class SystemPermissionTestCase(unittest.TestCase): + ''' + Tests for system permissons + ''' + + @classmethod + def setUpClass(cls) -> None: + super().setUpClass() + + # Create preparatory data + with tmss_test_env.create_tmss_client() as client: + cluster_url = client.get_path_as_json_object('/cluster/1')['url'] + + # setup: first create an observation, so the pipeline can have input. + obs_task_blueprint_data = test_data_creator.TaskBlueprint( + template_url=client.get_task_template(name="target observation")['url']) + obs_task_blueprint = test_data_creator.post_data_and_get_response_as_json_object(obs_task_blueprint_data, + '/task_blueprint/') + obs_subtask_template = client.get_subtask_template("observation control") + obs_spec = get_default_json_object_for_schema(obs_subtask_template['schema']) + obs_spec['stations']['digital_pointings'][0]['subbands'] = [0] + + obs_subtask_data = test_data_creator.Subtask(specifications_template_url=obs_subtask_template['url'], + specifications_doc=obs_spec, + cluster_url=cluster_url, + task_blueprint_url=obs_task_blueprint['url'], + raw_feedback='Observation.Correlator.channelWidth=3051.7578125') + obs_subtask = test_data_creator.post_data_and_get_response_as_json_object(obs_subtask_data, '/subtask/') + cls.obs_subtask_id = obs_subtask['id'] + obs_subtask_output_url = test_data_creator.post_data_and_get_url( + test_data_creator.SubtaskOutput(subtask_url=obs_subtask['url']), '/subtask_output/') + test_data_creator.post_data_and_get_url( + test_data_creator.Dataproduct(filename="L%s_SB000.MS" % obs_subtask['id'], + subtask_output_url=obs_subtask_output_url), '/dataproduct/') + + # Create test_data_creator as regular user + cls.test_data_creator = TMSSRESTTestDataCreator(BASE_URL, requests.auth.HTTPBasicAuth('paulus', 'pauluspass')) + response = requests.get(cls.test_data_creator.django_api_url + '/', auth=cls.test_data_creator.auth) + + # Populate permissions + tmss_test_env.populate_permissions() + + # Retrieve TO observer system role + cls.to_observer_group = Group.objects.get(name='TO observer') + + + def test_SubtaskViewSet_has_TMSSPermissions_in_permission_classes(self): + # Assert SubtaskViewSet has the TMSSPermissions permission class. + is_in_permission_classes = False + for p in SubtaskViewSet.get_permissions(SubtaskViewSet): + # Check TMSSPermission is included in the SubtaskViewSet permission classes + if not is_in_permission_classes: + if type(p).__name__ is TMSSPermissions.__name__: + # Check that it is actually an instance of TMSSPermissions + self.assertTrue(isinstance(p, TMSSPermissions)) + is_in_permission_classes = True + + self.assertTrue(is_in_permission_classes, + msg='SubtaskViewSet does not have %s in its permission classes.' % TMSSPermissions.__name__) + + + def test_Subtask_cannot_get_progress_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.get_progress_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the get_progress_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='get_progress_subtask')) + # Assert Paulus does not have the get_progress_subtask permission + self.assertFalse(user.has_perm('tmssapp.get_progress_subtask')) + + # Try to get_progress subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/get_progress/' % self.obs_subtask_id, 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_get_progress_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.get_progress_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the get_progress_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='get_progress_subtask')) + # Assert Paulus has the get_progress_subtask permission + self.assertTrue(user.has_perm('tmssapp.get_progress_subtask')) + + # Try to get_progress subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/get_progress/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_input_dataproducts_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.input_dataproducts_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the input_dataproducts_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='input_dataproducts_subtask')) + # Assert Paulus does not have the input_dataproducts_subtask permission + self.assertFalse(user.has_perm('tmssapp.input_dataproducts_subtask')) + + # Try to input_dataproducts subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/input_dataproducts/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_input_dataproducts_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.input_dataproducts_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the input_dataproducts_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='input_dataproducts_subtask')) + # Assert Paulus has the input_dataproducts_subtask permission + self.assertTrue(user.has_perm('tmssapp.input_dataproducts_subtask')) + + # Try to input_dataproducts subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/input_dataproducts/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_output_dataproducts_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.output_dataproducts_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the output_dataproducts_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='output_dataproducts_subtask')) + # Assert Paulus does not have the output_dataproducts_subtask permission + self.assertFalse(user.has_perm('tmssapp.output_dataproducts_subtask')) + + # Try to output_dataproducts subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/output_dataproducts/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_output_dataproducts_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.output_dataproducts_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the output_dataproducts_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='output_dataproducts_subtask')) + # Assert Paulus has the output_dataproducts_subtask permission + self.assertTrue(user.has_perm('tmssapp.output_dataproducts_subtask')) + + # Try to output_dataproducts subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/output_dataproducts/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_parset_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.parset_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the parset_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='parset_subtask')) + # Assert Paulus does not have the parset_subtask permission + self.assertFalse(user.has_perm('tmssapp.parset_subtask')) + + # Try to parset subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/parset/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_parset_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.parset_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the parset_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='parset_subtask')) + # Assert Paulus has the parset_subtask permission + self.assertTrue(user.has_perm('tmssapp.parset_subtask')) + + # Try to parset subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/parset/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_predecessors_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.predecessors_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the predecessors_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='predecessors_subtask')) + # Assert Paulus does not have the predecessors_subtask permission + self.assertFalse(user.has_perm('tmssapp.predecessors_subtask')) + + # Try to predecessors subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, + BASE_URL + '/subtask/%s/predecessors/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_predecessors_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.predecessors_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the predecessors_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='predecessors_subtask')) + # Assert Paulus has the predecessors_subtask permission + self.assertTrue(user.has_perm('tmssapp.predecessors_subtask')) + + # Try to predecessors subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/predecessors/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_process_feedback_and_set_to_finished_if_complete_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.process_feedback_and_set_to_finished_if_complete_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the process_feedback_and_set_to_finished_if_complete_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='process_feedback_and_set_to_finished_if_complete_subtask')) + # Assert Paulus does not have the process_feedback_and_set_to_finished_if_complete_subtask permission + self.assertFalse(user.has_perm('tmssapp.process_feedback_and_set_to_finished_if_complete_subtask')) + + # Try to process_feedback_and_set_to_finished_if_complete subtask and assert Paulus can't do it without the TO observer group permissions. + response = POST_and_assert_expected_response(self, BASE_URL + '/subtask/%s/process_feedback_and_set_to_finished_if_complete/' % self.obs_subtask_id, + {}, 403, {}, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_process_feedback_and_set_to_finished_if_complete_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.process_feedback_and_set_to_finished_if_complete_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the process_feedback_and_set_to_finished_if_complete_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='process_feedback_and_set_to_finished_if_complete_subtask')) + # Assert Paulus has the process_feedback_and_set_to_finished_if_complete_subtask permission + self.assertTrue(user.has_perm('tmssapp.process_feedback_and_set_to_finished_if_complete_subtask')) + + # Set subtask status to finishing, so it can process feedback and set to finished. + with tmss_test_env.create_tmss_client() as client: + client.set_subtask_status(self.obs_subtask_id, 'finishing') + + # Try to process_feedback_and_set_to_finished_if_complete subtask and assert Paulus can do it within the TO observer group permissions. + response = POST_and_assert_expected_response(self, BASE_URL + '/subtask/%s/process_feedback_and_set_to_finished_if_complete/' % self.obs_subtask_id, + {}, 200, {}, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_without_to_observer_group( + self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask')) + # Assert Paulus does not have the reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask permission + self.assertFalse(user.has_perm('tmssapp.reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask')) + + # Try to reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_with_to_observer_group( + self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask')) + # Assert Paulus has the reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask permission + self.assertTrue(user.has_perm('tmssapp.reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete_subtask')) + + # Try to reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/reprocess_raw_feedback_for_subtask_and_set_to_finished_if_complete/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_schedule_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.schedule_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the schedule_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='schedule_subtask')) + # Assert Paulus does not have the schedule_subtask permission + self.assertFalse(user.has_perm('tmssapp.schedule_subtask')) + + # Try to schedule subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, + BASE_URL + '/subtask/%s/schedule/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_schedule_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.schedule_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the schedule_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='schedule_subtask')) + # Assert Paulus has the schedule_subtask permission + self.assertTrue(user.has_perm('tmssapp.schedule_subtask')) + + # Set subtask status to defined, so it can be scheduled. + with tmss_test_env.create_tmss_client() as client: + client.set_subtask_status(self.obs_subtask_id, 'defined') + + # Try to schedule subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/schedule/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_state_log_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.state_log_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the state_log_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='state_log_subtask')) + # Assert Paulus does not have the state_log_subtask permission + self.assertFalse(user.has_perm('tmssapp.state_log_subtask')) + + # Try to state_log subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/state_log/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_state_log_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.state_log_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the state_log_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='state_log_subtask')) + # Assert Paulus has the state_log_subtask permission + self.assertTrue(user.has_perm('tmssapp.state_log_subtask')) + + # Try to state_log subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/state_log/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_successors_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.successors_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the successors_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='successors_subtask')) + # Assert Paulus does not have the successors_subtask permission + self.assertFalse(user.has_perm('tmssapp.successors_subtask')) + + # Try to successors subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/successors/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_successors_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.successors_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the successors_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='successors_subtask')) + # Assert Paulus has the successors_subtask permission + self.assertTrue(user.has_perm('tmssapp.successors_subtask')) + + # Try to successors subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/successors/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_task_log_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.task_log_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the task_log_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='task_log_subtask')) + # Assert Paulus does not have the task_log_subtask permission + self.assertFalse(user.has_perm('tmssapp.task_log_subtask')) + + # Try to task_log subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/task_log/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_task_log_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.task_log_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the task_log_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='task_log_subtask')) + # Assert Paulus has the task_log_subtask permission + self.assertTrue(user.has_perm('tmssapp.task_log_subtask')) + + # Try to task_log subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/task_log/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_transformed_output_dataproduct_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.transformed_output_dataproduct_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the transformed_output_dataproduct_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='transformed_output_dataproduct_subtask')) + # Assert Paulus does not have the transformed_output_dataproduct_subtask permission + self.assertFalse(user.has_perm('tmssapp.transformed_output_dataproduct_subtask')) + + # Try to transformed_output_dataproduct subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/transformed_output_dataproduct/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_transformed_output_dataproduct_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.transformed_output_dataproduct_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the transformed_output_dataproduct_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='transformed_output_dataproduct_subtask')) + # Assert Paulus has the transformed_output_dataproduct_subtask permission + self.assertTrue(user.has_perm('tmssapp.transformed_output_dataproduct_subtask')) + + # Try to transformed_output_dataproduct subtask and assert Paulus can do it within the TO observer group permissions. + with tmss_test_env.create_tmss_client() as client: + # NOTE: A 404 Not found error is returned if input_dataproducts are missing. It is not related to user's permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/transformed_output_dataproduct?input_dataproduct_id=%s' % ( + self.obs_subtask_id, 1), + 404, + auth=self.test_data_creator.auth) + + + def test_Subtask_cannot_unschedule_without_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while user.has_perm('tmssapp.unschedule_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the unschedule_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='unschedule_subtask')) + # Assert Paulus does not have the unschedule_subtask permission + self.assertFalse(user.has_perm('tmssapp.unschedule_subtask')) + + # Try to unschedule subtask and assert Paulus can't do it without the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/unschedule/' % self.obs_subtask_id, + 403, + auth=self.test_data_creator.auth) + + + def test_Subtask_can_unschedule_with_to_observer_group(self): + user = User.objects.get(username='paulus') + user.groups.set([self.to_observer_group]) + + # refresh user to update cache, see: https://docs.djangoproject.com/en/3.0/topics/auth/default/#permission-caching + user = User.objects.get(username='paulus') + while not user.has_perm('tmssapp.unschedule_subtask'): + user = User.objects.get(username='paulus') + + # Assert TO observer group has the unschedule_subtask permission + self.assertIsNotNone(self.to_observer_group.permissions.all().filter(codename='unschedule_subtask')) + # Assert Paulus has the unschedule_subtask permission + self.assertTrue(user.has_perm('tmssapp.unschedule_subtask')) + + # Set subtask status to scheduled, so it can be unscheduled. + with tmss_test_env.create_tmss_client() as client: + client.set_subtask_status(self.obs_subtask_id, 'scheduled') + + # Try to unschedule subtask and assert Paulus can do it within the TO observer group permissions. + response = GET_and_assert_equal_expected_code(self, BASE_URL + '/subtask/%s/unschedule/' % self.obs_subtask_id, + 200, + auth=self.test_data_creator.auth) + + +if __name__ == "__main__": + logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', + level=logging.INFO) + unittest.main() diff --git a/SAS/TMSS/backend/test/t_permissions_system_roles.run b/SAS/TMSS/backend/test/t_permissions_system_roles.run new file mode 100755 index 0000000000000000000000000000000000000000..f0bb28dd8eb18b7260e5e3a3af9dfe0dbe00c70d --- /dev/null +++ b/SAS/TMSS/backend/test/t_permissions_system_roles.run @@ -0,0 +1,6 @@ +#!/bin/bash + +# Run the unit test +source python-coverage.sh +python_coverage_test "*tmss*" t_permissions_system_roles.py + diff --git a/SAS/TMSS/backend/test/t_permissions_system_roles.sh b/SAS/TMSS/backend/test/t_permissions_system_roles.sh new file mode 100755 index 0000000000000000000000000000000000000000..54ab6cca1cae579743b5f8f95a9ccd13f4536858 --- /dev/null +++ b/SAS/TMSS/backend/test/t_permissions_system_roles.sh @@ -0,0 +1,3 @@ +#!/bin/sh + +./runctest.sh t_permissions_system_roles \ No newline at end of file diff --git a/SAS/TMSS/backend/test/t_tmssapp_specification_REST_API.py b/SAS/TMSS/backend/test/t_tmssapp_specification_REST_API.py index c43fc1ae28505d5528cd4a05b754cd3d282f2f0c..9fb0b24d6fdf1f97882ec4dc2a5157646dcd5247 100755 --- a/SAS/TMSS/backend/test/t_tmssapp_specification_REST_API.py +++ b/SAS/TMSS/backend/test/t_tmssapp_specification_REST_API.py @@ -2911,7 +2911,7 @@ class ExtendedViewTestCase(unittest.TestCase): # assert that task templates inside task blueprints are expanded self.assertIn('schema', r['task_blueprints'][0]['specifications_template']) - +# todo: move to t_permissions (I tried, but it broke) class CyclePermissionTestCase(unittest.TestCase): @classmethod @@ -3010,7 +3010,7 @@ class SchedulingUnitObservingStrategyTemplateTestCase(unittest.TestCase): GET_and_assert_in_expected_response_result_list(self, BASE_URL + '/scheduling_set/%s/scheduling_unit_observing_strategy_template/' % scheduling_set.id, template_test_data, 1) - +# todo: move to t_permissions (I tried, but it broke) class SystemRolePermissionTestCase(unittest.TestCase): @classmethod @@ -3123,10 +3123,17 @@ class SystemRolePermissionTestCase(unittest.TestCase): self.assertFalse(user.has_perm('tmssapp.view_project')) - GET_and_assert_equal_expected_code(self, BASE_URL + '/project/', 403, auth=self.test_data_creator.auth) + # Note: with just the model permissions, you'd expect a permission denied error on a listing. + # But since we do also allow viewing projects based on project permissions (users should be able to list + # their own projects), we get an empty list instead. + #GET_and_assert_equal_expected_code(self, BASE_URL + '/project/', 403, auth=self.test_data_creator.auth) + list = GET_and_assert_equal_expected_code(self, BASE_URL + '/project/', 200, auth=self.test_data_creator.auth) + self.assertEqual(list['count'], 0) GET_and_assert_equal_expected_code(self, self.project_url, 403, auth=self.test_data_creator.auth) + + def test_Project_can_be_viewed_with_scientist_group(self): user = User.objects.get(username='paulus') user.groups.set([self.scientist_group]) diff --git a/SAS/TMSS/backend/test/tmss_test_data_rest.py b/SAS/TMSS/backend/test/tmss_test_data_rest.py index b3357b2bf9d0b83160e1245e5109f153c44f7348..8d59aa434be32e177cffc564ec6b1f6edb44e938 100644 --- a/SAS/TMSS/backend/test/tmss_test_data_rest.py +++ b/SAS/TMSS/backend/test/tmss_test_data_rest.py @@ -612,7 +612,7 @@ class TMSSRESTTestDataCreator(): return self._cluster_url - def Subtask(self, cluster_url=None, task_blueprint_url=None, specifications_template_url=None, specifications_doc=None, state:str="defining", start_time: datetime=None, stop_time: datetime=None, raw_feedack:str =None): + def Subtask(self, cluster_url=None, task_blueprint_url=None, specifications_template_url=None, specifications_doc=None, state:str="defining", start_time: datetime=None, stop_time: datetime=None, raw_feedback:str =None): if cluster_url is None: cluster_url = self.cached_cluster_url @@ -646,7 +646,7 @@ class TMSSRESTTestDataCreator(): "tags": ["TMSS", "TESTING"], "do_cancel": datetime.utcnow().isoformat(), "cluster": cluster_url, - "raw_feedback": raw_feedack} + "raw_feedback": raw_feedback} @property def cached_subtask_url(self):