Coverage for backend/django/core/auxiliary/views/CopyFlowsheetView.py: 87%
59 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.db import transaction
2from drf_spectacular.utils import OpenApiParameter, OpenApiTypes, extend_schema
3from rest_framework import status
4from rest_framework.decorators import api_view
5from rest_framework.response import Response
7from authentication.user.models import User
8from core.auxiliary.models.Flowsheet import Flowsheet
9from core.auxiliary.models.Project import Project
10from core.auxiliary.serializers.FlowsheetCloneValidationSerializer import (
11 CloneRequestSerializer,
12 FlowsheetCloneValidationErrorSerializer,
13 ScenarioResultCopySummarySerializer,
14)
15from core.auxiliary.services.flowsheet_states import (
16 FlowsheetStateCloneError,
17 clone_to_new_flowsheet,
18 clone_validation_error_data,
19 summarize_scenario_results,
20)
21from core.managers import has_flowsheet_read_access, has_flowsheet_write_access
22from core.validation import (
23 api_view_validate,
24 bypass_write_access_checks,
25 sanitize_flowsheet_id,
26)
29@api_view_validate
30@extend_schema(
31 parameters=[
32 OpenApiParameter(
33 name="flowsheet",
34 type=OpenApiTypes.INT,
35 location=OpenApiParameter.QUERY,
36 description="ID of the flowsheet whose duplicate is being previewed",
37 ),
38 ],
39 responses={200: ScenarioResultCopySummarySerializer},
40)
41@api_view(["GET"])
42def copy_flowsheet_preview(request) -> Response:
43 """Return scenario-result counts before a user confirms duplication."""
45 flowsheet_id = sanitize_flowsheet_id(request.GET.get("flowsheet"))
46 if not has_flowsheet_read_access(request.user, flowsheet_id): 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true
47 return Response(
48 data={"error": "You do not have access to this flowsheet"},
49 status=status.HTTP_403_FORBIDDEN,
50 )
51 try:
52 flowsheet = Flowsheet.objects.select_related("current_state").get(
53 pk=flowsheet_id
54 )
55 except Flowsheet.DoesNotExist:
56 return Response(
57 data={"error": "Flowsheet not found"},
58 status=status.HTTP_404_NOT_FOUND,
59 )
60 summary = summarize_scenario_results(
61 flowsheet_state_ids=[flowsheet.current_state_id]
62 )
63 return Response(ScenarioResultCopySummarySerializer(summary).data)
66@api_view_validate
67@extend_schema(
68 parameters=[
69 OpenApiParameter(
70 name="flowsheet",
71 type=OpenApiTypes.INT,
72 location=OpenApiParameter.QUERY,
73 description="ID of the flowsheet to copy",
74 ),
75 OpenApiParameter(
76 name="repair_legacy_formula_units",
77 type=OpenApiTypes.BOOL,
78 location=OpenApiParameter.QUERY,
79 required=False,
80 description="Repair deterministic legacy unit mentions before retrying",
81 ),
82 OpenApiParameter(
83 name="include_scenario_results",
84 type=OpenApiTypes.BOOL,
85 location=OpenApiParameter.QUERY,
86 required=False,
87 description="Copy retained scenario Solution values",
88 ),
89 ],
90 responses={
91 200: OpenApiTypes.INT,
92 422: FlowsheetCloneValidationErrorSerializer,
93 },
94)
95@api_view(["POST"])
96def copy_flowsheet(request) -> Response:
97 """Duplicate an accessible flowsheet or report repairable source problems."""
99 flowsheet_id = request.GET.get("flowsheet")
100 flowsheet_id = sanitize_flowsheet_id(flowsheet_id)
101 input_serializer = CloneRequestSerializer(data=request.query_params)
102 input_serializer.is_valid(raise_exception=True)
103 repair_legacy_formula_units = input_serializer.validated_data[
104 "repair_legacy_formula_units"
105 ]
106 include_scenario_results = input_serializer.validated_data[
107 "include_scenario_results"
108 ]
110 user: User = request.user
111 if not has_flowsheet_read_access(user, flowsheet_id):
112 return Response(
113 data={"error": "You do not have access to this flowsheet"},
114 status=status.HTTP_403_FORBIDDEN,
115 )
116 if repair_legacy_formula_units and not has_flowsheet_write_access(
117 user, flowsheet_id
118 ):
119 return Response(
120 data={"error": "You do not have write access to repair this flowsheet"},
121 status=status.HTTP_403_FORBIDDEN,
122 )
124 try:
125 old_flowsheet = Flowsheet.objects.get(pk=flowsheet_id)
126 except Flowsheet.DoesNotExist:
127 return Response(
128 data={"error": "Flowsheet not found"},
129 status=status.HTTP_404_NOT_FOUND,
130 )
132 # Create a project and access row, then add the copied flowsheet into that project.
133 try:
134 with transaction.atomic():
135 target_project = None
136 if (
137 old_flowsheet.project_id is not None
138 and not has_flowsheet_write_access(user, flowsheet_id)
139 and old_flowsheet.project.owner_id != user.id
140 ):
141 # Read-only users receive a personal project for the duplicate.
142 target_project = Project.create_empty(
143 name=f"{old_flowsheet.name} (copy)"[:64],
144 owner=user,
145 active_flowsheet=None,
146 is_starred=False,
147 is_binned=False,
148 binned_at=None,
149 )
150 with bypass_write_access_checks():
151 # Source reads and target writes cross flowsheet access scopes after
152 # the request-level permission check above.
153 new_flowsheet = clone_to_new_flowsheet(
154 source_flowsheet=old_flowsheet,
155 user=user,
156 target_project=target_project,
157 repair_legacy_formula_units=repair_legacy_formula_units,
158 include_scenario_results=include_scenario_results,
159 )
160 except FlowsheetStateCloneError as exc:
161 if not exc.is_user_repairable: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 raise
163 return Response(
164 data=clone_validation_error_data(exc, flowsheet=old_flowsheet),
165 status=status.HTTP_422_UNPROCESSABLE_ENTITY,
166 )
168 active_project = target_project or new_flowsheet.project
169 if active_project is not None: 169 ↛ 172line 169 didn't jump to line 172 because the condition on line 169 was always true
170 active_project.set_active_flowsheet(new_flowsheet)
172 return Response(data=new_flowsheet.pk, status=status.HTTP_200_OK)