forked from HAL9000/cleveragents-core
freemo
631e847497
Resolves issue #3443: the `agents plan rollback` confirmation prompt was missing the checkpoint's descriptive label, relative creation time, and side-effects count (decisions invalidated / child plans cancelled). Changes: - Added `_format_relative_time(dt)` helper to produce human-readable relative timestamps (e.g. "42 minutes ago", "2 hours ago"). - In `rollback_plan`, moved `get_container()` / `checkpoint_service()` before the confirmation block so checkpoint metadata is available. - When `--yes` is not passed, `svc.get_checkpoint()` is called to fetch the checkpoint label (`metadata.reason`) and creation time. - Decisions created after the checkpoint are counted via `decision_service.list_decisions()`; `subplan_spawn` / `subplan_parallel_spawn` decisions are counted as child plans. - Side-effects line is printed before the prompt when counts > 0. - Falls back to the original simple prompt if checkpoint metadata cannot be fetched (e.g. checkpoint not found). - Updated existing rollback mock helpers to wire `get_checkpoint` and `decision_service` properly. - Added 3 new Behave scenarios covering label display, side-effects display, and fallback behaviour. Closes #3443
953 lines
32 KiB
Python
953 lines
32 KiB
Python
"""Step definitions for plan_cli_coverage_r2.feature.
|
|
|
|
Targets remaining uncovered lines in cleveragents/cli/commands/plan.py:
|
|
- Line 724: build >5 changes truncation ("... and N more")
|
|
- Lines 737-738: build PlanError handler
|
|
- Lines 740-741: build CleverAgentsError handler
|
|
- Lines 832-833: apply CleverAgentsError handler
|
|
- Lines 879-880: new ValidationError handler
|
|
- Lines 882-883: new CleverAgentsError handler
|
|
- Lines 930-931: current CleverAgentsError handler
|
|
- Lines 1007-1008: list CleverAgentsError handler
|
|
- Lines 1094-1095: continue CleverAgentsError handler
|
|
- Lines 1108-1115: _get_lifecycle_service body
|
|
- Lines 1223-1245: Multi-project scopes rendering in _print_lifecycle_plan
|
|
- Lines 1469-1481: use_action execution_environment validation
|
|
- Lines 1565-1578: execute_plan execution_environment validation
|
|
- Lines 2588, 2593, 2598: resume_plan rich output optional branches
|
|
- Lines 2616-2618: resume CleverAgentsError handler
|
|
- Lines 2691-2762: rollback_plan full command body
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from rich.console import Console
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands import plan as plan_module
|
|
from cleveragents.cli.commands.plan import app as plan_app
|
|
from cleveragents.core.exceptions import (
|
|
BusinessRuleViolation,
|
|
CleverAgentsError,
|
|
PlanError,
|
|
ResourceNotFoundError,
|
|
ValidationError,
|
|
)
|
|
from cleveragents.domain.models.core.checkpoint import (
|
|
Checkpoint,
|
|
CheckpointMetadata,
|
|
RollbackResult,
|
|
)
|
|
from cleveragents.domain.models.core.multi_project import (
|
|
ChangeSetSummary,
|
|
MultiProjectMetadata,
|
|
ProjectScope,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
from cleveragents.domain.models.core.resume import ResumeSummary
|
|
|
|
_ULID_A = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
_ULID_B = "01ARZ3NDEKTSV4RRFFQ69G5FBV"
|
|
_ULID_CP = "01ARZ3NDEKTSV4RRFFQ69G5FCV"
|
|
|
|
_PATCH_CONTAINER = "cleveragents.application.container.get_container"
|
|
_PATCH_GET_LIFECYCLE = "cleveragents.cli.commands.plan._get_lifecycle_service"
|
|
_PATCH_GET_PROJECT = "cleveragents.cli.commands.plan._get_current_project"
|
|
_PATCH_GET_APPLY = "cleveragents.cli.commands.plan._get_apply_service"
|
|
|
|
|
|
def _make_legacy_project() -> SimpleNamespace:
|
|
return SimpleNamespace(name="test-project", path="/tmp/test")
|
|
|
|
|
|
def _make_legacy_plan(
|
|
*,
|
|
name: str = "my-plan",
|
|
status: str = "active",
|
|
current: bool = True,
|
|
) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
name=name,
|
|
status=status,
|
|
current=current,
|
|
created_at=datetime(2025, 1, 1),
|
|
prompt="do stuff",
|
|
)
|
|
|
|
|
|
def _make_legacy_change(file_path: str, operation: str = "modify") -> SimpleNamespace:
|
|
return SimpleNamespace(file_path=file_path, operation=operation)
|
|
|
|
|
|
def _make_lifecycle_plan(
|
|
*,
|
|
plan_id: str = _ULID_A,
|
|
name: str = "local/test-plan",
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
processing_state: ProcessingState = ProcessingState.QUEUED,
|
|
project_links: list[ProjectLink] | None = None,
|
|
error_message: str | None = None,
|
|
error_details: dict[str, str] | None = None,
|
|
read_only: bool = False,
|
|
multi_project_metadata: MultiProjectMetadata | None = None,
|
|
execution_environment: str | None = None,
|
|
) -> Plan:
|
|
plan = Plan(
|
|
identity=PlanIdentity(plan_id=plan_id),
|
|
namespaced_name=NamespacedName.parse(name),
|
|
action_name="local/test-action",
|
|
description="Coverage test plan",
|
|
definition_of_done=None,
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
phase=phase,
|
|
processing_state=processing_state,
|
|
project_links=project_links or [],
|
|
timestamps=PlanTimestamps(
|
|
created_at=datetime(2025, 6, 15, 10, 0, 0),
|
|
updated_at=datetime(2025, 6, 15, 11, 0, 0),
|
|
),
|
|
error_message=error_message,
|
|
error_details=error_details,
|
|
reusable=True,
|
|
read_only=read_only,
|
|
created_by=None,
|
|
multi_project_metadata=multi_project_metadata,
|
|
)
|
|
if execution_environment:
|
|
plan.execution_environment = execution_environment
|
|
return plan
|
|
|
|
|
|
def _setup_legacy_mocks(
|
|
context: Context,
|
|
*,
|
|
plan_service_attrs: dict[str, Any] | None = None,
|
|
project_service_attrs: dict[str, Any] | None = None,
|
|
) -> list:
|
|
"""Set up standard legacy container mocks and return patches for cleanup."""
|
|
mock_container = MagicMock()
|
|
mock_plan_svc = MagicMock()
|
|
mock_project_svc = MagicMock()
|
|
mock_project_svc.get_current_project.return_value = _make_legacy_project()
|
|
|
|
if plan_service_attrs:
|
|
for key, val in plan_service_attrs.items():
|
|
setattr(mock_plan_svc, key, val)
|
|
|
|
mock_container.plan_service.return_value = mock_plan_svc
|
|
mock_container.project_service.return_value = mock_project_svc
|
|
|
|
# actor_registry for build/tell
|
|
mock_container.actor_registry.return_value = MagicMock()
|
|
mock_container.actor_service.return_value = MagicMock()
|
|
|
|
patches = []
|
|
p_container = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p_project = patch(_PATCH_GET_PROJECT, return_value=_make_legacy_project())
|
|
p_container.start()
|
|
p_project.start()
|
|
patches.extend([p_container.stop, p_project.stop])
|
|
|
|
context.r2cov_plan_svc = mock_plan_svc
|
|
context.r2cov_container = mock_container
|
|
return patches
|
|
|
|
|
|
# ======================================================================
|
|
# Given — CLI runner
|
|
# ======================================================================
|
|
|
|
|
|
@given("a r2cov CLI runner")
|
|
def step_r2cov_cli_runner(context: Context) -> None:
|
|
context.r2cov_runner = CliRunner()
|
|
context.r2cov_result = None
|
|
if not hasattr(context, "_r2cov_cleanups"):
|
|
context._r2cov_cleanups = []
|
|
|
|
def _cleanup(ctx: Context) -> None:
|
|
for fn in getattr(ctx, "_r2cov_cleanups", []):
|
|
fn()
|
|
ctx._r2cov_cleanups = []
|
|
|
|
context.add_cleanup(_cleanup, context)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Legacy build mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked legacy container for r2cov build returning 8 changes")
|
|
def step_mock_build_8_changes(context: Context) -> None:
|
|
changes = [_make_legacy_change(f"src/file{i}.py") for i in range(8)]
|
|
build_plan = MagicMock(return_value=changes)
|
|
patches = _setup_legacy_mocks(
|
|
context, plan_service_attrs={"build_plan": build_plan}
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
@given("a mocked legacy container for r2cov build that raises PlanError")
|
|
def step_mock_build_plan_error(context: Context) -> None:
|
|
build_plan = MagicMock(side_effect=PlanError("Build failed unexpectedly"))
|
|
patches = _setup_legacy_mocks(
|
|
context, plan_service_attrs={"build_plan": build_plan}
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
@given("a mocked legacy container for r2cov build that raises CleverAgentsError")
|
|
def step_mock_build_clever_error(context: Context) -> None:
|
|
build_plan = MagicMock(side_effect=CleverAgentsError("service unavailable"))
|
|
patches = _setup_legacy_mocks(
|
|
context, plan_service_attrs={"build_plan": build_plan}
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Legacy apply mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked legacy container for r2cov apply that raises CleverAgentsError")
|
|
def step_mock_apply_clever_error(context: Context) -> None:
|
|
get_pending = MagicMock(return_value=[_make_legacy_change("f.py")])
|
|
apply_changes = MagicMock(side_effect=CleverAgentsError("apply failed"))
|
|
patches = _setup_legacy_mocks(
|
|
context,
|
|
plan_service_attrs={
|
|
"get_pending_changes": get_pending,
|
|
"apply_changes": apply_changes,
|
|
},
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Legacy new mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked legacy container for r2cov new that raises ValidationError")
|
|
def step_mock_new_validation_error(context: Context) -> None:
|
|
new_plan = MagicMock(side_effect=ValidationError("Invalid plan name"))
|
|
patches = _setup_legacy_mocks(context, plan_service_attrs={"new_plan": new_plan})
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
@given("a mocked legacy container for r2cov new that raises CleverAgentsError")
|
|
def step_mock_new_clever_error(context: Context) -> None:
|
|
new_plan = MagicMock(side_effect=CleverAgentsError("new plan failed"))
|
|
patches = _setup_legacy_mocks(context, plan_service_attrs={"new_plan": new_plan})
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Legacy current mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked legacy container for r2cov current that raises CleverAgentsError")
|
|
def step_mock_current_clever_error(context: Context) -> None:
|
|
get_current = MagicMock(side_effect=CleverAgentsError("DB connection lost"))
|
|
patches = _setup_legacy_mocks(
|
|
context, plan_service_attrs={"get_current_plan": get_current}
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Legacy list mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked legacy container for r2cov list that raises CleverAgentsError")
|
|
def step_mock_list_clever_error(context: Context) -> None:
|
|
list_plans = MagicMock(side_effect=CleverAgentsError("list failed"))
|
|
patches = _setup_legacy_mocks(
|
|
context, plan_service_attrs={"list_plans": list_plans}
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Legacy continue mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked legacy container for r2cov continue that raises CleverAgentsError")
|
|
def step_mock_continue_clever_error(context: Context) -> None:
|
|
get_current = MagicMock(side_effect=CleverAgentsError("continue failed"))
|
|
patches = _setup_legacy_mocks(
|
|
context, plan_service_attrs={"get_current_plan": get_current}
|
|
)
|
|
context._r2cov_cleanups.extend(patches)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — _get_lifecycle_service construction
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked container for r2cov lifecycle service construction")
|
|
def step_mock_container_for_lifecycle(context: Context) -> None:
|
|
mock_container = MagicMock()
|
|
mock_settings = MagicMock()
|
|
mock_container.settings.return_value = mock_settings
|
|
mock_container.plan_lifecycle_service.return_value = MagicMock(
|
|
settings=mock_settings
|
|
)
|
|
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
context.r2cov_mock_settings = mock_settings
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Multi-project scopes plan
|
|
# ======================================================================
|
|
|
|
|
|
@given("a lifecycle plan with multi-project scopes for r2cov")
|
|
def step_plan_with_multi_project_scopes(context: Context) -> None:
|
|
scopes = [
|
|
ProjectScope(
|
|
project_name="proj-alpha",
|
|
alias="alpha",
|
|
read_only=False,
|
|
changeset_summary=ChangeSetSummary(
|
|
project_name="proj-alpha",
|
|
files_changed=3,
|
|
files_added=1,
|
|
files_deleted=0,
|
|
total_lines_changed=42,
|
|
),
|
|
),
|
|
ProjectScope(
|
|
project_name="proj-beta",
|
|
alias=None,
|
|
read_only=True,
|
|
changeset_summary=None,
|
|
),
|
|
]
|
|
metadata = MultiProjectMetadata(project_scopes=scopes)
|
|
context.r2cov_lifecycle_plan = _make_lifecycle_plan(
|
|
multi_project_metadata=metadata,
|
|
)
|
|
|
|
|
|
@given("a lifecycle plan with multi-project scopes with validation for r2cov")
|
|
def step_plan_with_validation_scopes(context: Context) -> None:
|
|
scopes = [
|
|
ProjectScope(
|
|
project_name="proj-gamma",
|
|
changeset_summary=ChangeSetSummary(
|
|
project_name="proj-gamma",
|
|
files_changed=5,
|
|
files_added=2,
|
|
files_deleted=1,
|
|
total_lines_changed=100,
|
|
validation_passed=True,
|
|
),
|
|
),
|
|
]
|
|
metadata = MultiProjectMetadata(project_scopes=scopes)
|
|
context.r2cov_lifecycle_plan = _make_lifecycle_plan(
|
|
multi_project_metadata=metadata,
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Lifecycle service for use_action
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked lifecycle service for r2cov use action")
|
|
def step_mock_lifecycle_for_use(context: Context) -> None:
|
|
mock_service = MagicMock()
|
|
mock_action = MagicMock()
|
|
mock_action.namespaced_name = "local/test-action"
|
|
mock_service.get_action_by_name.return_value = mock_action
|
|
mock_plan = _make_lifecycle_plan()
|
|
mock_service.use_action.return_value = mock_plan
|
|
|
|
p = patch(_PATCH_GET_LIFECYCLE, return_value=mock_service)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
context.r2cov_service = mock_service
|
|
context.r2cov_plan = mock_plan
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Lifecycle service for execute_plan
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked lifecycle service for r2cov execute with a valid plan")
|
|
def step_mock_lifecycle_for_execute(context: Context) -> None:
|
|
mock_service = MagicMock()
|
|
plan = _make_lifecycle_plan(
|
|
processing_state=ProcessingState.COMPLETE,
|
|
)
|
|
mock_service.get_plan.return_value = plan
|
|
mock_service.execute_plan.return_value = plan
|
|
|
|
p = patch(_PATCH_GET_LIFECYCLE, return_value=mock_service)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
p_exec = patch(
|
|
"cleveragents.cli.commands.plan._get_plan_executor",
|
|
return_value=MagicMock(),
|
|
)
|
|
p_exec.start()
|
|
context._r2cov_cleanups.append(p_exec.stop)
|
|
|
|
context.r2cov_service = mock_service
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Resume service mocks
|
|
# ======================================================================
|
|
|
|
|
|
@given("a mocked resume service for r2cov with all optional fields")
|
|
def step_mock_resume_all_fields(context: Context) -> None:
|
|
summary = ResumeSummary(
|
|
plan_id="PLAN123",
|
|
phase="execute",
|
|
processing_state="errored",
|
|
last_completed_step=3,
|
|
next_step_index=4,
|
|
total_steps=10,
|
|
decision_id="DEC-001",
|
|
last_checkpoint_id="CP-001",
|
|
sandbox_ref="sandbox-abc-123",
|
|
)
|
|
mock_lifecycle = MagicMock()
|
|
|
|
p_lifecycle = patch(_PATCH_GET_LIFECYCLE, return_value=mock_lifecycle)
|
|
p_lifecycle.start()
|
|
context._r2cov_cleanups.append(p_lifecycle.stop)
|
|
|
|
p_resume = patch(
|
|
"cleveragents.application.services.plan_resume_service.PlanResumeService",
|
|
autospec=False,
|
|
)
|
|
mock_resume_cls = p_resume.start()
|
|
context._r2cov_cleanups.append(p_resume.stop)
|
|
mock_resume_instance = MagicMock()
|
|
mock_resume_instance.resume_plan.return_value = summary
|
|
mock_resume_cls.return_value = mock_resume_instance
|
|
|
|
|
|
@given("a mocked resume service for r2cov that raises CleverAgentsError")
|
|
def step_mock_resume_clever_error(context: Context) -> None:
|
|
mock_lifecycle = MagicMock()
|
|
|
|
p_lifecycle = patch(_PATCH_GET_LIFECYCLE, return_value=mock_lifecycle)
|
|
p_lifecycle.start()
|
|
context._r2cov_cleanups.append(p_lifecycle.stop)
|
|
|
|
p_resume = patch(
|
|
"cleveragents.application.services.plan_resume_service.PlanResumeService",
|
|
autospec=False,
|
|
)
|
|
mock_resume_cls = p_resume.start()
|
|
context._r2cov_cleanups.append(p_resume.stop)
|
|
mock_resume_instance = MagicMock()
|
|
mock_resume_instance.resume_plan.side_effect = CleverAgentsError("resume failed")
|
|
mock_resume_cls.return_value = mock_resume_instance
|
|
|
|
|
|
# ======================================================================
|
|
# Given — Checkpoint service for rollback
|
|
# ======================================================================
|
|
|
|
|
|
def _make_rollback_result() -> RollbackResult:
|
|
return RollbackResult(
|
|
restored_files_count=3,
|
|
changed_paths=["src/a.py", "src/b.py", "src/c.py"],
|
|
from_checkpoint_id=_ULID_CP,
|
|
)
|
|
|
|
|
|
def _make_checkpoint(
|
|
*,
|
|
label: str = "before-auth-refactor",
|
|
created_at: datetime | None = None,
|
|
) -> Checkpoint:
|
|
"""Create a Checkpoint domain object for use in rollback test mocks."""
|
|
return Checkpoint(
|
|
checkpoint_id=_ULID_CP,
|
|
plan_id=_ULID_A,
|
|
sandbox_ref="abc123",
|
|
created_at=created_at or datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC),
|
|
metadata=CheckpointMetadata(reason=label),
|
|
)
|
|
|
|
|
|
def _make_rollback_container(
|
|
*,
|
|
rollback_result: RollbackResult | None = None,
|
|
rollback_side_effect: Exception | None = None,
|
|
) -> MagicMock:
|
|
"""Build a mock container with checkpoint_service and decision_service wired."""
|
|
mock_container = MagicMock()
|
|
mock_cp_svc = MagicMock()
|
|
mock_cp_svc.get_checkpoint.return_value = _make_checkpoint()
|
|
if rollback_side_effect is not None:
|
|
mock_cp_svc.selective_rollback.side_effect = rollback_side_effect
|
|
else:
|
|
mock_cp_svc.selective_rollback.return_value = (
|
|
rollback_result or _make_rollback_result()
|
|
)
|
|
mock_container.checkpoint_service.return_value = mock_cp_svc
|
|
mock_decision_svc = MagicMock()
|
|
mock_decision_svc.list_decisions.return_value = []
|
|
mock_container.decision_service.return_value = mock_decision_svc
|
|
return mock_container
|
|
|
|
|
|
@given("a mocked checkpoint service for r2cov rollback that succeeds")
|
|
def step_mock_rollback_success(context: Context) -> None:
|
|
mock_container = _make_rollback_container()
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback that raises BusinessRuleViolation"
|
|
)
|
|
def step_mock_rollback_business_rule(context: Context) -> None:
|
|
mock_container = _make_rollback_container(
|
|
rollback_side_effect=BusinessRuleViolation("Plan is already applied")
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback that raises ResourceNotFoundError"
|
|
)
|
|
def step_mock_rollback_not_found(context: Context) -> None:
|
|
mock_container = _make_rollback_container(
|
|
rollback_side_effect=ResourceNotFoundError("Checkpoint not found")
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given("a mocked checkpoint service for r2cov rollback that raises ValidationError")
|
|
def step_mock_rollback_validation(context: Context) -> None:
|
|
mock_container = _make_rollback_container(
|
|
rollback_side_effect=ValidationError("Invalid checkpoint")
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given("a mocked checkpoint service for r2cov rollback that raises CleverAgentsError")
|
|
def step_mock_rollback_clever_error(context: Context) -> None:
|
|
mock_container = _make_rollback_container(
|
|
rollback_side_effect=CleverAgentsError("unknown rollback failure")
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
# ======================================================================
|
|
# When — Legacy commands
|
|
# ======================================================================
|
|
|
|
|
|
@when("I invoke r2cov legacy build")
|
|
def step_invoke_legacy_build(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["build"])
|
|
|
|
|
|
@when("I invoke r2cov legacy apply with --yes")
|
|
def step_invoke_legacy_apply_yes(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["apply", "--yes"])
|
|
|
|
|
|
@when('I invoke r2cov legacy new "{name}"')
|
|
def step_invoke_legacy_new(context: Context, name: str) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["new", name])
|
|
|
|
|
|
@when("I invoke r2cov legacy current")
|
|
def step_invoke_legacy_current(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["current"])
|
|
|
|
|
|
@when("I invoke r2cov legacy list")
|
|
def step_invoke_legacy_list(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["list"])
|
|
|
|
|
|
@when("I invoke r2cov legacy continue")
|
|
def step_invoke_legacy_continue(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["continue"])
|
|
|
|
|
|
# ======================================================================
|
|
# When — _get_lifecycle_service
|
|
# ======================================================================
|
|
|
|
|
|
@when("I call r2cov _get_lifecycle_service")
|
|
def step_call_get_lifecycle_service(context: Context) -> None:
|
|
from cleveragents.cli.commands.plan import _get_lifecycle_service
|
|
|
|
context.r2cov_lifecycle_result = _get_lifecycle_service()
|
|
|
|
|
|
# ======================================================================
|
|
# When — _print_lifecycle_plan
|
|
# ======================================================================
|
|
|
|
|
|
@when("I invoke r2cov _print_lifecycle_plan on the plan")
|
|
def step_invoke_print_lifecycle_plan(context: Context) -> None:
|
|
from cleveragents.cli.commands.plan import _print_lifecycle_plan
|
|
|
|
# Capture console output
|
|
buf = io.StringIO()
|
|
capture_console = Console(file=buf, width=200, force_terminal=True)
|
|
original = plan_module.console
|
|
plan_module.console = capture_console
|
|
try:
|
|
_print_lifecycle_plan(context.r2cov_lifecycle_plan, title="Test Plan")
|
|
finally:
|
|
plan_module.console = original
|
|
context.r2cov_result = SimpleNamespace(
|
|
exit_code=0,
|
|
output=buf.getvalue(),
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# When — use_action with execution_environment
|
|
# ======================================================================
|
|
|
|
|
|
@when('I invoke r2cov use with execution-environment "{env}"')
|
|
def step_invoke_use_exec_env(context: Context, env: str) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["use", "local/test-action", "--execution-environment", env],
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# When — execute_plan with execution_environment
|
|
# ======================================================================
|
|
|
|
|
|
@when('I invoke r2cov execute with execution-environment "{env}"')
|
|
def step_invoke_execute_exec_env(context: Context, env: str) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["execute", _ULID_A, "--execution-environment", env],
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# When — resume plan
|
|
# ======================================================================
|
|
|
|
|
|
@when('I invoke r2cov plan resume "{plan_id}" in rich format')
|
|
def step_invoke_resume_rich(context: Context, plan_id: str) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["resume", plan_id],
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# When — rollback plan
|
|
# ======================================================================
|
|
|
|
|
|
@when("I invoke r2cov plan rollback with --yes")
|
|
def step_invoke_rollback_yes(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["rollback", "--yes", _ULID_A, _ULID_CP],
|
|
)
|
|
|
|
|
|
@when("I invoke r2cov plan rollback with --yes and --format json")
|
|
def step_invoke_rollback_yes_json(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["rollback", "--yes", "--format", "json", _ULID_A, _ULID_CP],
|
|
)
|
|
|
|
|
|
@when("I invoke r2cov plan rollback without --yes and decline")
|
|
def step_invoke_rollback_decline(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["rollback", _ULID_A, _ULID_CP],
|
|
input="n\n",
|
|
)
|
|
|
|
|
|
@when("I invoke r2cov plan rollback without --yes and accept")
|
|
def step_invoke_rollback_accept(context: Context) -> None:
|
|
context.r2cov_result = context.r2cov_runner.invoke(
|
|
plan_app,
|
|
["rollback", _ULID_A, _ULID_CP],
|
|
input="y\n",
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# Given — enriched confirmation prompt mocks (issue #3443)
|
|
# ======================================================================
|
|
|
|
|
|
@given('a mocked checkpoint service for r2cov rollback with label "{label}"')
|
|
def step_mock_rollback_with_label(context: Context, label: str) -> None:
|
|
mock_container = _make_rollback_container()
|
|
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
|
|
_make_checkpoint(label=label)
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback with {decisions:d} decisions "
|
|
"and {child_plans:d} child plan"
|
|
)
|
|
def step_mock_rollback_with_decisions(
|
|
context: Context, decisions: int, child_plans: int
|
|
) -> None:
|
|
from datetime import timedelta
|
|
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.decision import Decision, DecisionType
|
|
|
|
cp_time = datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC)
|
|
after_time = cp_time + timedelta(minutes=5)
|
|
|
|
# Build decision list: (decisions - child_plans) regular + child_plans spawns
|
|
decision_list = []
|
|
regular_count = decisions - child_plans
|
|
for i in range(regular_count):
|
|
decision_list.append(
|
|
Decision(
|
|
decision_id=str(ULID()),
|
|
plan_id=_ULID_A,
|
|
sequence_number=i,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
question="What to do?",
|
|
chosen_option="Do it",
|
|
created_at=after_time,
|
|
)
|
|
)
|
|
for i in range(child_plans):
|
|
decision_list.append(
|
|
Decision(
|
|
decision_id=str(ULID()),
|
|
plan_id=_ULID_A,
|
|
sequence_number=regular_count + i,
|
|
decision_type=DecisionType.SUBPLAN_SPAWN,
|
|
question="Spawn child?",
|
|
chosen_option="Yes",
|
|
created_at=after_time,
|
|
)
|
|
)
|
|
|
|
mock_container = _make_rollback_container()
|
|
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
|
|
_make_checkpoint(created_at=cp_time)
|
|
)
|
|
mock_container.decision_service.return_value.list_decisions.return_value = (
|
|
decision_list
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback where get_checkpoint raises not found"
|
|
)
|
|
def step_mock_rollback_get_checkpoint_not_found(context: Context) -> None:
|
|
mock_container = _make_rollback_container()
|
|
mock_container.checkpoint_service.return_value.get_checkpoint.side_effect = (
|
|
ResourceNotFoundError("Checkpoint not found")
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback with {decisions:d} decisions "
|
|
"and no child plans"
|
|
)
|
|
def step_mock_rollback_with_decisions_no_child_plans(
|
|
context: Context, decisions: int
|
|
) -> None:
|
|
from datetime import timedelta
|
|
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.decision import Decision, DecisionType
|
|
|
|
cp_time = datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC)
|
|
after_time = cp_time + timedelta(minutes=5)
|
|
|
|
decision_list = [
|
|
Decision(
|
|
decision_id=str(ULID()),
|
|
plan_id=_ULID_A,
|
|
sequence_number=i,
|
|
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
|
|
question="What to do?",
|
|
chosen_option="Do it",
|
|
created_at=after_time,
|
|
)
|
|
for i in range(decisions)
|
|
]
|
|
|
|
mock_container = _make_rollback_container()
|
|
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
|
|
_make_checkpoint(created_at=cp_time)
|
|
)
|
|
mock_container.decision_service.return_value.list_decisions.return_value = (
|
|
decision_list
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback with {child_plans:d} "
|
|
"parallel spawn child plans"
|
|
)
|
|
def step_mock_rollback_with_parallel_spawn(context: Context, child_plans: int) -> None:
|
|
from datetime import timedelta
|
|
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.decision import Decision, DecisionType
|
|
|
|
cp_time = datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC)
|
|
after_time = cp_time + timedelta(minutes=5)
|
|
|
|
decision_list = [
|
|
Decision(
|
|
decision_id=str(ULID()),
|
|
plan_id=_ULID_A,
|
|
sequence_number=i,
|
|
decision_type=DecisionType.SUBPLAN_PARALLEL_SPAWN,
|
|
question="Spawn parallel child?",
|
|
chosen_option="Yes",
|
|
created_at=after_time,
|
|
)
|
|
for i in range(child_plans)
|
|
]
|
|
|
|
mock_container = _make_rollback_container()
|
|
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
|
|
_make_checkpoint(created_at=cp_time)
|
|
)
|
|
mock_container.decision_service.return_value.list_decisions.return_value = (
|
|
decision_list
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
@given(
|
|
"a mocked checkpoint service for r2cov rollback with a future checkpoint timestamp"
|
|
)
|
|
def step_mock_rollback_with_future_timestamp(context: Context) -> None:
|
|
from datetime import timedelta
|
|
|
|
future_time = datetime.now(UTC) + timedelta(hours=1)
|
|
mock_container = _make_rollback_container()
|
|
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
|
|
_make_checkpoint(created_at=future_time)
|
|
)
|
|
p = patch(_PATCH_CONTAINER, return_value=mock_container)
|
|
p.start()
|
|
context._r2cov_cleanups.append(p.stop)
|
|
|
|
|
|
# ======================================================================
|
|
# Then — common assertions
|
|
# ======================================================================
|
|
|
|
|
|
@then("the r2cov command should exit normally")
|
|
def step_exit_normally(context: Context) -> None:
|
|
result = context.r2cov_result
|
|
assert result is not None, "No command result captured"
|
|
assert result.exit_code == 0, (
|
|
f"Expected exit code 0, got {result.exit_code}.\nOutput:\n{result.output}"
|
|
)
|
|
|
|
|
|
@then("the r2cov command should abort")
|
|
def step_exit_abort(context: Context) -> None:
|
|
result = context.r2cov_result
|
|
assert result is not None, "No command result captured"
|
|
assert result.exit_code != 0, (
|
|
f"Expected non-zero exit code, got {result.exit_code}.\n"
|
|
f"Output:\n{result.output}"
|
|
)
|
|
|
|
|
|
@then('the r2cov output should contain "{text}"')
|
|
def step_output_contains(context: Context, text: str) -> None:
|
|
result = context.r2cov_result
|
|
assert result is not None, "No command result captured"
|
|
assert text in result.output, (
|
|
f"Expected '{text}' in output.\nActual output:\n{result.output}"
|
|
)
|
|
|
|
|
|
@then("the r2cov lifecycle service should be returned")
|
|
def step_lifecycle_returned(context: Context) -> None:
|
|
assert context.r2cov_lifecycle_result is not None, (
|
|
"Expected a lifecycle service instance"
|
|
)
|