forked from HAL9000/cleveragents-core
365 lines
13 KiB
Python
365 lines
13 KiB
Python
"""Step definitions for plan_model_uncovered_lines.feature.
|
|
|
|
Covers Plan hierarchy properties (is_subplan, is_root_plan, depth,
|
|
has_subplans) and SubplanFailureHandler decision logic (should_stop_others,
|
|
should_retry). All step names are prefixed with "uncovered-lines" to
|
|
avoid collisions with other step files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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 for test fixtures (Crockford base32, 26 chars)
|
|
# ---------------------------------------------------------------------------
|
|
_ULID_SELF = "01HGZ6FE0AQDYTR4BXVQZ6EA01"
|
|
_ULID_PARENT = "01HGZ6FE0AQDYTR4BXVQZ6EB01"
|
|
_ULID_ROOT = "01HGZ6FE0AQDYTR4BXVQZ6EC01"
|
|
_ULID_SUBPLAN = "01HGZ6FE0AQDYTR4BXVQZ6ED01"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_plan(
|
|
plan_id: str = _ULID_SELF,
|
|
parent_plan_id: str | None = None,
|
|
root_plan_id: str | None = None,
|
|
subplan_statuses: list[SubplanStatus] | None = None,
|
|
) -> Plan:
|
|
"""Build a minimal Plan for hierarchy property testing."""
|
|
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="uncovered-plan"),
|
|
action_name="local/uncovered-action",
|
|
description="Plan for uncovered-lines testing",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
subplan_statuses=subplan_statuses or [],
|
|
)
|
|
|
|
|
|
def _make_subplan_status(
|
|
error: str | None = None,
|
|
attempt_number: int = 1,
|
|
) -> SubplanStatus:
|
|
"""Build a SubplanStatus in ERRORED state for failure-handler testing."""
|
|
return SubplanStatus(
|
|
subplan_id=_ULID_SUBPLAN,
|
|
action_name="local/uncovered-sub-action",
|
|
status=ProcessingState.ERRORED,
|
|
error=error,
|
|
attempt_number=attempt_number,
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# Subplan hierarchy: is_subplan
|
|
# ===================================================================
|
|
|
|
|
|
@given("an uncovered-lines plan that was created independently")
|
|
def step_uncovered_create_plan_no_parent(context: Context) -> None:
|
|
"""Create a plan without a parent (standalone)."""
|
|
context.plan = _make_plan(parent_plan_id=None)
|
|
|
|
|
|
@then("the uncovered-lines plan should not be recognized as a subplan")
|
|
def step_uncovered_verify_not_subplan(context: Context) -> None:
|
|
"""Assert ``is_subplan`` is False."""
|
|
assert context.plan.is_subplan is False, (
|
|
f"Expected is_subplan=False, got {context.plan.is_subplan}"
|
|
)
|
|
|
|
|
|
@given("an uncovered-lines plan that was spawned by another plan")
|
|
def step_uncovered_create_plan_with_parent(context: Context) -> None:
|
|
"""Create a plan that has a parent_plan_id set."""
|
|
context.plan = _make_plan(parent_plan_id=_ULID_PARENT)
|
|
|
|
|
|
@then("the uncovered-lines plan should be recognized as a subplan")
|
|
def step_uncovered_verify_is_subplan(context: Context) -> None:
|
|
"""Assert ``is_subplan`` is True."""
|
|
assert context.plan.is_subplan is True, (
|
|
f"Expected is_subplan=True, got {context.plan.is_subplan}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# Subplan hierarchy: is_root_plan
|
|
# ===================================================================
|
|
|
|
|
|
@given("an uncovered-lines plan with no designated root")
|
|
def step_uncovered_create_plan_no_root(context: Context) -> None:
|
|
"""Create a plan with root_plan_id=None (treated as root)."""
|
|
context.plan = _make_plan(root_plan_id=None)
|
|
|
|
|
|
@then("the uncovered-lines plan should be recognized as a root plan")
|
|
def step_uncovered_verify_is_root(context: Context) -> None:
|
|
"""Assert ``is_root_plan`` is True."""
|
|
assert context.plan.is_root_plan is True, (
|
|
f"Expected is_root_plan=True, got {context.plan.is_root_plan}"
|
|
)
|
|
|
|
|
|
@given("an uncovered-lines plan whose designated root is itself")
|
|
def step_uncovered_create_plan_root_self(context: Context) -> None:
|
|
"""Create a plan whose root_plan_id equals its own plan_id."""
|
|
context.plan = _make_plan(
|
|
plan_id=_ULID_SELF,
|
|
root_plan_id=_ULID_SELF,
|
|
)
|
|
|
|
|
|
@given("an uncovered-lines plan whose designated root is a different plan")
|
|
def step_uncovered_create_plan_root_other(context: Context) -> None:
|
|
"""Create a plan whose root_plan_id differs from its own plan_id."""
|
|
context.plan = _make_plan(
|
|
plan_id=_ULID_SELF,
|
|
root_plan_id=_ULID_ROOT,
|
|
)
|
|
|
|
|
|
@then("the uncovered-lines plan should not be recognized as a root plan")
|
|
def step_uncovered_verify_not_root(context: Context) -> None:
|
|
"""Assert ``is_root_plan`` is False."""
|
|
assert context.plan.is_root_plan is False, (
|
|
f"Expected is_root_plan=False, got {context.plan.is_root_plan}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# Subplan hierarchy: depth
|
|
# ===================================================================
|
|
|
|
|
|
@then("the uncovered-lines plan hierarchy depth should be {expected:d}")
|
|
def step_uncovered_verify_depth(context: Context, expected: int) -> None:
|
|
"""Assert the plan ``depth`` property equals *expected*."""
|
|
assert context.plan.depth == expected, (
|
|
f"Expected depth={expected}, got {context.plan.depth}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# Subplan hierarchy: has_subplans
|
|
# ===================================================================
|
|
|
|
|
|
@given("an uncovered-lines plan that has no child work tracked")
|
|
def step_uncovered_create_plan_no_children(context: Context) -> None:
|
|
"""Create a plan with an empty subplan_statuses list."""
|
|
context.plan = _make_plan(subplan_statuses=[])
|
|
|
|
|
|
@then("the uncovered-lines plan should report having no subplans")
|
|
def step_uncovered_verify_no_subplans(context: Context) -> None:
|
|
"""Assert ``has_subplans`` is False."""
|
|
assert context.plan.has_subplans is False, (
|
|
f"Expected has_subplans=False, got {context.plan.has_subplans}"
|
|
)
|
|
|
|
|
|
@given("an uncovered-lines plan that has tracked child work")
|
|
def step_uncovered_create_plan_with_children(context: Context) -> None:
|
|
"""Create a plan with at least one SubplanStatus entry."""
|
|
status = _make_subplan_status()
|
|
context.plan = _make_plan(subplan_statuses=[status])
|
|
|
|
|
|
@then("the uncovered-lines plan should report having subplans")
|
|
def step_uncovered_verify_has_subplans(context: Context) -> None:
|
|
"""Assert ``has_subplans`` is True."""
|
|
assert context.plan.has_subplans is True, (
|
|
f"Expected has_subplans=True, got {context.plan.has_subplans}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# Failure handler: should_stop_others
|
|
# ===================================================================
|
|
|
|
|
|
@given(
|
|
"an uncovered-lines subplan configuration with fail-fast enabled and parallel work"
|
|
)
|
|
def step_uncovered_config_failfast_parallel(context: Context) -> None:
|
|
"""SubplanConfig with fail_fast=True and PARALLEL execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=True,
|
|
execution_mode=ExecutionMode.PARALLEL,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an uncovered-lines subplan configuration with fail-fast disabled and sequential work"
|
|
)
|
|
def step_uncovered_config_no_failfast_sequential(context: Context) -> None:
|
|
"""SubplanConfig with fail_fast=False and SEQUENTIAL execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=False,
|
|
execution_mode=ExecutionMode.SEQUENTIAL,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an uncovered-lines subplan configuration with fail-fast disabled and parallel work"
|
|
)
|
|
def step_uncovered_config_no_failfast_parallel(context: Context) -> None:
|
|
"""SubplanConfig with fail_fast=False and PARALLEL execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=False,
|
|
execution_mode=ExecutionMode.PARALLEL,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an uncovered-lines subplan configuration with fail-fast disabled and dependency-ordered work"
|
|
)
|
|
def step_uncovered_config_no_failfast_dependency(context: Context) -> None:
|
|
"""SubplanConfig with fail_fast=False and DEPENDENCY_ORDERED execution."""
|
|
context.subplan_config = SubplanConfig(
|
|
fail_fast=False,
|
|
execution_mode=ExecutionMode.DEPENDENCY_ORDERED,
|
|
)
|
|
|
|
|
|
@given("an uncovered-lines subplan that has failed")
|
|
def step_uncovered_create_failed_status(context: Context) -> None:
|
|
"""Record a SubplanStatus in ERRORED state with a generic error."""
|
|
context.failed_status = _make_subplan_status(
|
|
error="GenericError: something failed",
|
|
)
|
|
|
|
|
|
@when("the uncovered-lines failure handler evaluates whether to halt remaining work")
|
|
def step_uncovered_call_should_stop_others(context: Context) -> None:
|
|
"""Invoke ``SubplanFailureHandler.should_stop_others``."""
|
|
handler = SubplanFailureHandler()
|
|
context.should_stop_result = handler.should_stop_others(
|
|
config=context.subplan_config,
|
|
failed_status=context.failed_status,
|
|
)
|
|
|
|
|
|
@then("the uncovered-lines remaining work should be halted")
|
|
def step_uncovered_verify_should_stop_true(context: Context) -> None:
|
|
"""Assert should_stop_others returned True."""
|
|
assert context.should_stop_result is True, (
|
|
f"Expected should_stop_others=True, got {context.should_stop_result}"
|
|
)
|
|
|
|
|
|
@then("the uncovered-lines remaining work should continue")
|
|
def step_uncovered_verify_should_stop_false(context: Context) -> None:
|
|
"""Assert should_stop_others returned False."""
|
|
assert context.should_stop_result is False, (
|
|
f"Expected should_stop_others=False, got {context.should_stop_result}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# Failure handler: should_retry
|
|
# ===================================================================
|
|
|
|
|
|
@given("an uncovered-lines subplan configuration with retries disabled")
|
|
def step_uncovered_config_retry_disabled(context: Context) -> None:
|
|
"""SubplanConfig with retry_failed=False."""
|
|
context.subplan_config = SubplanConfig(retry_failed=False)
|
|
|
|
|
|
@given(
|
|
"an uncovered-lines subplan configuration allowing up to {max_retries:d} retries"
|
|
)
|
|
def step_uncovered_config_retry_enabled(context: Context, max_retries: int) -> None:
|
|
"""SubplanConfig with retry_failed=True and given max_retries."""
|
|
context.subplan_config = SubplanConfig(
|
|
retry_failed=True,
|
|
max_retries=max_retries,
|
|
)
|
|
|
|
|
|
@given('an uncovered-lines subplan that failed with error "{error_msg}"')
|
|
def step_uncovered_create_status_with_error(context: Context, error_msg: str) -> None:
|
|
"""Record a SubplanStatus with the specified error on attempt 1."""
|
|
context.failed_status = _make_subplan_status(
|
|
error=error_msg,
|
|
attempt_number=1,
|
|
)
|
|
|
|
|
|
@given(
|
|
'an uncovered-lines subplan on attempt {attempt:d} that failed with error "{error_msg}"'
|
|
)
|
|
def step_uncovered_create_status_attempt_error(
|
|
context: Context,
|
|
attempt: int,
|
|
error_msg: str,
|
|
) -> None:
|
|
"""Record a SubplanStatus on the given attempt with specified error."""
|
|
context.failed_status = _make_subplan_status(
|
|
error=error_msg,
|
|
attempt_number=attempt,
|
|
)
|
|
|
|
|
|
@given(
|
|
"an uncovered-lines subplan on attempt {attempt:d} that failed with no error message"
|
|
)
|
|
def step_uncovered_create_status_no_error(context: Context, attempt: int) -> None:
|
|
"""Record a SubplanStatus with error=None on the given attempt."""
|
|
context.failed_status = _make_subplan_status(
|
|
error=None,
|
|
attempt_number=attempt,
|
|
)
|
|
|
|
|
|
@when("the uncovered-lines failure handler evaluates whether to retry the failed work")
|
|
def step_uncovered_call_should_retry(context: Context) -> None:
|
|
"""Invoke ``SubplanFailureHandler.should_retry``."""
|
|
handler = SubplanFailureHandler()
|
|
context.should_retry_result = handler.should_retry(
|
|
config=context.subplan_config,
|
|
status=context.failed_status,
|
|
)
|
|
|
|
|
|
@then("the uncovered-lines failed work should be retried")
|
|
def step_uncovered_verify_should_retry_true(context: Context) -> None:
|
|
"""Assert should_retry returned True."""
|
|
assert context.should_retry_result is True, (
|
|
f"Expected should_retry=True, got {context.should_retry_result}"
|
|
)
|
|
|
|
|
|
@then("the uncovered-lines failed work should not be retried")
|
|
def step_uncovered_verify_should_retry_false(context: Context) -> None:
|
|
"""Assert should_retry returned False."""
|
|
assert context.should_retry_result is False, (
|
|
f"Expected should_retry=False, got {context.should_retry_result}"
|
|
)
|