Coverage for postrfp/model/fsm.py: 95%

1from datetime import datetime 

2from typing import Optional, List, Set, TYPE_CHECKING 

3from sqlalchemy import VARCHAR, ForeignKey, UniqueConstraint, Index, func, String 

4from sqlalchemy.orm import ( 

5 Mapped, 

6 mapped_column, 

7 relationship, 

8 validates, 

9) # Import validates 

10from sqlalchemy.ext.associationproxy import association_proxy, AssociationProxy 

11 

12 

13from postrfp.model.helpers import random_string 

14from postrfp.authorisation import perms 

15from postrfp.model.meta import Base 

16from postrfp.shared.types import PermString 

17from postrfp.shared.constants import CEL_EXPRESSION_MAX_LENGTH 

18 

19if TYPE_CHECKING: 

20 from postrfp.model import Organisation 

21 

22 

23VALID_ACTIONS = perms.ALL_PERMISSIONS 

24 

25CODE_LENGTH = 12 # String length for Status Code 

26 

27 

28class Workflow(Base): 

29 """Stores metadata about a Finite State Machine definition.""" 

30 

31 __tablename__ = "workflows" 

32 

33 # Indices and constraints 

34 __table_args__ = ( 

35 UniqueConstraint("title", "entity_type", "organisation_id", "version"), 

36 ) 

37 

38 id: Mapped[int] = mapped_column(primary_key=True) 

39 title: Mapped[str] = mapped_column(String(128), nullable=False) 

40 description: Mapped[Optional[str]] = mapped_column( 

41 String(1024), default=None, nullable=True 

42 ) 

43 entity_type: Mapped[str] = mapped_column(String(64), nullable=False) 

44 organisation_id: Mapped[str] = mapped_column( 

45 ForeignKey( 

46 "organisations.id", 

47 ondelete="CASCADE", 

48 ), 

49 nullable=False, 

50 ) 

51 version: Mapped[int] = mapped_column(default=1, nullable=False) 

52 is_active: Mapped[bool] = mapped_column(default=True, nullable=False) 

53 initial_status_code: Mapped[str] = mapped_column( 

54 String(CODE_LENGTH), nullable=False 

55 ) 

56 

57 date_created: Mapped[datetime] = mapped_column( 

58 server_default=func.utc_timestamp(), nullable=False 

59 ) 

60 date_modified: Mapped[datetime] = mapped_column( 

61 server_default=func.utc_timestamp(), 

62 server_onupdate=func.utc_timestamp(), 

63 nullable=False, 

64 ) 

65 

66 # Relationships 

67 organisation: Mapped["Organisation"] = relationship( 

68 "Organisation", foreign_keys=[organisation_id] 

69 ) 

70 statuses: Mapped[List["Status"]] = relationship( 

71 back_populates="workflow", cascade="all, delete-orphan" 

72 ) 

73 # Add overlaps parameter to status_query 

74 status_query = relationship("Status", lazy="dynamic", overlaps="statuses") 

75 

76 transitions: Mapped[List["Transition"]] = relationship( 

77 back_populates="workflow", cascade="all, delete-orphan" 

78 ) 

79 

80 transitions_query = relationship( 

81 "Transition", lazy="dynamic", overlaps="transitions" 

82 ) 

83 

84 def status_by_name(self, status_name: str) -> "Status": 

85 return self.status_query.filter_by(name=status_name).scalar() 

86 

87 def get_initial_status(self) -> "Status": 

88 initial_status = self.status_query.filter_by( 

89 code=self.initial_status_code 

90 ).scalar() 

91 if initial_status is None: 

92 raise ValueError( 

93 f"Initial status with code '{self.initial_status_code}' not found in workflow '{self.title}'" 

94 ) 

95 return initial_status 

96 

97 def __repr__(self): 

98 return f"<Workflow #{self.id} '{self.title}'>" 

99 

100 def validate(self): 

101 """ 

102 A basic check of the structure of this workflow 

103 """ 

104 

105 if len(self.statuses) != len({s.name for s in self.statuses}): 

106 raise ValueError("Duplicate status name found") 

107 

108 status_codes = {s.code for s in self.statuses} 

109 if len(self.statuses) != len(status_codes): 

110 raise ValueError("Duplicate status code found") 

111 

112 if len(self.transitions) != len({t.name for t in self.transitions}): 

113 raise ValueError("Duplicate transition name found") 

114 

115 # Find orphans 

116 involved_status_codes = set() 

117 for trans in self.transitions: 

118 # Use status codes directly from related objects 

119 involved_status_codes.add(trans.source_status.code) 

120 involved_status_codes.add(trans.target_status.code) 

121 

122 all_status_codes = {s.code for s in self.statuses} 

123 uninvolved_status_codes = all_status_codes - involved_status_codes 

124 

125 if len(uninvolved_status_codes) > 0: 

126 raise ValueError( 

127 f"Statuses not involved in any transition, code(s): {uninvolved_status_codes}" 

128 ) 

129 

130 

131class Status(Base): 

132 """Represents a status value for an entity. A "state" in a finite state machine.""" 

133 

134 __tablename__ = "statuses" 

135 

136 __table_args__ = ( 

137 UniqueConstraint("workflow_id", "code"), 

138 UniqueConstraint("workflow_id", "name"), 

139 UniqueConstraint( 

140 "id", "workflow_id" 

141 ), # Required for FSMEntity foreign key constraint 

142 ) 

