Coverage for backend/django/core/auxiliary/models/FlowsheetState.py: 92%

1from django.conf import settings 

2from django.core.exceptions import ValidationError 

3from django.db import models 

4from django.db.models import Q 

5from core.auxiliary.on_delete import protect_root_grouping_unless_state_deleting 

6 

7 

8class FlowsheetStateRole(models.TextChoices): 

9 """Lifecycle roles for a relational flowsheet aggregate.""" 

10 

11 WORKING = "working", "Working" 

12 REVISION = "revision", "Revision" 

13 STAGING = "staging", "Staging" 

14 RETIRING = "retiring", "Retiring" 

15 

16 

17class FlowsheetRevisionKind(models.TextChoices): 

18 """User-visible reasons for retaining an immutable flowsheet state.""" 

19 

20 MANUAL = "manual", "Manual" 

21 AUTO_SOLVE = "auto_solve", "After solve" 

22 BEFORE_RESTORE = "before_restore", "Before restore" 

23 

24 

25class FlowsheetState(models.Model): 

26 """Structured content owned by a stable :class:`Flowsheet` identity. 

27 

28 Ordinary application code addresses the stable flowsheet and is scoped to 

29 its sole working state. Revision and restore services use the explicit 

30 state identity to clone immutable historical aggregates. 

31 """ 

32 

33 flowsheet = models.ForeignKey( 

34 "Flowsheet", 

35 on_delete=models.CASCADE, 

36 related_name="states", 

37 ) 

38 role = models.CharField( 

39 max_length=16, 

40 choices=FlowsheetStateRole.choices, 

41 default=FlowsheetStateRole.WORKING, 

42 ) 

43 revision_number = models.PositiveBigIntegerField(null=True, blank=True) 

44 revision_kind = models.CharField( 

45 max_length=24, 

46 choices=FlowsheetRevisionKind.choices, 

47 null=True, 

48 blank=True, 

49 ) 

50 label = models.CharField(max_length=64, null=True, blank=True) 

51 created_by = models.ForeignKey( 

52 settings.AUTH_USER_MODEL, 

53 on_delete=models.SET_NULL, 

54 related_name="created_flowsheet_states", 

55 null=True, 

56 blank=True, 

57 ) 

58 restored_from_revision_number = models.PositiveBigIntegerField( 

59 null=True, 

60 blank=True, 

61 ) 

62 root_grouping = models.ForeignKey( 

63 "flowsheetInternals_graphicData.Grouping", 

64 on_delete=protect_root_grouping_unless_state_deleting, 

65 related_name="root_for_states", 

66 null=True, 

67 blank=True, 

68 ) 

69 build_version = models.CharField(max_length=32, null=True, blank=True) 

70 build_date = models.CharField(max_length=32, null=True, blank=True) 

71 source_saved_at = models.DateTimeField(null=True, blank=True) 

72 created_at = models.DateTimeField(auto_now_add=True) 

73 

74 class Meta: 

75 constraints = [ 

76 models.UniqueConstraint( 

77 fields=["flowsheet", "revision_number"], 

78 condition=Q(revision_number__isnull=False), 

79 name="unique_flowsheet_revision_number", 

80 ), 

81 models.UniqueConstraint( 

82 fields=["flowsheet"], 

83 condition=Q(role=FlowsheetStateRole.WORKING), 

84 name="unique_working_state_per_flowsheet", 

85 ), 

86 models.CheckConstraint( 

87 check=( 

88 Q( 

89 role=FlowsheetStateRole.REVISION, 

90 revision_number__isnull=False, 

91 revision_kind__isnull=False, 

92 ) 

93 | Q( 

94 role__in=( 

95 FlowsheetStateRole.WORKING, 

96 FlowsheetStateRole.STAGING, 

97 FlowsheetStateRole.RETIRING, 

98 ), 

99 revision_number__isnull=True, 

100 revision_kind__isnull=True, 

101 label__isnull=True, 

102 ) 

103 ), 

104 name="flowsheet_state_role_metadata", 

105 ), 

106 models.CheckConstraint( 

107 check=( 

108 Q( 

109 role=FlowsheetStateRole.REVISION, 

110 revision_kind=FlowsheetRevisionKind.MANUAL, 

111 label__isnull=False, 

112 ) 

113 & ~Q(label="") 

114 | Q( 

115 role=FlowsheetStateRole.REVISION, 

116 revision_kind__in=( 

117 FlowsheetRevisionKind.AUTO_SOLVE, 

118 FlowsheetRevisionKind.BEFORE_RESTORE, 

119 ), 

120 ) 

121 & ~Q(label="") 

122 | Q(role__in=( 

123 FlowsheetStateRole.WORKING, 

124 FlowsheetStateRole.STAGING, 

125 FlowsheetStateRole.RETIRING, 

126 )) 

127 ), 

128 name="flowsheet_revision_label_contract", 

129 ), 

130 ] 

131 indexes = [ 

132 models.Index(fields=["flowsheet", "role"]), 

133 models.Index(fields=["flowsheet", "-revision_number"]), 

134 ] 

135 

136 def clean(self): 

137 """Validate state metadata and cross-row ownership invariants.""" 

138 

139 super().clean() 

140 if self.role == FlowsheetStateRole.REVISION: 

141 if self.revision_kind == FlowsheetRevisionKind.MANUAL: 

142 if not self.label or not self.label.strip(): 

143 raise ValidationError({"label": "Manual revisions require a name."}) 

144 elif self.label is not None and not self.label.strip(): 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 raise ValidationError({"label": "Revision names cannot be blank."}) 

146 

147 if self.root_grouping_id is not None: 

148 grouping_state_id = getattr(self.root_grouping, "flowsheet_state_id", None) 

149 if grouping_state_id != self.pk: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true

150 raise ValidationError( 

151 {"root_grouping": "The root grouping must belong to this flowsheet state."} 

152 ) 

153 

154 def save(self, *args, **kwargs): 

155 """Normalize and validate lifecycle metadata before persistence.""" 

156 

157 if self.revision_kind == FlowsheetRevisionKind.MANUAL and self.label is not None: 

158 # Field max-length validation must see the product value after 

159 # whitespace normalization, not the raw request-sized string. 

160 self.label = self.label.strip() 

161 self.full_clean() 

162 return super().save(*args, **kwargs) 

163 

164 @property 

165 def is_working(self) -> bool: 

166 """Return whether ordinary request-scoped code may access this state.""" 

167 

168 return self.role == FlowsheetStateRole.WORKING