Coverage for backend/django/core/auxiliary/models/Project.py: 92%
92 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.core.exceptions import ValidationError
2from django.db import models, router, transaction
3from typing import TYPE_CHECKING
5from authentication.user.models import User
6from authentication.user.AccessTable import AccessTable
8if TYPE_CHECKING:
9 from core.auxiliary.models.Flowsheet import Flowsheet
12PROJECT_FOLDER_STRUCTURE_FIELDS = {"owner", "owner_id", "folder", "folder_id"}
15class ProjectQuerySet(models.QuerySet):
16 """Reject raw relational rewrites that bypass Project validation.
18 Folder clearing is intrinsically safe, but every non-null folder assignment
19 and every owner rewrite must use model instances, where ``Project.save``
20 enforces the ownership invariant. Expressions are rejected as well because
21 a ``Value`` or ``Case`` can otherwise smuggle in a foreign folder ID.
22 """
24 _STRUCTURAL_FIELDS = PROJECT_FOLDER_STRUCTURE_FIELDS
26 def bulk_create(
27 self,
28 objs,
29 batch_size=None,
30 ignore_conflicts=False,
31 update_conflicts=False,
32 update_fields=None,
33 unique_fields=None,
34 ):
35 """Validate folder ownership before bypassing model ``save`` hooks."""
37 conflict_update_fields = {
38 getattr(field, "name", field) for field in (update_fields or ())
39 }
40 if update_conflicts and PROJECT_FOLDER_STRUCTURE_FIELDS.intersection(
41 conflict_update_fields
42 ):
43 raise TypeError(
44 "Project bulk upserts cannot rewrite owner or folder fields."
45 )
47 projects = list(objs)
48 folder_ids = {
49 project.folder_id
50 for project in projects
51 if project.folder_id is not None
52 }
53 if folder_ids: 53 ↛ 74line 53 didn't jump to line 74 because the condition on line 53 was always true
54 from core.auxiliary.models.ProjectFolder import ProjectFolder
56 folder_owner_ids = dict(
57 ProjectFolder.objects.using(self.db)
58 .filter(pk__in=folder_ids)
59 .values_list("pk", "owner_id")
60 )
61 for project in projects: 61 ↛ 74line 61 didn't jump to line 74 because the loop on line 61 didn't complete
62 if ( 62 ↛ 61line 62 didn't jump to line 61 because the condition on line 62 was always true
63 project.folder_id is not None
64 and folder_owner_ids.get(project.folder_id) != project.owner_id
65 ):
66 raise ValidationError(
67 {
68 "folder": (
69 "A project and its folder must have the same owner."
70 )
71 }
72 )
74 return super().bulk_create(
75 projects,
76 batch_size=batch_size,
77 ignore_conflicts=ignore_conflicts,
78 update_conflicts=update_conflicts,
79 update_fields=update_fields,
80 unique_fields=unique_fields,
81 )
83 def update(self, **kwargs):
84 unsafe_fields = {
85 field
86 for field, value in kwargs.items()
87 if field in self._STRUCTURAL_FIELDS
88 and value is not None
89 }
90 if "owner" in kwargs or "owner_id" in kwargs: 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 unsafe_fields.update({"owner", "owner_id"} & kwargs.keys())
92 if unsafe_fields:
93 fields = ", ".join(sorted(unsafe_fields))
94 raise TypeError(
95 f"Project queryset updates cannot assign {fields}; "
96 "save validated Project instances instead."
97 )
98 return super().update(**kwargs)
101class Project(models.Model):
102 name = models.CharField(max_length=64, default="Project")
103 owner = models.ForeignKey(User, on_delete=models.CASCADE, related_name="projects", null=True)
104 active_flowsheet = models.ForeignKey("Flowsheet", default=None, null=True, on_delete=models.SET_NULL, related_name="projectsActiveFlowsheet")
105 folder = models.ForeignKey(
106 "ProjectFolder",
107 null=True,
108 blank=True,
109 on_delete=models.SET_NULL,
110 related_name="projects",
111 )
112 binned_at = models.DateTimeField(null=True)
113 is_starred = models.BooleanField(default=False)
114 is_binned = models.BooleanField(default=False)
115 created_at = models.DateTimeField(auto_now_add=True)
116 updated_at = models.DateTimeField(auto_now=True)
118 # Runtime-accessed relations
119 flowsheets: models.Manager["Flowsheet"]
121 objects = ProjectQuerySet.as_manager()
123 @classmethod
124 @transaction.atomic
125 def create_empty(cls, **kwargs):
126 """
127 Create a project shell and owner access row without creating a default flowsheet.
129 Copy/template flows use this because they insert copied flowsheets before
130 deciding which one should become active.
131 """
132 project_number = Project.objects.count() + 1
133 owner = kwargs.get('owner')
135 field_values = {
136 'name': kwargs.get('name', f'Project-{project_number}'),
137 'owner': owner,
138 'active_flowsheet': kwargs.get('active_flowsheet', None),
139 'folder': kwargs.get('folder'),
140 'is_starred': kwargs.get('is_starred', False),
141 'is_binned': kwargs.get('is_binned', False),
142 'binned_at': kwargs.get('binned_at', None),
143 }
145 instance = Project.objects.create(**field_values)
147 AccessTable.objects.get_or_create(
148 user=instance.owner,
149 project=instance,
150 defaults={"read_only": False},
151 )
153 return instance
155 @classmethod
156 @transaction.atomic
157 def create(cls, **kwargs):
158 from core.auxiliary.models.Flowsheet import Flowsheet
160 instance = cls.create_empty(**kwargs)
162 # Create a default flowsheet for the project
163 active_flowsheet = Flowsheet.create(owner=instance.owner, project=instance)
164 instance.active_flowsheet = active_flowsheet
165 instance.save(update_fields=["active_flowsheet"])
167 return instance
169 def _validate_structure(self, *, using: str) -> None:
170 """Freeze ownership and enforce same-owner folder membership."""
172 if self.pk is not None:
173 persisted = (
174 type(self).objects.using(using)
175 .filter(pk=self.pk)
176 .values("owner_id")
177 .first()
178 )
179 if (
180 persisted is not None
181 and persisted["owner_id"] != self.owner_id
182 ):
183 raise ValidationError(
184 {"owner": "Project owner cannot be changed after creation."}
185 )
187 if self.folder_id is None:
188 return
190 from core.auxiliary.models.ProjectFolder import ProjectFolder
192 folder_owner_id = (
193 ProjectFolder.objects.using(using)
194 .filter(pk=self.folder_id)
195 .values_list("owner_id", flat=True)
196 .first()
197 )
198 if folder_owner_id != self.owner_id:
199 raise ValidationError(
200 {"folder": "A project and its folder must have the same owner."}
201 )
203 def save(self, *args, **kwargs):
204 """Persist a project only after enforcing relational ownership."""
206 update_fields = kwargs.get("update_fields")
207 validates_structure = (
208 self._state.adding
209 or update_fields is None
210 or bool(PROJECT_FOLDER_STRUCTURE_FIELDS.intersection(update_fields))
211 )
212 if validates_structure:
213 using = (
214 kwargs.get("using")
215 or self._state.db
216 or router.db_for_write(type(self), instance=self)
217 )
218 self._validate_structure(using=using)
219 return super().save(*args, **kwargs)
221 def clean(self):
222 if (self.active_flowsheet is not None and self.active_flowsheet.project_id != self.pk): 222 ↛ 223line 222 didn't jump to line 223 because the condition on line 222 was never true
223 raise ValidationError(
224 {"active_flowsheet": "Default flowsheet must belong to the same project."}
225 )
226 using = self._state.db or router.db_for_write(type(self), instance=self)
227 self._validate_structure(using=using)
229 def get_configurations(self):
230 """
231 Get all flowsheet configurations for the project.
232 """
233 if hasattr(self, "prefetched_configurations"):
234 return self.prefetched_configurations
235 return (
236 self.flowsheets
237 .order_by("created_at", "id")
238 )
240 def set_active_flowsheet(self, flowsheet):
241 """Set a fresh regular flowsheet active through the transition service."""
243 from core.auxiliary.services.flowsheet_template_transitions import (
244 activate_regular_flowsheet,
245 )
247 if flowsheet is not None and flowsheet.project_id != self.pk: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 raise ValidationError(
249 {"active_flowsheet": "Active flowsheet must belong to the same project."}
250 )
252 active_flowsheet = activate_regular_flowsheet(
253 project=self,
254 candidate=flowsheet,
255 )
256 self.active_flowsheet = active_flowsheet