From 5febbbc93bb4627cece8f8b9c8fd2024dc66c585 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sat, 6 Jun 2026 16:23:02 -0400 Subject: [PATCH] fix(plan-cli): fix lint and BDD step issues for plan rollback command - Add list-mode to rollback_plan: when no checkpoint ID given, list available checkpoints instead of aborting (fixes feature file scenario) - Add CleverAgentsError import to rollback_plan function scope - Rewrite plan_cli_rollback_steps.py with correct mocking pattern: patch get_container, use plan_app with ["rollback", ...] args, :S parse modifiers to avoid AmbiguousStep, proper exception hierarchy - Rename 4 conflicting @then step patterns to be rollback-specific: "the rollback output should be valid JSON/YAML", "the plan rollback should succeed", "no rollback confirmation prompt should be shown" - Fix JSON/YAML assertion steps to check format_output envelope structure (data is nested under "data" key in the envelope) - Update plan_cli_rollback.feature to match renamed step patterns ISSUES CLOSED: #9612 --- features/plan_cli_rollback.feature | 16 +- features/steps/plan_cli_rollback_steps.py | 492 +++++++++++++++------- src/cleveragents/cli/commands/plan.py | 45 +- 3 files changed, 382 insertions(+), 171 deletions(-) diff --git a/features/plan_cli_rollback.feature b/features/plan_cli_rollback.feature index 4034aa1c4..7334f1844 100644 --- a/features/plan_cli_rollback.feature +++ b/features/plan_cli_rollback.feature @@ -32,14 +32,14 @@ Feature: Plan rollback CLI command Given a plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" exists And the plan has 2 checkpoints When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV --format json" - Then the output should be valid JSON + Then the rollback output should be valid JSON And the JSON should contain a list of checkpoints Scenario: List checkpoints with YAML output format Given a plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" exists And the plan has 2 checkpoints When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV --format yaml" - Then the output should be valid YAML + Then the rollback output should be valid YAML And the YAML should contain a list of checkpoints Scenario: List checkpoints with plain output format @@ -58,7 +58,7 @@ Feature: Plan rollback CLI command And the plan has a checkpoint "01BRZ4NFEKTSV4RRFFQ69G5FAV" And the plan is not applied When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV 01BRZ4NFEKTSV4RRFFQ69G5FAV" with confirmation - Then the rollback should succeed + Then the plan rollback should succeed And the output should show rollback summary Scenario: Restore plan to checkpoint with --yes flag @@ -66,15 +66,15 @@ Feature: Plan rollback CLI command And the plan has a checkpoint "01BRZ4NFEKTSV4RRFFQ69G5FAV" And the plan is not applied When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV 01BRZ4NFEKTSV4RRFFQ69G5FAV --yes" - Then the rollback should succeed - And no confirmation prompt should be shown + Then the plan rollback should succeed + And no rollback confirmation prompt should be shown Scenario: Restore plan to checkpoint with --to-checkpoint option Given a plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" exists And the plan has a checkpoint "01BRZ4NFEKTSV4RRFFQ69G5FAV" And the plan is not applied When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV --to-checkpoint 01BRZ4NFEKTSV4RRFFQ69G5FAV --yes" - Then the rollback should succeed + Then the plan rollback should succeed Scenario: Restore plan to non-existent checkpoint Given a plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" exists @@ -98,7 +98,7 @@ Feature: Plan rollback CLI command And the plan has a checkpoint "01BRZ4NFEKTSV4RRFFQ69G5FAV" And the plan is not applied When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV 01BRZ4NFEKTSV4RRFFQ69G5FAV --yes --format json" - Then the output should be valid JSON + Then the rollback output should be valid JSON And the JSON should contain rollback summary Scenario: Restore plan with YAML output format @@ -106,7 +106,7 @@ Feature: Plan rollback CLI command And the plan has a checkpoint "01BRZ4NFEKTSV4RRFFQ69G5FAV" And the plan is not applied When I run "agents plan rollback 01ARZ3NDEKTSV4RRFFQ69G5FAV 01BRZ4NFEKTSV4RRFFQ69G5FAV --yes --format yaml" - Then the output should be valid YAML + Then the rollback output should be valid YAML And the YAML should contain rollback summary # ─────────────────────────────────────────────────────────── diff --git a/features/steps/plan_cli_rollback_steps.py b/features/steps/plan_cli_rollback_steps.py index b6c0c65aa..caa659c59 100644 --- a/features/steps/plan_cli_rollback_steps.py +++ b/features/steps/plan_cli_rollback_steps.py @@ -1,52 +1,172 @@ """Step definitions for plan CLI rollback command feature.""" + from __future__ import annotations import json +from datetime import UTC, datetime +from unittest.mock import MagicMock, patch from behave import given, then, when from behave.runner import Context from typer.testing import CliRunner import yaml -from cleveragents.cli.main import app as cli_app +from cleveragents.cli.commands.plan import app as plan_app +from cleveragents.core.exceptions import ( + BusinessRuleViolation, + CleverAgentsError, + ResourceNotFoundError, +) +from cleveragents.domain.models.core.checkpoint import Checkpoint, CheckpointMetadata + +_PATCH_CONTAINER = "cleveragents.application.container.get_container" +_RUNNER = CliRunner() + + +def _make_test_checkpoint( + checkpoint_id: str, + plan_id: str, + index: int = 0, +) -> Checkpoint: + """Build a Checkpoint domain object for testing.""" + return Checkpoint( + checkpoint_id=checkpoint_id, + plan_id=plan_id, + sandbox_ref="test-sandbox", + checkpoint_type="manual", + created_at=datetime(2026, 4, 15, 0, index, 0, tzinfo=UTC), + metadata=CheckpointMetadata( + reason=f"checkpoint-{index}", + source_tool="test", + phase="execute", + ), + ) + + +def _build_mock_container(context: Context) -> MagicMock: + """Build a MagicMock container from context state.""" + mock_container = MagicMock() + mock_cp_svc = MagicMock() + mock_lifecycle_svc = MagicMock() + + plan_id = getattr(context, "plan_id", None) + checkpoint_id = getattr(context, "checkpoint_id", None) + + if plan_id and hasattr(context, "plans") and plan_id in context.plans: + plan_data = context.plans[plan_id] + raw_checkpoints = plan_data.get("checkpoints", []) + cp_objects = [ + _make_test_checkpoint( + checkpoint_id=cp["id"], + plan_id=plan_id, + index=i, + ) + for i, cp in enumerate(raw_checkpoints) + ] + mock_cp_svc.list_checkpoints.return_value = cp_objects + # Use fallback prompt text for confirmation flows + mock_cp_svc.get_checkpoint.side_effect = CleverAgentsError( + "metadata unavailable" + ) + + if plan_data.get("applied", False): + mock_lifecycle_svc.rollback_plan.side_effect = BusinessRuleViolation( + "plan is already applied" + ) + else: + mock_result = MagicMock() + mock_result.from_checkpoint_id = checkpoint_id or "" + mock_result.restored_files_count = len(raw_checkpoints) + mock_result.changed_paths = [ + f"file{i}.py" for i in range(len(raw_checkpoints)) + ] + mock_result.changes_reverted = None + mock_lifecycle_svc.rollback_plan.return_value = mock_result + + if checkpoint_id: + matching = [ + cp for cp in cp_objects if cp.checkpoint_id == checkpoint_id + ] + if not matching: + mock_lifecycle_svc.rollback_plan.side_effect = ( + ResourceNotFoundError( + resource_type="checkpoint", resource_id=checkpoint_id + ) + ) + elif plan_id == "INVALID_PLAN_ID": + mock_cp_svc.list_checkpoints.side_effect = CleverAgentsError("invalid plan ID") + mock_lifecycle_svc.rollback_plan.side_effect = CleverAgentsError( + "invalid plan ID" + ) + else: + mock_cp_svc.list_checkpoints.side_effect = ResourceNotFoundError( + resource_type="plan", resource_id=plan_id or "unknown" + ) + mock_lifecycle_svc.rollback_plan.side_effect = ResourceNotFoundError( + resource_type="plan", resource_id=plan_id or "unknown" + ) + + mock_container.checkpoint_service.return_value = mock_cp_svc + mock_container.plan_lifecycle_service.return_value = mock_lifecycle_svc + return mock_container + + +def _invoke_rollback( + context: Context, + args: list[str], + input_text: str | None = None, +) -> None: + """Invoke the rollback CLI command with a mocked container.""" + mock_container = _build_mock_container(context) + with patch(_PATCH_CONTAINER, return_value=mock_container): + result = _RUNNER.invoke(plan_app, args, input=input_text) + context.last_exit_code = result.exit_code + context.last_stdout = result.output + context.last_stderr = str(result.exception) if result.exception else "" + + +# ---------------------------------------------------------------------- +# Given steps +# ---------------------------------------------------------------------- @given('a plan "{plan_id}" exists') def step_plan_exists(context: Context, plan_id: str) -> None: """Create a plan with the given ID.""" context.plan_id = plan_id - if not hasattr(context, 'plans'): + if not hasattr(context, "plans"): context.plans = {} context.plans[plan_id] = { - 'id': plan_id, - 'checkpoints': [], - 'applied': False, + "id": plan_id, + "checkpoints": [], + "applied": False, } -@given('the plan has {count:d} checkpoints') +@given("the plan has {count:d} checkpoints") def step_plan_has_checkpoints(context: Context, count: int) -> None: - """Add checkpoints to the plan.""" + """Add N auto-generated checkpoints to the plan.""" plan_id = context.plan_id if plan_id not in context.plans: - context.plans[plan_id] = {'id': plan_id, 'checkpoints': [], 'applied': False} - + context.plans[plan_id] = {"id": plan_id, "checkpoints": [], "applied": False} for i in range(count): checkpoint_id = f"01BRZ4NFEKTSV4RRFFQ69G5F{i:02d}" - context.plans[plan_id]['checkpoints'].append({ - 'id': checkpoint_id, - 'created_at': f"2026-04-15T00:{i:02d}:00Z", - 'reason': f"checkpoint-{i}", - }) + context.plans[plan_id]["checkpoints"].append( + { + "id": checkpoint_id, + "created_at": f"2026-04-15T00:{i:02d}:00Z", + "reason": f"checkpoint-{i}", + } + ) -@given('the plan has no checkpoints') +@given("the plan has no checkpoints") def step_plan_has_no_checkpoints(context: Context) -> None: """Ensure the plan has no checkpoints.""" plan_id = context.plan_id if plan_id not in context.plans: - context.plans[plan_id] = {'id': plan_id, 'checkpoints': [], 'applied': False} - context.plans[plan_id]['checkpoints'] = [] + context.plans[plan_id] = {"id": plan_id, "checkpoints": [], "applied": False} + context.plans[plan_id]["checkpoints"] = [] @given('the plan has a checkpoint "{checkpoint_id}"') @@ -54,144 +174,175 @@ def step_plan_has_checkpoint(context: Context, checkpoint_id: str) -> None: """Add a specific checkpoint to the plan.""" plan_id = context.plan_id if plan_id not in context.plans: - context.plans[plan_id] = {'id': plan_id, 'checkpoints': [], 'applied': False} - - context.plans[plan_id]['checkpoints'].append({ - 'id': checkpoint_id, - 'created_at': "2026-04-15T00:00:00Z", - 'reason': "test-checkpoint", - }) + context.plans[plan_id] = {"id": plan_id, "checkpoints": [], "applied": False} + context.plans[plan_id]["checkpoints"].append( + { + "id": checkpoint_id, + "created_at": "2026-04-15T00:00:00Z", + "reason": "test-checkpoint", + } + ) context.checkpoint_id = checkpoint_id -@given('the plan is not applied') +@given("the plan is not applied") def step_plan_not_applied(context: Context) -> None: """Mark the plan as not applied.""" plan_id = context.plan_id if plan_id not in context.plans: - context.plans[plan_id] = {'id': plan_id, 'checkpoints': [], 'applied': False} - context.plans[plan_id]['applied'] = False + context.plans[plan_id] = {"id": plan_id, "checkpoints": [], "applied": False} + context.plans[plan_id]["applied"] = False -@given('the plan is marked as applied') +@given("the plan is marked as applied") def step_plan_marked_applied(context: Context) -> None: """Mark the plan as applied.""" plan_id = context.plan_id if plan_id not in context.plans: - context.plans[plan_id] = {'id': plan_id, 'checkpoints': [], 'applied': False} - context.plans[plan_id]['applied'] = True + context.plans[plan_id] = {"id": plan_id, "checkpoints": [], "applied": False} + context.plans[plan_id]["applied"] = True -def _invoke_command(context: Context, input_text: str | None = None) -> None: - """Helper to invoke the CLI via Typer's CliRunner and store results on context.""" - runner = CliRunner() - args = getattr(context, 'command', []) - # If the list starts with the top-level program name, drop it - if args and args[0] == 'agents': - args = args[1:] - - if input_text is not None: - result = runner.invoke(cli_app, args, input=input_text) - else: - result = runner.invoke(cli_app, args) - - context.last_exit_code = result.exit_code - context.last_stdout = result.output - context.last_stderr = str(result.exception) if result.exception else "" +# ---------------------------------------------------------------------- +# When steps +# ---------------------------------------------------------------------- -@when('I run "agents plan rollback {plan_id}"') +@when('I run "agents plan rollback {plan_id:S}"') def step_run_rollback_list(context: Context, plan_id: str) -> None: - """Run the rollback command in list mode.""" + """Run the rollback command in list mode (no checkpoint ID).""" context.plan_id = plan_id - context.command = ['agents', 'plan', 'rollback', plan_id] - context.last_command = ' '.join(context.command) - _invoke_command(context) + _invoke_rollback(context, ["rollback", plan_id]) -@when('I run "agents plan rollback {plan_id} {checkpoint_id}"') -def step_run_rollback_restore(context: Context, plan_id: str, checkpoint_id: str) -> None: - """Run the rollback command in restore mode.""" +@when('I run "agents plan rollback {plan_id:S} {checkpoint_id:S}"') +def step_run_rollback_restore( + context: Context, plan_id: str, checkpoint_id: str +) -> None: + """Run the rollback command in restore mode (no flags).""" context.plan_id = plan_id context.checkpoint_id = checkpoint_id - context.command = ['agents', 'plan', 'rollback', plan_id, checkpoint_id] - context.last_command = ' '.join(context.command) - _invoke_command(context) + _invoke_rollback(context, ["rollback", plan_id, checkpoint_id]) -@when('I run "agents plan rollback {plan_id} {checkpoint_id} --yes"') -def step_run_rollback_restore_yes(context: Context, plan_id: str, checkpoint_id: str) -> None: - """Run the rollback command in restore mode with --yes flag.""" +@when('I run "agents plan rollback {plan_id:S} {checkpoint_id:S} --yes"') +def step_run_rollback_restore_yes( + context: Context, plan_id: str, checkpoint_id: str +) -> None: + """Run the rollback command with --yes to skip confirmation.""" context.plan_id = plan_id context.checkpoint_id = checkpoint_id - context.command = ['agents', 'plan', 'rollback', plan_id, checkpoint_id, '--yes'] - context.last_command = ' '.join(context.command) - _invoke_command(context) + _invoke_rollback(context, ["rollback", plan_id, checkpoint_id, "--yes"]) -@when('I run "agents plan rollback {plan_id} --format {fmt}"') +@when('I run "agents plan rollback {plan_id:S} {checkpoint_id:S}" with confirmation') +def step_run_rollback_with_confirmation( + context: Context, plan_id: str, checkpoint_id: str +) -> None: + """Run the rollback command and confirm the prompt.""" + context.plan_id = plan_id + context.checkpoint_id = checkpoint_id + _invoke_rollback(context, ["rollback", plan_id, checkpoint_id], input_text="y\n") + + +@when( + 'I run "agents plan rollback {plan_id:S} {checkpoint_id:S}" and cancel confirmation' +) +def step_run_rollback_cancel_confirmation( + context: Context, plan_id: str, checkpoint_id: str +) -> None: + """Run the rollback command and cancel the confirmation prompt.""" + context.plan_id = plan_id + context.checkpoint_id = checkpoint_id + _invoke_rollback(context, ["rollback", plan_id, checkpoint_id], input_text="n\n") + + +@when('I run "agents plan rollback {plan_id:S} --format {fmt:S}"') def step_run_rollback_with_format(context: Context, plan_id: str, fmt: str) -> None: - """Run the rollback command with a specific format.""" + """Run the rollback (list mode) with a specific output format.""" context.plan_id = plan_id context.format = fmt - context.command = ['agents', 'plan', 'rollback', plan_id, '--format', fmt] - context.last_command = ' '.join(context.command) - _invoke_command(context) + _invoke_rollback(context, ["rollback", plan_id, "--format", fmt]) -@when('I run "agents plan rollback {plan_id} {checkpoint_id} --yes --format {fmt}"') -def step_run_rollback_restore_with_format(context: Context, plan_id: str, checkpoint_id: str, fmt: str) -> None: - """Run the rollback command in restore mode with format.""" +@when( + 'I run "agents plan rollback {plan_id:S} {checkpoint_id:S} --yes --format {fmt:S}"' +) +def step_run_rollback_restore_with_format( + context: Context, plan_id: str, checkpoint_id: str, fmt: str +) -> None: + """Run the rollback command in restore mode with a specific format.""" context.plan_id = plan_id context.checkpoint_id = checkpoint_id context.format = fmt - context.command = ['agents', 'plan', 'rollback', plan_id, checkpoint_id, '--yes', '--format', fmt] - context.last_command = ' '.join(context.command) - _invoke_command(context) + _invoke_rollback( + context, ["rollback", plan_id, checkpoint_id, "--yes", "--format", fmt] + ) -@then('the output should list {count:d} checkpoints') +@when( + 'I run "agents plan rollback {plan_id:S} --to-checkpoint {checkpoint_id:S} --yes"' +) +def step_run_rollback_to_checkpoint( + context: Context, plan_id: str, checkpoint_id: str +) -> None: + """Run the rollback command using the --to-checkpoint named option.""" + context.plan_id = plan_id + context.checkpoint_id = checkpoint_id + _invoke_rollback( + context, ["rollback", plan_id, "--to-checkpoint", checkpoint_id, "--yes"] + ) + + +# ---------------------------------------------------------------------- +# Then steps +# ---------------------------------------------------------------------- + + +@then("the output should list {count:d} checkpoints") def step_output_lists_checkpoints(context: Context, count: int) -> None: """Verify the output lists the expected number of checkpoints.""" output = context.last_stdout - checkpoint_count = output.count('01BRZ4NFEKTSV4RRFFQ69G5F') - assert checkpoint_count == count, f"Expected {count} checkpoints, found {checkpoint_count}" + checkpoint_count = output.count("01BRZ4NFEKTSV4RRFFQ69G5F") + assert checkpoint_count == count, ( + f"Expected {count} checkpoints, found {checkpoint_count}" + ) -@then('each checkpoint should show ID, timestamp, and metadata') +@then("each checkpoint should show ID, timestamp, and metadata") def step_checkpoint_shows_metadata(context: Context) -> None: """Verify each checkpoint shows required metadata.""" output = context.last_stdout - assert '01BRZ4NFEKTSV4RRFFQ69G5F' in output, "Checkpoint ID not found in output" - assert '2026-04-15' in output or 'checkpoint' in output.lower(), "Checkpoint metadata not found" + assert "01BRZ4NFEKTSV4RRFFQ69G5F" in output, "Checkpoint ID not found in output" + assert "2026-04-15" in output or "checkpoint" in output.lower(), ( + "Checkpoint metadata not found" + ) -@then('the output should indicate no checkpoints available') +@then("the output should indicate no checkpoints available") def step_output_no_checkpoints(context: Context) -> None: - """Verify the output indicates no checkpoints.""" + """Verify the output indicates no checkpoints are available.""" output = context.last_stdout.lower() - assert any(phrase in output for phrase in [ - 'no checkpoints', - 'empty', - '0 checkpoints', - ]), f"Output does not indicate no checkpoints: {context.last_stdout}" + assert any( + phrase in output for phrase in ["no checkpoints", "empty", "0 checkpoints"] + ), f"Output does not indicate no checkpoints: {context.last_stdout}" -@then('the output should be formatted as a rich table') +@then("the output should be formatted as a rich table") def step_output_rich_table(context: Context) -> None: - """Verify the output is formatted as a rich table.""" - output = context.last_stdout - assert len(output) > 0, "Output is empty" + """Verify the output is non-empty (rich table rendered).""" + assert len(context.last_stdout) > 0, "Output is empty" -@then('the table should contain checkpoint IDs and metadata') +@then("the table should contain checkpoint IDs and metadata") def step_table_contains_metadata(context: Context) -> None: - """Verify the table contains checkpoint IDs and metadata.""" - output = context.last_stdout - assert '01BRZ4NFEKTSV4RRFFQ69G5F' in output, "Checkpoint ID not in table" + """Verify the table contains checkpoint IDs.""" + assert "01BRZ4NFEKTSV4RRFFQ69G5F" in context.last_stdout, ( + "Checkpoint ID not in table" + ) -@then('the output should be valid JSON') +@then("the rollback output should be valid JSON") def step_output_valid_json(context: Context) -> None: """Verify the output is valid JSON.""" try: @@ -200,15 +351,16 @@ def step_output_valid_json(context: Context) -> None: raise AssertionError(f"Output is not valid JSON: {e}") from e -@then('the JSON should contain a list of checkpoints') +@then("the JSON should contain a list of checkpoints") def step_json_contains_checkpoints(context: Context) -> None: - """Verify the JSON contains a list of checkpoints.""" + """Verify the JSON contains a checkpoints list.""" assert isinstance(context.json_output, dict), "JSON output should be a dict" - assert 'checkpoints' in context.json_output, "JSON should contain 'checkpoints' key" - assert isinstance(context.json_output['checkpoints'], list), "Checkpoints should be a list" + inner = context.json_output.get("data", context.json_output) + assert "checkpoints" in inner, "JSON should contain 'checkpoints' key" + assert isinstance(inner["checkpoints"], list), "Checkpoints should be a list" -@then('the output should be valid YAML') +@then("the rollback output should be valid YAML") def step_output_valid_yaml(context: Context) -> None: """Verify the output is valid YAML.""" try: @@ -217,59 +369,60 @@ def step_output_valid_yaml(context: Context) -> None: raise AssertionError(f"Output is not valid YAML: {e}") from e -@then('the YAML should contain a list of checkpoints') +@then("the YAML should contain a list of checkpoints") def step_yaml_contains_checkpoints(context: Context) -> None: - """Verify the YAML contains a list of checkpoints.""" + """Verify the YAML contains a checkpoints list.""" assert isinstance(context.yaml_output, dict), "YAML output should be a dict" - assert 'checkpoints' in context.yaml_output, "YAML should contain 'checkpoints' key" - assert isinstance(context.yaml_output['checkpoints'], list), "Checkpoints should be a list" + inner = context.yaml_output.get("data", context.yaml_output) + assert "checkpoints" in inner, "YAML should contain 'checkpoints' key" + assert isinstance(inner["checkpoints"], list), "Checkpoints should be a list" -@then('the output should be plain text') +@then("the output should be plain text") def step_output_plain_text(context: Context) -> None: - """Verify the output is plain text.""" + """Verify the output is non-empty plain text.""" assert len(context.last_stdout) > 0, "Output is empty" -@then('the text should contain checkpoint IDs') +@then("the text should contain checkpoint IDs") def step_text_contains_checkpoint_ids(context: Context) -> None: """Verify the text contains checkpoint IDs.""" - assert '01BRZ4NFEKTSV4RRFFQ69G5F' in context.last_stdout, "Checkpoint ID not in output" + assert "01BRZ4NFEKTSV4RRFFQ69G5F" in context.last_stdout, ( + "Checkpoint ID not in output" + ) -@then('the rollback should succeed') +@then("the plan rollback should succeed") def step_rollback_succeeds(context: Context) -> None: - """Verify the rollback succeeded.""" - assert context.last_exit_code == 0, f"Rollback failed with exit code {context.last_exit_code}: {context.last_stderr}" + """Verify the rollback succeeded (exit code 0).""" + assert context.last_exit_code == 0, ( + f"Rollback failed with exit code {context.last_exit_code}: {context.last_stdout}" + ) -@then('the output should show rollback summary') +@then("the output should show rollback summary") def step_output_shows_summary(context: Context) -> None: """Verify the output shows a rollback summary.""" output = context.last_stdout.lower() - assert any(phrase in output for phrase in [ - 'rollback', - 'restored', - 'summary', - ]), f"Output does not show rollback summary: {context.last_stdout}" + assert any(phrase in output for phrase in ["rollback", "restored", "summary"]), ( + f"Output does not show rollback summary: {context.last_stdout}" + ) -@then('no confirmation prompt should be shown') +@then("no rollback confirmation prompt should be shown") def step_no_confirmation_prompt(context: Context) -> None: - """Verify no confirmation prompt was shown.""" + """Verify the --yes flag bypassed the confirmation prompt.""" assert context.last_exit_code == 0, "Command should succeed with --yes flag" -@then('the command should fail with not found error') +@then("the command should fail with not found error") def step_command_fails_not_found(context: Context) -> None: - """Verify the command fails with not found error.""" + """Verify the command fails with a not-found error.""" assert context.last_exit_code != 0, "Command should fail" error_output = (context.last_stdout + context.last_stderr).lower() - assert any(phrase in error_output for phrase in [ - 'not found', - 'does not exist', - 'not found', - ]), f"Error message does not indicate not found: {context.last_stderr}" + assert any(phrase in error_output for phrase in ["not found", "does not exist"]), ( + f"Error message does not indicate not found: {context.last_stdout}" + ) @then('the command should fail with "{error_message}"') @@ -277,64 +430,87 @@ def step_command_fails_with_message(context: Context, error_message: str) -> Non """Verify the command fails with a specific error message.""" assert context.last_exit_code != 0, "Command should fail" error_output = (context.last_stdout + context.last_stderr).lower() - assert error_message.lower() in error_output, f"Error message '{error_message}' not found in: {context.last_stderr}" + assert error_message.lower() in error_output, ( + f"Error message '{error_message}' not found in: {context.last_stdout}" + ) -@then('the JSON should contain rollback summary') +@then("the JSON should contain rollback summary") def step_json_contains_rollback_summary(context: Context) -> None: - """Verify the JSON contains rollback summary.""" + """Verify the JSON contains a rollback_summary key.""" assert isinstance(context.json_output, dict), "JSON output should be a dict" - assert 'rollback_summary' in context.json_output, "JSON should contain 'rollback_summary' key" + inner = context.json_output.get("data", context.json_output) + assert "rollback_summary" in inner, "JSON should contain 'rollback_summary' key" -@then('the YAML should contain rollback summary') +@then("the YAML should contain rollback summary") def step_yaml_contains_rollback_summary(context: Context) -> None: - """Verify the YAML contains rollback summary.""" + """Verify the YAML contains a rollback_summary key.""" assert isinstance(context.yaml_output, dict), "YAML output should be a dict" - assert 'rollback_summary' in context.yaml_output, "YAML should contain 'rollback_summary' key" + inner = context.yaml_output.get("data", context.yaml_output) + assert "rollback_summary" in inner, "YAML should contain 'rollback_summary' key" -@then('the rollback should be atomic') +@then("the rollback should be atomic") def step_rollback_atomic(context: Context) -> None: - """Verify the rollback is atomic.""" + """Verify the rollback completed atomically (exit code 0).""" assert context.last_exit_code == 0, "Rollback should succeed atomically" -@then('the plan state should be consistent') +@then("the plan state should be consistent") def step_plan_state_consistent(context: Context) -> None: - """Verify the plan state is consistent.""" + """Verify the plan state is consistent after rollback.""" assert context.last_exit_code == 0, "Plan state should be consistent after rollback" -@then('the output should show the number of restored files') +@then("the output should show the number of restored files") def step_output_shows_restored_files(context: Context) -> None: - """Verify the output shows the number of restored files.""" + """Verify the output mentions restored files.""" output = context.last_stdout.lower() - assert any(phrase in output for phrase in [ - 'restored', - 'files', - 'changes', - ]), f"Output does not show restored files: {context.last_stdout}" + assert any(phrase in output for phrase in ["restored", "files", "changes"]), ( + f"Output does not show restored files: {context.last_stdout}" + ) -@then('the output should list changed paths') +@then("the output should list changed paths") def step_output_lists_changed_paths(context: Context) -> None: """Verify the output lists changed paths.""" output = context.last_stdout.lower() - assert any(phrase in output for phrase in [ - 'changed', - 'paths', - 'files', - ]), f"Output does not list changed paths: {context.last_stdout}" + assert any(phrase in output for phrase in ["changed", "paths", "files"]), ( + f"Output does not list changed paths: {context.last_stdout}" + ) -@then('the command should fail with validation error') +@then("the command should fail with validation error") def step_command_fails_validation(context: Context) -> None: - """Verify the command fails with validation error.""" + """Verify the command fails with a validation error.""" assert context.last_exit_code != 0, "Command should fail" error_output = (context.last_stdout + context.last_stderr).lower() - assert any(phrase in error_output for phrase in [ - 'invalid', - 'validation', - 'error', - ]), f"Error message does not indicate validation error: {context.last_stderr}" + assert any( + phrase in error_output for phrase in ["invalid", "validation", "error"] + ), f"Error message does not indicate validation error: {context.last_stdout}" + + +@then("the output should list available checkpoints") +def step_output_lists_available_checkpoints(context: Context) -> None: + """Verify the output lists available checkpoints (list mode).""" + output = context.last_stdout.lower() + assert any(phrase in output for phrase in ["checkpoint", "no checkpoints"]), ( + f"Output does not list checkpoints: {context.last_stdout}" + ) + + +@then("the rollback should be cancelled") +def step_rollback_cancelled(context: Context) -> None: + """Verify the rollback was cancelled by the user.""" + assert context.last_exit_code != 0, "Rollback should be cancelled" + output = context.last_stdout.lower() + assert any(phrase in output for phrase in ["cancelled", "aborted"]), ( + f"Output does not indicate rollback was cancelled: {context.last_stdout}" + ) + + +@then("the plan state should remain unchanged") +def step_plan_state_unchanged(context: Context) -> None: + """Verify the plan state did not change (rollback was not completed).""" + assert context.last_exit_code != 0, "Rollback should not have completed" diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 0dcd15c97..3cead2b5f 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -3967,6 +3967,7 @@ def rollback_plan( from cleveragents.application.container import get_container from cleveragents.core.exceptions import ( BusinessRuleViolation, + CleverAgentsError, ) from cleveragents.core.exceptions import ( ResourceNotFoundError as RNF, @@ -3975,11 +3976,45 @@ def rollback_plan( # Resolve checkpoint from positional arg or --to-checkpoint option resolved_checkpoint_id = checkpoint_id or to_checkpoint if not resolved_checkpoint_id: - console.print( - "[red]Error:[/red] checkpoint ID required. " - "Provide as positional argument or via --to-checkpoint." - ) - raise typer.Abort() + # List mode: no checkpoint ID given — show available checkpoints + container = get_container() + svc = container.checkpoint_service() + try: + checkpoints = svc.list_checkpoints(plan_id) + except RNF as exc: + console.print(f"[red]Not found:[/red] {exc.message}") + raise typer.Abort() from exc + except CleverAgentsError as exc: + console.print(f"[red]Error:[/red] {exc.message}") + raise typer.Abort() from exc + checkpoints = sorted(checkpoints, key=lambda cp: cp.created_at) + if fmt != OutputFormat.RICH.value: + data = { + "checkpoints": [ + { + "id": cp.checkpoint_id, + "created_at": cp.created_at.isoformat(), + "reason": cp.metadata.reason, + } + for cp in checkpoints + ] + } + console.print(format_output(data, fmt)) + elif not checkpoints: + console.print(f"[dim]No checkpoints available for plan {plan_id}.[/dim]") + else: + table = Table(show_header=True, expand=False) + table.add_column("ID", style="cyan") + table.add_column("Created", style="green") + table.add_column("Reason") + for cp in checkpoints: + table.add_row( + cp.checkpoint_id, + cp.created_at.isoformat(), + cp.metadata.reason or "", + ) + console.print(table) + return container = get_container() # Use PlanLifecycleService for the actual rollback (service layer pattern).