Coverage for postrfp/fsm/schemas.py: 99%

1from typing import List, Optional, Literal, Set, Annotated 

2from pydantic import ( 

3 BaseModel, 

4 Field, 

5 ConfigDict, 

6 model_validator, 

7 computed_field, 

8 StringConstraints, 

9) 

10 

11from postrfp.shared.types import PermString 

12from postrfp.shared.serial.common import Pagination 

13 

14 

15class DbModel(BaseModel): 

16 """Base model for database schemas.""" 

17 

18 model_config = ConfigDict( 

19 from_attributes=True, 

20 populate_by_name=True, 

21 use_enum_values=True, 

22 arbitrary_types_allowed=True, 

23 ) 

24 

25 

26class IdSchema(BaseModel): 

27 """Schema for returning an ID.""" 

28 

29 id: int = Field(..., description="Unique identifier") 

30 

31 

32# Base schemas for simple requests/responses 

33class StatusSchema(DbModel): 

34 """Schema for state data compatible with editor.""" 

35 

36 id: Optional[int] = None 

37 name: str = Field( 

38 ..., max_length=64, description="Human-readable name of the state" 

39 ) 

40 code: str = Field(max_length=64, description="System identifier for the state") 

41 colour: Optional[ 

42 Annotated[str, StringConstraints(max_length=7, pattern=r"^#[0-9A-Fa-f]{6}$")] 

43 ] = Field(None, description="Hex color code (#RRGGBB)", examples=["#1E90FF"]) 

44 

45 status_actions: list[PermString] = [] 

46 description: Optional[str] = Field(max_length=512, default=None) 

47 

48 

49class TransitionSchema(DbModel): 

50 """Schema for transition data compatible with editor.""" 

51 

52 id: Optional[int] = None 

53 name: str = Field(..., max_length=64) 

54 

55 source_status: StatusSchema 

56 target_status: StatusSchema 

57 

58 guard_policy: Optional[str] = Field( 

59 default=None, description="CEL expression for guard condition" 

60 ) 

61 

62 @computed_field # type: ignore[misc] 

63 @property 

64 def source(self) -> str: 

65 return self.source_status.code 

66 

67 @computed_field # type: ignore[misc] 

68 @property 

69 def target(self) -> str: 

70 return self.target_status.code 

71 

72 

73class WorkflowSchema(DbModel): 

74 """Schema for definition compatible with editor.""" 

75 

76 id: Optional[int] = Field(None, description="ID of the FSM definition") 

77 title: str = Field(..., max_length=128) 

78 description: str | None = Field(None, max_length=1024) 

79 entity_type: str = Field(..., max_length=64) 

80 initial_status_code: Optional[str] = Field( 

81 default="Not Set", max_length=64, description="Code of the initial state" 

82 ) 

83 current_state: Optional[str] = Field( 

84 None, description="Current state code (if applicable)" 

85 ) 

86 

87 context_spec: Optional[str] = Field( 

88 None, description="JSON schema for context validation" 

89 ) 

90 version: int = 1 

91 is_active: bool = True 

92 

93 organisation_id: str = Field( 

94 ..., description="ID of the organisation that owns this " 

95 ) 

96 statuses: List[StatusSchema] = Field(default_factory=list, min_length=1) 

97 transitions: List[TransitionSchema] = [] 

98 

99 @model_validator(mode="after") 

100 def validate_workflow_integrity(self) -> "WorkflowSchema": 

101 """ 

102 Validate the workflow structure: 

103 - Status names must be unique 

104 - Status codes must be unique 

105 - Transition names must be unique 

106 - Each status must be involved in at least one transition (no orphans) 

107 - If transitions exist, statuses must exist too 

108 - All transition source and target codes must refer to valid statuses 

109 """ 

110 if self.transitions and not self.statuses: 

111 raise ValueError("Workflow contains transitions but no statuses") 

112 

113 if not self.statuses: 

114 return self 

115 

116 status_names = [status.name for status in self.statuses] 

117 if len(status_names) != len(set(status_names)): 

118 raise ValueError("Duplicate status name found") 

119 

120 status_codes = [status.code for status in self.statuses] 

121 if len(status_codes) != len(set(status_codes)): 

122 raise ValueError("Duplicate status code found") 

123 

124 if not self.transitions: 

125 return self 

126 

127 transition_names = [transition.name for transition in self.transitions] 

128 if len(transition_names) != len(set(transition_names)): 

129 raise ValueError("Duplicate transition name found") 

130 

131 valid_status_codes = set(status.code for status in self.statuses) 

132 

133 for transition in self.transitions: 

134 if transition.source not in valid_status_codes: 

135 raise ValueError( 

136 f"Transition '{transition.name}' references invalid source status code: '{transition.source}'" 

137 ) 

138 if transition.target not in valid_status_codes: 

139 raise ValueError( 

140 f"Transition '{transition.name}' references invalid target status code: '{transition.target}'" 

141 ) 

142 

143 involved_status_codes: Set[str] = set() 

144 

145 for transition in self.transitions: 

146 involved_status_codes.add(transition.source) 

147 involved_status_codes.add(transition.target) 

148 

149 all_status_codes = valid_status_codes 

150 uninvolved_status_codes = all_status_codes - involved_status_codes 

151 if uninvolved_status_codes: 

152 raise ValueError( 

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

154 ) 

155 

156 return self 

157 

158 

159class WorkflowSummary(DbModel): 

160 """Schema for a summary of a workflow definition.""" 

161 

162 id: int = Field(..., description="ID of the FSM definition") 

163 title: str = Field(..., max_length=128) 

164 entity_type: str = Field(..., max_length=64) 

165 version: int = 1 

166 is_active: bool = True 

167 

168 

169class WorkflowList(DbModel): 

170 """Schema for a list of transitions.""" 

171 

172 data: List[WorkflowSummary] 

173 pagination: Pagination 

174 

175 

176class TransitionRequest(BaseModel): 

177 entity_id: int 

178 entity_type: Literal["Project", "Issue"] 

179 transition_name: str 

180 

181 

182class TransitionResult(BaseModel): 

183 """ 

184 Result returned from running an Entity transition 

185 """ 

186 

187 transition_permitted: bool 

188 message: Optional[str] = Field( 

189 None, description="Message returned from the Angst server" 

190 ) 

191 job_ref: Optional[str] = Field( 

192 None, description="Job ID of any HTTP calls invoked on the Angst server" 

193 ) 

194 

195 

196class EntityTypeSchema(DbModel): 

197 """Schema for FSM entity types.""" 

198 

199 name: str = Field(..., description="Entity class name") 

200 table_name: str = Field(..., description="SQLAlchemy table name") 

201 context_schema: dict = Field(..., description="JSON schema for context validation") 

202 

203 

204class EntityTypeList(DbModel): 

205 """Schema for a list of FSM entity types.""" 

206 

207 data: List[EntityTypeSchema]