"""Step definitions for plan lifecycle model validation tests. Covers ERRORED terminality, per-phase state constraints, rejection of invalid phase/state combinations, deserialization coercion for legacy DB rows, and terminal-state guard behaviour per the specification. """ from __future__ import annotations from typing import Any from behave import given, then, when from behave.runner import Context from pydantic import ValidationError from cleveragents.domain.models.core.plan import ( NamespacedName, Plan, PlanIdentity, PlanPhase, ProcessingState, ) _ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV" def _make_plan( phase: PlanPhase, processing_state: ProcessingState, ) -> Plan: """Build a Plan with the given phase and processing state.""" return Plan( identity=PlanIdentity(plan_id=_ULID), namespaced_name=NamespacedName( server=None, namespace="local", name="plmv-test" ), description="Plan lifecycle model validation test", action_name="local/plmv-action", phase=phase, processing_state=processing_state, ) _PHASE_MAP: dict[str, PlanPhase] = { "ACTION": PlanPhase.ACTION, "STRATEGIZE": PlanPhase.STRATEGIZE, "EXECUTE": PlanPhase.EXECUTE, "APPLY": PlanPhase.APPLY, } _STATE_MAP: dict[str, ProcessingState] = { "QUEUED": ProcessingState.QUEUED, "PROCESSING": ProcessingState.PROCESSING, "ERRORED": ProcessingState.ERRORED, "COMPLETE": ProcessingState.COMPLETE, "CANCELLED": ProcessingState.CANCELLED, "APPLIED": ProcessingState.APPLIED, "CONSTRAINED": ProcessingState.CONSTRAINED, } # ---- Given steps ---- @given("a plmv plan in {phase} phase with {state} state") def step_create_plmv_plan(context: Context, phase: str, state: str) -> None: """Create a plan with the given phase and processing state.""" plan = _make_plan(_PHASE_MAP[phase], _STATE_MAP[state]) if not hasattr(context, "plmv_plans"): context.plmv_plans = [] context.plmv_plans.append(plan) context.plmv_plan = plan # ---- When steps ---- @when("I try to create a plmv plan with phase {phase} and state {state}") def step_try_create_invalid_plmv_plan(context: Context, phase: str, state: str) -> None: """Attempt to create a plan that should fail validation.""" context.plmv_error = None try: _make_plan(_PHASE_MAP[phase], _STATE_MAP[state]) except (ValidationError, ValueError) as exc: context.plmv_error = exc # ---- Then steps ---- @then("the plmv plan should be terminal") def step_plmv_plan_terminal(context: Context) -> None: """Assert the plan is in a terminal state.""" assert context.plmv_plan.is_terminal, ( f"Expected plan to be terminal, but is_terminal=False " f"(phase={context.plmv_plan.phase.value}, " f"state={context.plmv_plan.processing_state.value})" ) @then("the plmv plan should not be terminal") def step_plmv_plan_not_terminal(context: Context) -> None: """Assert the plan is NOT in a terminal state.""" assert not context.plmv_plan.is_terminal, ( f"Expected plan to NOT be terminal, but is_terminal=True " f"(phase={context.plmv_plan.phase.value}, " f"state={context.plmv_plan.processing_state.value})" ) @then('a plmv validation error should be raised containing "{fragment}"') def step_plmv_validation_error(context: Context, fragment: str) -> None: """Assert that a validation error was raised with the expected message.""" assert context.plmv_error is not None, ( "Expected a validation error, but none was raised" ) msg = str(context.plmv_error) assert fragment in msg, f"Expected error to contain {fragment!r}, got: {msg}" @then("all plmv plans should be valid") def step_all_plmv_plans_valid(context: Context) -> None: """Assert that all previously created plans are valid.""" assert hasattr(context, "plmv_plans"), "No plmv_plans found on context" assert len(context.plmv_plans) > 0, "plmv_plans list is empty" for plan in context.plmv_plans: assert plan is not None, "Plan should not be None" # ---- Assignment-order regression steps (T6) ---- @when( "I plmv transition the plan to {phase} phase with {state} state" " using state-first order" ) def step_plmv_transition_state_first(context: Context, phase: str, state: str) -> None: """Mutate an existing plan's phase and state using the safe state-first ordering. Verifies that ``validate_assignment=True`` does not trigger the phase-state validator on the intermediate state. """ plan = context.plmv_plan plan.processing_state = _STATE_MAP[state] plan.phase = _PHASE_MAP[phase] context.plmv_plan = plan @then("the plmv plan should have phase {phase} and state {state}") def step_plmv_plan_phase_state(context: Context, phase: str, state: str) -> None: """Assert the plan's current phase and processing state.""" plan = context.plmv_plan assert plan.phase == _PHASE_MAP[phase], ( f"Expected phase {phase}, got {plan.phase.value}" ) assert plan.processing_state == _STATE_MAP[state], ( f"Expected state {state}, got {plan.processing_state.value}" ) # ---- Phase-first assignment negative test ---- @when( "I plmv transition the plan to {phase} phase with {state} state" " using phase-first order" ) def step_plmv_transition_phase_first(context: Context, phase: str, state: str) -> None: """Attempt phase-first mutation which should fail the validator when the intermediate phase/state combination is invalid. """ context.plmv_error = None try: plan = context.plmv_plan plan.phase = _PHASE_MAP[phase] plan.processing_state = _STATE_MAP[state] context.plmv_plan = plan except (ValidationError, ValueError) as exc: context.plmv_error = exc # ---- Terminal-state guard steps ---- @then("the plmv plan should not be cancellable") def step_plmv_not_cancellable(context: Context) -> None: """Assert the plan is terminal and therefore not cancellable.""" assert context.plmv_plan.is_terminal, ( "Expected plan to be terminal (not cancellable)" ) @then("the plmv plan should not be pausable") def step_plmv_not_pausable(context: Context) -> None: """Assert the plan is terminal and therefore not pausable.""" assert context.plmv_plan.is_terminal, "Expected plan to be terminal (not pausable)" @then("the plmv plan should not be resumable via lifecycle") def step_plmv_not_resumable_lifecycle(context: Context) -> None: """Assert the plan is terminal and therefore not resumable via ``PlanLifecycleService.resume_plan()``. """ assert context.plmv_plan.is_terminal, ( "Expected plan to be terminal (not resumable via lifecycle)" ) # ---- Deserialization coercion steps ---- @when('I plmv deserialize a plan with phase "{phase}" and state "{state}"') def step_plmv_deserialize(context: Context, phase: str, state: str) -> None: """Build a ``LifecyclePlanModel`` ORM row with the given phase/state and call ``to_domain()`` to exercise the defensive coercion logic. """ from cleveragents.infrastructure.database.models import ( LifecyclePlanModel, ) model = LifecyclePlanModel() model.plan_id = _ULID model.namespaced_name = "local/plmv-deser" model.description = "deserialization coercion test" model.action_name = "local/plmv-action" model.phase = phase model.processing_state = state model.attempt = 1 model.created_at = "2025-01-01T00:00:00" model.updated_at = "2025-01-01T00:00:00" model.reusable = True model.read_only = False model.sandbox_refs_json = "[]" model.tags_json = "[]" model.reversion_count = 0 model.last_completed_step = -1 context.plmv_deserialized_plan = model.to_domain() @then("the plmv deserialized plan should have phase {phase} and state {state}") def step_plmv_deserialized_phase_state( context: Context, phase: str, state: str ) -> None: """Assert the deserialized plan's phase and state after coercion.""" plan: Any = context.plmv_deserialized_plan assert plan.phase == _PHASE_MAP[phase], ( f"Expected phase {phase}, got {plan.phase.value}" ) assert plan.processing_state == _STATE_MAP[state], ( f"Expected state {state}, got {plan.processing_state.value}" )