#!/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.""" _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, list) assert len(parsed) >= 8 # At least 8 built-in profiles print("list-json-ok") def test_remove_profile() -> None: """Test removing a custom profile.""" _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}" assert "removed" in result.output.lower() 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)