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 socket 

24import logging 

25from dataclasses import dataclass, field 

26from enum import Enum 

27from typing import Optional 

28 

29logger = logging.getLogger(__name__) 

30 

31 

32class Severity(str, Enum): 

33 CRITICAL = "critical" # Server MUST NOT start 

34 ERROR = "error" # Feature degraded 

35 WARNING = "warning" # Non-blocking concern 

36 OK = "ok" 

37 

38 

39@dataclass 

40class Issue: 

41 component: str 

42 message: str 

43 severity: Severity 

44 suggestion: str = "" 

45 

46 

47@dataclass 

48class ValidationReport: 

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

50 

51 @property 

52 def has_critical(self) -> bool: 

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

54 

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

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

57 

58 def report(self) -> str: 

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

60 for issue in self.issues: 

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

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

63 if issue.suggestion: 

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

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

66 

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

68 if Severity.CRITICAL in statuses: 

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

70 elif Severity.ERROR in statuses: 

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

72 else: 

73 lines.append(" RESULT: OK") 

74 return "\n".join(lines) 

75 

76 

77# ── Checks ────────────────────────────────────────────────────────────────── 

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("env", f"{var} not set", Severity.WARNING, 

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

90 

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

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

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

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

95 

96 

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

98 for path in paths: 

99 try: 

100 os.makedirs(path, exist_ok=True) 

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

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

103 f.write("ok") 

104 os.remove(test_file) 

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

106 except PermissionError: 

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

108 "Fix permissions or change AGENTOS_LOG_DIR / AGENTOS_DATA_DIR") 

109 except OSError as e: 

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

111 

112 

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

114 """Quick TCP connectivity check.""" 

115 from urllib.parse import urlparse 

116 parsed = urlparse(url) 

117 host = parsed.hostname or "localhost" 

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

119 

120 try: 

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

122 sock.close() 

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

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

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

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

127 

128 

129# ── Public API ────────────────────────────────────────────────────────────── 

130 

131def validate_startup( 

132 data_dir: Optional[str] = None, 

133 log_dir: Optional[str] = None, 

134) -> ValidationReport: 

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

136 

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

138 """ 

139 report = ValidationReport() 

140 

141 _check_env_vars(report) 

142 

143 disk_paths = [ 

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

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

146 ] 

147 _check_disk(report, disk_paths) 

148 

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

150 if db_url: 

151 _check_connectivity(report, "DB", db_url) 

152 

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

154 if redis_url: 

155 _check_connectivity(report, "Redis", redis_url) 

156 

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

158 if otlp: 

159 _check_connectivity(report, "OTLP", otlp) 

160 

161 logger.info(report.report()) 

162 return report 

163 

164 

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