Coverage for backend/django/core/auxiliary/services/flowsheet_states/errors.py: 93%
78 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Structured failures shared by flowsheet clone services and API adapters."""
3import logging
4from collections import defaultdict
5from dataclasses import dataclass, replace
6from enum import StrEnum
9CLONE_VALIDATION_ERROR_CODE = "flowsheet_clone_validation_failed"
10MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY = 25
11logger = logging.getLogger(__name__)
14class CloneDeficiencyType(StrEnum):
15 """Stable classes of source-data problems that a user can repair."""
17 FORMULA_REFERENCES = "formula_references"
18 FLOWSHEET_RELATIONSHIPS = "flowsheet_relationships"
19 FLOWSHEET_STRUCTURE = "flowsheet_structure"
22@dataclass(frozen=True)
23class CloneDeficiencyDetail:
24 """Product-safe location of one repairable problem in a flowsheet."""
26 deficiency_type: CloneDeficiencyType
27 flowsheet_id: int | None
28 flowsheet_name: str | None
29 object_name: str
30 object_type: str
31 field_name: str
32 reference_name: str | None = None
35class FlowsheetStateCloneError(ValueError):
36 """Raised when a source aggregate cannot be cloned without corruption."""
38 def __init__(
39 self,
40 message: str,
41 *,
42 deficiency_type: CloneDeficiencyType | None = None,
43 source_flowsheet_id: int | None = None,
44 source_flowsheet_name: str | None = None,
45 deficiency_counts: dict[CloneDeficiencyType, int] | None = None,
46 details: tuple[CloneDeficiencyDetail, ...] = (),
47 ) -> None:
48 super().__init__(message)
49 if deficiency_counts is None:
50 deficiency_counts = {deficiency_type: 1} if deficiency_type else {}
51 self.deficiency_counts = dict(deficiency_counts)
52 self.deficiency_type = deficiency_type or next(
53 iter(self.deficiency_counts), None
54 )
55 self.source_flowsheet_id = source_flowsheet_id
56 self.source_flowsheet_name = source_flowsheet_name
57 self.details = details
59 @property
60 def is_user_repairable(self) -> bool:
61 """Return whether this failure describes source data the user can fix."""
63 return bool(self.deficiency_counts)
65 def for_source_flowsheet(self, flowsheet) -> "FlowsheetStateCloneError":
66 """Return the same failure with product-safe source flowsheet context."""
68 details = tuple(
69 replace(
70 detail,
71 flowsheet_id=detail.flowsheet_id or flowsheet.pk,
72 flowsheet_name=detail.flowsheet_name or flowsheet.name,
73 )
74 for detail in self.details
75 )
76 return type(self)(
77 str(self),
78 deficiency_type=self.deficiency_type,
79 source_flowsheet_id=flowsheet.pk,
80 source_flowsheet_name=flowsheet.name,
81 deficiency_counts=self.deficiency_counts,
82 details=details,
83 )
86def merge_clone_validation_errors(
87 errors: list[FlowsheetStateCloneError],
88) -> FlowsheetStateCloneError:
89 """Combine repairable failures while retaining only bounded safe details."""
91 if not errors or any(not error.is_user_repairable for error in errors): 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 raise ValueError("Only repairable clone failures can be merged.")
94 counts: defaultdict[CloneDeficiencyType, int] = defaultdict(int)
95 for error in errors:
96 for deficiency_type, count in error.deficiency_counts.items():
97 counts[deficiency_type] += count
99 details_by_type: defaultdict[
100 CloneDeficiencyType, list[CloneDeficiencyDetail]
101 ] = defaultdict(list)
102 for deficiency_type in CloneDeficiencyType:
103 per_error_details = [
104 [
105 detail
106 for detail in error.details
107 if detail.deficiency_type == deficiency_type
108 ]
109 for error in errors
110 ]
111 detail_index = 0
112 while ( 112 ↛ 102line 112 didn't jump to line 102 because the condition on line 112 was always true
113 len(details_by_type[deficiency_type])
114 < MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY
115 ):
116 added = False
117 for error_details in per_error_details:
118 if detail_index < len(error_details):
119 details_by_type[deficiency_type].append(
120 error_details[detail_index]
121 )
122 added = True
123 if ( 123 ↛ 127line 123 didn't jump to line 127 because the condition on line 123 was never true
124 len(details_by_type[deficiency_type])
125 == MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY
126 ):
127 break
128 if not added:
129 break
130 detail_index += 1
132 details = tuple(
133 detail
134 for deficiency_type in CloneDeficiencyType
135 for detail in details_by_type[deficiency_type]
136 )
137 first = errors[0]
138 return FlowsheetStateCloneError(
139 f"Source data has {sum(counts.values())} repairable clone deficiencies.",
140 source_flowsheet_id=first.source_flowsheet_id,
141 source_flowsheet_name=first.source_flowsheet_name,
142 deficiency_counts=dict(counts),
143 details=details,
144 )
147def clone_validation_error_data(
148 error: FlowsheetStateCloneError,
149 *,
150 flowsheet=None,
151) -> dict:
152 """Build the safe API payload without exposing internal model or object IDs."""
154 if not error.is_user_repairable: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 raise ValueError("Only repairable clone failures have a validation payload.")
157 flowsheet_id = flowsheet.pk if flowsheet is not None else error.source_flowsheet_id
158 flowsheet_name = (
159 flowsheet.name if flowsheet is not None else error.source_flowsheet_name
160 )
161 logger.warning(
162 "Flowsheet clone validation failed for flowsheet %s: %s",
163 flowsheet_id,
164 error,
165 )
166 details_by_type: defaultdict[
167 CloneDeficiencyType, list[CloneDeficiencyDetail]
168 ] = defaultdict(list)
169 for detail in error.details:
170 details_by_type[detail.deficiency_type].append(detail)
172 categories = []
173 for deficiency_type in CloneDeficiencyType:
174 total_count = error.deficiency_counts.get(deficiency_type, 0)
175 if total_count == 0:
176 continue
177 details = details_by_type[deficiency_type][
178 :MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY
179 ]
180 categories.append(
181 {
182 "code": deficiency_type.value,
183 "total_count": total_count,
184 "items": [
185 {
186 "flowsheet_id": detail.flowsheet_id or flowsheet_id,
187 "flowsheet_name": detail.flowsheet_name or flowsheet_name,
188 "object_name": detail.object_name,
189 "object_type": detail.object_type,
190 "field_name": detail.field_name,
191 "reference_name": detail.reference_name,
192 }
193 for detail in details
194 ],
195 "truncated": (
196 len(details) == MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY
197 and total_count > len(details)
198 ),
199 }
200 )
202 return {
203 "code": CLONE_VALIDATION_ERROR_CODE,
204 "categories": categories,
205 "flowsheet_id": flowsheet_id,
206 "flowsheet_name": flowsheet_name,
207 }