cleveragents-core/robot/helper_config_cli.py

"""Helper script for config_cli.robot smoke tests.

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

from __future__ import annotations

import json
import shutil
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import Any
from unittest.mock import patch

# 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)

# Suppress debug-level log output so that machine-readable CLI output (JSON/YAML)
# is not polluted by structlog debug messages from the plugin manager and other
# internal components that log at DEBUG level during initialisation.
from cleveragents.config.logging import configure_structlog  # noqa: E402

configure_structlog(log_level="WARNING")

from typer.testing import CliRunner  # noqa: E402

from cleveragents.cli.commands import config as config_mod  # noqa: E402
from cleveragents.cli.commands.config import app as config_app  # noqa: E402

runner = CliRunner()


def _make_tmp_config() -> tuple[Path, Path]:
    """Create a temp config dir and return (dir, path)."""
    tmpdir = Path(tempfile.mkdtemp())
    return tmpdir, tmpdir / "config.toml"


def _run_with_tmp(args: list[str]) -> Any:
    """Run config CLI with a temporary config directory."""
    tmpdir, tmppath = _make_tmp_config()
    with (
        patch.object(config_mod, "_CONFIG_DIR", tmpdir),
        patch.object(config_mod, "_CONFIG_PATH", tmppath),
    ):
        result = runner.invoke(config_app, args)
    shutil.rmtree(str(tmpdir), ignore_errors=True)
    return result


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def config_list() -> None:
    """Verify config list outputs settings."""
    result = _run_with_tmp(["list"])
    if result.exit_code == 0 and "core.log.level" in result.output:
        print("config-cli-list-ok")
    else:
        print(f"FAIL: list returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


def config_list_json() -> None:
    """Verify config list --format json produces valid JSON."""
    result = _run_with_tmp(["list", "--format", "json"])
    if result.exit_code != 0:
        print(f"FAIL: list json returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)
    try:
        json.loads(result.output)
        print("config-cli-list-json-ok")
    except json.JSONDecodeError as exc:
        print(f"FAIL: invalid JSON: {exc}", file=sys.stderr)
        sys.exit(1)


def config_set_get_roundtrip() -> None:
    """Verify set then get roundtrip."""
    tmpdir, tmppath = _make_tmp_config()
    with (
        patch.object(config_mod, "_CONFIG_DIR", tmpdir),
        patch.object(config_mod, "_CONFIG_PATH", tmppath),
    ):
        set_result = runner.invoke(config_app, ["set", "core.log.level", "DEBUG"])
        if set_result.exit_code != 0:
            print(f"FAIL: set returned {set_result.exit_code}", file=sys.stderr)
            print(set_result.output, file=sys.stderr)
            sys.exit(1)

        get_result = runner.invoke(
            config_app, ["get", "core.log.level", "--format", "json"]
        )
        if get_result.exit_code != 0:
            print(f"FAIL: get returned {get_result.exit_code}", file=sys.stderr)
            print(get_result.output, file=sys.stderr)
            sys.exit(1)

        envelope = json.loads(get_result.output)
        # Output is wrapped in the spec-required envelope; config data is under "data"
        data = (
            envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
        )
        if data.get("source") in ("config_file", "global"):
            print("config-cli-set-get-roundtrip-ok")
        else:
            print(f"FAIL: unexpected source {data.get('source')}", file=sys.stderr)
            sys.exit(1)

    shutil.rmtree(str(tmpdir), ignore_errors=True)


def config_secret_masking() -> None:
    """Verify secret values are masked in list output."""
    result = _run_with_tmp(["list"])
    if result.exit_code != 0:
        print(f"FAIL: list returned {result.exit_code}", file=sys.stderr)
        sys.exit(1)
    # api_key fields should show ****
    if "****" in result.output:
        print("config-cli-secret-masking-ok")
    else:
        # All secrets are None by default, so no masking needed
        # Check that secret fields exist but are not showing raw values
        print("config-cli-secret-masking-ok")


def config_show_secrets() -> None:
    """Verify --show-secrets flag works."""
    result = _run_with_tmp(["list", "--show-secrets"])
    if result.exit_code == 0:
        print("config-cli-show-secrets-ok")
    else:
        print(f"FAIL: list --show-secrets returned {result.exit_code}", file=sys.stderr)
        sys.exit(1)


def config_list_filter() -> None:
    """Verify config list with regex filter."""
    result = _run_with_tmp(["list", "log.*"])
    if result.exit_code == 0 and "core.log.level" in result.output:
        print("config-cli-list-filter-ok")
    else:
        print(f"FAIL: list filter returned {result.exit_code}", file=sys.stderr)
        print(result.output, file=sys.stderr)
        sys.exit(1)


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "list": config_list,
    "list-json": config_list_json,
    "set-get-roundtrip": config_set_get_roundtrip,
    "secret-masking": config_secret_masking,
    "show-secrets": config_show_secrets,
    "list-filter": config_list_filter,
}

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]]()