cleveragents-core/robot/helper_automation_profile_cli.py

#!/usr/bin/env python3
"""Robot Framework helper for automation-profile CLI smoke tests.

Invoked by Robot tests to exercise the automation-profile CLI commands
in isolation using the Typer CliRunner with a per-test in-memory SQLite
database, fully isolated for parallel execution.

The module-level ``_get_service()`` in the production code calls
``get_container()`` which requires full DI wiring.  We monkey-patch it
to return an ``AutomationProfileService`` backed by a fresh in-memory
SQLite ``AutomationProfileRepository`` so the tests can run without
container setup and without colliding across pabot workers.
"""

from __future__ import annotations

import json
import logging
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

# Ensure the 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 structlog/retry noise that pollutes stdout in the
# CliRunner capture buffer.  Must happen before any cleveragents imports
# that configure structlog.
logging.getLogger().setLevel(logging.WARNING)

from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402
from typer.testing import CliRunner  # noqa: E402

import cleveragents.cli.commands.automation_profile as _ap_mod  # noqa: E402
from cleveragents.application.services.automation_profile_service import (  # noqa: E402
    AutomationProfileService,
)
from cleveragents.infrastructure.database.models import Base  # noqa: E402
from cleveragents.infrastructure.database.repositories import (  # noqa: E402
    AutomationProfileRepository,
)

profile_app = _ap_mod.app

_runner = CliRunner()

_VALID_YAML = """\
name: acme/robot-test
description: Robot test profile
schema_version: "1.0"
decompose_task: 0.8
create_tool: 0.7
select_tool: 1.0
edit_code: 0.6
execute_command: 0.8
create_file: 0.7
delete_content: 0.8
access_network: 0.9
install_dependency: 0.7
modify_config: 0.0
approve_plan: 0.6
safety:
  require_sandbox: true
  require_checkpoints: true
  allow_unsafe_tools: false
"""


def _extract_json(text: str) -> object:
    """Extract and parse the first valid JSON object or array from *text*.

    ``format_output`` writes JSON directly to ``sys.stdout`` which the
    CliRunner captures.  Structlog debug messages may precede the JSON
    payload.  This function scans for the first ``{`` or ``[`` and
    parses from there.
    """
    for i, ch in enumerate(text):
        if ch in ("{", "["):
            try:
                return json.loads(text[i:])
            except json.JSONDecodeError:
                continue
    raise ValueError(f"No valid JSON found in output:\n{text[:500]}")


