Coverage for backend/django/core/state_validation.py: 94%
149 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.core.exceptions import FieldDoesNotExist, ValidationError
2from django.db.models.expressions import BaseExpression
3from django.db.models import QuerySet
4from django.db.models.signals import m2m_changed, pre_delete, pre_save
5from django.dispatch import receiver
8def model_has_flowsheet_state(model) -> bool:
9 """Return whether a concrete model declares direct state ownership."""
11 try:
12 model._meta.get_field("flowsheet_state")
13 except FieldDoesNotExist:
14 return False
15 return True
18def validate_state_owned_relations(instance) -> None:
19 """Require direct state-owned foreign keys to remain in one aggregate.
21 The operational audit allowlist also carries stable ``flowsheet`` identity.
22 For those models, require nullable state provenance to belong to that same
23 identity whenever it is present.
24 """
26 if not model_has_flowsheet_state(type(instance)):
27 return
28 state_id = instance.flowsheet_state_id
29 if state_id is None:
30 return
32 errors = {}
33 try:
34 flowsheet_field = instance._meta.get_field("flowsheet")
35 except FieldDoesNotExist:
36 flowsheet_field = None
37 if flowsheet_field is not None and flowsheet_field.is_relation:
38 state_flowsheet_id = (
39 instance._meta.get_field("flowsheet_state").remote_field.model._base_manager
40 .filter(pk=state_id)
41 .values_list("flowsheet_id", flat=True)
42 .first()
43 )
44 if state_flowsheet_id != instance.flowsheet_id:
45 errors["flowsheet_state"] = (
46 "State provenance must belong to the stable flowsheet."
47 )
49 for field in instance._meta.concrete_fields:
50 if (
51 not field.is_relation
52 or field.name == "flowsheet_state"
53 or field.remote_field is None
54 or not model_has_flowsheet_state(field.remote_field.model)
55 ):
56 continue
58 related_id = getattr(instance, field.attname)
59 if related_id is None:
60 continue
61 if field.is_cached(instance):
62 related_state_id = getattr(instance, field.name).flowsheet_state_id
63 else:
64 related_state_id = (
65 field.remote_field.model._base_manager
66 .filter(pk=related_id)
67 .values_list("flowsheet_state_id", flat=True)
68 .first()
69 )
70 if related_state_id != state_id:
71 errors[field.name] = "Referenced row must belong to the same flowsheet state."
73 if errors:
74 raise ValidationError(errors)
77def validate_state_owned_relations_bulk(instances) -> None:
78 """Validate a batch with a bounded number of relation queries.
80 Bulk clone/import paths routinely contain thousands of rows. Grouping
81 referenced primary keys by relation keeps validation proportional to the
82 model shape rather than to the number of inserted rows.
83 """
85 instances = list(instances)
86 if not instances:
87 return
88 model = type(instances[0])
89 if not model_has_flowsheet_state(model): 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 return
91 if any(type(instance) is not model for instance in instances): 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 raise TypeError("State-owned bulk validation requires one model type.")
94 errors = {}
95 state_ids = {instance.flowsheet_state_id for instance in instances}
96 non_null_state_ids = {state_id for state_id in state_ids if state_id is not None}
98 try:
99 flowsheet_field = model._meta.get_field("flowsheet")
100 except FieldDoesNotExist:
101 flowsheet_field = None
102 if flowsheet_field is not None and flowsheet_field.is_relation and non_null_state_ids:
103 state_model = model._meta.get_field("flowsheet_state").remote_field.model
104 state_owners = dict(
105 state_model._base_manager.filter(pk__in=non_null_state_ids).values_list(
106 "pk", "flowsheet_id"
107 )
108 )
109 if any(
110 instance.flowsheet_state_id is not None
111 and state_owners.get(instance.flowsheet_state_id) != instance.flowsheet_id
112 for instance in instances
113 ):
114 errors["flowsheet_state"] = (
115 "State provenance must belong to the stable flowsheet."
116 )
118 for field in model._meta.concrete_fields:
119 if (
120 not field.is_relation
121 or field.name == "flowsheet_state"
122 or field.remote_field is None
123 or not model_has_flowsheet_state(field.remote_field.model)
124 ):
125 continue
126 relation_ids = {
127 getattr(instance, field.attname)
128 for instance in instances
129 if getattr(instance, field.attname) is not None
130 }
131 if not relation_ids:
132 continue
133 related_states = dict(
134 field.remote_field.model._base_manager.filter(pk__in=relation_ids).values_list(
135 "pk", "flowsheet_state_id"
136 )
137 )
138 if any(
139 related_states.get(getattr(instance, field.attname))
140 != instance.flowsheet_state_id
141 for instance in instances
142 if getattr(instance, field.attname) is not None
143 ):
144 errors[field.name] = "Referenced row must belong to the same flowsheet state."
146 if errors:
147 raise ValidationError(errors)
150def validate_state_mutation(instance) -> None:
151 """Reject ordinary saves/deletes outside the captured working state."""
153 if not model_has_flowsheet_state(type(instance)):
154 return
155 from core.managers import StaleFlowsheetState
156 from core.validation import get_current_flowsheet, write_access_checks_are_bypassed
158 if write_access_checks_are_bypassed():
159 return
160 state_id = instance.flowsheet_state_id
161 if state_id is None:
162 return
163 ctx = get_current_flowsheet() or {}
164 captured_state_id = ctx.get("flowsheet_state")
165 if captured_state_id is not None and captured_state_id != state_id: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 raise StaleFlowsheetState()
168 state_model = instance._meta.get_field("flowsheet_state").remote_field.model
169 is_current = state_model._base_manager.filter(
170 pk=state_id,
171 role="working",
172 flowsheet__current_state_id=state_id,
173 ).exists()
174 if not is_current:
175 raise StaleFlowsheetState()
178def validate_state_owned_relation_update(queryset, update_values: dict) -> None:
179 """Validate relation assignments performed through ``QuerySet.update``.
181 Django does not emit ``pre_save`` for bulk updates, so relation assignments
182 need an explicit aggregate-boundary check. Expressions are rejected for
183 state-owned relations because their target state cannot be proven before
184 the update executes.
185 """
187 if not model_has_flowsheet_state(queryset.model): 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 return
190 fields_by_update_name = {
191 update_name: field
192 for field in queryset.model._meta.concrete_fields
193 for update_name in {field.name, field.attname}
194 }
195 relation_updates = []
196 for update_name, value in update_values.items():
197 field = fields_by_update_name.get(update_name)
198 if (
199 field is not None
200 and field.is_relation
201 and field.remote_field is not None
202 and model_has_flowsheet_state(field.remote_field.model)
203 ):
204 relation_updates.append((update_name, value, field))
205 if not relation_updates:
206 return
208 queryset_state_ids = set(
209 queryset.values_list("flowsheet_state_id", flat=True).distinct()
210 )
211 if not queryset_state_ids:
212 return
213 errors = {}
215 for update_name, value, field in relation_updates:
216 if value is None:
217 continue
218 if isinstance(value, BaseExpression): 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true
219 errors[field.name] = (
220 "Expression updates are not supported for state-owned relations."
221 )
222 continue
224 related_state_id = getattr(value, "flowsheet_state_id", None)
225 related_id = getattr(value, "pk", value)
226 if related_state_id is None: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 related_state_id = (
228 field.remote_field.model._base_manager
229 .filter(pk=related_id)
230 .values_list("flowsheet_state_id", flat=True)
231 .first()
232 )
234 if related_state_id is None or queryset_state_ids != {related_state_id}:
235 errors[field.name] = (
236 "Referenced row must belong to the same flowsheet state."
237 )
239 if errors:
240 raise ValidationError(errors)
243@receiver(pre_save, dispatch_uid="validate_state_owned_relations")
244def validate_state_owned_relations_before_save(sender, instance, raw=False, **kwargs):
245 """Apply aggregate validation to ordinary model saves across all apps."""
247 if raw: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 return
249 validate_state_mutation(instance)
250 validate_state_owned_relations(instance)
253@receiver(pre_delete, dispatch_uid="validate_state_owned_delete")
254def validate_state_owned_delete(sender, instance, origin=None, **kwargs):
255 """Guard direct instance/queryset deletes while allowing validated cascades."""
257 if not model_has_flowsheet_state(sender):
258 return
259 if getattr(origin, "_state_mutation_validated", False):
260 return
261 is_direct_origin = origin is instance or (
262 isinstance(origin, QuerySet) and origin.model is sender
263 )
264 if is_direct_origin:
265 validate_state_mutation(instance)
268@receiver(m2m_changed, dispatch_uid="validate_state_owned_many_to_many")
269def validate_state_owned_many_to_many(
270 sender,
271 instance,
272 action,
273 reverse,
274 model,
275 pk_set,
276 **kwargs,
277):
278 """Prevent state-owned many-to-many links from crossing aggregates."""
280 if action in {"pre_remove", "pre_clear"}:
281 if model_has_flowsheet_state(type(instance)):
282 validate_state_mutation(instance)
283 return
284 if action != "pre_add" or not pk_set:
285 return
286 instance_is_state_owned = model_has_flowsheet_state(type(instance))
287 target_is_state_owned = model_has_flowsheet_state(model)
288 if not instance_is_state_owned or not target_is_state_owned:
289 return
291 validate_state_mutation(instance)
292 target_state_ids = set(
293 model._base_manager.filter(pk__in=pk_set).values_list(
294 "flowsheet_state_id", flat=True
295 )
296 )
297 if target_state_ids != {instance.flowsheet_state_id}:
298 raise ValidationError(
299 "Many-to-many rows must belong to the same flowsheet state."
300 )