Coverage for agentos/tests/test_config.py: 0%

1"""Tests for agentos.core.config — ConfigManager, ConfigSource, helpers.""" 

2 

3import json 

4import os 

5import tempfile 

6from dataclasses import dataclass 

7 

8import pytest 

9 

10from agentos.core.config import ( 

11 ConfigError, 

12 ConfigManager, 

13 ConfigNotFoundError, 

14 ConfigSource, 

15 SourceType, 

16 _coerce, 

17 _flatten_dict, 

18 _parse_env_value, 

19) 

20 

21# ============================================================================ 

22# _parse_env_value 

23# ============================================================================ 

24 

25class TestParseEnvValue: 

26 def test_bool_true(self): 

27 assert _parse_env_value("true") is True 

28 assert _parse_env_value("yes") is True 

29 assert _parse_env_value("on") is True 

30 

31 def test_bool_false(self): 

32 assert _parse_env_value("false") is False 

33 assert _parse_env_value("no") is False 

34 assert _parse_env_value("off") is False 

35 

36 def test_null(self): 

37 assert _parse_env_value("null") is None 

38 assert _parse_env_value("none") is None 

39 assert _parse_env_value("") is None 

40 

41 def test_int(self): 

42 assert _parse_env_value("42") == 42 

43 assert _parse_env_value("-10") == -10 

44 

45 def test_float(self): 

46 assert _parse_env_value("3.14") == 3.14 

47 

48 def test_json(self): 

49 assert _parse_env_value('[1,2,3]') == [1, 2, 3] 

50 assert _parse_env_value('{"a":1}') == {"a": 1} 

51 

52 def test_fallback_to_string(self): 

53 assert _parse_env_value("hello world") == "hello world" 

54 

55 

56# ============================================================================ 

57# _flatten_dict 

58# ============================================================================ 

59 

60class TestFlattenDict: 

61 def test_flat(self): 

62 assert _flatten_dict({"a": 1, "b": 2}) == {"a": 1, "b": 2} 

63 

64 def test_nested(self): 

65 d = {"database": {"host": "localhost", "port": 5432}} 

66 result = _flatten_dict(d) 

67 assert result == {"database_host": "localhost", "database_port": 5432} 

68 

69 def test_deep_nested(self): 

70 d = {"a": {"b": {"c": 1}}} 

71 assert _flatten_dict(d) == {"a_b_c": 1} 

72 

73 def test_skip_environ_style(self): 

74 d = {"DB_HOST": "1", "nested": {"x": 2}} 

75 result = _flatten_dict(d) 

76 assert "db_host" in result 

77 assert "nested_x" in result 

78 

79 def test_skip_list_value(self): 

80 d = {"items": [1, 2, 3]} 

81 assert _flatten_dict(d) == {"items": [1, 2, 3]} 

82 

83 

84# ============================================================================ 

85# _coerce 

86# ============================================================================ 

87 

88class TestCoerce: 

89 def test_none(self): 

90 assert _coerce(None, str) is None 

91 

92 def test_bool_from_str(self): 

93 assert _coerce("true", bool) is True 

94 

95 def test_bool_already_bool(self): 

96 assert _coerce(True, bool) is True 

97 

98 def test_int(self): 

99 assert _coerce("42", int) == 42 

100 

101 def test_float(self): 

102 assert _coerce("3.14", float) == 3.14 

103 

104 def test_str(self): 

105 assert _coerce(42, str) == "42" 

106 

107 def test_list_from_str(self): 

108 assert _coerce("a,b,c", list) == ["a", "b", "c"] 

109 

110 def test_list_already_list(self): 

111 assert _coerce([1, 2], list) == [1, 2] 

112 

113 def test_dict(self): 

114 assert _coerce({"a": 1}, dict) == {"a": 1} 

115 

116 def test_optional_pass(self): 

117 assert _coerce("hello", str | None) == "hello" 

118 assert _coerce(None, str | None) is None 

119 

120 def test_dataclass_coerce(self): 

121 @dataclass 

122 class Sub: 

123 x: int = 1 

124 result = _coerce({"x": 42}, Sub) 

125 assert result.x == 42 

126 

127 

128# ============================================================================ 

129# ConfigSource 

130# ============================================================================ 

131 

132class TestConfigSource: 

133 def test_from_dict(self): 

