cleveragents-core/robot/helper_config_project_scope.py

"""Helper script for config_project_scope.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)

from typer.testing import CliRunner  # noqa: E402

from cleveragents.application.services.config_service import (  # noqa: E402
    ConfigLevel,
    ConfigService,
)
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()


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


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


def _cleanup(tmpdir: Path) -> None:
    """Remove the temp directory ignoring errors."""
    shutil.rmtree(str(tmpdir), ignore_errors=True)


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


def set_get_roundtrip() -> None:
    """Verify project-scoped set followed by get returns the value."""
    svc, tmpdir = _make_service()
    try:
        svc.set_project_value("my-project", "core.automation-profile", "manual")
        result = svc.resolve("core.automation-profile", project_name="my-project")
        if result.value == "manual" and result.source == ConfigLevel.PROJECT:
            print("config-project-scope-set-get-ok")
        else:
            print(
                f"FAIL: expected manual/project, got {result.value}/{result.source}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        _cleanup(tmpdir)


def project_overrides_global() -> None:
    """Verify that project-scoped value overrides global in resolution."""
    svc, tmpdir = _make_service()
    try:
        svc.set_value("core.automation-profile", "supervised")
        svc.set_project_value("my-project", "core.automation-profile", "full-auto")
        result = svc.resolve("core.automation-profile", project_name="my-project")
        if result.value == "full-auto" and result.source == ConfigLevel.PROJECT:
            print("config-project-overrides-global-ok")
        else:
            print(
                f"FAIL: expected full-auto/project, got {result.value}/{result.source}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        _cleanup(tmpdir)


def list_overrides_only() -> None:
    """Verify get_project_overrides returns only project-scoped keys."""
    svc, tmpdir = _make_service()
    try:
        svc.set_value("core.log.level", "DEBUG")
        svc.set_project_value("proj-a", "core.automation-profile", "trusted")
        svc.set_project_value("proj-a", "plan.concurrency", "8")
        overrides = svc.get_project_overrides("proj-a")
        if len(overrides) == 2 and "core.log.level" not in overrides:
            print("config-project-list-overrides-ok")
        else:
            print(
                f"FAIL: expected 2 keys without core.log.level, got {overrides}",
                file=sys.stderr,
            )
            sys.exit(1)
    finally:
        _cleanup(tmpdir)


def cli_roundtrip() -> None:
    """Verify CLI --project flag for set and get."""
    tmpdir = Path(tempfile.mkdtemp())
    tmppath = tmpdir / "config.toml"
    try:
        with (
            patch.object(config_mod, "_CONFIG_DIR", tmpdir),
            patch.object(config_mod, "_CONFIG_PATH", tmppath),
        ):
            set_result = runner.invoke(
                config_app,
                ["set", "core.automation-profile", "manual", "--project", "cli-proj"],
            )
            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.automation-profile",
                    "--project",
                    "cli-proj",
                    "--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)

            data: dict[str, Any] = json.loads(get_result.output)
            if data.get("source") == "project":
                print("config-project-cli-roundtrip-ok")
            else:
                print(
                    f"FAIL: unexpected source {data.get('source')}",
                    file=sys.stderr,
                )
                sys.exit(1)
    finally:
        _cleanup(tmpdir)


def non_scopable_rejected() -> None:
    """Verify that a non-project-scopable key is rejected."""
    svc, tmpdir = _make_service()
    try:
        try:
            svc.set_project_value("proj", "core.data-dir", "/tmp/bad")
            print("FAIL: expected ValueError", file=sys.stderr)
            sys.exit(1)
        except ValueError as exc:
            if "not project-scopable" in str(exc):
                print("config-project-non-scopable-ok")
            else:
                print(f"FAIL: unexpected error: {exc}", file=sys.stderr)
                sys.exit(1)
    finally:
        _cleanup(tmpdir)


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "set-get-roundtrip": set_get_roundtrip,
    "project-overrides-global": project_overrides_global,
    "list-overrides-only": list_overrides_only,
    "cli-roundtrip": cli_roundtrip,
    "non-scopable-rejected": non_scopable_rejected,
}

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