fix: updated tests to override env variables they need instead of relying on the current values from .env
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
"""Centralized environment variable access for Zen MCP Server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from dotenv import dotenv_values, load_dotenv
|
|
except ImportError: # pragma: no cover - optional dependency
|
|
dotenv_values = None # type: ignore[assignment]
|
|
load_dotenv = None # type: ignore[assignment]
|
|
|
|
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
_ENV_PATH = _PROJECT_ROOT / ".env"
|
|
|
|
_DOTENV_VALUES: dict[str, str | None] = {}
|
|
_FORCE_ENV_OVERRIDE = False
|
|
|
|
|
|
def _read_dotenv_values() -> dict[str, str | None]:
|
|
if dotenv_values is not None and _ENV_PATH.exists():
|
|
loaded = dotenv_values(_ENV_PATH)
|
|
return dict(loaded)
|
|
return {}
|
|
|
|
|
|
def _compute_force_override(values: Mapping[str, str | None]) -> bool:
|
|
raw = (values.get("ZEN_MCP_FORCE_ENV_OVERRIDE") or "false").strip().lower()
|
|
return raw == "true"
|
|
|
|
|
|
def reload_env(dotenv_mapping: Mapping[str, str | None] | None = None) -> None:
|
|
"""Reload .env values and recompute override semantics.
|
|
|
|
Args:
|
|
dotenv_mapping: Optional mapping used instead of reading the .env file.
|
|
Intended for tests; when provided, load_dotenv is not invoked.
|
|
"""
|
|
|
|
global _DOTENV_VALUES, _FORCE_ENV_OVERRIDE
|
|
|
|
if dotenv_mapping is not None:
|
|
_DOTENV_VALUES = dict(dotenv_mapping)
|
|
_FORCE_ENV_OVERRIDE = _compute_force_override(_DOTENV_VALUES)
|
|
return
|
|
|
|
_DOTENV_VALUES = _read_dotenv_values()
|
|
_FORCE_ENV_OVERRIDE = _compute_force_override(_DOTENV_VALUES)
|
|
|
|
if load_dotenv is not None and _ENV_PATH.exists():
|
|
load_dotenv(dotenv_path=_ENV_PATH, override=_FORCE_ENV_OVERRIDE)
|
|
|
|
|
|
reload_env()
|
|
|
|
|
|
def env_override_enabled() -> bool:
|
|
"""Return True when ZEN_MCP_FORCE_ENV_OVERRIDE is enabled via the .env file."""
|
|
|
|
return _FORCE_ENV_OVERRIDE
|
|
|
|
|
|
def get_env(key: str, default: str | None = None) -> str | None:
|
|
"""Retrieve environment variables respecting ZEN_MCP_FORCE_ENV_OVERRIDE."""
|
|
|
|
if env_override_enabled():
|
|
if key in _DOTENV_VALUES:
|
|
value = _DOTENV_VALUES[key]
|
|
return value if value is not None else default
|
|
return default
|
|
|
|
return os.getenv(key, default)
|
|
|
|
|
|
def get_env_bool(key: str, default: bool = False) -> bool:
|
|
"""Boolean helper that respects override semantics."""
|
|
|
|
raw_default = "true" if default else "false"
|
|
raw_value = get_env(key, raw_default)
|
|
return (raw_value or raw_default).strip().lower() == "true"
|
|
|
|
|
|
def get_all_env() -> dict[str, str | None]:
|
|
"""Expose the loaded .env mapping for diagnostics/logging."""
|
|
|
|
return dict(_DOTENV_VALUES)
|