134 s = ConfigSource.from_dict({"a": 1, "b": "hello"}) 

135 assert s.source_type == SourceType.DICT 

136 assert s.data == {"a": 1, "b": "hello"} 

137 assert s.precedence == 500 

138 

139 def test_from_env(self): 

140 os.environ["TEST_CFG_KEY"] = "42" 

141 s = ConfigSource.from_env(prefix="TEST_CFG_") 

142 assert "key" in s.data 

143 assert s.data["key"] == 42 

144 del os.environ["TEST_CFG_KEY"] 

145 

146 def test_from_json(self): 

147 with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: 

148 json.dump({"host": "localhost", "port": 5432}, f) 

149 f.flush() 

150 s = ConfigSource.from_json(f.name) 

151 os.unlink(f.name) 

152 assert s.data["host"] == "localhost" 

153 assert s.data["port"] == 5432 

154 

155 def test_from_dotenv(self): 

156 with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f: 

157 f.write("HOST=localhost\nPORT=8080\nDEBUG=true\n") 

158 f.flush() 

159 s = ConfigSource.from_dotenv(f.name) 

160 os.unlink(f.name) 

161 assert s.data["host"] == "localhost" 

162 assert s.data["port"] == 8080 

163 assert s.data["debug"] is True 

164 

165 def test_from_dotenv_not_found(self): 

166 with pytest.raises(ConfigError): 

167 ConfigSource.from_dotenv("/nonexistent/path/.env") 

168 

169 def test_from_dotenv_not_found_override(self): 

170 s = ConfigSource.from_dotenv("/nonexistent/path/.env", override=True) 

171 assert s.data == {} 

172 

173 def test_from_yaml_not_found(self): 

174 with pytest.raises(ConfigError): 

175 ConfigSource.from_yaml("/nonexistent/config.yaml") 

176 

177 def test_from_json_not_found(self): 

178 with pytest.raises(ConfigError): 

179 ConfigSource.from_json("/nonexistent/config.json") 

180 

181 

182# ============================================================================ 

183# ConfigManager 

184# ============================================================================ 

185 

186class TestConfigManager: 

187 def test_get_basic(self): 

188 cm = ConfigManager(auto_env=False) 

189 cm.add_source(ConfigSource.from_dict({"host": "localhost"})) 

190 assert cm.get("host") == "localhost" 

191 

192 def test_get_missing_raises(self): 

193 cm = ConfigManager(auto_env=False) 

194 with pytest.raises(ConfigNotFoundError): 

195 cm.get("nonexistent") 

196 

197 def test_get_default(self): 

198 cm = ConfigManager(auto_env=False) 

199 assert cm.get("missing", "default") == "default" 

200 

201 def test_get_str(self): 

202 cm = ConfigManager(auto_env=False) 

203 cm.add_source(ConfigSource.from_dict({"port": 8080})) 

204 assert cm.get_str("port") == "8080" 

205 

206 def test_get_int(self): 

207 cm = ConfigManager(auto_env=False) 

208 cm.add_source(ConfigSource.from_dict({"port": "8080"})) 

209 assert cm.get_int("port") == 8080 

210 

211 def test_get_float(self): 

212 cm = ConfigManager(auto_env=False) 

213 cm.add_source(ConfigSource.from_dict({"rate": "3.14"})) 

214 assert cm.get_float("rate") == 3.14 

215 

216 def test_get_bool(self): 

217 cm = ConfigManager(auto_env=False) 

218 cm.add_source(ConfigSource.from_dict({"debug": "true", "verbose": "yes"})) 

219 assert cm.get_bool("debug") is True 

220 assert cm.get_bool("verbose") is True 

221 

222 def test_get_bool_real_false(self): 

223 cm = ConfigManager(auto_env=False) 

224 cm.add_source(ConfigSource.from_dict({"debug": False})) 

225 assert cm.get_bool("debug") is False 

226 

227 def test_get_list_from_str(self): 

228 cm = ConfigManager(auto_env=False) 

229 cm.add_source(ConfigSource.from_dict({"hosts": "a,b,c"})) 

230 assert cm.get_list("hosts") == ["a", "b", "c"] 

231 

232 def test_get_list_already_list(self): 

233 cm = ConfigManager(auto_env=False) 

