Coverage for backend/django/core/auxiliary/services/flowsheet_states/validation.py: 83%
120 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Read-only validation of embedded references before clone writes begin."""
3from __future__ import annotations
5import re
6from collections import defaultdict
8from django.apps import apps
9from django.core.exceptions import ObjectDoesNotExist
10from django.db import models
11from django.db.models import Q
13from core.auxiliary.models.FlowsheetState import FlowsheetStateRole
14from core.validation import bypass_write_access_checks
16from .errors import (
17 MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY,
18 CloneDeficiencyDetail,
19 CloneDeficiencyType,
20 FlowsheetStateCloneError,
21)
22from .registry import CloneProfile, registry
25_FORMULA_REFERENCE = re.compile(r"\((?P<kind>unit|prop)(?P<id>\d+)[^)]*\)")
26_FORMULA_MENTION = re.compile(
27 r"@\[[^]]*\]\((?P<kind>unit|prop)(?P<id>\d+)[^)]*\)"
28)
29_REFERENCE_LABEL = re.compile(r"@\[(?P<label>[^]]*)\]\s*$")
30_FORMULA_FIELDS = frozenset({"formula", "property_formula"})
31_MAX_PRODUCT_LABEL_LENGTH = 120
34def formula_text_field_names(model) -> tuple[str, ...]:
35 """Return concrete text fields that contain remappable formula syntax."""
37 return tuple(
38 field.name
39 for field in model._meta.concrete_fields
40 if field.name in _FORMULA_FIELDS
41 and isinstance(field, (models.CharField, models.TextField))
42 )
45def _manager(model):
46 """Return a manager that includes deleted rows and every state role."""
48 return getattr(model, "all_states", model._base_manager)
51def _readable_model_name(model) -> str:
52 """Return a product-readable model label from Django metadata."""
54 return str(model._meta.verbose_name).replace("_", " ").title()
57def _safe_label(value: object | None, fallback: str) -> str:
58 """Bound stored display text before returning it in a validation response."""
60 label = str(value).strip() if value is not None else ""
61 return (label or fallback)[:_MAX_PRODUCT_LABEL_LENGTH]
64def _formula_owner(
65 row,
66 field_name: str,
67 property_value_model,
68) -> tuple[str, str, str]:
69 """Describe the place a user should open to repair a stored formula."""
71 if isinstance(row, property_value_model): 71 ↛ 92line 71 didn't jump to line 92 because the condition on line 71 was always true
72 try:
73 property_info = row.property
74 simulation_object = (
75 property_info.set.simulationObject if property_info else None
76 )
77 except ObjectDoesNotExist:
78 property_info = None
79 simulation_object = None
80 return (
81 _safe_label(
82 getattr(simulation_object, "componentName", None),
83 "Flowsheet item",
84 ),
85 "Flowsheet item",
86 _safe_label(
87 getattr(property_info, "displayName", None),
88 "Formula",
89 ),
90 )
92 for attribute in (
93 "displayName",
94 "name",
95 "label",
96 "title",
97 "metric_key",
98 "line_key",
99 ):
100 value = getattr(row, attribute, None)
101 if value:
102 return (
103 _safe_label(value, _readable_model_name(type(row))),
104 _readable_model_name(type(row)),
105 field_name.replace("_", " ").title(),
106 )
107 return (
108 _readable_model_name(type(row)),
109 _readable_model_name(type(row)),
110 field_name.replace("_", " ").title(),
111 )
114def _formula_rows(*, source_state, profile: CloneProfile):
115 """Yield clone-owned rows and their populated formula fields."""
117 property_value_model = apps.get_model("core_auxiliary", "PropertyValue")
118 for model in registry.copy_models(profile):
119 formula_fields = formula_text_field_names(model)
120 if not formula_fields:
121 continue
123 has_formula = Q()
124 for field_name in formula_fields:
125 has_formula |= Q(**{f"{field_name}__isnull": False}) & ~Q(
126 **{field_name: ""}
127 )
128 queryset = _manager(model).filter(
129 has_formula,
130 flowsheet_state=source_state,
131 )
132 if model is property_value_model:
133 queryset = queryset.select_related(
134 "property__set__simulationObject"
135 )
136 for row in queryset:
137 yield model, row, formula_fields
140def repair_legacy_formula_unit_references(
141 *,
142 source_state,
143 profile: CloneProfile,
144) -> int:
145 """Repair deterministic unit mentions left stale by the legacy clone path.
147 Legacy clones remapped each property-value mention but retained the source
148 simulation-object ID in the immediately preceding unit mention. A valid
149 property mention identifies its owning simulation object unambiguously, so
150 replacing only that paired unit ID cannot change formula evaluation.
152 Incomplete formulas, unpaired mentions, missing properties, and properties
153 without an owning simulation object are deliberately left for manual repair.
154 """
156 profile = CloneProfile(profile)
157 if profile == CloneProfile.MODULE_SUBTREE: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 return 0
159 if source_state.role != FlowsheetStateRole.WORKING: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 raise ValueError("Legacy formula repair is restricted to the current state.")
162 property_value_model = apps.get_model("core_auxiliary", "PropertyValue")
163 property_owners = {
164 property_value.pk: property_value.property.set.simulationObject_id
165 for property_value in _manager(property_value_model)
166 .filter(
167 flowsheet_state=source_state,
168 property__set__simulationObject__isnull=False,
169 property__set__simulationObject__flowsheet_state=source_state,
170 )
171 .select_related("property__set__simulationObject")
172 }
174 repaired_references = 0
175 for model, row, formula_fields in _formula_rows(
176 source_state=source_state,
177 profile=profile,
178 ):
179 if (
180 model is property_value_model
181 and row.property is not None
182 and row.property.formula_incomplete
183 ):
184 continue
186 updates = {}
187 for field_name in formula_fields:
188 text = getattr(row, field_name)
189 if not text: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 continue
192 mentions = list(_FORMULA_MENTION.finditer(text))
193 replacements = []
194 for unit_mention, property_mention in zip(mentions, mentions[1:]):
195 if ( 195 ↛ 200line 195 didn't jump to line 200 because the condition on line 195 was never true
196 unit_mention.group("kind") != "unit"
197 or property_mention.group("kind") != "prop"
198 or text[unit_mention.end() : property_mention.start()].strip()
199 ):
200 continue
201 owner_id = property_owners.get(int(property_mention.group("id")))
202 if owner_id is None or int(unit_mention.group("id")) == owner_id: 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 continue
204 replacements.append(
205 (
206 unit_mention.start("id"),
207 unit_mention.end("id"),
208 str(owner_id),
209 )
210 )
212 if not replacements: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 continue
214 repaired_text = text
215 for start, end, replacement in reversed(replacements):
216 repaired_text = (
217 repaired_text[:start] + replacement + repaired_text[end:]
218 )
219 updates[field_name] = repaired_text
220 repaired_references += len(replacements)
222 if updates: 222 ↛ 175line 222 didn't jump to line 175 because the condition on line 222 was always true
223 with bypass_write_access_checks():
224 _manager(model).filter(
225 pk=row.pk,
226 flowsheet_state=source_state,
227 ).update(**updates)
229 return repaired_references
232def validate_state_for_clone(*, source_state, profile: CloneProfile) -> None:
233 """Raise one bounded, categorized error for invalid formula references."""
235 profile = CloneProfile(profile)
236 if profile == CloneProfile.MODULE_SUBTREE:
237 return
239 simulation_object_model = apps.get_model(
240 "flowsheetInternals_unitops", "SimulationObject"
241 )
242 property_value_model = apps.get_model("core_auxiliary", "PropertyValue")
243 valid_references = {
244 "unit": set(
245 _manager(simulation_object_model)
246 .filter(flowsheet_state=source_state)
247 .values_list("pk", flat=True)
248 ),
249 "prop": set(
250 _manager(property_value_model)
251 .filter(flowsheet_state=source_state)
252 .values_list("pk", flat=True)
253 ),
254 }
256 counts: defaultdict[CloneDeficiencyType, int] = defaultdict(int)
257 details: list[CloneDeficiencyDetail] = []
258 seen: set[tuple] = set()
259 for model, row, formula_fields in _formula_rows(
260 source_state=source_state,
261 profile=profile,
262 ):
263 for field_name in formula_fields:
264 text = getattr(row, field_name)
265 if not text: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 continue
267 for match in _FORMULA_REFERENCE.finditer(text):
268 if (
269 int(match.group("id"))
270 in valid_references[match.group("kind")]
271 ):
272 continue
273 label_match = _REFERENCE_LABEL.search(text[: match.start()])
274 reference_name = (
275 _safe_label(label_match.group("label"), "Unavailable item")
276 if label_match
277 else None
278 )
279 object_name, object_type, product_field_name = _formula_owner(
280 row, field_name, property_value_model
281 )
282 key = (
283 model._meta.label_lower,
284 row.pk,
285 field_name,
286 match.group("kind"),
287 match.group("id"),
288 )
289 if key in seen: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 continue
291 seen.add(key)
292 counts[CloneDeficiencyType.FORMULA_REFERENCES] += 1
293 if len(details) < MAX_CLONE_DEFICIENCY_DETAILS_PER_CATEGORY:
294 details.append(
295 CloneDeficiencyDetail(
296 deficiency_type=CloneDeficiencyType.FORMULA_REFERENCES,
297 flowsheet_id=source_state.flowsheet_id,
298 flowsheet_name=source_state.flowsheet.name,
299 object_name=object_name,
300 object_type=object_type,
301 field_name=product_field_name,
302 reference_name=reference_name,
303 )
304 )
306 if counts:
307 raise FlowsheetStateCloneError(
308 f"Source state has {sum(counts.values())} invalid formula references.",
309 deficiency_counts=dict(counts),
310 source_flowsheet_id=source_state.flowsheet_id,
311 source_flowsheet_name=source_state.flowsheet.name,
312 details=tuple(details),
313 )