268 lines
8.6 KiB
Python
268 lines
8.6 KiB
Python
"""Helper script for subplan model Robot Framework smoke tests.
|
|
|
|
Exercises SubplanConfig, SubplanStatus, SubplanFailureHandler, and Plan
|
|
hierarchy properties without requiring the service layer or persistence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
from cleveragents.domain.models.core.plan import (
|
|
ExecutionMode,
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
SubplanConfig,
|
|
SubplanFailureHandler,
|
|
SubplanMergeStrategy,
|
|
SubplanStatus,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Valid ULID constants
|
|
# ---------------------------------------------------------------------------
|
|
_ROOT_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
_PARENT_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
|
|
_CHILD_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAX"
|
|
_GRANDCHILD_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAY"
|
|
_SUB_STATUS_ULID = "01KH72MDGQMPRW3R5WPZVB8KC1"
|
|
|
|
|
|
def _make_plan(
|
|
*,
|
|
plan_id: str = _ROOT_ULID,
|
|
parent_plan_id: str | None = None,
|
|
root_plan_id: str | None = None,
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
processing_state: ProcessingState = ProcessingState.QUEUED,
|
|
subplan_config: SubplanConfig | None = None,
|
|
subplan_statuses: list[SubplanStatus] | None = None,
|
|
) -> Plan:
|
|
"""Create a Plan for testing 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(
|
|
server=None, namespace="local", name="robot-subplan-test"
|
|
),
|
|
description="Robot subplan test plan",
|
|
action_name="local/robot-subplan-action",
|
|
phase=phase,
|
|
processing_state=processing_state,
|
|
subplan_config=subplan_config,
|
|
subplan_statuses=subplan_statuses or [],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command handlers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _root_identity() -> None:
|
|
"""Verify root plan identity flags."""
|
|
plan = _make_plan()
|
|
assert plan.is_root_plan, "Expected root plan"
|
|
assert not plan.is_subplan, "Expected not subplan"
|
|
assert plan.depth == 0, f"Expected depth 0, got {plan.depth}"
|
|
print("root-identity-ok")
|
|
|
|
|
|
def _child_identity() -> None:
|
|
"""Verify child plan identity flags."""
|
|
plan = _make_plan(
|
|
plan_id=_CHILD_ULID,
|
|
parent_plan_id=_PARENT_ULID,
|
|
root_plan_id=_ROOT_ULID,
|
|
)
|
|
assert plan.is_subplan, "Expected subplan"
|
|
assert not plan.is_root_plan, "Expected not root plan"
|
|
assert plan.depth == -1, f"Expected depth -1, got {plan.depth}"
|
|
print("child-identity-ok")
|
|
|
|
|
|
def _hierarchy_propagation() -> None:
|
|
"""Verify root_plan_id propagates through three levels."""
|
|
root = _make_plan(plan_id=_ROOT_ULID, root_plan_id=_ROOT_ULID)
|
|
parent = _make_plan(
|
|
plan_id=_PARENT_ULID,
|
|
parent_plan_id=_ROOT_ULID,
|
|
root_plan_id=_ROOT_ULID,
|
|
)
|
|
grandchild = _make_plan(
|
|
plan_id=_GRANDCHILD_ULID,
|
|
parent_plan_id=_PARENT_ULID,
|
|
root_plan_id=_ROOT_ULID,
|
|
)
|
|
assert root.is_root_plan
|
|
assert parent.is_subplan
|
|
assert grandchild.is_subplan
|
|
assert grandchild.identity.root_plan_id == root.identity.plan_id
|
|
print("hierarchy-propagation-ok")
|
|
|
|
|
|
def _config_defaults() -> None:
|
|
"""Verify SubplanConfig defaults."""
|
|
cfg = SubplanConfig()
|
|
assert cfg.execution_mode == ExecutionMode.SEQUENTIAL
|
|
assert cfg.merge_strategy == SubplanMergeStrategy.GIT_THREE_WAY
|
|
assert cfg.max_parallel == 5
|
|
assert cfg.fail_fast is False
|
|
assert cfg.retry_failed is True
|
|
assert cfg.max_retries == 2
|
|
assert cfg.timeout_per_subplan_seconds is None
|
|
print("config-defaults-ok")
|
|
|
|
|
|
def _config_custom() -> None:
|
|
"""Verify SubplanConfig accepts custom settings."""
|
|
cfg = SubplanConfig(
|
|
execution_mode=ExecutionMode.PARALLEL,
|
|
merge_strategy=SubplanMergeStrategy.FAIL_ON_CONFLICT,
|
|
max_parallel=10,
|
|
fail_fast=True,
|
|
retry_failed=True,
|
|
max_retries=4,
|
|
timeout_per_subplan_seconds=300,
|
|
)
|
|
assert cfg.execution_mode == ExecutionMode.PARALLEL
|
|
assert cfg.merge_strategy == SubplanMergeStrategy.FAIL_ON_CONFLICT
|
|
assert cfg.max_parallel == 10
|
|
assert cfg.fail_fast is True
|
|
assert cfg.max_retries == 4
|
|
assert cfg.timeout_per_subplan_seconds == 300
|
|
print("config-custom-ok")
|
|
|
|
|
|
def _failure_stop_others() -> None:
|
|
"""Verify SubplanFailureHandler stops others when fail_fast=True."""
|
|
config = SubplanConfig(
|
|
execution_mode=ExecutionMode.PARALLEL,
|
|
fail_fast=True,
|
|
)
|
|
status = SubplanStatus(
|
|
subplan_id=_SUB_STATUS_ULID,
|
|
action_name="local/sub-task",
|
|
status=ProcessingState.ERRORED,
|
|
error="TimeoutError: timed out",
|
|
)
|
|
handler = SubplanFailureHandler()
|
|
assert handler.should_stop_others(config, status) is True
|
|
print("failure-stop-others-ok")
|
|
|
|
|
|
def _failure_retry_retriable() -> None:
|
|
"""Verify SubplanFailureHandler retries retriable errors."""
|
|
config = SubplanConfig(retry_failed=True, max_retries=2)
|
|
status = SubplanStatus(
|
|
subplan_id=_SUB_STATUS_ULID,
|
|
action_name="local/sub-task",
|
|
status=ProcessingState.ERRORED,
|
|
error="TimeoutError: timed out",
|
|
attempt_number=1,
|
|
)
|
|
handler = SubplanFailureHandler()
|
|
assert handler.should_retry(config, status) is True
|
|
print("failure-retry-retriable-ok")
|
|
|
|
|
|
def _failure_skip_non_retriable() -> None:
|
|
"""Verify SubplanFailureHandler does not retry non-retriable errors."""
|
|
config = SubplanConfig(retry_failed=True, max_retries=2)
|
|
status = SubplanStatus(
|
|
subplan_id=_SUB_STATUS_ULID,
|
|
action_name="local/sub-task",
|
|
status=ProcessingState.ERRORED,
|
|
error="ConfigurationError: bad config",
|
|
attempt_number=1,
|
|
)
|
|
handler = SubplanFailureHandler()
|
|
assert handler.should_retry(config, status) is False
|
|
print("failure-skip-non-retriable-ok")
|
|
|
|
|
|
def _cli_dict_child() -> None:
|
|
"""Verify as_cli_dict includes parent_plan_id for child plan."""
|
|
plan = _make_plan(
|
|
plan_id=_CHILD_ULID,
|
|
parent_plan_id=_PARENT_ULID,
|
|
root_plan_id=_ROOT_ULID,
|
|
)
|
|
cli = plan.as_cli_dict()
|
|
assert "parent_plan_id" in cli, "Expected parent_plan_id in CLI dict"
|
|
assert cli["parent_plan_id"] == _PARENT_ULID
|
|
print("cli-dict-child-ok")
|
|
|
|
|
|
def _cli_dict_subplan_count() -> None:
|
|
"""Verify as_cli_dict includes subplan_count when subplans exist."""
|
|
statuses = [
|
|
SubplanStatus(
|
|
subplan_id=_SUB_STATUS_ULID,
|
|
action_name="local/sub-task",
|
|
status=ProcessingState.QUEUED,
|
|
),
|
|
]
|
|
plan = _make_plan(subplan_statuses=statuses)
|
|
cli = plan.as_cli_dict()
|
|
assert "subplan_count" in cli, "Expected subplan_count in CLI dict"
|
|
assert cli["subplan_count"] == 1
|
|
print("cli-dict-subplan-count-ok")
|
|
|
|
|
|
def _plan_with_config() -> None:
|
|
"""Verify Plan model accepts subplan config and statuses."""
|
|
config = SubplanConfig(
|
|
execution_mode=ExecutionMode.SEQUENTIAL,
|
|
merge_strategy=SubplanMergeStrategy.SEQUENTIAL_APPLY,
|
|
)
|
|
statuses = [
|
|
SubplanStatus(
|
|
subplan_id=_SUB_STATUS_ULID,
|
|
action_name="local/sub-task",
|
|
target_resources=["src/main.py"],
|
|
status=ProcessingState.QUEUED,
|
|
),
|
|
]
|
|
plan = _make_plan(subplan_config=config, subplan_statuses=statuses)
|
|
assert plan.has_subplans
|
|
assert plan.subplan_config is not None
|
|
assert plan.subplan_config.execution_mode == ExecutionMode.SEQUENTIAL
|
|
assert len(plan.subplan_statuses) == 1
|
|
print("plan-with-config-ok")
|
|
|
|
|
|
def main() -> None:
|
|
"""Dispatch command from sys.argv."""
|
|
if len(sys.argv) < 2:
|
|
raise SystemExit("Expected command argument")
|
|
command = sys.argv[1]
|
|
commands: dict[str, object] = {
|
|
"root-identity": _root_identity,
|
|
"child-identity": _child_identity,
|
|
"hierarchy-propagation": _hierarchy_propagation,
|
|
"config-defaults": _config_defaults,
|
|
"config-custom": _config_custom,
|
|
"failure-stop-others": _failure_stop_others,
|
|
"failure-retry-retriable": _failure_retry_retriable,
|
|
"failure-skip-non-retriable": _failure_skip_non_retriable,
|
|
"cli-dict-child": _cli_dict_child,
|
|
"cli-dict-subplan-count": _cli_dict_subplan_count,
|
|
"plan-with-config": _plan_with_config,
|
|
}
|
|
if command not in commands:
|
|
raise SystemExit(f"Unknown command: {command}")
|
|
func = commands[command]
|
|
if callable(func):
|
|
func()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|