234 cm.add_source(ConfigSource.from_dict({"hosts": ["a", "b"]})) 

235 assert cm.get_list("hosts") == ["a", "b"] 

236 

237 def test_get_dict(self): 

238 cm = ConfigManager(auto_env=False) 

239 cm.add_source(ConfigSource.from_dict({"db": {"host": "localhost"}})) 

240 assert cm.get_dict("db") == {"host": "localhost"} 

241 

242 def test_keys(self): 

243 cm = ConfigManager(auto_env=False) 

244 cm.add_source(ConfigSource.from_dict({"a": 1, "b": 2})) 

245 assert cm.keys() == ["a", "b"] 

246 

247 def test_to_dict(self): 

248 cm = ConfigManager(auto_env=False) 

249 cm.add_source(ConfigSource.from_dict({"a": 1})) 

250 assert cm.to_dict() == {"a": 1} 

251 

252 def test_mask_secrets(self): 

253 cm = ConfigManager(auto_env=False) 

254 cm.add_source(ConfigSource.from_dict({"password": "secret123", "host": "localhost"})) 

255 cm.mark_secret("password") 

256 d = cm.to_dict(mask_secrets=True) 

257 assert d["password"] == "***MASKED***" 

258 assert d["host"] == "localhost" 

259 

260 def test_precedence(self): 

261 cm = ConfigManager(auto_env=False) 

262 cm.add_source(ConfigSource.from_dict({"key": "low"}, precedence=500)) 

263 cm.add_source(ConfigSource.from_dict({"key": "high"}, precedence=100)) 

264 assert cm.get("key") == "high" 

265 

266 def test_set(self): 

267 cm = ConfigManager(auto_env=False) 

268 cm.add_source(ConfigSource.from_dict({"key": "old"})) 

269 cm.set("key", "new") 

270 assert cm.get("key") == "new" 

271 

272 def test_on_change(self): 

273 cm = ConfigManager(auto_env=False) 

274 changes = [] 

275 

276 def cb(key, old, new): 

277 changes.append((key, old, new)) 

278 

279 cm.on_change(cb) 

280 cm.add_source(ConfigSource.from_dict({"key": "old"})) 

281 cm.set("key", "new") 

282 assert changes == [("key", "old", "new")] 

283 

284 def test_reload(self): 

285 cm = ConfigManager(auto_env=False) 

286 cm.add_source(ConfigSource.from_dict({"key": "v1"})) 

287 assert cm.get("key") == "v1" 

288 cm.reload() 

289 assert cm.get("key") == "v1" 

290 

291 def test_repr(self): 

292 cm = ConfigManager(auto_env=False) 

293 cm.add_source(ConfigSource.from_dict({"a": 1})) 

294 r = repr(cm) 

295 assert "ConfigManager" in r 

296 assert "sources=1" in r 

297 

298 def test_custom_separator(self): 

299 cm = ConfigManager(auto_env=False) 

300 cm.add_source(ConfigSource.from_dict({"items": "a|b|c"})) 

301 assert cm.get_list("items", separator="|") == ["a", "b", "c"] 

302 

303 def test_auto_env(self): 

304 os.environ["AGENTOS_TEST_X"] = "hello" 

305 cm = ConfigManager(auto_env=True, env_prefix="AGENTOS_") 

306 assert cm.get("test_x") == "hello" 

307 del os.environ["AGENTOS_TEST_X"] 

308 

309 def test_bind_dataclass(self): 

310 @dataclass 

311 class AppConfig: 

312 host: str = "0.0.0.0" 

313 port: int = 8080 

314 debug: bool = False 

315 

316 cm = ConfigManager(auto_env=False) 

317 cm.add_source(ConfigSource.from_dict({"host": "10.0.0.1", "port": "3000", "debug": "true"})) 

318 cfg = cm.bind(AppConfig) 

319 assert cfg.host == "10.0.0.1" 

320 assert cfg.port == 3000 

321 assert cfg.debug is True 

322 

323 def test_bind_defaults(self): 

324 @dataclass 

325 class AppConfig: 

326 host: str = "0.0.0.0" 

327 port: int = 8080 

328 

329 cm = ConfigManager(auto_env=False) 

330 cfg = cm.bind(AppConfig) 

331 assert cfg.host == "0.0.0.0" 

332 assert cfg.port == 8080