forked from HAL9000/cleveragents-core
CoreRasurae
325 lines
11 KiB
Python
325 lines
11 KiB
Python
"""Step definitions for plan hierarchy and subplan failure handling tests."""
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.plan import (
|
|
ExecutionMode,
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
SubplanConfig,
|
|
SubplanFailureHandler,
|
|
SubplanStatus,
|
|
)
|
|
|
|
# Valid ULIDs (Crockford base32, 26 chars)
|
|
ULID_A = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
|
|
ULID_B = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
|
|
ULID_C = "01HGZ6FE0AQDYTR4BXVQZ6EC00"
|
|
|
|
|
|
def _make_plan(
|
|
plan_id=ULID_A,
|
|
parent_plan_id=None,
|
|
root_plan_id=None,
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
subplan_statuses=None,
|
|
):
|
|
"""Helper to create a Plan with minimal required fields."""
|
|
return Plan(
|
|
identity=PlanIdentity(
|
|
plan_id=plan_id,
|
|
parent_plan_id=parent_plan_id,
|
|
root_plan_id=root_plan_id,
|
|
),
|
|
namespaced_name=NamespacedName(namespace="local", name="test-plan"),
|
|
action_name="local/test-action",
|
|
description="A test plan for coverage",
|
|
phase=phase,
|
|
processing_state=processing_state,
|
|
subplan_statuses=subplan_statuses or [],
|
|
)
|
|
|
|
|
|
def _make_subplan_status(
|
|
subplan_id=ULID_B,
|
|
error=None,
|
|
attempt_number=1,
|
|
):
|
|
"""Helper to create a SubplanStatus."""
|
|
return SubplanStatus(
|
|
subplan_id=subplan_id,
|
|
action_name="local/test-action",
|
|
error=error,
|
|
attempt_number=attempt_number,
|
|
status=ProcessingState.ERRORED,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan hierarchy: is_subplan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a plan that was created independently")
|
|
def step_create_plan_no_parent(context: Context) -> None:
|
|
"""Create a plan without a parent."""
|
|
context.plan = _make_plan(parent_plan_id=None)
|
|
|
|
|
|
@then("the plan should not be recognized as a subplan")
|
|
def step_verify_not_subplan(context: Context) -> None:
|
|
"""Verify the plan is not a subplan."""
|
|
assert context.plan.is_subplan is False, (
|
|
f"Expected is_subplan=False, got {context.plan.is_subplan}"
|
|
)
|
|
|
|
|
|
@given("a plan that was spawned by another plan")
|
|
def step_create_plan_with_parent(context: Context) -> None:
|
|
"""Create a plan that was spawned by another plan."""
|
|
context.plan = _make_plan(parent_plan_id=ULID_B)
|
|
|
|
|
|
@then("the plan should be recognized as a subplan")
|
|
def step_verify_is_subplan(context: Context) -> None:
|
|
"""Verify the plan is a subplan."""
|
|
assert context.plan.is_subplan is True, (
|
|
f"Expected is_subplan=True, got {context.plan.is_subplan}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan hierarchy: is_root_plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a plan with no designated root")
|
|
def step_create_plan_no_root(context: Context) -> None:
|
|
"""Create a plan without a designated root."""
|
|
context.plan = _make_plan(root_plan_id=None)
|
|
|
|
|
|
@then("the plan should be recognized as a root plan")
|
|
def step_verify_is_root_plan(context: Context) -> None:
|
|
"""Verify the plan is a root plan."""
|
|
assert context.plan.is_root_plan is True, (
|
|
f"Expected is_root_plan=True, got {context.plan.is_root_plan}"
|
|
)
|
|
|
|
|
|
@given("a plan whose designated root is itself")
|
|
def step_create_plan_root_equals_self(context: Context) -> None:
|
|
"""Create a plan whose designated root is itself."""
|
|
context.plan = _make_plan(plan_id=ULID_A, root_plan_id=ULID_A)
|
|
|
|
|
|
@given("a plan whose designated root is a different plan")
|
|
def step_create_plan_root_differs(context: Context) -> None:
|
|
"""Create a plan whose designated root is a different plan."""
|
|
context.plan = _make_plan(plan_id=ULID_A, root_plan_id=ULID_C)
|
|
|
|
|
|
@then("the plan should not be recognized as a root plan")
|
|
def step_verify_not_root_plan(context: Context) -> None:
|
|
"""Verify the plan is not a root plan."""
|
|
assert context.plan.is_root_plan is False, (
|
|
f"Expected is_root_plan=False, got {context.plan.is_root_plan}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan hierarchy: depth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the plan hierarchy depth should be 0")
|
|
def step_verify_depth_zero(context: Context) -> None:
|
|
"""Verify the plan hierarchy depth is 0."""
|
|
assert context.plan.depth == 0, f"Expected depth=0, got {context.plan.depth}"
|
|
|
|
|
|
@then("the plan hierarchy depth should be -1")
|
|
def step_verify_depth_negative_one(context: Context) -> None:
|
|
"""Verify the plan hierarchy depth is -1 (placeholder)."""
|
|
assert context.plan.depth == -1, f"Expected depth=-1, got {context.plan.depth}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan hierarchy: has_subplans
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a plan that has no child work tracked")
|
|
def step_create_plan_no_subplans(context: Context) -> None:
|
|
"""Create a plan with no child work tracked."""
|
|
context.plan = _make_plan(subplan_statuses=[])
|
|
|
|
|
|
@then("the plan should report having no subplans")
|
|
def step_verify_no_subplans(context: Context) -> None:
|
|
"""Verify the plan reports having no subplans."""
|
|
assert context.plan.has_subplans is False, (
|
|
f"Expected has_subplans=False, got {context.plan.has_subplans}"
|
|
)
|
|
|
|
|
|
@given("a plan that has tracked child work")
|
|
def step_create_plan_with_subplans(context: Context) -> None:
|
|
"""Create a plan with tracked child work."""
|
|
status = _make_subplan_status()
|
|
context.plan = _make_plan(subplan_statuses=[status])
|
|
|
|
|
|
@then("the plan should report having subplans")
|
|
def step_verify_has_subplans(context: Context) -> None:
|
|
"""Verify the plan reports having subplans."""
|
|
assert context.plan.has_subplans is True, (
|
|
f"Expected has_subplans=True, got {context.plan.has_subplans}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Failure handler: should_stop_others
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a subplan configuration with fail-fast enabled and parallel work")
|
|
def step_config_failfast_parallel(context: Context) -> None:
|
|
"""Configure subplans with fail-fast enabled and parallel execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=True,
|
|
execution_mode=ExecutionMode.PARALLEL,
|
|
)
|
|
|
|
|
|
@given("a subplan configuration with fail-fast disabled and sequential work")
|
|
def step_config_no_failfast_sequential(context: Context) -> None:
|
|
"""Configure subplans with fail-fast disabled and sequential execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=False,
|
|
execution_mode=ExecutionMode.SEQUENTIAL,
|
|
)
|
|
|
|
|
|
@given("a subplan configuration with fail-fast disabled and parallel work")
|
|
def step_config_no_failfast_parallel(context: Context) -> None:
|
|
"""Configure subplans with fail-fast disabled and parallel execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=False,
|
|
execution_mode=ExecutionMode.PARALLEL,
|
|
)
|
|
|
|
|
|
@given("a subplan configuration with fail-fast disabled and dependency-ordered work")
|
|
def step_config_no_failfast_dependency(context: Context) -> None:
|
|
"""Configure subplans with fail-fast disabled and dependency-ordered execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=False,
|
|
execution_mode=ExecutionMode.DEPENDENCY_ORDERED,
|
|
)
|
|
|
|
|
|
@given("a subplan that has failed")
|
|
def step_create_failed_status(context: Context) -> None:
|
|
"""Record a failed subplan with a generic error."""
|
|
context.failed_status = _make_subplan_status(error="GenericError: something failed")
|
|
|
|
|
|
@when("the failure handler evaluates whether to halt remaining work")
|
|
def step_call_should_stop_others(context: Context) -> None:
|
|
"""Ask the failure handler whether remaining work should be halted."""
|
|
handler = SubplanFailureHandler()
|
|
context.should_stop_result = handler.should_stop_others(
|
|
config=context.subplan_config,
|
|
failed_status=context.failed_status,
|
|
)
|
|
|
|
|
|
@then("the remaining work should be halted")
|
|
def step_verify_should_stop_true(context: Context) -> None:
|
|
"""Verify the failure handler decided to halt remaining work."""
|
|
assert context.should_stop_result is True, (
|
|
f"Expected should_stop_others=True, got {context.should_stop_result}"
|
|
)
|
|
|
|
|
|
@then("the remaining work should continue")
|
|
def step_verify_should_stop_false(context: Context) -> None:
|
|
"""Verify the failure handler decided to allow remaining work to continue."""
|
|
assert context.should_stop_result is False, (
|
|
f"Expected should_stop_others=False, got {context.should_stop_result}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Failure handler: should_retry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a subplan configuration with retries disabled")
|
|
def step_config_retry_disabled(context: Context) -> None:
|
|
"""Configure subplans with retries disabled."""
|
|
context.subplan_config = SubplanConfig(retry_failed=False)
|
|
|
|
|
|
@given("a subplan configuration allowing up to {max_retries:d} retries")
|
|
def step_config_retry_enabled(context: Context, max_retries: int) -> None:
|
|
"""Configure subplans with retries enabled and a max retry count."""
|
|
context.subplan_config = SubplanConfig(
|
|
retry_failed=True,
|
|
max_retries=max_retries,
|
|
)
|
|
|
|
|
|
@given('a subplan that failed with error "{error_msg}"')
|
|
def step_create_status_with_error(context: Context, error_msg: str) -> None:
|
|
"""Record a failed subplan with a specific error message."""
|
|
context.failed_status = _make_subplan_status(error=error_msg, attempt_number=1)
|
|
|
|
|
|
@given('a subplan on attempt {attempt:d} that failed with error "{error_msg}"')
|
|
def step_create_status_with_attempt_and_error(
|
|
context: Context, attempt: int, error_msg: str
|
|
) -> None:
|
|
"""Record a failed subplan on a specific attempt with a specific error."""
|
|
context.failed_status = _make_subplan_status(
|
|
error=error_msg, attempt_number=attempt
|
|
)
|
|
|
|
|
|
@given("a subplan on attempt {attempt:d} that failed with no error message")
|
|
def step_create_status_with_no_error(context: Context, attempt: int) -> None:
|
|
"""Record a failed subplan with no error message."""
|
|
context.failed_status = _make_subplan_status(error=None, attempt_number=attempt)
|
|
|
|
|
|
@when("the failure handler evaluates whether to retry the failed work")
|
|
def step_call_should_retry(context: Context) -> None:
|
|
"""Ask the failure handler whether the failed work should be retried."""
|
|
handler = SubplanFailureHandler()
|
|
context.should_retry_result = handler.should_retry(
|
|
config=context.subplan_config,
|
|
status=context.failed_status,
|
|
)
|
|
|
|
|
|
@then("the failed work should be retried")
|
|
def step_verify_should_retry_true(context: Context) -> None:
|
|
"""Verify the failure handler decided to retry the failed work."""
|
|
assert context.should_retry_result is True, (
|
|
f"Expected should_retry=True, got {context.should_retry_result}"
|
|
)
|
|
|
|
|
|
@then("the failed work should not be retried")
|
|
def step_verify_should_retry_false(context: Context) -> None:
|
|
"""Verify the failure handler decided not to retry the failed work."""
|
|
assert context.should_retry_result is False, (
|
|
f"Expected should_retry=False, got {context.should_retry_result}"
|
|
)
|