Coverage for oracle / oci_fsdr_mcp_server / models.py: 100.00%

1""" 

2Copyright (c) 2025, Oracle and/or its affiliates. 

3Licensed under the Universal Permissive License v1.0 as shown at 

4https://oss.oracle.com/licenses/upl. 

5 

6Pydantic response models for OCI Full Stack Disaster Recovery MCP tools. 

7 

8These models wrap the OCI SDK's raw to_dict() output into typed, clean 

9structures that are easier for LLMs to reason about. Noisy SDK fields 

10(empty tag dicts, verbose headers) are omitted. 

11""" 

12 

13from __future__ import annotations 

14 

15from typing import Any, Dict, Literal, Optional 

16 

17from pydantic import BaseModel, ConfigDict, Field 

18 

19from .consts import FSDR_OPERATION_NAMES 

20 

21DrProtectionGroupLifecycleState = Literal[ 

22 "CREATING", 

23 "UPDATING", 

24 "ACTIVE", 

25 "INACTIVE", 

26 "DELETING", 

27 "DELETED", 

28 "FAILED", 

29 "NEEDS_ATTENTION", 

30 "UNKNOWN_ENUM_VALUE", 

31] 

32 

33DrProtectionGroupRole = Literal[ 

34 "PRIMARY", 

35 "STANDBY", 

36 "UNCONFIGURED", 

37 "UNKNOWN_ENUM_VALUE", 

38] 

39 

40DrPlanLifecycleState = DrProtectionGroupLifecycleState 

41 

42DrPlanType = Literal[ 

43 "SWITCHOVER", 

44 "FAILOVER", 

45 "START_DRILL", 

46 "STOP_DRILL", 

47 "UNKNOWN_ENUM_VALUE", 

48] 

49 

50DrPlanExecutionLifecycleState = Literal[ 

51 "ACCEPTED", 

52 "IN_PROGRESS", 

53 "WAITING", 

54 "PAUSING", 

55 "PAUSED", 

56 "RESUMING", 

57 "CANCELING", 

58 "CANCELED", 

59 "SUCCEEDED", 

60 "FAILED", 

61 "DELETING", 

62 "DELETED", 

63 "UNKNOWN_ENUM_VALUE", 

64] 

65 

66DrPlanExecutionType = Literal[ 

67 "SWITCHOVER", 

68 "SWITCHOVER_PRECHECK", 

69 "FAILOVER", 

70 "FAILOVER_PRECHECK", 

71 "START_DRILL", 

72 "START_DRILL_PRECHECK", 

73 "STOP_DRILL", 

74 "STOP_DRILL_PRECHECK", 

75 "UNKNOWN_ENUM_VALUE", 

76] 

77 

78WorkRequestStatus = Literal[ 

79 "ACCEPTED", 

80 "IN_PROGRESS", 

81 "WAITING", 

82 "CANCELING", 

83 "CANCELED", 

84 "SUCCEEDED", 

85 "FAILED", 

86 "NEEDS_ATTENTION", 

87 "UNKNOWN_ENUM_VALUE", 

88] 

89 

90WorkRequestOperationType = Literal[ 

91 "CREATE_AUTOMATIC_DR_CONFIGURATION", 

92 "UPDATE_AUTOMATIC_DR_CONFIGURATION", 

93 "DELETE_AUTOMATIC_DR_CONFIGURATION", 

94 "CREATE_DR_PROTECTION_GROUP", 

95 "UPDATE_DR_PROTECTION_GROUP", 

96 "DELETE_DR_PROTECTION_GROUP", 

97 "MOVE_DR_PROTECTION_GROUP", 

98 "UPDATE_ROLE_DR_PROTECTION_GROUP", 

99 "ASSOCIATE_DR_PROTECTION_GROUP", 

100 "DISASSOCIATE_DR_PROTECTION_GROUP", 

101 "CREATE_DR_PLAN", 

102 "UPDATE_DR_PLAN", 

103 "DELETE_DR_PLAN", 

104 "REFRESH_DR_PLAN", 

105 "VERIFY_DR_PLAN", 

106 "CREATE_DR_PLAN_EXECUTION", 

107 "UPDATE_DR_PLAN_EXECUTION", 

108 "DELETE_DR_PLAN_EXECUTION", 

109 "CANCEL_DR_PLAN_EXECUTION", 

110 "PAUSE_DR_PLAN_EXECUTION", 

111 "RESUME_DR_PLAN_EXECUTION", 

112 "RETRY_DR_PLAN_EXECUTION", 

113 "IGNORE_DR_PLAN_EXECUTION", 

114 "UNKNOWN_ENUM_VALUE", 

115] 

