Select Git revision
schedulingunitflow.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
schedulingunitflow.py 13.39 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
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
#Viewsets and serializers to access intermediate steps of the QA Workflow
#through DRF
class QAReportingTOViewSet(WorkflowViewSet):
queryset = models.QAReportingTO.objects.all()
serializer_class = serializers.QAReportingTOSerializer
class QAReportingSOSViewSet(WorkflowViewSet):
queryset = models.QAReportingSOS.objects.all()
serializer_class = serializers.QAReportingSOSSerializer
class PIVerificationViewSet(WorkflowViewSet):
queryset = models.PIVerification.objects.all()
serializer_class = serializers.PIVerificationSerializer
class DecideAcceptanceViewSet(WorkflowViewSet):
queryset = models.DecideAcceptance.objects.all()
serializer_class = serializers.DecideAcceptanceSerializer
class SchedulingUnitProcessViewSet(WorkflowViewSet):
queryset = models.SchedulingUnitProcess.objects.all()
serializer_class = serializers.SchedulingUnitProcessSerializer
class SchedulingUnitTaskViewSet(WorkflowViewSet):
queryset = Task.objects.all()
serializer_class = serializers.SchedulingUnitTaskSerializer
class QAReportingTOView(FlowMixin, generic.CreateView):
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, generic.CreateView):
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, generic.CreateView):
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, generic.CreateView):
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, generic.CreateView):
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,
viewsets.GenericViewSet):
queryset = Task.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 = Task.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
Task.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,
viewsets.GenericViewSet):
queryset = Task.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:
Task.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,
viewsets.GenericViewSet):
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:
data = serialize('json', models.SchedulingUnitProcess.objects.filter(id=self.kwargs['qa_scheduling_unit_process_id'])[0].active_tasks())
return HttpResponse(data, content_type='application/json')
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,
viewsets.GenericViewSet):
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()
# 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)