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