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 os 

23import sys 

24import socket 

25import logging 

26from dataclasses import dataclass, field 

27from enum import Enum 

28from typing import Optional 

29 

30logger = logging.getLogger(__name__) 

31 

32 

33class Severity(str, Enum): 

34 CRITICAL = "critical" # Server MUST NOT start 

35 ERROR = "error" # Feature degraded 

36 WARNING = "warning" # Non-blocking concern 

37 OK = "ok" 

38 

39 

40@dataclass 

41class Issue: 

42 component: str 

43 message: str 

44 severity: Severity 

45 suggestion: str = "" 

46 

47 

48@dataclass 

49class ValidationReport: 

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

51 

52 @property 

53 def has_critical(self) -> bool: 

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

55 

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

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

58 

59 def report(self) -> str: 

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

61 for issue in self.issues: 

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

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

64 if issue.suggestion: 

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

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

67 

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

69 if Severity.CRITICAL in statuses: 

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

71 elif Severity.ERROR in statuses: 

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

73 else: 

74 lines.append(" RESULT: OK") 

75 return "\n".join(lines) 

76 

77 

78# ── Checks ────────────────────────────────────────────────────────────────── 

79 

80def _check_env_vars(report: ValidationReport): 

81 required = ["AGENTOS_SECRET_KEY"] 

82 optional = { 

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

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

85 "AGENTOS_OTLP_ENDPOINT": "Distributed tracing disabled", 

86 } 

87 for var in required: 

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

89 report.add("env", f"{var} not set", Severity.WARNING, 

90 f"Set {var} for production; using default for dev") 

91 

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

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

94 report.add("env", f"{var} not set — {hint}", Severity.WARNING, 

95 f"Set {var} for full production readiness") 

96 

97 

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

99 for path in paths: 

100 try: 

101 os.makedirs(path, exist_ok=True) 

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

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

104 f.write("ok") 

105 os.remove(test_file) 

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

107 except PermissionError: 

108 report.add("disk", f"Cannot write to {path}", Severity.CRITICAL, 

109 "Fix permissions or change AGENTOS_LOG_DIR / AGENTOS_DATA_DIR") 

110 except OSError as e: 

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

112 

113 

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

115 """Quick TCP connectivity check.""" 

116 from urllib.parse import urlparse 

117 parsed = urlparse(url) 

118 host = parsed.hostname or "localhost" 

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

120 

121 try: 

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

123 sock.close() 

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

125 except (socket.timeout, ConnectionRefusedError, OSError) as e: 

126 report.add("connectivity", f"{name} unreachable ({host}:{port}): {e}", 

127 Severity.WARNING, f"Verify {name} is running or disable related features") 

128 

129 

130# ── Public API ────────────────────────────────────────────────────────────── 

131 

132def validate_startup( 

133 data_dir: Optional[str] = None, 

134 log_dir: Optional[str] = None, 

135) -> ValidationReport: 

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

137 

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

139 """ 

140 report = ValidationReport() 

141 

142 _check_env_vars(report) 

143 

144 disk_paths = [ 

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

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

147 ] 

148 _check_disk(report, disk_paths) 

149 

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

151 if db_url: 

152 _check_connectivity(report, "DB", db_url) 

153 

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

155 if redis_url: 

156 _check_connectivity(report, "Redis", redis_url) 

157 

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

159 if otlp: 

160 _check_connectivity(report, "OTLP", otlp) 

161 

162 logger.info(report.report()) 

163 return report 

164 

165 

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