forked from HAL9000/cleveragents-core
freemo
2eb31a598c
Added targeted Behave BDD feature files and step definitions to improve unit test coverage for: - decision_service.py: Full coverage of all 7 service methods (18 scenarios) - plan_apply_service.py: Branch coverage for handle_merge_failure (2 scenarios) - plan_executor.py: Edge cases for rollback, checkpoint, and parse_steps (15 scenarios) - cli/commands/plan.py: Uncovered region lines 1950-2273 (23 scenarios) - repositories.py: Remaining missed branches and lines (14 scenarios) - sandbox/checkpoint.py: Full coverage of CheckpointManager (26 scenarios) - langgraph/bridge.py: Remaining uncovered lines and branches (10 scenarios) - cli/commands/config.py: Safety net to maintain 100% coverage (42 scenarios) Total: 150 new scenarios, 596 steps, all passing. Also fixed a step definition collision in plan_lifecycle_coverage by renaming "the delete result should be false" to "the plan delete result should be false". ISSUES CLOSED: #475
570 lines
22 KiB
Python
570 lines
22 KiB
Python
"""Step definitions for plan_executor_edge_cases_coverage.feature.
|
|
|
|
Targets uncovered lines and branches in plan_executor.py:
|
|
- Line 378 / branch 377→378: _try_rollback_to_last_checkpoint with non-empty checkpoints
|
|
- Lines 410-411: _resolve_sandbox_for_checkpoint → _SandboxRootProxy fallback
|
|
- Lines 161-163 / branches 160→161, 162→163: _parse_steps non-empty path
|
|
- Lines 417-419: run_strategize function definition
|
|
|
|
All step names use the 'edge3' prefix to avoid collisions with existing step files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.services.plan_executor import (
|
|
ExecuteResult,
|
|
PlanExecutor,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
from cleveragents.infrastructure.sandbox.checkpoint import SandboxCheckpoint
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
EDGE3_PLAN_ID = "01KEDGE3PLANID00000000PLN"
|
|
EDGE3_ROOT_ID = "01KEDGE3ROOTID00000000RTD"
|
|
EDGE3_SANDBOX_ROOT = "/tmp/edge3-sandbox"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _edge3_make_plan(
|
|
*,
|
|
phase: PlanPhase = PlanPhase.EXECUTE,
|
|
state: ProcessingState = ProcessingState.QUEUED,
|
|
definition_of_done: str | None = "Implement feature\nWrite tests",
|
|
decision_root_id: str | None = EDGE3_ROOT_ID,
|
|
) -> MagicMock:
|
|
"""Build a mock plan object with sensible defaults."""
|
|
plan = MagicMock()
|
|
plan.phase = phase
|
|
plan.state = state
|
|
plan.definition_of_done = definition_of_done
|
|
plan.decision_root_id = decision_root_id
|
|
plan.invariants = []
|
|
plan.timestamps = PlanTimestamps()
|
|
plan.changeset_id = None
|
|
plan.sandbox_refs = []
|
|
plan.error_details = None
|
|
plan.read_only = False
|
|
return plan
|
|
|
|
|
|
def _edge3_make_lifecycle(plan: Any | None = None) -> MagicMock:
|
|
"""Build a mock lifecycle service."""
|
|
lcs = MagicMock()
|
|
if plan is not None:
|
|
lcs.get_plan.return_value = plan
|
|
lcs.start_strategize = MagicMock()
|
|
lcs.complete_strategize = MagicMock()
|
|
lcs.fail_strategize = MagicMock()
|
|
lcs.start_execute = MagicMock()
|
|
lcs.complete_execute = MagicMock()
|
|
lcs.fail_execute = MagicMock()
|
|
lcs._commit_plan = MagicMock()
|
|
return lcs
|
|
|
|
|
|
def _edge3_make_checkpoint_mock() -> SandboxCheckpoint:
|
|
"""Create a mock SandboxCheckpoint."""
|
|
from datetime import UTC, datetime
|
|
|
|
return SandboxCheckpoint(
|
|
checkpoint_id="01KEDGE3CPID0000000000CP",
|
|
sandbox_id="root-EDGE3PLAN",
|
|
plan_id="EDGE3PLAN",
|
|
phase="pre_execute",
|
|
created_at=datetime.now(tz=UTC),
|
|
metadata={},
|
|
snapshot_path="/tmp/edge3-snapshot",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_sandbox_for_checkpoint → _SandboxRootProxy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with checkpoint manager and sandbox root but no execution context"
|
|
)
|
|
def step_edge3_given_executor_sandbox_root_no_ctx(context: Context) -> None:
|
|
"""Create PlanExecutor with checkpoint_manager and sandbox_root, no exec context."""
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=_edge3_make_lifecycle(),
|
|
sandbox_root=EDGE3_SANDBOX_ROOT,
|
|
execution_context=None,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with checkpoint manager and sandbox root and context without sandbox manager"
|
|
)
|
|
def step_edge3_given_executor_ctx_no_sbmgr(context: Context) -> None:
|
|
"""Create PlanExecutor with exec context that has no sandbox_manager attr."""
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
# Execution context without sandbox_manager attribute
|
|
exec_ctx = MagicMock(spec=[]) # empty spec → no attributes
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=_edge3_make_lifecycle(),
|
|
sandbox_root=EDGE3_SANDBOX_ROOT,
|
|
execution_context=exec_ctx,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with checkpoint manager and sandbox root and context with empty sandboxes"
|
|
)
|
|
def step_edge3_given_executor_ctx_empty_sandboxes(context: Context) -> None:
|
|
"""Create PlanExecutor with exec context whose sandbox_manager returns []."""
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
exec_ctx = MagicMock()
|
|
exec_ctx.sandbox_manager = MagicMock()
|
|
exec_ctx.sandbox_manager.list_sandboxes.return_value = []
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=_edge3_make_lifecycle(),
|
|
sandbox_root=EDGE3_SANDBOX_ROOT,
|
|
execution_context=exec_ctx,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with checkpoint manager but no sandbox root and no execution context"
|
|
)
|
|
def step_edge3_given_executor_no_sandbox_no_ctx(context: Context) -> None:
|
|
"""Create PlanExecutor with checkpoint_manager but no sandbox_root and no exec context."""
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=_edge3_make_lifecycle(),
|
|
sandbox_root=None,
|
|
execution_context=None,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
|
|
|
|
@when('I edge3 resolve sandbox for checkpoint with plan id "{plan_id}"')
|
|
def step_edge3_resolve_sandbox(context: Context, plan_id: str) -> None:
|
|
"""Call _resolve_sandbox_for_checkpoint directly."""
|
|
context.edge3_resolved_sandbox = (
|
|
context.edge3_executor._resolve_sandbox_for_checkpoint(plan_id)
|
|
)
|
|
|
|
|
|
@then("the edge3 resolved sandbox should not be None")
|
|
def step_edge3_resolved_not_none(context: Context) -> None:
|
|
"""Verify resolved sandbox is not None."""
|
|
assert context.edge3_resolved_sandbox is not None, (
|
|
"Expected a sandbox-like object but got None"
|
|
)
|
|
|
|
|
|
@then("the edge3 resolved sandbox should have a synthetic sandbox id")
|
|
def step_edge3_resolved_has_sandbox_id(context: Context) -> None:
|
|
"""Verify the resolved sandbox has a sandbox_id starting with 'root-'."""
|
|
sid = context.edge3_resolved_sandbox.sandbox_id
|
|
assert sid.startswith("root-"), (
|
|
f"Expected sandbox_id starting with 'root-', got '{sid}'"
|
|
)
|
|
|
|
|
|
@then("the edge3 resolved sandbox context should have the sandbox path")
|
|
def step_edge3_resolved_has_sandbox_path(context: Context) -> None:
|
|
"""Verify the resolved sandbox context has the correct sandbox_path."""
|
|
ctx = context.edge3_resolved_sandbox.context
|
|
assert ctx is not None, "Expected context to be non-None"
|
|
assert ctx.sandbox_path == EDGE3_SANDBOX_ROOT, (
|
|
f"Expected sandbox_path='{EDGE3_SANDBOX_ROOT}', got '{ctx.sandbox_path}'"
|
|
)
|
|
|
|
|
|
@then("the edge3 resolved sandbox should be None")
|
|
def step_edge3_resolved_is_none(context: Context) -> None:
|
|
"""Verify resolved sandbox is None."""
|
|
assert context.edge3_resolved_sandbox is None, (
|
|
f"Expected None but got {context.edge3_resolved_sandbox}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _try_rollback_to_last_checkpoint with non-empty checkpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"the edge3 checkpoint manager has existing checkpoints that rollback successfully"
|
|
)
|
|
def step_edge3_given_checkpoints_rollback_ok(context: Context) -> None:
|
|
"""Configure checkpoint_manager to return checkpoints and succeed on rollback."""
|
|
cp = _edge3_make_checkpoint_mock()
|
|
context.edge3_checkpoint_mgr.list_checkpoints.return_value = [cp]
|
|
context.edge3_checkpoint_mgr.rollback_to.return_value = True
|
|
|
|
|
|
@given("the edge3 checkpoint manager has checkpoints but rollback raises an exception")
|
|
def step_edge3_given_checkpoints_rollback_raises(context: Context) -> None:
|
|
"""Configure checkpoint_manager with checkpoints but rollback_to raises."""
|
|
cp = _edge3_make_checkpoint_mock()
|
|
context.edge3_checkpoint_mgr.list_checkpoints.return_value = [cp]
|
|
context.edge3_checkpoint_mgr.rollback_to.side_effect = RuntimeError(
|
|
"edge3 rollback boom"
|
|
)
|
|
|
|
|
|
@given("the edge3 checkpoint manager returns no checkpoints")
|
|
def step_edge3_given_no_checkpoints(context: Context) -> None:
|
|
"""Configure checkpoint_manager to return empty checkpoints list."""
|
|
context.edge3_checkpoint_mgr.list_checkpoints.return_value = []
|
|
|
|
|
|
@when('I edge3 try rollback to last checkpoint for plan "{plan_id}"')
|
|
def step_edge3_try_rollback(context: Context, plan_id: str) -> None:
|
|
"""Call _try_rollback_to_last_checkpoint directly."""
|
|
context.edge3_rollback_result = (
|
|
context.edge3_executor._try_rollback_to_last_checkpoint(plan_id)
|
|
)
|
|
|
|
|
|
@then("the edge3 rollback result should be True")
|
|
def step_edge3_rollback_true(context: Context) -> None:
|
|
"""Verify rollback returned True."""
|
|
assert context.edge3_rollback_result is True, (
|
|
f"Expected True, got {context.edge3_rollback_result}"
|
|
)
|
|
|
|
|
|
@then("the edge3 rollback result should be False")
|
|
def step_edge3_rollback_false(context: Context) -> None:
|
|
"""Verify rollback returned False."""
|
|
assert context.edge3_rollback_result is False, (
|
|
f"Expected False, got {context.edge3_rollback_result}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _try_create_checkpoint via _SandboxRootProxy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the edge3 checkpoint manager accepts checkpoint creation")
|
|
def step_edge3_given_checkpoint_create_ok(context: Context) -> None:
|
|
"""Configure checkpoint_manager.create_checkpoint to return a checkpoint."""
|
|
cp = _edge3_make_checkpoint_mock()
|
|
context.edge3_checkpoint_mgr.create_checkpoint.return_value = cp
|
|
|
|
|
|
@given("the edge3 checkpoint manager raises on create_checkpoint")
|
|
def step_edge3_given_checkpoint_create_fails(context: Context) -> None:
|
|
"""Configure checkpoint_manager.create_checkpoint to raise."""
|
|
context.edge3_checkpoint_mgr.create_checkpoint.side_effect = RuntimeError(
|
|
"edge3 create boom"
|
|
)
|
|
|
|
|
|
@when('I edge3 try create checkpoint for plan "{plan_id}" with phase "{phase}"')
|
|
def step_edge3_try_create_checkpoint(
|
|
context: Context, plan_id: str, phase: str
|
|
) -> None:
|
|
"""Call _try_create_checkpoint directly."""
|
|
context.edge3_checkpoint_result = context.edge3_executor._try_create_checkpoint(
|
|
plan_id, phase
|
|
)
|
|
|
|
|
|
@then("the edge3 checkpoint result should not be None")
|
|
def step_edge3_checkpoint_not_none(context: Context) -> None:
|
|
"""Verify checkpoint creation returned a checkpoint."""
|
|
assert context.edge3_checkpoint_result is not None, (
|
|
"Expected a SandboxCheckpoint but got None"
|
|
)
|
|
|
|
|
|
@then("the edge3 checkpoint result should be None")
|
|
def step_edge3_checkpoint_is_none(context: Context) -> None:
|
|
"""Verify checkpoint creation returned None (non-fatal failure)."""
|
|
assert context.edge3_checkpoint_result is None, (
|
|
f"Expected None but got {context.edge3_checkpoint_result}"
|
|
)
|
|
|
|
|
|
@then("the edge3 checkpoint manager should have been called with sandbox path metadata")
|
|
def step_edge3_checkpoint_called_with_path(context: Context) -> None:
|
|
"""Verify create_checkpoint was called with sandbox_path in metadata."""
|
|
context.edge3_checkpoint_mgr.create_checkpoint.assert_called_once()
|
|
call_kwargs = context.edge3_checkpoint_mgr.create_checkpoint.call_args
|
|
# Keyword arg 'metadata' or positional
|
|
metadata = call_kwargs.kwargs.get("metadata") or call_kwargs[1].get("metadata", {})
|
|
assert "sandbox_path" in metadata, (
|
|
f"Expected 'sandbox_path' in metadata, got keys: {list(metadata.keys())}"
|
|
)
|
|
assert metadata["sandbox_path"] == EDGE3_SANDBOX_ROOT, (
|
|
f"Expected sandbox_path='{EDGE3_SANDBOX_ROOT}', got '{metadata['sandbox_path']}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stub execute with checkpoint rollback on failure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an edge3 mock lifecycle service for execute")
|
|
def step_edge3_given_lifecycle_execute(context: Context) -> None:
|
|
"""Create a mock lifecycle service for execute scenarios."""
|
|
context.edge3_lifecycle = _edge3_make_lifecycle()
|
|
context.edge3_plan_id = EDGE3_PLAN_ID
|
|
|
|
|
|
@given("an edge3 plan in Execute-Queued state with decision root")
|
|
def step_edge3_given_plan_execute_queued(context: Context) -> None:
|
|
"""Set up a plan in Execute-Queued state with decision root."""
|
|
plan = _edge3_make_plan(
|
|
phase=PlanPhase.EXECUTE,
|
|
state=ProcessingState.QUEUED,
|
|
definition_of_done="Implement feature\nWrite tests",
|
|
decision_root_id=EDGE3_ROOT_ID,
|
|
)
|
|
context.edge3_lifecycle.get_plan.return_value = plan
|
|
context.edge3_mock_plan = plan
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with checkpoint manager sandbox root and failing execute actor"
|
|
)
|
|
def step_edge3_given_executor_failing_stub_with_cp(context: Context) -> None:
|
|
"""Create PlanExecutor with checkpoint manager, sandbox root, and a failing execute actor."""
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=context.edge3_lifecycle,
|
|
sandbox_root=EDGE3_SANDBOX_ROOT,
|
|
execution_context=None,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
# Make execute actor always fail
|
|
context.edge3_executor._execute_actor = MagicMock()
|
|
context.edge3_executor._execute_actor.execute.side_effect = RuntimeError(
|
|
"edge3 stub execute boom"
|
|
)
|
|
|
|
|
|
@when("I edge3 call run execute expecting failure")
|
|
def step_edge3_run_execute_fail(context: Context) -> None:
|
|
"""Call run_execute expecting an exception."""
|
|
context.edge3_raised = None
|
|
try:
|
|
context.edge3_executor.run_execute(context.edge3_plan_id)
|
|
except Exception as exc:
|
|
context.edge3_raised = exc
|
|
|
|
|
|
@then("an edge3 exception should have been raised")
|
|
def step_edge3_exception_raised(context: Context) -> None:
|
|
"""Verify an exception was raised."""
|
|
assert context.edge3_raised is not None, "Expected an exception but none was raised"
|
|
|
|
|
|
@then("the edge3 checkpoint manager rollback_to should have been called")
|
|
def step_edge3_rollback_called(context: Context) -> None:
|
|
"""Verify checkpoint_manager.rollback_to was called."""
|
|
context.edge3_checkpoint_mgr.rollback_to.assert_called_once()
|
|
|
|
|
|
@then("the edge3 lifecycle should have called fail_execute for edge3")
|
|
def step_edge3_check_fail_execute(context: Context) -> None:
|
|
"""Verify lifecycle.fail_execute was called."""
|
|
context.edge3_lifecycle.fail_execute.assert_called_once()
|
|
call_args = context.edge3_lifecycle.fail_execute.call_args[0]
|
|
assert context.edge3_plan_id in call_args, (
|
|
f"Expected plan_id in fail_execute args, got {call_args}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Runtime execute with checkpoint rollback on failure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an edge3 mock execution context for runtime")
|
|
def step_edge3_given_exec_ctx_runtime(context: Context) -> None:
|
|
"""Create a mock execution context for runtime mode."""
|
|
from cleveragents.application.services.plan_execution_context import (
|
|
PlanExecutionContext,
|
|
)
|
|
|
|
context.edge3_exec_ctx = PlanExecutionContext(plan_id=EDGE3_PLAN_ID)
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with runtime context checkpoint manager sandbox root and failing runtime actor"
|
|
)
|
|
def step_edge3_given_executor_runtime_failing_with_cp(context: Context) -> None:
|
|
"""Create PlanExecutor with runtime context, checkpoint manager, and failing runtime."""
|
|
from cleveragents.application.services.plan_execution_context import (
|
|
RuntimeExecuteActor,
|
|
)
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
runner = ToolRunner(registry=ToolRegistry())
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=context.edge3_lifecycle,
|
|
tool_runner=runner,
|
|
sandbox_root=EDGE3_SANDBOX_ROOT,
|
|
execution_context=context.edge3_exec_ctx,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
# Patch RuntimeExecuteActor.execute to raise
|
|
patcher = patch.object(
|
|
RuntimeExecuteActor,
|
|
"execute",
|
|
side_effect=RuntimeError("edge3 runtime execute boom"),
|
|
)
|
|
patcher.start()
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(patcher.stop)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _parse_steps non-empty path exercised via run_strategize
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an edge3 mock lifecycle service for strategize")
|
|
def step_edge3_given_lifecycle_strategize(context: Context) -> None:
|
|
"""Create a mock lifecycle service for strategize scenarios."""
|
|
context.edge3_lifecycle = _edge3_make_lifecycle()
|
|
context.edge3_plan_id = EDGE3_PLAN_ID
|
|
|
|
|
|
@given('an edge3 plan in Strategize phase with multi-line definition "{defn}"')
|
|
def step_edge3_given_plan_strategize_multiline(context: Context, defn: str) -> None:
|
|
"""Set up a plan in Strategize phase with a multi-line definition."""
|
|
raw = defn.replace("\\n", "\n")
|
|
plan = _edge3_make_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
state=ProcessingState.QUEUED,
|
|
definition_of_done=raw,
|
|
decision_root_id=None,
|
|
)
|
|
context.edge3_lifecycle.get_plan.return_value = plan
|
|
context.edge3_mock_plan = plan
|
|
|
|
|
|
@given("an edge3 plan in Strategize phase with empty definition")
|
|
def step_edge3_given_plan_strategize_empty_defn(context: Context) -> None:
|
|
"""Set up a plan in Strategize phase with empty definition_of_done."""
|
|
plan = _edge3_make_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
state=ProcessingState.QUEUED,
|
|
definition_of_done="",
|
|
decision_root_id=None,
|
|
)
|
|
context.edge3_lifecycle.get_plan.return_value = plan
|
|
context.edge3_mock_plan = plan
|
|
|
|
|
|
@given("an edge3 PlanExecutor for strategize without execution context")
|
|
def step_edge3_given_executor_strategize_no_ctx(context: Context) -> None:
|
|
"""Create a PlanExecutor for strategize without execution context."""
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=context.edge3_lifecycle,
|
|
execution_context=None,
|
|
)
|
|
|
|
|
|
@when("I edge3 call run strategize successfully")
|
|
def step_edge3_run_strategize(context: Context) -> None:
|
|
"""Call run_strategize and store the result."""
|
|
context.edge3_raised = None
|
|
try:
|
|
context.edge3_strat_result = context.edge3_executor.run_strategize(
|
|
context.edge3_plan_id
|
|
)
|
|
except Exception as exc:
|
|
context.edge3_raised = exc
|
|
|
|
|
|
@then("the edge3 strategize result should have {n:d} decisions")
|
|
def step_edge3_check_decision_count(context: Context, n: int) -> None:
|
|
"""Verify the strategize result decision count."""
|
|
assert context.edge3_raised is None, f"Unexpected error: {context.edge3_raised}"
|
|
actual = len(context.edge3_strat_result.decisions)
|
|
assert actual == n, f"Expected {n} decisions, got {actual}"
|
|
|
|
|
|
@then("the edge3 lifecycle should have called complete_strategize for edge3")
|
|
def step_edge3_check_complete_strategize(context: Context) -> None:
|
|
"""Verify complete_strategize was called."""
|
|
context.edge3_lifecycle.complete_strategize.assert_called_once_with(
|
|
context.edge3_plan_id
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Successful stub execute with checkpoint creation via sandbox root proxy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"an edge3 PlanExecutor with checkpoint manager and sandbox root for stub execute"
|
|
)
|
|
def step_edge3_given_executor_cp_sandbox_stub(context: Context) -> None:
|
|
"""Create PlanExecutor with checkpoint manager and sandbox root for stub execute."""
|
|
context.edge3_checkpoint_mgr = MagicMock()
|
|
context.edge3_executor = PlanExecutor(
|
|
lifecycle_service=context.edge3_lifecycle,
|
|
sandbox_root=EDGE3_SANDBOX_ROOT,
|
|
execution_context=None,
|
|
checkpoint_manager=context.edge3_checkpoint_mgr,
|
|
)
|
|
|
|
|
|
@when("I edge3 call run execute successfully")
|
|
def step_edge3_run_execute_success(context: Context) -> None:
|
|
"""Call run_execute expecting success."""
|
|
context.edge3_raised = None
|
|
try:
|
|
context.edge3_exec_result = context.edge3_executor.run_execute(
|
|
context.edge3_plan_id
|
|
)
|
|
except Exception as exc:
|
|
context.edge3_raised = exc
|
|
|
|
|
|
@then("the edge3 execute result should be an ExecuteResult")
|
|
def step_edge3_check_exec_result_type(context: Context) -> None:
|
|
"""Verify the result is an ExecuteResult."""
|
|
assert context.edge3_raised is None, f"Unexpected error: {context.edge3_raised}"
|
|
assert isinstance(context.edge3_exec_result, ExecuteResult), (
|
|
f"Expected ExecuteResult, got {type(context.edge3_exec_result).__name__}"
|
|
)
|
|
|
|
|
|
@then(
|
|
"the edge3 checkpoint manager create_checkpoint should have been called at least twice"
|
|
)
|
|
def step_edge3_checkpoint_create_called_twice(context: Context) -> None:
|
|
"""Verify create_checkpoint was called at least twice (pre and post execute)."""
|
|
call_count = context.edge3_checkpoint_mgr.create_checkpoint.call_count
|
|
assert call_count >= 2, (
|
|
f"Expected create_checkpoint called >= 2 times, got {call_count}"
|
|
)
|