Coverage for agentos/config/validator.py: 29%

1"""AgentOS configuration validation — JSON Schema-based config integrity checks. 

2 

3Validates agentos.yaml and environment configurations at startup and reload. 

4""" 

5 

6from __future__ import annotations 

7 

8import json 

9from dataclasses import dataclass, field 

10from enum import Enum 

11from typing import Any 

12 

13# ── Schema definition ───────────────────────────────────────────────────────── 

14 

15AGENTOS_CONFIG_SCHEMA: dict = { 

16 "$schema": "https://json-schema.org/draft/2020-12/schema", 

17 "title": "AgentOS Configuration", 

18 "type": "object", 

19 "required": ["agentos"], 

20 "properties": { 

21 "agentos": { 

22 "type": "object", 

23 "required": ["version"], 

24 "properties": { 

25 "version": {"type": "string", "pattern": r"^\d+\.\d+\.\d+$"}, 

26 "name": {"type": "string", "minLength": 1}, 

27 "debug": {"type": "boolean"}, 

28 "models": { 

29 "type": "object", 

30 "properties": { 

31 "default_provider": { 

32 "type": "string", 

33 "enum": ["openai", "anthropic", "gemini", "deepseek"], 

34 }, 

35 "default_model": {"type": "string"}, 

36 "temperature": {"type": "number", "minimum": 0.0, "maximum": 2.0}, 

37 "max_retries": {"type": "integer", "minimum": 0, "maximum": 10}, 

38 "request_timeout": {"type": "integer", "minimum": 1, "maximum": 600}, 

39 }, 

40 }, 

41 "memory": { 

42 "type": "object", 

43 "properties": { 

44 "short_term_limit": {"type": "integer", "minimum": 1}, 

45 "long_term_backend": { 

46 "type": "string", 

47 "enum": ["chromadb", "faiss", "qdrant", "pinecone"], 

48 }, 

49 "summarization_threshold": {"type": "integer", "minimum": 100}, 

50 }, 

51 }, 

52 "server": { 

53 "type": "object", 

54 "properties": { 

55 "host": {"type": "string"}, 

56 "port": {"type": "integer", "minimum": 1, "maximum": 65535}, 

57 "workers": {"type": "integer", "minimum": 1, "maximum": 64}, 

58 "timeout_keep_alive": {"type": "integer", "minimum": 1}, 

59 }, 

60 }, 

61 "security": { 

62 "type": "object", 

63 "properties": { 

64 "sandbox": {"type": "boolean"}, 

65 "allowed_commands": { 

66 "type": "array", 

67 "items": {"type": "string"}, 

68 }, 

69 "pii_sanitizer": {"type": "boolean"}, 

70 "audit_log": {"type": "boolean"}, 

71 }, 

72 }, 

73 "benchmarks": { 

74 "type": "object", 

75 "properties": { 

76 "enabled": {"type": "boolean"}, 

77 "iterations": {"type": "integer", "minimum": 1}, 

78 "output_format": {"type": "string", "enum": ["json", "csv", "markdown"]}, 

79 }, 

80 }, 

81 }, 

82 }, 

83 }, 

84} 

85 

86 

87# ── Validation result ───────────────────────────────────────────────────────── 

88 

89 

90class ValidationLevel(Enum): 

91 """校验等级。""" 

92 

93 ERROR = "error" 

94 WARNING = "warning" 

95 INFO = "info" 

96 

97 

98@dataclass 

99class ValidationIssue: 

100 """校验问题。""" 

101 

102 level: ValidationLevel 

103 path: str 

104 message: str 

105 

106 

107@dataclass 

108class ValidationResult: 

109 """校验结果。""" 

110 

111 valid: bool = True 

112 issues: list[ValidationIssue] = field(default_factory=list) 

113 

114 @property 

115 def errors(self) -> list[ValidationIssue]: 

116 return [i for i in self.issues if i.level == ValidationLevel.ERROR] 

117 

118 @property 

119 def warnings(self) -> list[ValidationIssue]: 

120 return [i for i in self.issues if i.level == ValidationLevel.WARNING] 

121 

122 def add_error(self, path: str, message: str): 

123 self.issues.append(ValidationIssue(ValidationLevel.ERROR, path, message)) 

124 self.valid = False 

125 

126 def add_warning(self, path: str, message: str): 

127 self.issues.append(ValidationIssue(ValidationLevel.WARNING, path, message)) 

128 

