Coverage for gco / models / manifest_models.py: 99%

1""" 

2Manifest submission data models for GCO (Global Capacity Orchestrator on AWS). 

3 

4This module defines dataclasses for Kubernetes manifest processing: 

5- KubernetesManifest: Single Kubernetes resource definition 

6- ManifestSubmissionRequest: Request to submit one or more manifests 

7- ManifestSubmissionResponse: Response with processing results 

8- ResourceStatus: Status of a single resource operation 

9 

10These models are used by the manifest processor service to validate, 

11process, and track Kubernetes resource submissions. 

12""" 

13 

14from __future__ import annotations 

15 

16from dataclasses import dataclass 

17from typing import Any 

18 

19 

20@dataclass 

21class KubernetesManifest: 

22 """ 

23 Represents a single Kubernetes manifest. 

24 

25 Attributes: 

26 apiVersion: Kubernetes API version (e.g., 'apps/v1', 'v1') 

27 kind: Resource kind (e.g., 'Deployment', 'Service', 'ConfigMap') 

28 metadata: Resource metadata including name and namespace 

29 spec: Resource specification (for most resource types) 

30 data: Resource data (for ConfigMaps and Secrets) 

31 """ 

32 

33 apiVersion: str 

34 kind: str 

35 metadata: dict[str, Any] 

36 spec: dict[str, Any] | None = None 

37 data: dict[str, Any] | None = None 

38 

39 def __post_init__(self) -> None: 

40 """Validate Kubernetes manifest structure""" 

41 if not self.apiVersion: 

42 raise ValueError("apiVersion cannot be empty") 

43 

44 if not self.kind: 

45 raise ValueError("kind cannot be empty") 

46 

47 if not isinstance(self.metadata, dict): 

48 raise ValueError("metadata must be a dictionary") 

49 

50 if "name" not in self.metadata: 

51 raise ValueError("metadata must contain a 'name' field") 

52 

53 # Either spec or data should be present (or both for some resources) 

54 if self.spec is not None and not isinstance(self.spec, dict): 

55 raise ValueError("spec must be a dictionary") 

56 

57 if self.data is not None and not isinstance(self.data, dict): 

58 raise ValueError("data must be a dictionary") 

59 

60 # At least one of spec or data should be present for most resources 

61 if ( 

62 self.spec is None 

63 and self.data is None 

64 and self.kind not in ["Namespace", "ServiceAccount"] 

65 ): 

66 raise ValueError("Either spec or data must be provided for most resource types") 

67 

68 def get_name(self) -> str: 

69 """Get the resource name""" 

70 return str(self.metadata.get("name", "")) 

71 

72 def get_namespace(self) -> str: 

73 """Get the resource namespace""" 

74 return str(self.metadata.get("namespace", "default")) 

75 

76 def to_dict(self) -> dict[str, Any]: 

77 """Convert to dictionary representation""" 

78 result = {"apiVersion": self.apiVersion, "kind": self.kind, "metadata": self.metadata} 

79 

80 if self.spec is not None: 

81 result["spec"] = self.spec 

82 

83 if self.data is not None: 

84 result["data"] = self.data 

85 

86 return result 

87 

88 @classmethod 

89 def from_dict(cls, data: dict[str, Any]) -> KubernetesManifest: 

90 """Create from dictionary representation""" 

91 return cls( 

92 apiVersion=data["apiVersion"], 

93 kind=data["kind"], 

94 metadata=data["metadata"], 

95 spec=data.get("spec"), 

96 data=data.get("data"), 

97 ) 

98 

99 

100@dataclass 

101class ManifestSubmissionRequest: 

102 """Request to submit Kubernetes manifests""" 

103 

104 manifests: list[dict[str, Any]] 

105 namespace: str | None = None 

106 dry_run: bool = False 

107 validate: bool = True 

108 

109 def __post_init__(self) -> None: 

110 """Validate submission request""" 

111 if not self.manifests: 

112 raise ValueError("At least one manifest must be provided") 

113 

114 # Validate each manifest can be parsed 

115 for i, manifest_data in enumerate(self.manifests): 

116 try: 

117 KubernetesManifest.from_dict(manifest_data) 

118 except Exception as e: 

119 raise ValueError(f"Invalid manifest at index {i}: {e}") from e 

120 

121 def get_kubernetes_manifests(self) -> list[KubernetesManifest]: 

122 """Convert to KubernetesManifest objects""" 

123 return [KubernetesManifest.from_dict(manifest) for manifest in self.manifests] 

124 

125 def get_resource_count(self) -> int: 

126 """Get total number of resources to be created""" 

127 return len(self.manifests) 

128 

129 

130@dataclass 

131class ResourceStatus: 

132 """Status of a deployed Kubernetes resource""" 

133 

134 api_version: str 

135 kind: str 

136 name: str 

137 namespace: str 

138 status: str # 'created', 'updated', 'unchanged', 'failed' 

139 message: str | None = None 

140 uid: str | None = None # Kubernetes resource UID 

141 

142 def __post_init__(self) -> None: 

143 """Validate resource status""" 

144 valid_statuses = {"created", "updated", "unchanged", "failed", "deleted"} 

145 if self.status not in valid_statuses: 

146 raise ValueError(f"Status must be one of {valid_statuses}, got {self.status}") 

147 

148 if not self.name: 

149 raise ValueError("Resource name cannot be empty") 

150 

151 if not self.namespace: 

152 raise ValueError("Resource namespace cannot be empty") 

153 

154 def is_successful(self) -> bool: 

155 """Check if the resource operation was successful""" 

156 return self.status in {"created", "updated", "unchanged", "deleted"} 

157 

158 def get_resource_identifier(self) -> str: 

159 """Get unique identifier for this resource""" 

160 return f"{self.api_version}/{self.kind}/{self.namespace}/{self.name}" 

161 

162 

163@dataclass 

164class ManifestSubmissionResponse: 

165 """Response from manifest submission""" 

166 

167 success: bool 

168 cluster_id: str 

169 region: str 

170 resources: list[ResourceStatus] 

171 errors: list[str] | None = None 

172 

173 def __post_init__(self) -> None: 

174 """Validate submission response""" 

175 if not self.cluster_id: 

176 raise ValueError("Cluster ID cannot be empty") 

177 

178 if not self.region: 

179 raise ValueError("Region cannot be empty") 

180 

181 if not isinstance(self.resources, list): 

182 raise ValueError("Resources must be a list") 

183 

184 def get_successful_resources(self) -> list[ResourceStatus]: 

185 """Get list of successfully processed resources""" 

186 return [r for r in self.resources if r.is_successful()] 

187 

188 def get_failed_resources(self) -> list[ResourceStatus]: 

189 """Get list of failed resources""" 

190 return [r for r in self.resources if not r.is_successful()] 

191 

192 def get_summary(self) -> dict[str, int]: 

193 """Get summary of resource processing results""" 

194 summary = {"created": 0, "updated": 0, "unchanged": 0, "failed": 0} 

195 for resource in self.resources: 

196 summary[resource.status] += 1 

197 return summary