Coverage for agentos/config_validator.py: 0%

1""" 

2Startup configuration validator for AgentOS. 

3 

4Runs at server boot before accepting connections. Validates: 

5- Required env vars present 

6- Database connectivity (optional) 

7- Redis connectivity (optional) 

8- OTLP endpoint reachable (optional, timeout 3s) 

9- Disk write permissions on log/output dirs 

10- SSL/TLS cert validity if HTTPS enabled 

11 

12Usage: 

13 from agentos.config_validator import validate_startup 

14 

15 issues = validate_startup() 

16 if issues.has_critical: 

17 raise SystemExit(issues.report()) 

18""" 

19 

20from __future__ import annotations 

21 

22import logging 

23import os 

24import socket 

25from dataclasses import dataclass, field 

26from enum import StrEnum 

27 

28logger = logging.getLogger(__name__) 

29 

30 

31class Severity(StrEnum): 

32 CRITICAL = "critical" # Server MUST NOT start 

33 ERROR = "error" # Feature degraded 

34 WARNING = "warning" # Non-blocking concern 

35 OK = "ok" 

36 

37 

38@dataclass 

39class Issue: 

40 component: str 

41 message: str 

42 severity: Severity 

43 suggestion: str = "" 

44 

45 

46@dataclass 

47class ValidationReport: 

48 issues: list[Issue] = field(default_factory=list) 

49 

50 @property 

51 def has_critical(self) -> bool: 

52 return any(i.severity == Severity.CRITICAL for i in self.issues) 

53 

54 def add(self, component: str, message: str, severity: Severity, suggestion: str = ""): 

55 self.issues.append(Issue(component, message, severity, suggestion)) 

56 

57 def report(self) -> str: 

58 lines = [f"\n{'='*60}", " AgentOS Startup Validation", f"{'='*60}"] 

59 for issue in self.issues: 

60 tag = f"[{issue.severity.upper()}]" 

61 lines.append(f" {tag:12s} {issue.component}: {issue.message}") 

62 if issue.suggestion: 

63 lines.append(f" → {issue.suggestion}") 

64 lines.append(f"{'='*60}") 

65 

66 statuses = [i.severity for i in self.issues] 

67 if Severity.CRITICAL in statuses: 

68 lines.append(" RESULT: CRITICAL — server will NOT start") 

69 elif Severity.ERROR in statuses: 

70 lines.append(" RESULT: DEGRADED — some features unavailable") 

71 else: 

72 lines.append(" RESULT: OK") 

73 return "\n".join(lines) 

74 

75 

76# ── Checks ────────────────────────────────────────────────────────────────── 

77 

78 

79def _check_env_vars(report: ValidationReport): 

80 required = ["AGENTOS_SECRET_KEY"] 

81 optional = { 

82 "AGENTOS_DATABASE_URL": "Database-backed features disabled", 

83 "AGENTOS_REDIS_URL": "Distributed cache/locks disabled", 

84 "AGENTOS_OTLP_ENDPOINT": "Distributed tracing disabled", 

85 } 

86 for var in required: 

87 if not os.environ.get(var): 

88 report.add( 

89 "env", 

90 f"{var} not set", 

91 Severity.WARNING, 

92 f"Set {var} for production; using default for dev", 

93 ) 

94 

95 for var, hint in optional.items(): 

96 if not os.environ.get(var): 

97 report.add( 

98 "env", 

99 f"{var} not set — {hint}", 

100 Severity.WARNING, 

101 f"Set {var} for full production readiness", 

102 ) 

103 

104 

105def _check_disk(report: ValidationReport, paths: list[str]): 

106 for path in paths: 

107 try: 

108 os.makedirs(path, exist_ok=True) 

109 test_file = os.path.join(path, ".agentos_write_test") 

110 with open(test_file, "w") as f: 

111 f.write("ok") 

112 os.remove(test_file) 

113 report.add("disk", f"{path} writable", Severity.OK) 

114 except PermissionError: 

115 report.add( 

116 "disk", 

117 f"Cannot write to {path}", 

118 Severity.CRITICAL, 

119 "Fix permissions or change AGENTOS_LOG_DIR / AGENTOS_DATA_DIR", 

120 ) 

121 except OSError as e: 

122 report.add("disk", f"{path}: {e}", Severity.ERROR) 

123 

124 

125def _check_connectivity(report: ValidationReport, name: str, url: str, timeout: float = 3.0): 

126 """Quick TCP connectivity check.""" 

127 from urllib.parse import urlparse 

128 

129 parsed = urlparse(url) 

130 host = parsed.hostname or "localhost" 

131 port = parsed.port or (443 if parsed.scheme == "https" else 80) 

132 

133 try: 

134 sock = socket.create_connection((host, port), timeout=timeout) 

135 sock.close() 

136 report.add("connectivity", f"{name} reachable ({host}:{port})", Severity.OK) 

137 except (TimeoutError, ConnectionRefusedError, OSError) as e: 

138 report.add( 

139 "connectivity", 

140 f"{name} unreachable ({host}:{port}): {e}", 

141 Severity.WARNING, 

142 f"Verify {name} is running or disable related features", 

143 ) 

144 

145 

146# ── Public API ────────────────────────────────────────────────────────────── 

147 

148 

149def validate_startup( 

150 data_dir: str | None = None, 

151 log_dir: str | None = None, 

152) -> ValidationReport: 

153 """Run all startup checks and return a report. 

154 

155 Returns a ValidationReport — call `.has_critical` to decide whether to abort. 

156 """ 

157 report = ValidationReport() 

158 

159 _check_env_vars(report) 

160 

161 disk_paths = [ 

162 data_dir or os.environ.get("AGENTOS_DATA_DIR", "./data"), 

163 log_dir or os.environ.get("AGENTOS_LOG_DIR", "./logs"), 

164 ] 

165 _check_disk(report, disk_paths) 

166 

167 db_url = os.environ.get("AGENTOS_DATABASE_URL") 

168 if db_url: 

169 _check_connectivity(report, "DB", db_url) 

170 

171 redis_url = os.environ.get("AGENTOS_REDIS_URL") 

172 if redis_url: 

173 _check_connectivity(report, "Redis", redis_url) 

174 

175 otlp = os.environ.get("AGENTOS_OTLP_ENDPOINT") 

176 if otlp: 

177 _check_connectivity(report, "OTLP", otlp) 

178 

179 logger.info(report.report()) 

180 return report 

181 

182 

183__all__ = ["validate_startup", "ValidationReport", "Severity"]