143 

144 id: Mapped[int] = mapped_column(primary_key=True) 

145 workflow_id: Mapped[int] = mapped_column( 

146 ForeignKey( 

147 "workflows.id", 

148 ondelete="CASCADE", 

149 ), 

150 nullable=False, 

151 ) 

152 name: Mapped[str] = mapped_column(String(128), nullable=False) 

153 code: Mapped[str] = mapped_column( 

154 String(CODE_LENGTH), nullable=False, default=random_string 

155 ) 

156 description: Mapped[Optional[str]] = mapped_column(String(512)) 

157 

158 colour: Mapped[Optional[str]] = mapped_column( 

159 String(length=7), 

160 nullable=True, 

161 comment="Hex color code (#RRGGBB)", 

162 ) 

163 

164 # Relationships 

165 # Add overlaps parameter to workflow relationship 

166 workflow: Mapped[Workflow] = relationship( 

167 back_populates="statuses", overlaps="status_query" 

168 ) 

169 

170 actions: Mapped[List["StatusAction"]] = relationship( 

171 # Do not add lazy="joined" or the useful Set-like behaviour 

172 # is compromised by needed to call "unique()" on all queries 

173 back_populates="status", 

174 cascade="all, delete-orphan", 

175 collection_class=set, 

176 ) 

177 outgoing_transitions: Mapped[List["Transition"]] = relationship( 

178 foreign_keys="Transition.source_status_id", 

179 back_populates="source_status", 

180 ) 

181 incoming_transitions: Mapped[List["Transition"]] = relationship( 

182 foreign_keys="Transition.target_status_id", 

183 back_populates="target_status", 

184 ) 

185 

186 status_actions: AssociationProxy[Set[PermString]] = association_proxy( 

187 "actions", 

188 "action", 

189 creator=lambda action_name: StatusAction(action=action_name), 

190 ) 

191 

192 def has_action(self, action_name: PermString) -> bool: 

193 return action_name in self.status_actions 

194 

195 def __repr__(self): 

196 return f"<Status #{self.id} {self.name}>" 

197 

198 

199class Transition(Base): 

200 """Represents a transition between statuses.""" 

201 

202 __tablename__ = "transitions" 

203 

204 __table_args__ = ( 

205 UniqueConstraint("workflow_id", "source_status_id", "name"), 

206 Index("ix_transition_definition", "workflow_id"), 

207 Index("ix_transition_source", "source_status_id"), 

208 Index("ix_transition_target", "target_status_id"), 

209 ) 

210 

211 id: Mapped[int] = mapped_column(primary_key=True) 

212 workflow_id: Mapped[int] = mapped_column( 

213 ForeignKey( 

214 "workflows.id", 

215 ondelete="CASCADE", 

216 ), 

217 nullable=False, 

218 ) 

219 guard_policy: Mapped[Optional[str]] = mapped_column( 

220 VARCHAR(length=CEL_EXPRESSION_MAX_LENGTH), nullable=True 

221 ) 

222 

223 name: Mapped[str] = mapped_column(String(64), nullable=False) 

224 source_status_id: Mapped[int] = mapped_column( 

225 ForeignKey( 

226 "statuses.id", 

227 ondelete="CASCADE", 

228 ), 

229 nullable=False, 

230 ) 

231 

232 target_status_id: Mapped[int] = mapped_column( 

233 ForeignKey( 

234 "statuses.id", 

235 ondelete="CASCADE", 

236 ), 

237 nullable=False, 

238 ) 

239 

240 # Relationships 

241 workflow: Mapped[Workflow] = relationship( 

242 back_populates="transitions", overlaps="transitions_query" 

243 ) 

244 source_status: Mapped[Status] = relationship( 

245 foreign_keys=[source_status_id], back_populates="outgoing_transitions" 

246 ) 

247 target_status: Mapped[Status] = relationship( 

248 foreign_keys=[target_status_id], back_populates="incoming_transitions" 

249 ) 

250 

251 def has_functions(self) -> bool: 

252 """Does this Transition define any Javascript functions""" 

253 return self.guard_policy is not None 

254 

255 def __repr__(self): 

256 return f"<Transition #{self.id} {self.source_status.name} -> {self.target_status.name}>" 

257 

258 

259class StatusAction(Base): 

260 """Maps statuses to allowed actions/permissions.""" 

261 

262 __tablename__ = "status_actions" 

263 

264 __table_args__ = (UniqueConstraint("status_id", "action"),) 

265 

266 id: Mapped[int] = mapped_column(primary_key=True) 

267 status_id: Mapped[int] = mapped_column( 

268 ForeignKey( 

269 "statuses.id", 

270 ondelete="CASCADE", 

271 ), 

272 nullable=False, 

273 index=True, 

274 ) 

275 action: Mapped[PermString] = mapped_column(String(64), nullable=False, index=True) 

276 

277 status: Mapped[Status] = relationship(back_populates="actions") 

278 

279 @validates("action") 

280 def validate_action(self, key, action_value): 

281 """Ensure the action value is a valid action string.""" 

282 if action_value not in VALID_ACTIONS: 

283 raise ValueError( 

284 f"Invalid action: '{action_value}'. Must be one of {VALID_ACTIONS}" 

285 ) 

286 return action_value