feat(agent): load config from environment with validation

This commit is contained in:
2026-08-23 16:44:33 -05:00
parent 0284ab800a
commit 46365468fe
2 changed files with 98 additions and 0 deletions
+45
View File
@@ -0,0 +1,45 @@
"""Agent configuration loaded from environment variables."""
from __future__ import annotations
import os
from collections.abc import Mapping
from dataclasses import dataclass
class ConfigError(ValueError):
"""A required environment variable is missing or invalid."""
@dataclass(frozen=True)
class AgentConfig:
db_host: str
db_port: int
db_user: str
db_password: str
agent_db_name: str
firefly_url: str
firefly_token: str
heartbeat_interval_minutes: int
def _require(env: Mapping[str, str], name: str) -> str:
value = env.get(name, "").strip()
if not value:
raise ConfigError(f"missing required environment variable: {name}")
return value
def load_config(env: Mapping[str, str] | None = None) -> AgentConfig:
if env is None:
env = os.environ
return AgentConfig(
db_host=_require(env, "DB_HOST"),
db_port=int(env.get("DB_PORT", "5432")),
db_user=_require(env, "POSTGRES_USER"),
db_password=_require(env, "POSTGRES_PASSWORD"),
agent_db_name=env.get("AGENT_DB_NAME", "agentdb"),
firefly_url=_require(env, "FIREFLY_API_URL"),
firefly_token=_require(env, "AGENT_FIREFLY_TOKEN"),
heartbeat_interval_minutes=int(env.get("HEARTBEAT_INTERVAL_MINUTES", "60")),
)
+53
View File
@@ -0,0 +1,53 @@
import pytest
from agent.config import AgentConfig, ConfigError, load_config
FULL_ENV = {
"DB_HOST": "db",
"DB_PORT": "5433",
"POSTGRES_USER": "firefly",
"POSTGRES_PASSWORD": "s3cret",
"AGENT_DB_NAME": "agentdb2",
"FIREFLY_API_URL": "http://app:8080",
"AGENT_FIREFLY_TOKEN": "tok123",
"HEARTBEAT_INTERVAL_MINUTES": "15",
}
MINIMAL_ENV = {
"DB_HOST": "db",
"POSTGRES_USER": "firefly",
"POSTGRES_PASSWORD": "s3cret",
"FIREFLY_API_URL": "http://app:8080",
"AGENT_FIREFLY_TOKEN": "tok123",
}
def test_load_config_reads_all_vars():
cfg = load_config(FULL_ENV)
assert cfg == AgentConfig(
db_host="db",
db_port=5433,
db_user="firefly",
db_password="s3cret",
agent_db_name="agentdb2",
firefly_url="http://app:8080",
firefly_token="tok123",
heartbeat_interval_minutes=15,
)
def test_defaults_applied():
cfg = load_config(MINIMAL_ENV)
assert cfg.db_port == 5432
assert cfg.agent_db_name == "agentdb"
assert cfg.heartbeat_interval_minutes == 60
@pytest.mark.parametrize(
"missing",
["DB_HOST", "POSTGRES_USER", "POSTGRES_PASSWORD", "FIREFLY_API_URL", "AGENT_FIREFLY_TOKEN"],
)
def test_missing_required_var_names_the_var(missing):
env = {k: v for k, v in MINIMAL_ENV.items() if k != missing}
with pytest.raises(ConfigError, match=missing):
load_config(env)