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, Optional 

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": {"type": "string", "enum": ["openai", "anthropic", "gemini", "deepseek"]}, 

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

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

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

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

36 }, 

37 }, 

38 "memory": { 

39 "type": "object", 

40 "properties": { 

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

42 "long_term_backend": {"type": "string", "enum": ["chromadb", "faiss", "qdrant", "pinecone"]}, 

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

44 }, 

45 }, 

46 "server": { 

47 "type": "object", 

48 "properties": { 

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

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

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

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

53 }, 

54 }, 

55 "security": { 

56 "type": "object", 

57 "properties": { 

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

59 "allowed_commands": { 

60 "type": "array", 

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

62 }, 

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

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

65 }, 

66 }, 

67 "benchmarks": { 

68 "type": "object", 

69 "properties": { 

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

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

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

73 }, 

74 }, 

75 }, 

76 }, 

77 }, 

78} 

79 

80 

81# ── Validation result ───────────────────────────────────────────────────────── 

82 

83 

84class ValidationLevel(Enum): 

85 

86 """校验等级。""" 

87 

88 ERROR = "error" 

89 WARNING = "warning" 

90 INFO = "info" 

91 

92 

93@dataclass 

94class ValidationIssue: 

95 """校验问题。""" 

96 level: ValidationLevel 

97 path: str 

98 message: str 

99 

100 

101@dataclass 

102class ValidationResult: 

103 """校验结果。""" 

104 valid: bool = True 

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

106 

107 @property 

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

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

110 

111 @property 

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

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

114 

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

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

117 self.valid = False 

118 

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

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

121 

122 def __str__(self) -> str: 

123 if self.valid and not self.issues: 

124 return "Configuration valid" 

125 lines = [f"Configuration {'valid' if self.valid else 'invalid'} ({len(self.errors)} errors, {len(self.warnings)} warnings)"] 

126 for i in self.issues: 

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

128 return "\n".join(lines) 

129 

130 

131# ── Validator ───────────────────────────────────────────────────────────────── 

132 

133 

134def _validate_type(value: Any, expected: str, schema: dict) -> Optional[str]: 

135 """Return error string or None.""" 

136 type_map = { 

137 "string": str, "integer": int, "number": (int, float), 

138 "boolean": bool, "array": list, "object": dict, 

139 } 

140 py_type = type_map.get(expected) 

141 if py_type is None: 

142 return None 

143 if not isinstance(value, py_type): 

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

145 return None 

146 

147 

148def _walk_schema(config: dict, schema: dict, path: str = "", result: Optional[ValidationResult] = None) -> ValidationResult: 

149 if result is None: 

150 result = ValidationResult() 

151 

152 schema_type = schema.get("type") 

153 if schema_type == "object": 

154 if not isinstance(config, dict): 

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

156 return result 

157 # Required fields 

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

159 if req not in config: 

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

161 # Properties 

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

163 if prop in config: 

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

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

166 # Enum check for object itself (rare) 

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

168 err = _validate_type(config, schema_type, schema) 

169 if err: 

170 result.add_error(path, err) 

171 return result 

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

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

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

175 import re 

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

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

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

179 if config < schema["minimum"]: 

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

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

182 if config > schema["maximum"]: 

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

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

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

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

187 elif schema_type == "array": 

188 if not isinstance(config, list): 

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

190 return result 

191 

192 return result 

193 

194 

195def validate_config(config: dict, schema: Optional[dict] = None) -> ValidationResult: 

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

197 schema = schema or AGENTOS_CONFIG_SCHEMA 

198 return _walk_schema(config, schema) 

199 

200 

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

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

203 import os 

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

205 result = ValidationResult() 

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

207 return result 

208 

209 with open(file_path) as f: 

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

211 try: 

212 import yaml 

213 config = yaml.safe_load(f) 

214 except ImportError: 

215 import json 

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

217 else: 

218 config = json.load(f) 

219 

220 return validate_config(config) 

221 

222 

223def generate_schema_json() -> str: 

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

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