129 def __str__(self) -> str: 

130 if self.valid and not self.issues: 

131 return "Configuration valid" 

132 lines = [ 

133 f"Configuration {'valid' if self.valid else 'invalid'} ({len(self.errors)} errors, {len(self.warnings)} warnings)" 

134 ] 

135 for i in self.issues: 

136 lines.append(f" [{i.level.value}] {i.path}: {i.message}") 

137 return "\n".join(lines) 

138 

139 

140# ── Validator ───────────────────────────────────────────────────────────────── 

141 

142 

143def _validate_type(value: Any, expected: str, schema: dict) -> str | None: 

144 """Return error string or None.""" 

145 type_map = { 

146 "string": str, 

147 "integer": int, 

148 "number": (int, float), 

149 "boolean": bool, 

150 "array": list, 

151 "object": dict, 

152 } 

153 py_type = type_map.get(expected) 

154 if py_type is None: 

155 return None 

156 if not isinstance(value, py_type): 

157 return f"expected {expected}, got {type(value).__name__}" 

158 return None 

159 

160 

161def _walk_schema( 

162 config: dict, schema: dict, path: str = "", result: ValidationResult | None = None 

163) -> ValidationResult: 

164 if result is None: 

165 result = ValidationResult() 

166 

167 schema_type = schema.get("type") 

168 if schema_type == "object": 

169 if not isinstance(config, dict): 

170 result.add_error(path, f"expected object, got {type(config).__name__}") 

171 return result 

172 # Required fields 

173 for req in schema.get("required", []): 

174 if req not in config: 

175 result.add_error(f"{path}.{req}" if path else req, "required field missing") 

176 # Properties 

177 for prop, prop_schema in schema.get("properties", {}).items(): 

178 if prop in config: 

179 child_path = f"{path}.{prop}" if path else prop 

180 _walk_schema(config[prop], prop_schema, child_path, result) 

181 # Enum check for object itself (rare) 

182 elif schema_type in ("string", "integer", "number", "boolean"): 

183 err = _validate_type(config, schema_type, schema) 

184 if err: 

185 result.add_error(path, err) 

186 return result 

187 if "enum" in schema and config not in schema["enum"]: 

188 result.add_error(path, f"must be one of {schema['enum']}, got {config!r}") 

189 if "pattern" in schema and isinstance(config, str): 

190 import re 

191 

192 if not re.match(schema["pattern"], config): 

193 result.add_error(path, f"'{config}' does not match pattern {schema['pattern']}") 

194 if "minimum" in schema and isinstance(config, (int, float)): 

195 if config < schema["minimum"]: 

196 result.add_error(path, f"{config} < minimum {schema['minimum']}") 

197 if "maximum" in schema and isinstance(config, (int, float)): 

198 if config > schema["maximum"]: 

199 result.add_error(path, f"{config} > maximum {schema['maximum']}") 

200 if "minLength" in schema and isinstance(config, str): 

201 if len(config) < schema["minLength"]: 

202 result.add_error(path, f"length {len(config)} < min {schema['minLength']}") 

203 elif schema_type == "array": 

204 if not isinstance(config, list): 

205 result.add_error(path, f"expected array, got {type(config).__name__}") 

206 return result 

207 

208 return result 

209 

210 

211def validate_config(config: dict, schema: dict | None = None) -> ValidationResult: 

212 """Validate an AgentOS configuration dict against the built-in JSON Schema.""" 

213 schema = schema or AGENTOS_CONFIG_SCHEMA 

214 return _walk_schema(config, schema) 

215 

216 

217def validate_config_file(file_path: str) -> ValidationResult: 

218 """Load and validate an AgentOS configuration YAML/JSON file.""" 

219 import os 

220 

221 if not os.path.exists(file_path): 

222 result = ValidationResult() 

223 result.add_error("", f"config file not found: {file_path}") 

224 return result 

225 

226 with open(file_path) as f: 

227 if file_path.endswith((".yaml", ".yml")): 

228 try: 

229 import yaml 

230 

231 config = yaml.safe_load(f) 

232 except ImportError: 

233 import json 

234 

235 config = json.load(f) # fallback, may fail 

236 else: 

237 config = json.load(f) 

238 

239 return validate_config(config) 

240 

241 

242def generate_schema_json() -> str: 

243 """Return the AgentOS config JSON Schema as a formatted JSON string.""" 

244 return json.dumps(AGENTOS_CONFIG_SCHEMA, indent=2)