Coverage for backend/django/flowsheetInternals/unitops/viewsets/FlowsheetEditOperationViewSet.py: 100%

1from drf_spectacular.utils import OpenApiParameter, OpenApiTypes, extend_schema 

2from rest_framework import status 

3from rest_framework.decorators import action 

4from rest_framework.response import Response 

5 

6from core.viewset import ReadOnlyModelViewSet 

7from flowsheetInternals.unitops.models.FlowsheetEditOperation import ( 

8 FlowsheetEditOperation, 

9) 

10from flowsheetInternals.unitops.serializers.FlowsheetEditOperationSerializer import ( 

11 FlowsheetEditOperationSerializer, 

12 FlowsheetEditTransitionRequestSerializer, 

13 FlowsheetEditTransitionSerializer, 

14) 

15from flowsheetInternals.unitops.services.edit_operations.service import ( 

16 CURRENT_SCHEMA_VERSION, 

17 operation_response, 

18 transition_operation, 

19) 

20 

21 

22class FlowsheetEditOperationViewSet(ReadOnlyModelViewSet): 

23 """List and transition durable undo/redo operations for one flowsheet.""" 

24 

25 serializer_class = FlowsheetEditOperationSerializer 

26 

27 def get_queryset(self): 

28 return FlowsheetEditOperation.objects.filter( 

29 flowsheet_id=self.request.query_params.get("flowsheet"), 

30 schema_version=CURRENT_SCHEMA_VERSION, 

31 ).exclude(state=FlowsheetEditOperation.State.Invalidated) 

32 

33 @extend_schema( 

34 parameters=[ 

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

36 ] 

37 ) 

38 def list(self, request): 

39 return super().list(request) 

40 

41 @extend_schema( 

42 request=FlowsheetEditTransitionRequestSerializer, 

43 responses=FlowsheetEditTransitionSerializer, 

44 parameters=[ 

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

46 ], 

47 ) 

48 @action(detail=True, methods=["post"]) 

49 def undo(self, request, pk=None): 

50 """Apply the inverse state for the latest eligible operation.""" 

51 request_serializer = FlowsheetEditTransitionRequestSerializer(data=request.data) 

52 request_serializer.is_valid(raise_exception=True) 

53 operation = transition_operation( 

54 self.get_object(), 

55 undo=True, 

56 expected_revision=request_serializer.validated_data["expected_revision"], 

57 ) 

58 return Response(operation_response(operation), status=status.HTTP_200_OK) 

59 

60 @extend_schema( 

61 request=FlowsheetEditTransitionRequestSerializer, 

62 responses=FlowsheetEditTransitionSerializer, 

63 parameters=[ 

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

65 ], 

66 ) 

67 @action(detail=True, methods=["post"]) 

68 def redo(self, request, pk=None): 

69 """Reapply the forward state for the next eligible operation.""" 

70 request_serializer = FlowsheetEditTransitionRequestSerializer(data=request.data) 

71 request_serializer.is_valid(raise_exception=True) 

72 operation = transition_operation( 

73 self.get_object(), 

74 undo=False, 

75 expected_revision=request_serializer.validated_data["expected_revision"], 

76 ) 

77 return Response(operation_response(operation), status=status.HTTP_200_OK)