Files
cleveragents-core/robot/helper_cli_extensions.py
T
brent.edwards f3ddb48faf
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 16s
CI / quality (pull_request) Successful in 18s
CI / build (pull_request) Successful in 17s
CI / typecheck (pull_request) Successful in 32s
CI / security (pull_request) Successful in 33s
CI / unit_tests (pull_request) Has been cancelled
CI / integration_tests (pull_request) Has been cancelled
CI / benchmark-regression (pull_request) Has been cancelled
CI / coverage (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled
chore(pr): address code review feedback for PR #420
Added CHANGELOG entry for CLI extension test suite. Added Brent Edwards
to CONTRIBUTORS.md. Replaced broad except Exception with specific
ValidationError catch in benchmark. Moved import json to module
top-level in robot helper script.

ISSUES CLOSED: #326
2026-02-25 02:55:40 +00:00

424 lines
14 KiB
Python

"""Helper script for cli_extensions.robot smoke tests.
Each subcommand is a self-contained check that prints a sentinel on success.
"""
from __future__ import annotations
import json
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, 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.cli.commands.action import app as action_app # noqa: E402
from cleveragents.cli.commands.plan import app as plan_app # noqa: E402
from cleveragents.domain.models.core.action import ( # noqa: E402
Action,
ActionState,
)
from cleveragents.domain.models.core.plan import ( # noqa: E402
AutomationProfileProvenance,
AutomationProfileRef,
InvariantSource,
NamespacedName,
Plan,
PlanIdentity,
PlanInvariant,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"
def _mock_plan(
*,
automation_profile: AutomationProfileRef | None = None,
invariants: list[PlanInvariant] | None = None,
) -> Plan:
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=_PLAN_ULID),
namespaced_name=NamespacedName.parse("local/test-plan"),
description="Test plan for robot",
definition_of_done="All tests pass",
action_name="local/test-action",
phase=PlanPhase.STRATEGIZE,
processing_state=ProcessingState.QUEUED,
project_links=[ProjectLink(project_name="my-project")],
arguments={},
arguments_order=[],
automation_profile=automation_profile,
invariants=invariants or [],
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
created_by=None,
timestamps=PlanTimestamps(created_at=now, updated_at=now),
)
def _mock_action() -> Action:
return Action(
namespaced_name=NamespacedName.parse("local/test-action"),
description="Test action for robot",
long_description=None,
definition_of_done="All tests pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
reusable=True,
read_only=False,
state=ActionState.AVAILABLE,
created_by=None,
created_at=datetime.now(),
updated_at=datetime.now(),
)
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def plan_use_with_invariants() -> None:
"""Verify plan use with --invariant flags works."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan(
invariants=[
PlanInvariant(text="No warnings", source=InvariantSource.PLAN),
]
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--invariant",
"No warnings",
],
)
if result.exit_code == 0:
print("cli-ext-plan-use-invariants-ok")
else:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
def plan_use_with_automation_profile() -> None:
"""Verify plan use with --automation-profile works."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan(
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
)
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--automation-profile",
"trusted",
],
)
if result.exit_code == 0 and "trusted" in result.output:
print("cli-ext-plan-use-profile-ok")
else:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
def plan_use_actor_validation_valid() -> None:
"""Verify valid namespaced actors are accepted."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan()
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--strategy-actor",
"openai/gpt-4",
],
)
if result.exit_code == 0:
print("cli-ext-actor-valid-ok")
else:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
def plan_use_actor_validation_invalid() -> None:
"""Verify invalid actor format is rejected."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan()
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--strategy-actor",
"bad-format",
],
)
if result.exit_code != 0:
print("cli-ext-actor-invalid-ok")
else:
print(f"FAIL: expected non-zero exit, got {result.exit_code}")
sys.exit(1)
def plan_use_combined() -> None:
"""Verify plan use with profile + invariants together."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan(
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
invariants=[
PlanInvariant(text="No regressions", source=InvariantSource.PLAN),
],
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--automation-profile",
"trusted",
"--invariant",
"No regressions",
],
)
if result.exit_code == 0 and "trusted" in result.output:
print("cli-ext-plan-combined-ok")
else:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
def _mock_action_extended() -> Action:
"""Create an action with all optional fields set for extended show tests."""
return Action(
namespaced_name=NamespacedName.parse("local/test-action"),
description="Test action for robot extended",
long_description=None,
definition_of_done="All tests pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
estimation_actor="openai/gpt-4",
invariant_actor="anthropic/claude-3",
reusable=True,
read_only=False,
invariants=["No regressions", "Keep backward compat"],
inputs_schema={
"type": "object",
"properties": {"coverage": {"type": "integer"}},
},
automation_profile="trusted",
state=ActionState.AVAILABLE,
created_by=None,
created_at=datetime.now(),
updated_at=datetime.now(),
)
def action_show_extended() -> None:
"""Verify action show renders optional actors, invariants, inputs_schema."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action_extended()
with patch(
"cleveragents.cli.commands.action._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(action_app, ["show", "local/test-action"])
output = result.output
checks = [
("Estimation Actor" in output or "estimation_actor" in output),
("Invariant Actor" in output or "invariant_actor" in output),
("Invariants" in output or "invariants" in output),
("Inputs Schema" in output or "inputs_schema" in output),
]
if result.exit_code == 0 and all(checks):
print("cli-ext-action-show-extended-ok")
else:
print(f"FAIL: exit={result.exit_code} output={output}")
print(f"checks={checks}")
sys.exit(1)
def action_show_json() -> None:
"""Verify action show JSON output includes optional fields."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action_extended()
with patch(
"cleveragents.cli.commands.action._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
action_app, ["show", "local/test-action", "--format", "json"]
)
if result.exit_code != 0:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
try:
data = json.loads(result.output.strip())
except json.JSONDecodeError as exc:
print(f"FAIL: invalid JSON: {exc}\nOutput: {result.output}")
sys.exit(1)
required_keys = [
"estimation_actor",
"invariant_actor",
"inputs_schema",
"invariants",
]
missing = [k for k in required_keys if k not in data]
if missing:
print(f"FAIL: missing keys {missing} in JSON: {list(data.keys())}")
sys.exit(1)
print("cli-ext-action-show-json-ok")
def invariant_ordering() -> None:
"""Verify that multiple invariant flags preserve insertion order."""
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan(
invariants=[
PlanInvariant(text="Alpha", source=InvariantSource.PLAN),
PlanInvariant(text="Beta", source=InvariantSource.PLAN),
PlanInvariant(text="Gamma", source=InvariantSource.PLAN),
]
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--invariant",
"Alpha",
"--invariant",
"Beta",
"--invariant",
"Gamma",
],
)
if result.exit_code != 0:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
# Verify the service received invariants in order
call_args = mock_svc.use_action.call_args
invariants = call_args.kwargs.get("invariants") or call_args[1].get("invariants")
if invariants is None:
print("FAIL: no invariants passed to use_action")
sys.exit(1)
texts = [inv.text for inv in invariants]
if texts != ["Alpha", "Beta", "Gamma"]:
print(f"FAIL: ordering mismatch: {texts}")
sys.exit(1)
print("cli-ext-invariant-ordering-ok")
def profile_resolution() -> None:
"""Verify all builtin profiles resolve correctly."""
from cleveragents.domain.models.core.automation_profile import BUILTIN_PROFILES
for profile_name in sorted(BUILTIN_PROFILES):
mock_svc = MagicMock()
mock_svc.get_action_by_name.return_value = _mock_action()
mock_svc.use_action.return_value = _mock_plan(
automation_profile=AutomationProfileRef(
profile_name=profile_name,
provenance=AutomationProfileProvenance.PLAN,
)
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_svc,
):
result = runner.invoke(
plan_app,
[
"use",
"local/test-action",
"--automation-profile",
profile_name,
],
)
if result.exit_code != 0:
print(
f"FAIL: profile '{profile_name}' "
f"exit={result.exit_code} "
f"output={result.output}"
)
sys.exit(1)
print("cli-ext-profile-resolution-ok")
# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, object] = {
"plan-use-invariants": plan_use_with_invariants,
"plan-use-profile": plan_use_with_automation_profile,
"actor-valid": plan_use_actor_validation_valid,
"actor-invalid": plan_use_actor_validation_invalid,
"plan-combined": plan_use_combined,
"action-show-extended": action_show_extended,
"action-show-json": action_show_json,
"invariant-ordering": invariant_ordering,
"profile-resolution": profile_resolution,
}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
sys.exit(1)
fn = _COMMANDS[sys.argv[1]]
fn() # type: ignore[operator]