cleveragents-core/robot/helper_config_resolution.py

"""Helper script for config_resolution.robot end-to-end tests.

Each subcommand is a self-contained check that prints a sentinel on success.
"""

from __future__ import annotations

import os
import shutil
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.application.services.config_service import (  # noqa: E402
    ConfigLevel,
    ConfigService,
)


def _make_service() -> tuple[ConfigService, Path]:
    """Create a ConfigService with a temporary directory."""
    tmpdir = Path(tempfile.mkdtemp())
    svc = ConfigService(config_dir=tmpdir, config_path=tmpdir / "config.toml")
    return svc, tmpdir


def resolve_default() -> None:
    """Verify default resolution works."""
    svc, tmpdir = _make_service()
    try:
        result = svc.resolve("core.log_level")
        if result.value == "INFO" and result.source == ConfigLevel.DEFAULT:
            print("config-resolution-default-ok")
        else:
            print(
                f"FAIL: expected INFO/default, got {result.value}/{result.source}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        shutil.rmtree(str(tmpdir), ignore_errors=True)


def resolve_global() -> None:
    """Verify global config file overrides default."""
    svc, tmpdir = _make_service()
    try:
        svc.set_value("core.log_level", "DEBUG")
        result = svc.resolve("core.log_level")
        if result.value == "DEBUG" and result.source == ConfigLevel.GLOBAL:
            print("config-resolution-global-ok")
        else:
            print(
                f"FAIL: expected DEBUG/global, got {result.value}/{result.source}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        shutil.rmtree(str(tmpdir), ignore_errors=True)


def resolve_env() -> None:
    """Verify env var overrides global config."""
    svc, tmpdir = _make_service()
    env_key = "CLEVERAGENTS_CORE_LOG_LEVEL"
    old_val = os.environ.get(env_key)
    try:
        svc.set_value("core.log_level", "DEBUG")
        os.environ[env_key] = "WARNING"
        result = svc.resolve("core.log_level")
        if result.value == "WARNING" and result.source == ConfigLevel.ENV_VAR:
            print("config-resolution-env-ok")
        else:
            print(
                f"FAIL: expected WARNING/env_var, got {result.value}/{result.source}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        if old_val is None:
            os.environ.pop(env_key, None)
        else:
            os.environ[env_key] = old_val
        shutil.rmtree(str(tmpdir), ignore_errors=True)


def resolve_cli() -> None:
    """Verify CLI flag overrides env var."""
    svc, tmpdir = _make_service()
    env_key = "CLEVERAGENTS_CORE_LOG_LEVEL"
    old_val = os.environ.get(env_key)
    try:
        os.environ[env_key] = "WARNING"
        result = svc.resolve("core.log_level", cli_value="ERROR")
        if result.value == "ERROR" and result.source == ConfigLevel.CLI_FLAG:
            print("config-resolution-cli-ok")
        else:
            print(
                f"FAIL: expected ERROR/cli_flag, got {result.value}/{result.source}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        if old_val is None:
            os.environ.pop(env_key, None)
        else:
            os.environ[env_key] = old_val
        shutil.rmtree(str(tmpdir), ignore_errors=True)


def resolve_unknown_key() -> None:
    """Verify unknown key raises ValueError."""
    svc, tmpdir = _make_service()
    try:
        svc.resolve("bogus.nonexistent")
        print("FAIL: expected ValueError", file=sys.stderr)
        sys.exit(1)
    except ValueError:
        print("config-resolution-unknown-key-ok")
    finally:
        shutil.rmtree(str(tmpdir), ignore_errors=True)


def resolve_type_coercion() -> None:
    """Verify integer type coercion from env var."""
    svc, tmpdir = _make_service()
    env_key = "CLEVERAGENTS_CORE_SERVER_PORT"
    old_val = os.environ.get(env_key)
    try:
        os.environ[env_key] = "9090"
        result = svc.resolve("core.server_port")
        if result.value == 9090 and isinstance(result.value, int):
            print("config-resolution-type-coercion-ok")
        else:
            print(
                f"FAIL: expected 9090 (int), got {result.value} "
                f"({type(result.value).__name__})",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        if old_val is None:
            os.environ.pop(env_key, None)
        else:
            os.environ[env_key] = old_val
        shutil.rmtree(str(tmpdir), ignore_errors=True)


def registry_keys() -> None:
    """Verify the registry contains expected keys."""
    registry = ConfigService.registry()
    expected = ["core.log_level", "plan.max_retries", "provider.temperature"]
    for key in expected:
        if key not in registry:
            print(f"FAIL: {key} not in registry", file=sys.stderr)
            sys.exit(1)
    print("config-resolution-registry-ok")


def env_var_convention() -> None:
    """Verify env var naming convention."""
    env_name = ConfigService.env_var_for_key("core.log_level")
    if env_name == "CLEVERAGENTS_CORE_LOG_LEVEL":
        print("config-resolution-env-convention-ok")
    else:
        print(
            f"FAIL: expected CLEVERAGENTS_CORE_LOG_LEVEL, got {env_name}",
            file=sys.stderr,
        )
        sys.exit(1)


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], None]] = {
    "resolve-default": resolve_default,
    "resolve-global": resolve_global,
    "resolve-env": resolve_env,
    "resolve-cli": resolve_cli,
    "resolve-unknown-key": resolve_unknown_key,
    "resolve-type-coercion": resolve_type_coercion,
    "registry-keys": registry_keys,
    "env-var-convention": env_var_convention,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        sys.exit(2)
    _COMMANDS[sys.argv[1]]()