116 

117FsdrOperation = Literal[*FSDR_OPERATION_NAMES] 

118 

119 

120class DrProtectionGroupSummary(BaseModel): 

121 model_config = ConfigDict(extra="ignore") 

122 

123 id: str = Field(..., description="The OCID of the DR Protection Group.") 

124 display_name: str = Field( 

125 ..., description="A user-friendly name for the DR Protection Group." 

126 ) 

127 compartment_id: str = Field( 

128 ..., description="The OCID of the compartment containing the DR Protection Group." 

129 ) 

130 role: Optional[DrProtectionGroupRole] = Field( 

131 None, description="The DR Protection Group role in the paired topology." 

132 ) 

133 lifecycle_state: Optional[DrProtectionGroupLifecycleState] = Field( 

134 None, description="The current lifecycle state of the DR Protection Group." 

135 ) 

136 peer_id: Optional[str] = Field( 

137 None, description="The OCID of the associated peer DR Protection Group." 

138 ) 

139 peer_region: Optional[str] = Field( 

140 None, description="The OCI region identifier of the associated peer." 

141 ) 

142 time_created: Optional[str] = Field( 

143 None, description="The date and time the DR Protection Group was created." 

144 ) 

145 time_updated: Optional[str] = Field( 

146 None, description="The date and time the DR Protection Group was last updated." 

147 ) 

148 

149 @classmethod 

150 def from_sdk_dict(cls, d: Dict[str, Any]) -> "DrProtectionGroupSummary": 

151 return cls.model_validate(d) 

152 

153 

154class DrPlanSummary(BaseModel): 

155 model_config = ConfigDict(extra="ignore") 

156 

157 id: str = Field(..., description="The OCID of the DR Plan.") 

158 display_name: str = Field(..., description="A user-friendly name for the DR Plan.") 

159 type: DrPlanType = Field( 

160 ..., description="The type of DR workflow represented by the plan." 

161 ) 

162 lifecycle_state: Optional[DrPlanLifecycleState] = Field( 

163 None, description="The current lifecycle state of the DR Plan." 

164 ) 

165 dr_protection_group_id: Optional[str] = Field( 

166 None, description="The OCID of the DR Protection Group that owns the plan." 

167 ) 

168 peer_dr_protection_group_id: Optional[str] = Field( 

169 None, description="The OCID of the peer DR Protection Group for this plan." 

170 ) 

171 time_created: Optional[str] = Field( 

172 None, description="The date and time the DR Plan was created." 

173 ) 

174 time_updated: Optional[str] = Field( 

175 None, description="The date and time the DR Plan was last updated." 

176 ) 

177 

178 @classmethod 

179 def from_sdk_dict(cls, d: Dict[str, Any]) -> "DrPlanSummary": 

180 return cls.model_validate(d) 

181 

182 

183class DrPlanExecutionSummary(BaseModel): 

184 model_config = ConfigDict(extra="ignore") 

185 

186 id: str = Field(..., description="The OCID of the DR Plan Execution.") 

187 display_name: str = Field( 

188 ..., description="A user-friendly name for the DR Plan Execution." 

189 ) 

190 plan_execution_type: Optional[DrPlanExecutionType] = Field( 

191 None, description="The type of DR Plan Execution." 

192 ) 

193 lifecycle_state: Optional[DrPlanExecutionLifecycleState] = Field( 

194 None, description="The current lifecycle state of the DR Plan Execution." 

195 ) 