def _make_isolated_service() -> AutomationProfileService:
    """Create a fully isolated AutomationProfileService with an in-memory DB."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    repo = AutomationProfileRepository(session_factory=factory, auto_commit=True)
    return AutomationProfileService(repo=repo)


# Module-level service instance that is reset before each test.
_isolated_service: AutomationProfileService | None = None


def _get_service_override() -> AutomationProfileService:
    """Replacement for ``_ap_mod._get_service`` that returns the test service."""
    assert _isolated_service is not None, "Call _reset_service() first"
    return _isolated_service


def _reset_service() -> None:
    """Create a fresh isolated service for the next test."""
    global _isolated_service
    _isolated_service = _make_isolated_service()


def _invoke(args: list[str]):
    """Invoke the profile CLI with the patched service."""
    with patch.object(_ap_mod, "_get_service", _get_service_override):
        return _runner.invoke(profile_app, args)


def test_add_profile() -> None:
    """Test adding a profile via YAML config."""
    _reset_service()
    fd, path = tempfile.mkstemp(suffix=".yaml")
    try:
        with os.fdopen(fd, "w") as fh:
            fh.write(_VALID_YAML)
        result = _invoke(["add", "--config", path])
        assert result.exit_code == 0, f"add failed: {result.output}"
        print("add-profile-ok")
    finally:
        Path(path).unlink(missing_ok=True)


def test_show_profile() -> None:
    """Test showing a built-in profile."""
    _reset_service()
    result = _invoke(["show", "manual"])
    assert result.exit_code == 0, f"show failed: {result.output}"
    assert "manual" in result.output
    print("show-profile-ok")


def test_show_json() -> None:
    """Test showing a profile in JSON format."""
    _reset_service()
    result = _invoke(["show", "manual", "--format", "json"])
    assert result.exit_code == 0, f"show json failed: {result.output}"
    parsed = _extract_json(result.output)
    assert isinstance(parsed, dict)
    assert parsed["name"] == "manual"
    print("show-json-ok")


def test_show_yaml() -> None:
    """Test showing a profile in YAML format."""
    _reset_service()
    result = _invoke(["show", "manual", "--format", "yaml"])
    assert result.exit_code == 0, f"show yaml failed: {result.output}"
    assert "name: manual" in result.output
    print("show-yaml-ok")


def test_list_profiles() -> None:
    """Test listing all profiles."""
    _reset_service()
    result = _invoke(["list"])
    assert result.exit_code == 0, f"list failed: {result.output}"
    assert "manual" in result.output
    print("list-profiles-ok")


def test_list_json() -> None:
    """Test listing profiles in JSON format.

    Per docs/specification.md lines 16998-17017, the output must be a dict
    with a ``profiles`` key (list of simplified entries) and a ``summary``
    key with built_in/custom/total counts.
    """
    _reset_service()
    result = _invoke(["list", "--format", "json"])
    assert result.exit_code == 0, f"list json failed: {result.output}"
    parsed = _extract_json(result.output)
    assert isinstance(parsed, dict), (
        f"Expected dict at top level, got {type(parsed).__name__}. "
        f"Output: {result.output[:300]}"
    )
    # Validate profiles wrapper
    assert "profiles" in parsed, f"Missing 'profiles' key. Keys: {list(parsed.keys())}"
    profiles = parsed["profiles"]
    assert isinstance(profiles, list), (
        f"'profiles' must be a list, got {type(profiles)}"
    )
    assert len(profiles) >= 8, f"Expected >= 8 built-in profiles, got {len(profiles)}"
    # Validate each entry has only spec-required fields
    for entry in profiles:
        assert "name" in entry, f"Profile entry missing 'name': {entry}"
        assert "select_tool" in entry, f"Profile entry missing 'select_tool': {entry}"
        assert "sandbox" in entry, f"Profile entry missing 'sandbox': {entry}"
        assert "phase_transitions" not in entry, (
            "Full profile dict leaked into list output (found 'phase_transitions')"
        )
    # Validate summary
    assert "summary" in parsed, f"Missing 'summary' key. Keys: {list(parsed.keys())}"
    summary = parsed["summary"]
    assert "built_in" in summary, f"Missing 'built_in' in summary: {summary}"
    assert "total" in summary, f"Missing 'total' in summary: {summary}"
    assert summary["built_in"] >= 8, (
        f"Expected >= 8 built-in profiles in summary, got {summary['built_in']}"
    )
    print("list-json-ok")


def test_remove_profile() -> None:
    """Test removing a custom profile - verifies Profile Removed panel is rendered."""
    _reset_service()
    # First add a profile
    fd, path = tempfile.mkstemp(suffix=".yaml")
    try:
        with os.fdopen(fd, "w") as fh:
            fh.write(_VALID_YAML)
        add_result = _invoke(["add", "--config", path])
        assert add_result.exit_code == 0, f"add failed: {add_result.output}"
    finally:
        Path(path).unlink(missing_ok=True)

    # Then remove it (reuse same service so the profile persists)
    result = _invoke(["remove", "acme/robot-test", "--yes"])
    assert result.exit_code == 0, f"remove failed: {result.output}"
    # Verify the spec-required Profile Removed panel is rendered
    assert "Profile Removed" in result.output, (
        f"Expected 'Profile Removed' panel in output. Got: {result.output}"
    )
    assert "acme/robot-test" in result.output, (
        f"Expected profile name in panel output. Got: {result.output}"
    )
    assert "OK" in result.output, (
        f"Expected 'OK' success message in output. Got: {result.output}"
    )
    print("remove-profile-ok")


def test_remove_builtin_fails() -> None:
    """Test that removing a built-in profile fails."""
    _reset_service()
    result = _invoke(["remove", "manual", "--yes"])
    assert result.exit_code != 0, f"remove should have failed: {result.output}"
    print("remove-builtin-fails-ok")


if __name__ == "__main__":
    command = sys.argv[1] if len(sys.argv) > 1 else "all"

    tests = {
        "add": test_add_profile,
        "show": test_show_profile,
        "show-json": test_show_json,
        "show-yaml": test_show_yaml,
        "list": test_list_profiles,
        "list-json": test_list_json,
        "remove": test_remove_profile,
        "remove-builtin-fails": test_remove_builtin_fails,
    }

    if command == "all":
        for _name, test_fn in tests.items():
            test_fn()
        print("all-tests-ok")
    elif command in tests:
        tests[command]()
    else:
        print(f"Unknown test: {command}", file=sys.stderr)
        sys.exit(1)