Coverage for backend/core/auxiliary/viewsets/TaskViewSet.py: 66%

1from django.views.decorators.csrf import csrf_exempt 

2from drf_spectacular.types import OpenApiTypes 

3from drf_spectacular.utils import extend_schema, OpenApiParameter 

4from rest_framework import status 

5from rest_framework.decorators import action, authentication_classes 

6from rest_framework.response import Response 

7 

8from authentication.custom_drf_authentication import DaprApiTokenAuthentication 

9from common.models.general import TaskRunningEvent 

10from core.auxiliary.managers import TaskManager 

11from core.pagination import ViewSetPagination 

12from core.auxiliary.models.Task import Task 

13from core.auxiliary.serializers import TaskSerializer 

14from core.validation import api_view_ignore_access_control 

15from core.viewset import ReadOnlyModelViewSet 

16 

17 

18class TaskViewSet(ReadOnlyModelViewSet): 

19 serializer_class = TaskSerializer 

20 pagination_class = ViewSetPagination 

21 

22 # Filter to only show tasks for a specific flowsheet 

23 def get_queryset(self): 

24 queryset = Task.objects.all().select_related("metadata") 

25 

26 if self.action == 'list': 

27 non_child_tasks = queryset.filter(parent__isnull=True) 

28 return non_child_tasks 

29 

30 return queryset 

31 

32 @extend_schema( 

33 parameters=[ 

34 OpenApiParameter(name="flowsheet", required=True, type=OpenApiTypes.INT), 

35 ] 

36 ) 

37 def list(self, request): 

38 return super().list(request) 

39 

40 @extend_schema(responses=TaskSerializer(many=True)) 

41 @action(detail=True, methods=['get'], url_path='children') 

42 def get_task_children(self, request, pk): 

43 children = Task.objects.filter(parent_id=pk) 

44 

45 page = self.paginate_queryset(children) 

46 if page is not None: 

47 serializer = self.get_serializer(page, many=True) 

48 return self.get_paginated_response(serializer.data) 

49 

50 return Response(status=status.HTTP_400_BAD_REQUEST) 

51 

52 @action(detail=False, methods=['post'], url_path='running', authentication_classes=[DaprApiTokenAuthentication]) 

53 @extend_schema(exclude=True) 

54 @csrf_exempt 

55 @api_view_ignore_access_control 

56 def handle_task_running_event(self, request): 

57 task_running_event = TaskRunningEvent.model_validate(request.data) 

58 TaskManager.handle_task_running_event(task_running_event.data.task_id) 

59 

60 return Response(status=status.HTTP_200_OK)