freemo
7fb3fc76c8
- What was implemented
- Added PLAN_ROLLED_BACK event type to the EventType enum at src/cleveragents/infrastructure/events/types.py to properly represent successful rollbacks in the domain model.
- Implemented rollback_plan(plan_id: str, checkpoint_id: str) -> RollbackResult in PlanLifecycleService (src/cleveragents/application/services/plan_lifecycle_service.py) with:
- Plan state validation: rejects rollback when the plan is in terminal APPLIED or CANCELLED states.
- Delegation to CheckpointService.selective_rollback() to perform the actual rollback logic and obtain a RollbackResult.
- Emission of PLAN_ROLLED_BACK as a domain event to reflect the completed rollback.
- checkpoint_service is accepted as an optional constructor parameter; if not provided, a PlanError is raised to preserve backward compatibility.
- Updated CLI behavior in src/cleveragents/cli/commands/plan.py so agents plan rollback routes through PlanLifecycleService.rollback_plan() rather than calling CheckpointService.selective_rollback() directly.
- Updated PlanLifecycleService module docstring to include rollback_plan in the documented API.
- Added Behave feature file features/plan_lifecycle_rollback.feature with 11 scenarios covering state validation, domain events, and delegation.
- Added step implementations in features/steps/plan_lifecycle_rollback_steps.py to support the new scenarios.
- Key design decisions
- rollback_plan returns RollbackResult (the same result type produced by CheckpointService.selective_rollback) so the CLI can display rollback details consistently.
- Terminal states APPLIED and CANCELLED are disallowed for rollback to prevent inconsistent or invalid state transitions.
- checkpoint_service is optional in the PlanLifecycleService constructor; when omitted (None), a PlanError is raised to retain backward compatibility while signaling explicit dependency requirements.
- CLI UI remains powered by CheckpointService for metadata enrichment (e.g., confirmation prompts), but the actual rollback action is performed via PlanLifecycleService to ensure proper domain workflow and event emission.
- Technical implications
- All rollback logic now flows through the domain service layer (PlanLifecycleService) to preserve invariants and emit domain events, rather than allowing ad-hoc UI routes to bypass service validation.
- The UI can still retrieve checkpoint metadata for user confirmation, but the operation that modifies state uses the new rollback_plan pathway.
- Tests and behavior coverage were expanded via the new Behave feature and step implementations to validate state handling, events, and delegation.
- Affected modules/components
- src/cleveragents/infrastructure/events/types.py
- src/cleveragents/application/services/plan_lifecycle_service.py
- src/cleveragents/cli/commands/plan.py
- PlanLifecycleService module docstring
- features/plan_lifecycle_rollback.feature
- features/steps/plan_lifecycle_rollback_steps.py
ISSUES CLOSED: #3677
340 lines
13 KiB
Python
340 lines
13 KiB
Python
"""Helper script for cli_formats.robot smoke tests.
|
|
|
|
Each subcommand is a self-contained check that prints a sentinel on success.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import yaml
|
|
|
|
# Ensure local source tree is importable
|
|
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
from typer.testing import CliRunner # noqa: E402
|
|
|
|
from cleveragents.cli.commands.action import app as action_app # noqa: E402
|
|
from cleveragents.cli.commands.plan import app as plan_app # noqa: E402
|
|
from cleveragents.domain.models.core.action import Action, ActionState # noqa: E402
|
|
from cleveragents.domain.models.core.plan import ( # noqa: E402
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
runner = CliRunner()
|
|
_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
|
|
|
|
def _mock_action(name: str = "local/fmt-smoke") -> Action:
|
|
return Action(
|
|
namespaced_name=NamespacedName.parse(name),
|
|
description="Smoke test action",
|
|
long_description=None,
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
reusable=True,
|
|
read_only=False,
|
|
state=ActionState.AVAILABLE,
|
|
created_by=None,
|
|
created_at=datetime(2025, 1, 15, 10, 0, 0),
|
|
updated_at=datetime(2025, 1, 15, 10, 0, 0),
|
|
)
|
|
|
|
|
|
def _mock_plan(name: str = "local/fmt-smoke-plan") -> Plan:
|
|
now = datetime(2025, 1, 15, 10, 0, 0)
|
|
return Plan(
|
|
identity=PlanIdentity(plan_id=_ULID),
|
|
namespaced_name=NamespacedName.parse(name),
|
|
description="Smoke plan",
|
|
definition_of_done="Tests pass",
|
|
action_name="local/fmt-smoke",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
project_links=[ProjectLink(project_name="proj-a")],
|
|
timestamps=PlanTimestamps(created_at=now, updated_at=now),
|
|
created_by=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
def action_list_json() -> None:
|
|
svc = MagicMock()
|
|
svc.list_actions.return_value = [_mock_action("local/a1"), _mock_action("local/a2")]
|
|
with patch(
|
|
"cleveragents.cli.commands.action._get_lifecycle_service",
|
|
return_value=svc,
|
|
):
|
|
result = runner.invoke(action_app, ["list", "--format", "json"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
envelope = json.loads(result.output.strip())
|
|
# Output is wrapped in the spec-required envelope; actions array is under "data"
|
|
parsed = envelope.get("data", envelope)
|
|
assert isinstance(parsed, list) and len(parsed) == 2, (
|
|
f"Expected list of 2 actions, got: {type(parsed).__name__} "
|
|
f"with value: {parsed!r}"
|
|
)
|
|
print("cli-formats-action-list-json-ok")
|
|
|
|
|
|
def action_show_yaml() -> None:
|
|
svc = MagicMock()
|
|
svc.get_action_by_name.return_value = _mock_action()
|
|
with patch(
|
|
"cleveragents.cli.commands.action._get_lifecycle_service",
|
|
return_value=svc,
|
|
):
|
|
result = runner.invoke(
|
|
action_app, ["show", "local/fmt-smoke", "--format", "yaml"]
|
|
)
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
envelope = yaml.safe_load(result.output.strip())
|
|
# Output is wrapped in the spec-required envelope; action data is under "data"
|
|
parsed = envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
|
|
assert "namespaced_name" in parsed, (
|
|
f"Expected 'namespaced_name' in action data. Keys: "
|
|
f"{list(parsed.keys()) if isinstance(parsed, dict) else type(parsed)}"
|
|
)
|
|
print("cli-formats-action-show-yaml-ok")
|
|
|
|
|
|
def plan_list_json() -> None:
|
|
svc = MagicMock()
|
|
svc.list_plans.return_value = [_mock_plan()]
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=svc,
|
|
):
|
|
result = runner.invoke(plan_app, ["list", "--format", "json"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
envelope = json.loads(result.output.strip())
|
|
# Output is wrapped in the spec-required envelope; plans array is under "data"
|
|
parsed = envelope.get("data", envelope)
|
|
assert isinstance(parsed, list), (
|
|
f"Expected list for plan data, got {type(parsed).__name__}. "
|
|
f"Output: {result.output[:300]}"
|
|
)
|
|
assert "plan_id" in parsed[0], (
|
|
f"Expected 'plan_id' in first plan. Keys: {list(parsed[0].keys())}"
|
|
)
|
|
print("cli-formats-plan-list-json-ok")
|
|
|
|
|
|
def plan_status_plain() -> None:
|
|
svc = MagicMock()
|
|
svc.get_plan.return_value = _mock_plan()
|
|
with patch(
|
|
"cleveragents.cli.commands.plan._get_lifecycle_service",
|
|
return_value=svc,
|
|
):
|
|
result = runner.invoke(plan_app, ["status", _ULID, "--format", "plain"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
assert "plan_id:" in result.output
|
|
assert "processing_state:" in result.output
|
|
print("cli-formats-plan-status-plain-ok")
|
|
|
|
|
|
def global_format_json_version() -> None:
|
|
"""Test global --format json propagates to version command."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
result = runner.invoke(main_app, ["--format", "json", "version"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
envelope = json.loads(result.output.strip())
|
|
assert isinstance(envelope, dict)
|
|
# Output is wrapped in the spec-required envelope; version data is under "data"
|
|
parsed = envelope.get("data", envelope)
|
|
assert isinstance(parsed, dict), (
|
|
f"Expected dict for version data, got {type(parsed).__name__}"
|
|
)
|
|
assert "version" in parsed, (
|
|
f"Expected 'version' in version data. Keys: {list(parsed.keys())}"
|
|
)
|
|
print("cli-global-format-json-version-ok")
|
|
|
|
|
|
def global_format_yaml_version() -> None:
|
|
"""Test global --format yaml propagates to version command."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
result = runner.invoke(main_app, ["--format", "yaml", "version"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
envelope = yaml.safe_load(result.output.strip())
|
|
assert isinstance(envelope, dict)
|
|
# Output is wrapped in the spec-required envelope; version data is under "data"
|
|
parsed = envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
|
|
assert isinstance(parsed, dict), (
|
|
f"Expected dict for version data, got {type(parsed).__name__}"
|
|
)
|
|
assert "version" in parsed, (
|
|
f"Expected 'version' in version data. Keys: "
|
|
f"{list(parsed.keys()) if isinstance(parsed, dict) else type(parsed)}"
|
|
)
|
|
print("cli-global-format-yaml-version-ok")
|
|
|
|
|
|
def global_format_plain_version() -> None:
|
|
"""Test global --format plain propagates to version command."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
result = runner.invoke(main_app, ["--format", "plain", "version"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
assert "version:" in result.output
|
|
print("cli-global-format-plain-version-ok")
|
|
|
|
|
|
def global_format_json_info() -> None:
|
|
"""Test global --format json propagates to info command."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
result = runner.invoke(main_app, ["--format", "json", "info"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
assert isinstance(parsed, dict)
|
|
print("cli-global-format-json-info-ok")
|
|
|
|
|
|
def global_format_json_diagnostics() -> None:
|
|
"""Test global --format json propagates to diagnostics command."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
result = runner.invoke(main_app, ["--format", "json", "diagnostics"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
assert isinstance(parsed, dict)
|
|
print("cli-global-format-json-diagnostics-ok")
|
|
|
|
|
|
def global_format_shorthand() -> None:
|
|
"""Test global -f json shorthand propagates to version command."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
result = runner.invoke(main_app, ["-f", "json", "version"])
|
|
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
|
|
envelope = json.loads(result.output.strip())
|
|
assert isinstance(envelope, dict)
|
|
# Output is wrapped in the spec-required envelope; version data is under "data"
|
|
parsed = envelope.get("data", envelope)
|
|
assert isinstance(parsed, dict), (
|
|
f"Expected dict for version data, got {type(parsed).__name__}"
|
|
)
|
|
assert "version" in parsed, (
|
|
f"Expected 'version' in version data. Keys: {list(parsed.keys())}"
|
|
)
|
|
print("cli-global-format-shorthand-ok")
|
|
|
|
|
|
def global_format_all_six() -> None:
|
|
"""Test all six output formats work via the global --format flag."""
|
|
from cleveragents.cli.main import app as main_app
|
|
|
|
formats = ["json", "yaml", "plain", "rich", "table", "color"]
|
|
for fmt in formats:
|
|
result = runner.invoke(main_app, ["--format", fmt, "version"])
|
|
assert result.exit_code == 0, (
|
|
f"Format '{fmt}' failed: exit={result.exit_code}: {result.output}"
|
|
)
|
|
print("cli-global-format-all-six-ok")
|
|
|
|
|
|
def format_output_rich() -> None:
|
|
"""Verify format_output(data, 'rich') produces styled output, not JSON.
|
|
|
|
Regression test for issue #2921: rich format silently fell back to JSON.
|
|
"""
|
|
from cleveragents.cli.formatting import format_output
|
|
|
|
data = {"name": "smoke-test", "status": "active", "count": 42}
|
|
result = format_output(data, "rich")
|
|
|
|
# Must NOT be valid JSON (that was the bug)
|
|
try:
|
|
json.loads(result)
|
|
raise AssertionError(
|
|
f"format_output(data, 'rich') returned valid JSON — "
|
|
f"expected styled terminal output, got: {result!r}"
|
|
)
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass # Expected: rich output is not JSON
|
|
|
|
# Must contain the data keys
|
|
assert "name" in result, f"Rich output missing 'name': {result!r}"
|
|
assert "status" in result, f"Rich output missing 'status': {result!r}"
|
|
|
|
# Must contain ANSI codes or panel header (styled output)
|
|
has_ansi = "\033[" in result or "\x1b[" in result
|
|
has_panel = "Output" in result
|
|
assert has_ansi or has_panel, (
|
|
f"Rich output has no ANSI codes or panel header: {result!r}"
|
|
)
|
|
print("cli-formats-format-output-rich-ok")
|
|
|
|
|
|
def format_output_color() -> None:
|
|
"""Verify format_output(data, 'color') produces ANSI-colored output, not plain text.
|
|
|
|
Regression test for issue #2921: color format used plain renderer instead of color.
|
|
"""
|
|
from cleveragents.cli.formatting import format_output
|
|
|
|
data = {"name": "smoke-test", "status": "active", "count": 42}
|
|
result = format_output(data, "color")
|
|
|
|
# Must contain the data keys
|
|
assert "name" in result, f"Color output missing 'name': {result!r}"
|
|
assert "status" in result, f"Color output missing 'status': {result!r}"
|
|
|
|
# Must NOT be plain text only (that was the bug)
|
|
plain_text = "name: smoke-test\nstatus: active\ncount: 42"
|
|
assert result != plain_text, (
|
|
f"format_output(data, 'color') returned plain text — "
|
|
f"expected ANSI-colored output, got: {result!r}"
|
|
)
|
|
|
|
# Must contain ANSI codes or panel header (styled output)
|
|
has_ansi = "\033[" in result or "\x1b[" in result
|
|
has_panel = "Output" in result
|
|
assert has_ansi or has_panel, (
|
|
f"Color output has no ANSI codes or panel header: {result!r}"
|
|
)
|
|
print("cli-formats-format-output-color-ok")
|
|
|
|
|
|
_COMMANDS = {
|
|
"action-list-json": action_list_json,
|
|
"action-show-yaml": action_show_yaml,
|
|
"plan-list-json": plan_list_json,
|
|
"plan-status-plain": plan_status_plain,
|
|
"global-format-json-version": global_format_json_version,
|
|
"global-format-yaml-version": global_format_yaml_version,
|
|
"global-format-plain-version": global_format_plain_version,
|
|
"global-format-json-info": global_format_json_info,
|
|
"global-format-json-diagnostics": global_format_json_diagnostics,
|
|
"global-format-shorthand": global_format_shorthand,
|
|
"global-format-all-six": global_format_all_six,
|
|
"format-output-rich": format_output_rich,
|
|
"format-output-color": format_output_color,
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
|
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
|
|
sys.exit(2)
|
|
_COMMANDS[sys.argv[1]]()
|