196 dr_protection_group_id: Optional[str] = Field( 

197 None, 

198 description="The OCID of the DR Protection Group associated with the execution.", 

199 ) 

200 plan_id: Optional[str] = Field( 

201 None, description="The OCID of the DR Plan being executed." 

202 ) 

203 time_created: Optional[str] = Field( 

204 None, description="The date and time the DR Plan Execution was created." 

205 ) 

206 time_started: Optional[str] = Field( 

207 None, description="The date and time the DR Plan Execution started." 

208 ) 

209 time_ended: Optional[str] = Field( 

210 None, description="The date and time the DR Plan Execution ended." 

211 ) 

212 execution_duration_in_sec: Optional[int] = Field( 

213 None, 

214 description="The execution duration, in seconds, once the execution has ended.", 

215 ge=0, 

216 ) 

217 

218 @classmethod 

219 def from_sdk_dict(cls, d: Dict[str, Any]) -> "DrPlanExecutionSummary": 

220 return cls.model_validate(d) 

221 

222 

223class WorkRequestSummary(BaseModel): 

224 model_config = ConfigDict(extra="ignore") 

225 

226 id: str = Field(..., description="The OCID of the work request.") 

227 operation_type: Optional[WorkRequestOperationType] = Field( 

228 None, description="The OCI FSDR operation tracked by the work request." 

229 ) 

230 status: Optional[WorkRequestStatus] = Field( 

231 None, description="The current status of the work request." 

232 ) 

233 percent_complete: Optional[float] = Field( 

234 None, description="The percentage of work completed.", ge=0, le=100 

235 ) 

236 compartment_id: Optional[str] = Field( 

237 None, description="The OCID of the compartment associated with the work request." 

238 ) 

239 time_accepted: Optional[str] = Field( 

240 None, description="The date and time the work request was accepted." 

241 ) 

242 time_started: Optional[str] = Field( 

243 None, description="The date and time the work request started." 

244 ) 

245 time_finished: Optional[str] = Field( 

246 None, description="The date and time the work request finished." 

247 ) 

248 

249 @classmethod 

250 def from_sdk_dict(cls, d: Dict[str, Any]) -> "WorkRequestSummary": 

251 return cls.model_validate(d) 

252 

253 

254class OciResponseResult(BaseModel): 

255 """Normalized OCI SDK response.""" 

256 

257 model_config = ConfigDict(extra="ignore") 

258 

259 data: Optional[Any] = Field( 

260 None, description="Response data converted from the OCI SDK model." 

261 ) 

262 status: int = Field(..., description="HTTP status code returned by the OCI SDK.") 

263 headers: Dict[str, Any] = Field( 

264 default_factory=dict, description="Response headers returned by the OCI SDK." 

265 ) 

266 

267 

268class WriteResult(BaseModel): 

269 """Normalized response for write operations that trigger a work request.""" 

270 

271 model_config = ConfigDict(extra="ignore") 

272 

273 data: Optional[Any] = Field( 

274 None, description="Response data converted from the OCI SDK model." 

275 ) 

276 status: int = Field(..., description="HTTP status code returned by the OCI SDK.") 

277 headers: Dict[str, Any] = Field( 

278 default_factory=dict, description="Response headers returned by the OCI SDK." 

279 ) 

280 work_request_id: Optional[str] = Field( 

281 None, description="The opc-work-request-id header value, when present." 

282 ) 

283 

284 

285class ListResult(BaseModel): 

286 """Paginated list response.""" 

287 

288 items: list[Dict[str, Any]] = Field( 

289 ..., description="Items returned by the list operation." 

290 ) 

291 opc_next_page: Optional[str] = Field( 

292 None, description="Token for the next page of results, when present." 

293 ) 

294 total_items: int = Field( 

295 ..., description="The number of items included in this response.", ge=0 

296 ) 

297 

298 @classmethod 

299 def from_items( 

300 cls, 

301 items: list[Dict[str, Any]], 

302 next_page: Optional[str], 

303 ) -> "ListResult": 

304 return cls(items=items, opc_next_page=next_page, total_items=len(items))