Select Git revision
schedulingunitflow.py
TMSS-970: Fix merge conflicts
Mario Raciti authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
schedulingunitflow.py 16.11 KiB
from django.shortcuts import render, redirect
from rest_framework import viewsets, mixins, status
from rest_framework.response import Response
from rest_framework.decorators import action
from lofar.sas.tmss.tmss.workflowapp import models
from django.views import generic
from viewflow.flow.views import StartFlowMixin, FlowMixin
from viewflow.decorators import flow_start_view, flow_view
from viewflow.flow.views.utils import get_next_task_url
from django.forms import CharField, CheckboxInput
from django.forms.models import modelform_factory
#from viewflow.models import Task, Process
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from drf_yasg.inspectors import SwaggerAutoSchema
from drf_yasg.openapi import Parameter
from django.core.serializers import serialize
from django.http import HttpResponse, JsonResponse
from django.urls import NoReverseMatch
from viewflow.flow import views, viewset
from viewflow.flow.views.actions import BaseTaskActionView
from lofar.sas.tmss.tmss.workflowapp.viewsets.workflow_viewset import WorkflowViewSet
from lofar.sas.tmss.tmss.tmssapp.adapters.keycloak import get_users_by_role_in_project
from django.contrib.auth import get_user_model
User = get_user_model()
import uuid
from .. import forms, models, serializers, flows
import logging
logger = logging.getLogger(__name__)
import requests
from django.utils import timezone
from lofar.sas.tmss.tmss.tmssapp.viewsets.permissions import TMSSPermissions, IsProjectMemberFilterBackend
from rest_framework.filters import OrderingFilter
class PreventSaveByUnassignedUsersMixin():
def update(self, request, pk, **kwargs):
instance = self.serializer_class.Meta.model.objects.get(pk=pk)
if request.user != instance.owner:
raise Exception('Task=%s is not assigned to the requesting user, changes are only permitted by the user the task has been assigned to.' % self)
super(PreventSaveByUnassignedUsersMixin, self).save(request, pk, **kwargs)
class TMSSPermissionModelViewSet(viewsets.ModelViewSet):
permission_classes = (TMSSPermissions,)
filter_backends = (OrderingFilter, IsProjectMemberFilterBackend,)
def _get_permitted_methods(self, request):
# Django returns an "Allow" header that reflects what methods the model supports in principle, but not what
# the current user is actually has permission to perform. We use the "Access-Control-Allow-Methods" header
# to disclose read/write permission to the frontend, so that it can render its views accordingly.
allowed_methods = []
for method in ['GET', 'PUT', 'POST', 'PATCH', 'DELETE']:
request.method = method
if TMSSPermissions().has_permission(request=request, view=self):
allowed_methods.append(method)
return allowed_methods
def retrieve(self, request, pk=None, **kwargs):
response = super().retrieve(request, pk, **kwargs)
if "Access-Control-Allow-Methods" not in response:
response["Access-Control-Allow-Methods"] = ", ".join(self._get_permitted_methods(request))
return response
class TMSSPermissionGenericViewSet(viewsets.GenericViewSet):
permission_classes = (TMSSPermissions,)
filter_backends = (OrderingFilter, IsProjectMemberFilterBackend,)
#Viewsets and serializers to access intermediate steps of the QA Workflow
#through DRF
class QAReportingTOViewSet(WorkflowViewSet, TMSSPermissionModelViewSet):
queryset = models.QAReportingTO.objects.all()
serializer_class = serializers.QAReportingTOSerializer
class QAReportingSOSViewSet(WorkflowViewSet, TMSSPermissionModelViewSet):
queryset = models.QAReportingSOS.objects.all()
serializer_class = serializers.QAReportingSOSSerializer
class PIVerificationViewSet(WorkflowViewSet, TMSSPermissionModelViewSet):
queryset = models.PIVerification.objects.all()
serializer_class = serializers.PIVerificationSerializer
class DecideAcceptanceViewSet(WorkflowViewSet, TMSSPermissionModelViewSet):
queryset = models.DecideAcceptance.objects.all()
serializer_class = serializers.DecideAcceptanceSerializer
class SchedulingUnitProcessViewSet(WorkflowViewSet, TMSSPermissionModelViewSet):
queryset = models.SchedulingUnitProcess.objects.all()
serializer_class = serializers.SchedulingUnitProcessSerializer
class SchedulingUnitTaskViewSet(WorkflowViewSet, PreventSaveByUnassignedUsersMixin, TMSSPermissionModelViewSet):
queryset = models.SchedulingUnitTask.objects.all()
serializer_class = serializers.SchedulingUnitTaskSerializer
class TMSSPermissionView(generic.CreateView):
# todo: comment in to marshal permissions on the viewflow-internal views, if required.
# Not sure this doesn't break anything, so leaving this out for now.
# permission_classes = (TMSSPermissions,)
pass
class QAReportingTOView(FlowMixin, TMSSPermissionView):
template_name = 'qa_reporting.html'
model = models.QAReportingTO
#form_class=forms.QAReportingTO
fields = [
'operator_report', 'operator_accept'
]
def form_valid(self, form):
report_data = form.save(commit=False)
report_data.save()
self.activation.process.qa_reporting_to = report_data
self.activation.process.save()
self.activation_done()
try:
return redirect(self.get_success_url())
except NoReverseMatch as e:
return
def activation_done(self, *args, **kwargs):
"""Finish the task activation."""
logging.info('Activation done')
self.activation.done()
class QAReportingSOSView(FlowMixin, TMSSPermissionView):
template_name = 'qa_reporting.html'
model = models.QAReportingSOS
fields = [
'sos_report', 'quality_within_policy','sos_accept_show_pi'
]
def form_valid(self, form):
report_data = form.save(commit=False)
report_data.save()
self.activation.process.qa_reporting_sos = report_data
self.activation.process.save()
self.activation_done()
try:
return redirect(self.get_success_url())
except NoReverseMatch as e:
return
def activation_done(self, *args, **kwargs):
"""Finish the task activation."""
logging.info('Activation done')
self.activation.done()
class PIVerificationView(FlowMixin, TMSSPermissionView):
template_name = 'qa_reporting.html'
model = models.PIVerification
fields = [
'pi_report', 'pi_accept'
]
def form_valid(self, form):
report_data = form.save(commit=False)
report_data.save()
self.activation.process.pi_verification = report_data
self.activation.process.save()
self.activation_done()
try:
return redirect(self.get_success_url())
except NoReverseMatch as e:
return
def activation_done(self, *args, **kwargs):
"""Finish the task activation."""
logging.info('Activation done')
self.activation.done()
class DecideAcceptanceView(FlowMixin, TMSSPermissionView):
template_name = 'qa_reporting.html'
model = models.DecideAcceptance
fields = [
'sos_accept_after_pi'
]
def form_valid(self, form):
report_data = form.save(commit=False)
report_data.save()
self.activation.process.decide_acceptance = report_data
self.activation.process.save()
self.activation_done()
try:
return redirect(self.get_success_url())
except NoReverseMatch as e:
return
def activation_done(self, *args, **kwargs):
"""Finish the task activation."""
logging.info('Activation done')
self.activation.done()
class UnpinDataView(FlowMixin, TMSSPermissionView):
template_name = 'qa_reporting.html'
model = models.UnpinData
fields = [
'unpin_data'
]
def form_valid(self, form):
report_data = form.save(commit=False)
report_data.save()
self.activation.process.unpin_data = report_data
self.activation.process.save()
self.activation_done()
try:
return redirect(self.get_success_url())
except NoReverseMatch as e:
return
def activation_done(self, *args, **kwargs):
# TODO: Should Wait for data to be unpinned?
"""Finish the task activation."""
logging.info('Activation done')
self.activation.done()
class SchedulingUnitTaskAssignViewSet(mixins.CreateModelMixin,
#mixins.ListModelMixin,
#mixins.RetrieveModelMixin,
TMSSPermissionGenericViewSet):
queryset = models.SchedulingUnitTask.objects.all()
serializer_class = serializers.SchedulingUnitAssignTaskSerializer
@swagger_auto_schema(responses={200: 'Assign Scheduling Unit Task to the specified user',
403: 'forbidden',
422: 'error'},
operation_description="Assign a Scheduling Unit Task to an user")
def create(self, request, *args, **kwargs):
if 'qa_scheduling_unit_task_id' in kwargs:
try:
if request.GET.get('user_email'): # For some reason this is GET data, even though it's a POST action...
user, _ = User.objects.get_or_create(email=request.GET.get('user_email'), defaults={'username': 'user_%s' % uuid.uuid4()}) # Users that log in via Keycloak are matched with Django users with same email address
elif request.GET.get('project_role'): # For some reason this is GET data, even though it's a POST action...
task = models.SchedulingUnitTask.objects.filter(id=self.kwargs['qa_scheduling_unit_task_id'])[0]
scheduling_unit_blueprint = task.flow_process.su
project = scheduling_unit_blueprint.project.name
users = get_users_by_role_in_project(role=request.GET.get('project_role'), project=project)
if users and '@' in users[0]:
user_email = users[0]
else:
# Keycloak refers to users with particular project roles in a peculiar way, and TMSS has to map these
# to the user's email address first, so that they can be matched with Django users. Keycloak sometimes
# returns an unmappable user representation, in which case we get a human-readable string instead of
# an email address returned.
content = {'AttributeError': 'Cannot determine a user with role=%s in project=%s to assign this task to' % (request.GET.get('project_role'), project)}
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
user, _ = User.objects.get_or_create(email=user_email, defaults={'username': 'user_%s' % uuid.uuid4()})
else:
user = self.request.user
models.SchedulingUnitTask.objects.filter(id=self.kwargs['qa_scheduling_unit_task_id'])[0].activate().assign(user)
content = {'Assigned': 'Scheduling Unit Task assigned to user=%s' % user.email}
return Response(content, status=status.HTTP_200_OK)
except AttributeError:
content = {'AttributeError': 'Cannot assign the specified Scheduling Unit Task to a user'}
logger.exception(str(content))
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
except IndexError:
content = {'IndexError': 'No Scheduling Unit Task with the specified id'}
logger.exception(str(content))
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
class SchedulingUnitTaskUnassignViewSet(mixins.CreateModelMixin,
#mixins.ListModelMixin,
#mixins.RetrieveModelMixin,
TMSSPermissionGenericViewSet):
queryset = models.SchedulingUnitTask.objects.all()
serializer_class = serializers.SchedulingUnitUnassignTaskSerializer
@swagger_auto_schema(responses={200: '',
403: 'forbidden',
422: 'error'},
operation_description="Unassign a Scheduling Unit Task")
def create(self, request, *args, **kwargs):
if 'qa_scheduling_unit_task_id' in kwargs:
try:
models.SchedulingUnitTask.objects.filter(id=self.kwargs['qa_scheduling_unit_task_id'])[0].activate().unassign()
content = {'Unassign': 'Scheduling Unit Task unassigned'}
return Response(content, status=status.HTTP_200_OK)
except AttributeError:
content = {'Unassign': 'Cannot unassign the specified Scheduling Unit Task'}
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
except IndexError:
content = {'Unassign': 'No Scheduling Unit Task with the specified id'}
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
class SchedulingUnitGetActiveTasksViewSet(mixins.CreateModelMixin,
#mixins.ListModelMixin,
#mixins.RetrieveModelMixin,
TMSSPermissionGenericViewSet):
queryset = models.SchedulingUnitProcess.objects.all()
serializer_class = serializers.SchedulingUnitGetActiveTasksSerializer
@swagger_auto_schema(responses={200: 'List of non finished tasks.',
403: 'forbidden',
422: 'error'},
operation_description="Get the list of active tasks.")
def create(self, request, *args, **kwargs):
if 'qa_scheduling_unit_process_id' in kwargs:
try:
tasks = models.SchedulingUnitProcess.objects.filter(id=self.kwargs['qa_scheduling_unit_process_id'])[0].active_tasks()
return JsonResponse(list(tasks.values()), safe=False)
except IndexError:
content = {'Get Active Task(s)': 'No Process with the specified id'}
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
class SchedulingUnitTaskExecuteViewSet(mixins.CreateModelMixin,
#mixins.ListModelMixin,
#mixins.RetrieveModelMixin,
TMSSPermissionGenericViewSet):
queryset = models.SchedulingUnitProcess.objects.all()
serializer_class = serializers.SchedulingUnitGetActiveTasksSerializer
@swagger_auto_schema(responses={200: '',
403: 'forbidden',
422: 'error'},
operation_description="Unassign a Scheduling Unit Task")
def create(self, request, *args, **kwargs):
if 'qa_scheduling_unit_process_id' in kwargs:
try:
process= models.SchedulingUnitProcess.objects.get(pk=self.kwargs['qa_scheduling_unit_process_id'])
task = models.SchedulingUnitProcess.objects.get(pk=self.kwargs['qa_scheduling_unit_process_id']).active_tasks()[0]
view = task.flow_task._view_class.as_view()
act=task.activate()
act.prepare()
# todo: Is this action what is considered 'saving a workflow' (next to direct access to the task model)?
if request.user != task.owner:
raise Exception('Task=%s is not assigned to the requesting user, task can only be performed by the user the task has been assigned to.' % task)
# Prepare the POST request's fields
request._request.POST = request._request.POST.copy()
for field in request.data:
request._request.POST[field] = request.data[field]
request._request.POST['_viewflow_activation-started'] = timezone.now()
request._request.POST['_done'] = ''
response = view(request._request, flow_class=flows.SchedulingUnitFlow, flow_task=task.flow_task,
process_pk=process.pk, task_pk=task.pk)
content = {'Perform Task': 'Task Performed'}
return Response(content, status=status.HTTP_200_OK)
except AttributeError:
content = {'Perform Task': 'Cannot perform the active Scheduling Unit Task'}
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)
except IndexError:
content = {'Perform Task': 'No Scheduling Unit Process with the specified id'}
return Response(content, status=status.HTTP_422_UNPROCESSABLE_ENTITY)