test(cli): add Robot lifecycle CLI coverage

This commit is contained in:
2026-02-19 22:05:16 +00:00
parent 4243ac8cff
commit 4f90d7703b
8 changed files with 1148 additions and 17 deletions
+234
View File
@@ -0,0 +1,234 @@
"""ASV benchmarks for CLI Robot flow fixture setup overhead.
Measures the performance overhead of:
- Fixture setup for action creation (YAML write + parse)
- Mock service initialisation for plan lifecycle commands
- ChangeSet store creation and entry recording
- Full lifecycle fixture setup (action + plan + changeset + sandbox)
"""
from __future__ import annotations
import os
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch
try:
from cleveragents.cli.commands.action import app as action_app
from cleveragents.cli.commands.plan import app as plan_app
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.change import (
ChangeEntry,
ChangeOperation,
InMemoryChangeSetStore,
)
from cleveragents.domain.models.core.plan import (
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.cli.commands.action import app as action_app
from cleveragents.cli.commands.plan import app as plan_app
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.change import (
ChangeEntry,
ChangeOperation,
InMemoryChangeSetStore,
)
from cleveragents.domain.models.core.plan import (
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
from typer.testing import CliRunner
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZBN"
_VALID_YAML = """\
name: local/bench-lifecycle-action
description: Benchmark lifecycle action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: Benchmarks pass
"""
_runner = CliRunner()
def _mock_action(name: str = "local/bench-lifecycle-action") -> Action:
"""Create a minimal Action for benchmarks."""
return Action(
namespaced_name=NamespacedName.parse(name),
description="Benchmark lifecycle action",
long_description=None,
definition_of_done="Benchmarks pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
state=ActionState.AVAILABLE,
reusable=True,
read_only=False,
created_at=datetime.now(),
updated_at=datetime.now(),
created_by=None,
)
def _mock_plan(
phase: PlanPhase = PlanPhase.STRATEGIZE,
state: ProcessingState = ProcessingState.QUEUED,
) -> Plan:
"""Create a minimal Plan for benchmarks."""
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=_PLAN_ULID),
namespaced_name=NamespacedName.parse("local/bench-plan"),
description="Benchmark plan",
definition_of_done="Benchmarks pass",
action_name="local/bench-lifecycle-action",
phase=phase,
processing_state=state,
automation_level=AutomationLevel.MANUAL,
project_links=[ProjectLink(project_name="proj-bench")],
arguments={"target_coverage": 80},
arguments_order=["target_coverage"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
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),
)
class CLIRobotFixtureSetupSuite:
"""Benchmark fixture setup overhead for CLI Robot flow tests."""
def setup(self) -> None:
"""Prepare reusable mock objects."""
self._mock_service = MagicMock()
self._mock_service.create_action.return_value = _mock_action()
self._mock_service.get_action_by_name.return_value = _mock_action()
self._mock_service.use_action.return_value = _mock_plan()
self._mock_service.execute_plan.return_value = _mock_plan(
phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
)
self._mock_service.apply_plan.return_value = _mock_plan(
phase=PlanPhase.APPLY, state=ProcessingState.QUEUED
)
def time_yaml_fixture_write(self) -> None:
"""Benchmark writing a YAML fixture file."""
fd, path = tempfile.mkstemp(suffix=".yaml")
with os.fdopen(fd, "w") as fh:
fh.write(_VALID_YAML)
Path(path).unlink(missing_ok=True)
def time_mock_service_creation(self) -> None:
"""Benchmark creating and configuring a mock service."""
svc = MagicMock()
svc.create_action.return_value = _mock_action()
svc.get_action_by_name.return_value = _mock_action()
svc.use_action.return_value = _mock_plan()
def time_changeset_store_creation(self) -> None:
"""Benchmark creating an InMemoryChangeSetStore and starting a changeset."""
store = InMemoryChangeSetStore()
store.start(_PLAN_ULID)
def time_changeset_entry_recording(self) -> None:
"""Benchmark recording a change entry in the store."""
store = InMemoryChangeSetStore()
cid = store.start(_PLAN_ULID)
store.record(
cid,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-bench",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path="src/bench.py",
),
)
class CLIRobotLifecycleSuite:
"""Benchmark the full CLI lifecycle flow with mocked service."""
def setup(self) -> None:
"""Prepare mock service and YAML file."""
self._mock_service = MagicMock()
self._mock_service.create_action.return_value = _mock_action()
self._mock_service.get_action_by_name.return_value = _mock_action()
self._mock_service.use_action.return_value = _mock_plan()
self._mock_service.execute_plan.return_value = _mock_plan(
phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
)
self._mock_service.apply_plan.return_value = _mock_plan(
phase=PlanPhase.APPLY, state=ProcessingState.QUEUED
)
fd, self._yaml_path = tempfile.mkstemp(suffix=".yaml")
with os.fdopen(fd, "w") as fh:
fh.write(_VALID_YAML)
self._action_patcher = patch(
"cleveragents.cli.commands.action._get_lifecycle_service",
return_value=self._mock_service,
)
self._plan_patcher = patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=self._mock_service,
)
self._action_patcher.start()
self._plan_patcher.start()
def teardown(self) -> None:
"""Clean up patchers and temp file."""
self._action_patcher.stop()
self._plan_patcher.stop()
Path(self._yaml_path).unlink(missing_ok=True)
def time_action_create(self) -> None:
"""Benchmark action create via CLI."""
_runner.invoke(action_app, ["create", "--config", self._yaml_path])
def time_plan_use(self) -> None:
"""Benchmark plan use via CLI."""
_runner.invoke(plan_app, ["use", "local/bench-lifecycle-action", "proj-bench"])
def time_plan_execute(self) -> None:
"""Benchmark plan execute via CLI."""
_runner.invoke(plan_app, ["execute", _PLAN_ULID])
def time_plan_apply(self) -> None:
"""Benchmark plan lifecycle-apply via CLI."""
_runner.invoke(plan_app, ["lifecycle-apply", _PLAN_ULID])
def time_full_lifecycle_flow(self) -> None:
"""Benchmark the full create -> use -> execute -> apply flow."""
_runner.invoke(action_app, ["create", "--config", self._yaml_path])
_runner.invoke(plan_app, ["use", "local/bench-lifecycle-action", "proj-bench"])
_runner.invoke(plan_app, ["execute", _PLAN_ULID])
_runner.invoke(plan_app, ["lifecycle-apply", _PLAN_ULID])
+52
View File
@@ -318,6 +318,58 @@ nox -s integration_tests -- --suite robot/cli_lifecycle_e2e.robot
nox -s benchmark
```
### Robot Suite: `robot/cli_lifecycle.robot`
End-to-end Robot suite for the CLI lifecycle with git_worktree sandbox simulation
and ChangeSet capture verification. Uses `robot/helper_cli_lifecycle.py` as a
Python helper that exercises the full action → plan → execute → apply flow with
mocked service layer, simulated sandbox workspace, and `InMemoryChangeSetStore`
for change tracking.
**Positive tests:**
| Test Case | Description |
|-----------|-------------|
| Action Create Via Config CLI | Creates an action from YAML config |
| Plan Use Creates Strategize Plan With Sandbox | Uses action with sandbox workspace mock |
| Plan Execute With ChangeSet Capture | Executes plan, verifies changeset entry recorded |
| Plan Apply With ChangeSet Verification | Applies plan, verifies changeset summary |
| Full Lifecycle Action To Apply With Sandbox | End-to-end with sandbox + changeset |
**Negative tests:**
| Test Case | Description |
|-----------|-------------|
| Invalid Project Name Rejected | Invalid project name format handled gracefully |
| Missing Resource Returns Error | Non-existent action returns error |
| Invalid Arg Format Rejected | `--arg` without `=` separator rejected |
**Fixtures:**
- `helper_cli_lifecycle.py` creates temporary YAML config files and sandbox directories
- Uses `InMemoryChangeSetStore` for ChangeSet capture without database
- Each subcommand is self-contained; cleanup runs in `finally` blocks
**Aligned Behave feature:** `features/cli_lifecycle_robot_alignment.feature`
mirrors the Robot E2E flow to keep unit/integration expectations aligned.
**ASV benchmark:** `benchmarks/cli_robot_flow_bench.py` measures fixture setup
overhead (`CLIRobotFixtureSetupSuite`) and full lifecycle CLI throughput
(`CLIRobotLifecycleSuite`).
### Running the Robot CLI Lifecycle Suite
```bash
# Robot suite
nox -s integration_tests -- --suite robot/cli_lifecycle.robot
# Aligned Behave feature
nox -s unit_tests -- features/cli_lifecycle_robot_alignment.feature
# Benchmarks
nox -s benchmark
```
## Persistence Test Suites
The persistence layer (SQLAlchemy repositories for plans and actions) has dedicated test suites at both unit and integration levels.
@@ -0,0 +1,28 @@
Feature: CLI lifecycle Robot alignment
As a developer
I want a Behave scenario mirroring the Robot E2E lifecycle flow
So that unit and integration test expectations stay aligned
Background:
Given a robot alignment CLI runner
And a robot alignment mocked lifecycle service
Scenario: Full lifecycle action to apply with changeset verification
Given a robot alignment action "local/lifecycle-action" is created via config
When I run robot alignment plan use "local/lifecycle-action" on project "proj-a"
Then the robot alignment plan use should succeed
When I run robot alignment plan execute for the plan
Then the robot alignment plan execute should succeed
And the robot alignment changeset should have 1 entry with operation "create"
When I run robot alignment plan apply for the plan
Then the robot alignment plan apply should succeed
And the robot alignment changeset summary should have total 2
Scenario: Invalid arg format rejected via CLI
Given a robot alignment action "local/lifecycle-action" exists
When I run robot alignment plan use with invalid arg "bad_arg_no_equals"
Then the robot alignment CLI should reject the invalid arg format
Scenario: Missing action returns error via CLI
When I run robot alignment plan use for nonexistent action "local/nonexistent"
Then the robot alignment CLI should report action not found
@@ -0,0 +1,319 @@
"""Step definitions for CLI lifecycle Robot alignment feature.
Mirrors the Robot E2E lifecycle flow (action -> plan -> execute -> apply)
in Behave to keep unit and integration test expectations aligned.
All step names are prefixed with ``robot alignment`` to avoid
``AmbiguousStep`` conflicts with existing steps.
"""
from __future__ import annotations
import os
import tempfile
from datetime import datetime
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from typer.testing import CliRunner
from cleveragents.cli.commands.action import app as action_app
from cleveragents.cli.commands.plan import app as plan_app
from cleveragents.core.exceptions import NotFoundError
from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.change import (
ChangeEntry,
ChangeOperation,
InMemoryChangeSetStore,
)
from cleveragents.domain.models.core.plan import (
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8T"
_VALID_YAML = """\
name: local/lifecycle-action
description: Lifecycle test action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: All lifecycle tests pass
"""
def _mock_action(name: str = "local/lifecycle-action") -> Action:
"""Create a minimal valid Action for alignment tests."""
return Action(
namespaced_name=NamespacedName.parse(name),
description="Lifecycle test action",
long_description=None,
definition_of_done="All lifecycle tests pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
state=ActionState.AVAILABLE,
reusable=True,
read_only=False,
created_at=datetime.now(),
updated_at=datetime.now(),
created_by=None,
)
def _mock_plan(
phase: PlanPhase = PlanPhase.STRATEGIZE,
state: ProcessingState = ProcessingState.QUEUED,
) -> Plan:
"""Create a minimal valid Plan for alignment tests."""
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=_PLAN_ULID),
namespaced_name=NamespacedName.parse("local/lifecycle-plan"),
description="Lifecycle test plan",
definition_of_done="Tests pass",
action_name="local/lifecycle-action",
phase=phase,
processing_state=state,
automation_level=AutomationLevel.MANUAL,
project_links=[ProjectLink(project_name="proj-a")],
arguments={"target_coverage": 80},
arguments_order=["target_coverage"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
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),
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a robot alignment CLI runner")
def step_robot_alignment_runner(context: Context) -> None:
"""Set up the CLI runner."""
context.ra_runner = CliRunner()
@given("a robot alignment mocked lifecycle service")
def step_robot_alignment_service(context: Context) -> None:
"""Set up a mock PlanLifecycleService for the plan CLI."""
context.ra_mock_service = MagicMock()
context.ra_action_patcher = patch(
"cleveragents.cli.commands.action._get_lifecycle_service",
return_value=context.ra_mock_service,
)
context.ra_plan_patcher = patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=context.ra_mock_service,
)
context.ra_action_patcher.start()
context.ra_plan_patcher.start()
# Set up changeset store for tracking
context.ra_changeset_store = InMemoryChangeSetStore()
context.ra_changeset_id = context.ra_changeset_store.start(_PLAN_ULID)
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers = []
context._cleanup_handlers.append(context.ra_action_patcher.stop)
context._cleanup_handlers.append(context.ra_plan_patcher.stop)
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given('a robot alignment action "{name}" is created via config')
def step_robot_alignment_action_create(context: Context, name: str) -> None:
"""Create action via CLI config and configure mocks."""
action = _mock_action(name)
context.ra_mock_service.create_action.return_value = action
context.ra_mock_service.get_action_by_name.return_value = action
context.ra_mock_service.use_action.return_value = _mock_plan()
context.ra_mock_service.execute_plan.return_value = _mock_plan(
phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
)
context.ra_mock_service.apply_plan.return_value = _mock_plan(
phase=PlanPhase.APPLY, state=ProcessingState.QUEUED
)
fd, yaml_path = tempfile.mkstemp(suffix=".yaml")
with os.fdopen(fd, "w") as fh:
fh.write(_VALID_YAML)
context.ra_yaml_path = yaml_path
result = context.ra_runner.invoke(action_app, ["create", "--config", yaml_path])
context.ra_action_result = result
os.unlink(yaml_path)
assert result.exit_code == 0, (
f"action create failed ({result.exit_code}): {result.output}"
)
@given('a robot alignment action "{name}" exists')
def step_robot_alignment_action_exists(context: Context, name: str) -> None:
"""Set up an existing action mock."""
context.ra_mock_service.get_action_by_name.return_value = _mock_action(name)
context.ra_mock_service.use_action.return_value = _mock_plan()
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when('I run robot alignment plan use "{action}" on project "{project}"')
def step_robot_alignment_plan_use(context: Context, action: str, project: str) -> None:
"""Run plan use."""
context.ra_use_result = context.ra_runner.invoke(plan_app, ["use", action, project])
@when("I run robot alignment plan execute for the plan")
def step_robot_alignment_plan_execute(context: Context) -> None:
"""Run plan execute and record a changeset entry."""
# Record a change entry to mirror the Robot test
context.ra_changeset_store.record(
context.ra_changeset_id,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-a",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path="src/new_module.py",
),
)
context.ra_execute_result = context.ra_runner.invoke(
plan_app, ["execute", _PLAN_ULID]
)
@when("I run robot alignment plan apply for the plan")
def step_robot_alignment_plan_apply(context: Context) -> None:
"""Run plan apply and record an additional changeset entry."""
# Record a modify entry to bring total to 2
context.ra_changeset_store.record(
context.ra_changeset_id,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-a",
tool_name="builtin/file-edit",
operation=ChangeOperation.MODIFY,
path="src/existing.py",
),
)
context.ra_apply_result = context.ra_runner.invoke(
plan_app, ["lifecycle-apply", _PLAN_ULID]
)
@when('I run robot alignment plan use with invalid arg "{arg_str}"')
def step_robot_alignment_invalid_arg(context: Context, arg_str: str) -> None:
"""Run plan use with an invalid --arg format."""
context.ra_invalid_arg_result = context.ra_runner.invoke(
plan_app,
["use", "local/lifecycle-action", "proj-a", "--arg", arg_str],
)
@when('I run robot alignment plan use for nonexistent action "{action}"')
def step_robot_alignment_missing_action(context: Context, action: str) -> None:
"""Run plan use for an action that doesn't exist."""
context.ra_mock_service.get_action_by_name.side_effect = NotFoundError(
f"Action '{action}' not found"
)
context.ra_missing_result = context.ra_runner.invoke(
plan_app, ["use", action, "proj-a"]
)
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the robot alignment plan use should succeed")
def step_robot_alignment_plan_use_ok(context: Context) -> None:
"""Verify plan use succeeded."""
assert context.ra_use_result.exit_code == 0, (
f"plan use failed ({context.ra_use_result.exit_code}): "
f"{context.ra_use_result.output}"
)
@then("the robot alignment plan execute should succeed")
def step_robot_alignment_plan_execute_ok(context: Context) -> None:
"""Verify plan execute succeeded."""
assert context.ra_execute_result.exit_code == 0, (
f"plan execute failed ({context.ra_execute_result.exit_code}): "
f"{context.ra_execute_result.output}"
)
@then('the robot alignment changeset should have {count:d} entry with operation "{op}"')
def step_robot_alignment_changeset_entry(context: Context, count: int, op: str) -> None:
"""Verify changeset entries."""
cs = context.ra_changeset_store.get(context.ra_changeset_id)
assert cs is not None, "changeset not found"
assert len(cs.entries) == count, f"expected {count} entries, got {len(cs.entries)}"
assert cs.entries[0].operation == op, (
f"expected operation={op}, got {cs.entries[0].operation}"
)
@then("the robot alignment plan apply should succeed")
def step_robot_alignment_plan_apply_ok(context: Context) -> None:
"""Verify plan apply succeeded."""
assert context.ra_apply_result.exit_code == 0, (
f"plan apply failed ({context.ra_apply_result.exit_code}): "
f"{context.ra_apply_result.output}"
)
@then("the robot alignment changeset summary should have total {total:d}")
def step_robot_alignment_changeset_total(context: Context, total: int) -> None:
"""Verify changeset summary total."""
summary = context.ra_changeset_store.summarize(context.ra_changeset_id)
assert summary.get("total") == total, (
f"expected total={total}, got {summary.get('total')}"
)
@then("the robot alignment CLI should reject the invalid arg format")
def step_robot_alignment_invalid_arg_rejected(context: Context) -> None:
"""Verify the CLI rejected the invalid arg format."""
result = context.ra_invalid_arg_result
output_lower = result.output.lower()
assert (
result.exit_code != 0
or "invalid argument format" in output_lower
or "aborted" in output_lower
), f"expected rejection, got exit_code={result.exit_code}: {result.output}"
@then("the robot alignment CLI should report action not found")
def step_robot_alignment_action_not_found(context: Context) -> None:
"""Verify the CLI reported action not found."""
result = context.ra_missing_result
output_lower = result.output.lower()
assert (
result.exit_code != 0 or "not found" in output_lower or "error" in output_lower
), f"expected not-found error, got exit_code={result.exit_code}: {result.output}"
+16 -16
View File
@@ -1905,22 +1905,22 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [X] Quality [Brent]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [X] Git [Brent]: `git add .`
- [X] Git [Brent]: `git commit -m "test(cli): expand lifecycle command coverage"`
- [ ] Forgejo PR [Brent]: Open PR from `feature/m1-cli-tests` to `master` with description "Expand Behave + Robot CLI coverage for action/plan lifecycle commands and error paths.".
- [ ] Git [Brent]: `git checkout master`
- [ ] Git [Brent]: `git pull origin master`
- [ ] Git [Brent]: `git checkout -b feature/m1-cli-tests-robot`
- [ ] Git [Brent]: `git fetch origin && git merge origin/master`
- [ ] Tests (Robot) [Brent]: Add end-to-end Robot suite for action → plan → execute → apply on a local repo with git_worktree sandbox + ChangeSet capture verification.
- [ ] Tests (Robot) [Brent]: Add negative Robot cases for invalid project names, missing resources, and invalid `--arg` values to match CLI error outputs.
- [ ] Tests (Behave) [Brent]: Add one CLI scenario mirroring the Robot E2E flow to keep unit/integration expectations aligned.
- [ ] Docs [Brent]: Update `docs/development/testing.md` with Robot CLI suite entry and fixtures.
- [ ] Tests (ASV) [Brent]: Add `benchmarks/cli_robot_flow_bench.py` for CLI flow fixture setup overhead.
- [ ] Quality [Brent]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Quality [Brent]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [ ] Git [Brent]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [ ] Git [Brent]: `git commit -m "test(cli): add Robot lifecycle CLI coverage"`
- [ ] Git [Brent]: `git push -u origin feature/m1-cli-tests-robot`
- [ ] Forgejo PR [Brent]: Open PR from `feature/m1-cli-tests-robot` to `master` with description "Add Robot end-to-end lifecycle coverage for CLI flows with sandbox + ChangeSet assertions and aligned Behave smoke.".
- [X] Forgejo PR [Brent]: Open PR from `feature/m1-cli-tests` to `master` with description "Expand Behave + Robot CLI coverage for action/plan lifecycle commands and error paths.".
- [X] Git [Brent]: `git checkout master`
- [X] Git [Brent]: `git pull origin master`
- [X] Git [Brent]: `git checkout -b feature/m1-cli-tests-robot`
- [X] Git [Brent]: `git fetch origin && git merge origin/master`
- [X] Tests (Robot) [Brent]: Add end-to-end Robot suite for action → plan → execute → apply on a local repo with git_worktree sandbox + ChangeSet capture verification.
- [X] Tests (Robot) [Brent]: Add negative Robot cases for invalid project names, missing resources, and invalid `--arg` values to match CLI error outputs.
- [X] Tests (Behave) [Brent]: Add one CLI scenario mirroring the Robot E2E flow to keep unit/integration expectations aligned.
- [X] Docs [Brent]: Update `docs/development/testing.md` with Robot CLI suite entry and fixtures.
- [X] Tests (ASV) [Brent]: Add `benchmarks/cli_robot_flow_bench.py` for CLI flow fixture setup overhead.
- [X] Quality [Brent]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] Quality [Brent]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [X] Git [Brent]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [X] Git [Brent]: `git commit -m "test(cli): add Robot lifecycle CLI coverage"`
- [X] Git [Brent]: `git push -u origin feature/m1-cli-tests-robot`
- [X] Forgejo PR [Brent]: Open PR from `feature/m1-cli-tests-robot` to `master` with description "Add Robot end-to-end lifecycle coverage for CLI flows with sandbox + ChangeSet assertions and aligned Behave smoke.".
**Parallel Group A5: Plan Persistence (M1-critical)**
**PARALLEL SUBTRACK A5.beta [Jeff]**: DB rebaseline for Action phase + Apply terminal states
+84
View File
@@ -0,0 +1,84 @@
*** Settings ***
Documentation End-to-end Robot suite for CLI lifecycle: action -> plan -> execute -> apply
... with git_worktree sandbox simulation and ChangeSet capture verification.
... Also includes negative tests for invalid project names, missing resources,
... and invalid --arg values.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_cli_lifecycle.py
*** Test Cases ***
Action Create Via Config CLI
[Documentation] Create an action from a YAML config and verify success
[Tags] cli lifecycle positive
${result}= Run Process ${PYTHON} ${HELPER} action-create cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-action-create-ok
Plan Use Creates Strategize Plan With Sandbox
[Documentation] Use an action with simulated git_worktree sandbox workspace
[Tags] cli lifecycle sandbox positive
${result}= Run Process ${PYTHON} ${HELPER} plan-use-sandbox cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-plan-use-sandbox-ok
Plan Execute With ChangeSet Capture
[Documentation] Execute plan and verify ChangeSet entry is captured
[Tags] cli lifecycle changeset positive
${result}= Run Process ${PYTHON} ${HELPER} plan-execute-changeset cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-plan-execute-changeset-ok
Plan Apply With ChangeSet Verification
[Documentation] Apply plan and verify ChangeSet summary (creates + modifies)
[Tags] cli lifecycle changeset positive
${result}= Run Process ${PYTHON} ${HELPER} plan-apply-changeset cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-plan-apply-changeset-ok
Full Lifecycle Action To Apply With Sandbox
[Documentation] End-to-end: create action, use, execute, apply with sandbox + changeset
[Tags] cli lifecycle e2e positive
${result}= Run Process ${PYTHON} ${HELPER} full-lifecycle cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-full-e2e-sandbox-ok
Invalid Project Name Rejected
[Documentation] Verify that an invalid project name format is handled gracefully
[Tags] cli lifecycle negative
${result}= Run Process ${PYTHON} ${HELPER} invalid-project-name cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-invalid-project-name-ok
Missing Resource Returns Error
[Documentation] Verify that a non-existent action returns an appropriate error
[Tags] cli lifecycle negative
${result}= Run Process ${PYTHON} ${HELPER} missing-resource cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-missing-resource-ok
Invalid Arg Format Rejected
[Documentation] Verify that --arg without '=' separator is rejected
[Tags] cli lifecycle negative
${result}= Run Process ${PYTHON} ${HELPER} invalid-arg-format cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} robot-lifecycle-invalid-arg-format-ok
+411
View File
@@ -0,0 +1,411 @@
"""Robot helper for CLI lifecycle E2E with sandbox + ChangeSet verification.
Each subcommand is self-contained and prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
"""
from __future__ import annotations
import os
import shutil
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import NoReturn
from unittest.mock import MagicMock, patch
# Ensure the src directory is on the import path.
_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 Action, ActionState # noqa: E402
from cleveragents.domain.models.core.change import ( # noqa: E402
ChangeEntry,
ChangeOperation,
InMemoryChangeSetStore,
SpecChangeSet,
)
from cleveragents.domain.models.core.plan import ( # noqa: E402
AutomationLevel,
AutomationProfileProvenance,
AutomationProfileRef,
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
runner = CliRunner()
_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8R"
_VALID_YAML = """\
name: local/lifecycle-action
description: Lifecycle test action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: All lifecycle tests pass
"""
def _mock_action(name: str = "local/lifecycle-action") -> Action:
"""Create a minimal valid Action."""
return Action(
namespaced_name=NamespacedName.parse(name),
description="Lifecycle test action",
long_description=None,
definition_of_done="All lifecycle tests pass",
strategy_actor="openai/gpt-4",
execution_actor="openai/gpt-4",
state=ActionState.AVAILABLE,
reusable=True,
read_only=False,
created_at=datetime.now(),
updated_at=datetime.now(),
created_by=None,
)
def _mock_plan(
phase: PlanPhase = PlanPhase.STRATEGIZE,
state: ProcessingState = ProcessingState.QUEUED,
plan_id: str = _PLAN_ULID,
) -> Plan:
"""Create a minimal valid Plan."""
now = datetime.now()
return Plan(
identity=PlanIdentity(plan_id=plan_id),
namespaced_name=NamespacedName.parse("local/lifecycle-plan"),
description="Lifecycle test plan",
definition_of_done="Tests pass",
action_name="local/lifecycle-action",
phase=phase,
processing_state=state,
automation_level=AutomationLevel.MANUAL,
project_links=[ProjectLink(project_name="proj-a")],
arguments={"target_coverage": 80},
arguments_order=["target_coverage"],
automation_profile=AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
),
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 _write_yaml(content: str) -> str:
"""Write YAML content to a temp file and return the path."""
fd, path = tempfile.mkstemp(suffix=".yaml")
with os.fdopen(fd, "w") as fh:
fh.write(content)
return path
def _make_store() -> tuple[InMemoryChangeSetStore, str]:
"""Create an InMemoryChangeSetStore with one changeset started."""
store = InMemoryChangeSetStore()
cid = store.start(_PLAN_ULID)
return store, cid
def _fail(msg: str) -> NoReturn:
"""Print failure message and exit (never returns)."""
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
# ---------------------------------------------------------------------------
# Positive subcommands
# ---------------------------------------------------------------------------
def action_create() -> None:
"""Verify action create from config file."""
svc = MagicMock()
svc.create_action.return_value = _mock_action()
yaml_path = _write_yaml(_VALID_YAML)
try:
with patch(
"cleveragents.cli.commands.action._get_lifecycle_service", return_value=svc
):
result = runner.invoke(action_app, ["create", "--config", yaml_path])
if result.exit_code != 0:
_fail(f"action create rc={result.exit_code}\n{result.output}")
print("robot-lifecycle-action-create-ok")
finally:
os.unlink(yaml_path)
def plan_use_sandbox() -> None:
"""Verify plan use with simulated git_worktree sandbox workspace."""
svc = MagicMock()
svc.get_action_by_name.return_value = _mock_action()
svc.use_action.return_value = _mock_plan()
sandbox_dir = tempfile.mkdtemp(prefix="robot_sandbox_")
try:
sb = Path(sandbox_dir)
(sb / ".git").mkdir()
(sb / "src").mkdir()
(sb / "src" / "main.py").write_text("# placeholder\n")
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
):
result = runner.invoke(
plan_app, ["use", "local/lifecycle-action", "proj-a"]
)
if result.exit_code != 0:
_fail(f"plan use rc={result.exit_code}\n{result.output}")
if not sb.exists():
_fail("sandbox directory missing")
print("robot-lifecycle-plan-use-sandbox-ok")
finally:
shutil.rmtree(sandbox_dir, ignore_errors=True)
def plan_execute_changeset() -> None:
"""Verify plan execute and ChangeSet capture records entries."""
svc = MagicMock()
svc.execute_plan.return_value = _mock_plan(phase=PlanPhase.EXECUTE)
store, cid = _make_store()
store.record(
cid,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-a",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path="src/new_module.py",
),
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
):
result = runner.invoke(plan_app, ["execute", _PLAN_ULID])
if result.exit_code != 0:
_fail(f"plan execute rc={result.exit_code}\n{result.output}")
cs: SpecChangeSet | None = store.get(cid)
if cs is None:
_fail("changeset not found")
if len(cs.entries) != 1:
_fail(f"changeset entry count={len(cs.entries)}")
if cs.entries[0].operation != ChangeOperation.CREATE:
_fail(f"expected CREATE, got {cs.entries[0].operation}")
print("robot-lifecycle-plan-execute-changeset-ok")
def plan_apply_changeset() -> None:
"""Verify plan apply with ChangeSet summary verification."""
svc = MagicMock()
svc.apply_plan.return_value = _mock_plan(phase=PlanPhase.APPLY)
store, cid = _make_store()
store.record(
cid,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-a",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path="src/new_module.py",
),
)
store.record(
cid,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-a",
tool_name="builtin/file-edit",
operation=ChangeOperation.MODIFY,
path="src/existing.py",
),
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
):
result = runner.invoke(plan_app, ["lifecycle-apply", _PLAN_ULID])
if result.exit_code != 0:
_fail(f"plan apply rc={result.exit_code}\n{result.output}")
summary = store.summarize(cid)
if summary.get("total") != 2:
_fail(f"total={summary.get('total')}")
if summary.get("creates") != 1:
_fail(f"creates={summary.get('creates')}")
if summary.get("modifies") != 1:
_fail(f"modifies={summary.get('modifies')}")
print("robot-lifecycle-plan-apply-changeset-ok")
def full_lifecycle() -> None:
"""End-to-end: action create -> plan use -> execute -> apply with sandbox."""
svc = MagicMock()
svc.create_action.return_value = _mock_action()
svc.get_action_by_name.return_value = _mock_action()
svc.use_action.return_value = _mock_plan()
svc.execute_plan.return_value = _mock_plan(phase=PlanPhase.EXECUTE)
svc.apply_plan.return_value = _mock_plan(phase=PlanPhase.APPLY)
sandbox_dir = tempfile.mkdtemp(prefix="robot_full_sandbox_")
yaml_path = _write_yaml(_VALID_YAML)
store, cid = _make_store()
store.record(
cid,
ChangeEntry(
plan_id=_PLAN_ULID,
resource_id="res-a",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path="src/generated.py",
),
)
try:
with (
patch(
"cleveragents.cli.commands.action._get_lifecycle_service",
return_value=svc,
),
patch(
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=svc,
),
):
r1 = runner.invoke(action_app, ["create", "--config", yaml_path])
assert r1.exit_code == 0, f"action create: {r1.output}"
r2 = runner.invoke(plan_app, ["use", "local/lifecycle-action", "proj-a"])
assert r2.exit_code == 0, f"plan use: {r2.output}"
r3 = runner.invoke(plan_app, ["execute", _PLAN_ULID])
assert r3.exit_code == 0, f"plan execute: {r3.output}"
r4 = runner.invoke(plan_app, ["lifecycle-apply", _PLAN_ULID])
assert r4.exit_code == 0, f"plan apply: {r4.output}"
cs = store.get(cid)
assert cs is not None, "changeset not found"
assert len(cs.entries) == 1, f"entries={len(cs.entries)}"
print("robot-lifecycle-full-e2e-sandbox-ok")
except AssertionError as exc:
_fail(str(exc))
finally:
os.unlink(yaml_path)
shutil.rmtree(sandbox_dir, ignore_errors=True)
# ---------------------------------------------------------------------------
# Negative subcommands
# ---------------------------------------------------------------------------
def invalid_project_name() -> None:
"""Verify invalid project name format is handled."""
svc = MagicMock()
svc.get_action_by_name.return_value = _mock_action()
svc.use_action.side_effect = ValueError("Invalid project name")
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
):
result = runner.invoke(plan_app, ["use", "local/lifecycle-action", "bad name!"])
if (
result.exit_code != 0
or "error" in result.output.lower()
or svc.use_action.called
):
print("robot-lifecycle-invalid-project-name-ok")
else:
_fail("expected error for invalid project name")
def missing_resource() -> None:
"""Verify missing action/resource returns appropriate error."""
from cleveragents.core.exceptions import NotFoundError
svc = MagicMock()
svc.get_action_by_name.side_effect = NotFoundError(
"Action 'local/nonexistent' not found"
)
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
):
result = runner.invoke(plan_app, ["use", "local/nonexistent", "proj-a"])
out = result.output.lower()
if result.exit_code != 0 or "error" in out or "not found" in out:
print("robot-lifecycle-missing-resource-ok")
else:
_fail(f"expected error, rc={result.exit_code}\n{result.output}")
def invalid_arg_format() -> None:
"""Verify --arg without '=' separator is rejected."""
svc = MagicMock()
svc.get_action_by_name.return_value = _mock_action()
with patch(
"cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc
):
result = runner.invoke(
plan_app,
["use", "local/lifecycle-action", "proj-a", "--arg", "no_equals"],
)
out = result.output.lower()
if (
result.exit_code != 0
or "invalid argument format" in out
or "aborted" in out
):
print("robot-lifecycle-invalid-arg-format-ok")
else:
_fail(f"expected rejection, rc={result.exit_code}\n{result.output}")
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, object] = {
"action-create": action_create,
"plan-use-sandbox": plan_use_sandbox,
"plan-execute-changeset": plan_execute_changeset,
"plan-apply-changeset": plan_apply_changeset,
"full-lifecycle": full_lifecycle,
"invalid-project-name": invalid_project_name,
"missing-resource": missing_resource,
"invalid-arg-format": invalid_arg_format,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(f"Usage: helper_cli_lifecycle.py <{'|'.join(_COMMANDS)}>")
return 1
command = sys.argv[1]
handler = _COMMANDS.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
handler() # type: ignore[operator]
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -207,7 +207,10 @@ class AuditService:
if since is not None:
query = query.filter(AuditLogModel.created_at >= since)
query = query.order_by(AuditLogModel.created_at.desc()).limit(limit)
query = query.order_by(
AuditLogModel.created_at.desc(),
AuditLogModel.id.desc(),
).limit(limit)
return [self._row_to_entry(row) for row in query.all()]