Coverage for backend/django/core/auxiliary/views/CopyFlowsheetView.py: 92%

1 

2from rest_framework.response import Response 

3from drf_spectacular.utils import extend_schema, OpenApiParameter, OpenApiTypes 

4from rest_framework.decorators import api_view 

5from core.validation import api_view_validate, sanitize_flowsheet_id, bypass_write_access_checks 

6from core.auxiliary.methods.CopyFlowsheet import copy_flowsheet_data 

7from authentication.user.models import User 

8from core.auxiliary.models.Flowsheet import Flowsheet 

9from core.managers import has_flowsheet_read_access 

10 

11@api_view_validate 

12@extend_schema(parameters=[ 

13 OpenApiParameter(name='flowsheet', type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, description="ID of the flowsheet to copy"), 

14], responses={200: OpenApiTypes.INT}) 

15@api_view(['POST']) 

16def copy_flowsheet(request) -> Response: 

17 flowsheet_id = request.GET.get("flowsheet") 

18 flowsheet_id = sanitize_flowsheet_id(flowsheet_id) 

19 

20 user: User = request.user 

21 if not has_flowsheet_read_access(user, flowsheet_id): 

22 return Response(data={"error": "You do not have access to this flowsheet"}, status=403) 

23 

24 try: 

25 old_flowsheet = Flowsheet.objects.get(pk=flowsheet_id) 

26 except Flowsheet.DoesNotExist: 

27 return Response(data={"error": "Flowsheet not found"}, status=404) 

28 with bypass_write_access_checks(): 

29 # The copy operation reads from the source flowsheet context while 

30 # creating objects for a brand new flowsheet owned by the requester. 

31 # That would trip the normal manager-level write guard unless we bypass 

32 # it for this already-validated internal flow. 

33 new_flowsheet = copy_flowsheet_data(old_flowsheet, user=user) 

34 

35 return Response(data=new_flowsheet.pk, status=200)