Dev User
6b5568af1d
The base step 'I create a plan in strategize phase' was renamed to 'I create a PlanModel in strategize phase' to avoid AmbiguousStep collision. This commit updates compound step definitions that use the base step: - 'I create a plan in strategize phase with errored state' - 'I create a plan in strategize phase with processing state' - 'I create a plan in strategize phase with cancelled state'
787 lines
27 KiB
Python
787 lines
27 KiB
Python
"""Step definitions for Plan v3 domain model tests."""
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.domain.models.core.plan import (
|
|
AutomationProfileProvenance,
|
|
AutomationProfileRef,
|
|
InvariantSource,
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanInvariant,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
can_transition,
|
|
)
|
|
|
|
# NamespacedName Steps
|
|
|
|
|
|
@when('I parse the namespaced name "{full_name}"')
|
|
def step_parse_namespaced_name(context: Context, full_name: str) -> None:
|
|
"""Parse a full namespaced name string."""
|
|
context.namespaced_name = NamespacedName.parse(full_name)
|
|
|
|
|
|
@then('the parsed namespace should be "{expected}"')
|
|
def step_check_namespace(context: Context, expected: str) -> None:
|
|
"""Check the namespace matches expected value."""
|
|
assert context.namespaced_name.namespace == expected, (
|
|
f"Expected namespace '{expected}', got '{context.namespaced_name.namespace}'"
|
|
)
|
|
|
|
|
|
@then('the parsed item name should be "{expected}"')
|
|
def step_check_name(context: Context, expected: str) -> None:
|
|
"""Check the name matches expected value."""
|
|
assert context.namespaced_name.name == expected, (
|
|
f"Expected name '{expected}', got '{context.namespaced_name.name}'"
|
|
)
|
|
|
|
|
|
@then("the parsed server should be empty")
|
|
def step_check_server_empty(context: Context) -> None:
|
|
"""Check the server is None."""
|
|
assert context.namespaced_name.server is None, (
|
|
f"Expected server to be None, got '{context.namespaced_name.server}'"
|
|
)
|
|
|
|
|
|
@then('the parsed server should be "{expected}"')
|
|
def step_check_server(context: Context, expected: str) -> None:
|
|
"""Check the server matches expected value."""
|
|
assert context.namespaced_name.server == expected, (
|
|
f"Expected server '{expected}', got '{context.namespaced_name.server}'"
|
|
)
|
|
|
|
|
|
@given(
|
|
'I have a namespaced name with server "{server}" namespace "{namespace}" and name "{name}"'
|
|
)
|
|
def step_create_namespaced_name_with_server(
|
|
context: Context, server: str, namespace: str, name: str
|
|
) -> None:
|
|
"""Create a namespaced name with all components."""
|
|
context.namespaced_name = NamespacedName(
|
|
server=server, namespace=namespace, name=name
|
|
)
|
|
|
|
|
|
@given('I have a namespaced name with namespace "{namespace}" and name "{name}"')
|
|
def step_create_namespaced_name_without_server(
|
|
context: Context, namespace: str, name: str
|
|
) -> None:
|
|
"""Create a namespaced name without server."""
|
|
context.namespaced_name = NamespacedName(
|
|
server=None, namespace=namespace, name=name
|
|
)
|
|
|
|
|
|
@given('I create a namespaced name with an empty namespace and name "{name}"')
|
|
def step_create_namespaced_name_empty_namespace(context: Context, name: str) -> None:
|
|
"""Create a namespaced name with an empty namespace."""
|
|
context.namespaced_name = NamespacedName(server=None, namespace="", name=name)
|
|
|
|
|
|
@then('the namespaced string representation should be "{expected}"')
|
|
def step_check_string_representation(context: Context, expected: str) -> None:
|
|
"""Check the string representation matches expected."""
|
|
result = str(context.namespaced_name)
|
|
assert result == expected, f"Expected '{expected}', got '{result}'"
|
|
|
|
|
|
@when(
|
|
'I try to create a namespaced name with namespace "{namespace}" and name "{name}"'
|
|
)
|
|
def step_try_create_invalid_namespaced_name(
|
|
context: Context, namespace: str, name: str
|
|
) -> None:
|
|
"""Attempt to create a namespaced name that might fail validation."""
|
|
context.error = None
|
|
try:
|
|
context.namespaced_name = NamespacedName(
|
|
server=None, namespace=namespace, name=name
|
|
)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
# Note: "a validation error should be raised" step is defined in domain_models_steps.py
|
|
|
|
|
|
# PlanPhase Steps
|
|
|
|
|
|
@then('the plan phases should be in order "{expected_order}"')
|
|
def step_check_phase_order(context: Context, expected_order: str) -> None:
|
|
"""Verify that phases are in the expected order."""
|
|
expected_phases = [p.strip() for p in expected_order.split(",")]
|
|
actual_phases = [phase.value for phase in PlanPhase]
|
|
assert actual_phases == expected_phases, (
|
|
f"Expected phases {expected_phases}, got {actual_phases}"
|
|
)
|
|
|
|
|
|
# Plan Creation Steps
|
|
|
|
|
|
def _default_plan(
|
|
*,
|
|
identity: PlanIdentity | None = None,
|
|
namespaced_name: NamespacedName | None = None,
|
|
description: str = "Test plan description",
|
|
definition_of_done: str | None = None,
|
|
action_name: str = "local/test-action",
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
processing_state: ProcessingState = ProcessingState.QUEUED,
|
|
strategy_actor: str | None = None,
|
|
execution_actor: str | None = None,
|
|
created_by: str | None = None,
|
|
reusable: bool = True,
|
|
read_only: bool = False,
|
|
error_message: str | None = None,
|
|
) -> Plan:
|
|
"""Helper to create a Plan with sensible defaults for testing.
|
|
|
|
All plans require ``action_name`` and start in STRATEGIZE phase.
|
|
"""
|
|
if identity is None:
|
|
identity = PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV")
|
|
if namespaced_name is None:
|
|
namespaced_name = NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
)
|
|
return Plan(
|
|
identity=identity,
|
|
namespaced_name=namespaced_name,
|
|
description=description,
|
|
definition_of_done=definition_of_done,
|
|
action_name=action_name,
|
|
phase=phase,
|
|
processing_state=processing_state,
|
|
strategy_actor=strategy_actor,
|
|
execution_actor=execution_actor,
|
|
created_by=created_by,
|
|
reusable=reusable,
|
|
read_only=read_only,
|
|
error_message=error_message,
|
|
)
|
|
|
|
|
|
@given("I create a new plan")
|
|
def step_create_new_plan(context: Context) -> None:
|
|
"""Create a new plan with default values."""
|
|
context.plan = _default_plan()
|
|
|
|
|
|
@given('I create a new plan with description "{description}"')
|
|
def step_create_plan_with_description(context: Context, description: str) -> None:
|
|
"""Create a new plan with specific description."""
|
|
context.plan = _default_plan(description=description)
|
|
|
|
|
|
@given("I create a PlanModel in strategize phase")
|
|
def step_create_plan_strategize_phase(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
|
|
|
|
@given(
|
|
"I create a plan in strategize phase with action state set and no processing state"
|
|
)
|
|
def step_create_plan_strategize_with_action_state(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase with QUEUED state."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
|
|
|
|
@given("I create a PlanModel in strategize phase with errored state")
|
|
def step_create_plan_strategize_errored(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase with ERRORED state."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.ERRORED,
|
|
error_message="Test error",
|
|
)
|
|
|
|
|
|
@given("I create a PlanModel in strategize phase with processing state")
|
|
def step_create_plan_strategize_processing(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase with PROCESSING state."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.PROCESSING,
|
|
)
|
|
|
|
|
|
@given("I create a plan in applied phase")
|
|
def step_create_plan_applied_phase(context: Context) -> None:
|
|
"""Create a plan in APPLY phase with APPLIED processing state (terminal)."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.APPLY,
|
|
processing_state=ProcessingState.APPLIED,
|
|
)
|
|
|
|
|
|
@given("I create a plan in apply phase as terminal")
|
|
def step_create_plan_apply_terminal(context: Context) -> None:
|
|
"""Create a plan in APPLY phase with APPLIED processing state (terminal)."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.APPLY,
|
|
processing_state=ProcessingState.APPLIED,
|
|
)
|
|
|
|
|
|
@given("I create a new plan in action phase")
|
|
def step_create_plan_action_phase(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase (formerly ACTION phase)."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
|
|
|
|
@given("I create a plan in action phase without action state")
|
|
def step_create_plan_action_without_state(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase without explicit state."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
|
|
|
|
@given("I create a plan in action phase with available state")
|
|
def step_create_plan_action_available(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase with COMPLETE state (can transition)."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.COMPLETE,
|
|
)
|
|
|
|
|
|
@given("I create a PlanModel in strategize phase with cancelled state")
|
|
def step_create_plan_strategize_cancelled(context: Context) -> None:
|
|
"""Create a plan in STRATEGIZE phase with CANCELLED processing state."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.CANCELLED,
|
|
)
|
|
|
|
|
|
@given("I create a plan in action phase with archived state")
|
|
def step_create_plan_action_archived(context: Context) -> None:
|
|
"""Create a plan in APPLY phase with APPLIED state (terminal, formerly ACTION/ARCHIVED)."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.APPLY,
|
|
processing_state=ProcessingState.APPLIED,
|
|
)
|
|
|
|
|
|
@given("I create a plan in execute phase with complete processing state")
|
|
def step_create_plan_execute_complete(context: Context) -> None:
|
|
"""Create a plan in EXECUTE phase with COMPLETE processing_state."""
|
|
context.plan = _default_plan(
|
|
phase=PlanPhase.EXECUTE,
|
|
processing_state=ProcessingState.COMPLETE,
|
|
)
|
|
|
|
|
|
@given("I create a plan with an unknown phase value")
|
|
def step_create_plan_unknown_phase(context: Context) -> None:
|
|
"""Create a plan with a phase outside the enum."""
|
|
context.plan = Plan.model_construct(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
description="Test plan",
|
|
action_name="local/test-action",
|
|
phase="mystery",
|
|
processing_state=None,
|
|
)
|
|
|
|
|
|
# Plan State Checks
|
|
|
|
|
|
@then('the plan phase should be "{expected}"')
|
|
def step_check_plan_phase(context: Context, expected: str) -> None:
|
|
"""Check the plan phase matches expected."""
|
|
actual = context.plan.phase.value
|
|
assert actual == expected, f"Expected phase '{expected}', got '{actual}'"
|
|
|
|
|
|
@then('the plan state should be "{expected}"')
|
|
def step_check_plan_state(context: Context, expected: str) -> None:
|
|
"""Check the derived plan state matches expected."""
|
|
state = context.plan.state
|
|
actual = state.value
|
|
assert actual == expected, f"Expected plan state '{expected}', got '{actual}'"
|
|
|
|
|
|
@then('the plan processing state should be "{expected}"')
|
|
def step_check_processing_state(context: Context, expected: str) -> None:
|
|
"""Check the plan state (formerly processing_state, now unified state)."""
|
|
assert context.plan.state is not None, "State is None"
|
|
actual = context.plan.state.value
|
|
assert actual == expected, f"Expected state '{expected}', got '{actual}'"
|
|
|
|
|
|
@then("the plan processing state should be empty")
|
|
def step_check_processing_state_empty(context: Context) -> None:
|
|
"""Processing state no longer exists as separate field; assert state is present."""
|
|
assert context.plan.state is not None, "Expected state to exist on plan"
|
|
|
|
|
|
@then("the plan should not be terminal")
|
|
def step_check_plan_not_terminal(context: Context) -> None:
|
|
"""Check the plan is not in a terminal state."""
|
|
assert not context.plan.is_terminal, "Expected plan to not be terminal"
|
|
|
|
|
|
@then("the plan should be terminal")
|
|
def step_check_plan_terminal(context: Context) -> None:
|
|
"""Check the plan is in a terminal state."""
|
|
assert context.plan.is_terminal, "Expected plan to be terminal"
|
|
|
|
|
|
@then("the plan should be errored")
|
|
def step_check_plan_errored(context: Context) -> None:
|
|
"""Check the plan is in an errored state."""
|
|
assert context.plan.is_errored, "Expected plan to be errored"
|
|
|
|
|
|
@then("the plan should not be errored")
|
|
def step_check_plan_not_errored(context: Context) -> None:
|
|
"""Check the plan is not in an errored state."""
|
|
assert not context.plan.is_errored, "Expected plan to not be errored"
|
|
|
|
|
|
@then("the plan can transition to next phase")
|
|
def step_check_plan_can_transition(context: Context) -> None:
|
|
"""Check the plan can transition to the next phase."""
|
|
assert context.plan.can_transition_to_next_phase, (
|
|
"Expected plan to be able to transition to next phase"
|
|
)
|
|
|
|
|
|
@then("the plan cannot transition to next phase")
|
|
def step_check_plan_cannot_transition(context: Context) -> None:
|
|
"""Check the plan cannot transition to the next phase."""
|
|
assert not context.plan.can_transition_to_next_phase, (
|
|
"Expected plan to not be able to transition to next phase"
|
|
)
|
|
|
|
|
|
@then('the next phase should be "{expected}"')
|
|
def step_check_next_phase(context: Context, expected: str) -> None:
|
|
"""Check the next phase matches expected."""
|
|
next_phase = context.plan.get_next_phase()
|
|
assert next_phase is not None, "Expected next phase to exist"
|
|
assert next_phase.value == expected, (
|
|
f"Expected next phase '{expected}', got '{next_phase.value}'"
|
|
)
|
|
|
|
|
|
@then("the next phase should be empty")
|
|
def step_check_next_phase_empty(context: Context) -> None:
|
|
"""Check there is no next phase."""
|
|
next_phase = context.plan.get_next_phase()
|
|
assert next_phase is None, f"Expected no next phase, got '{next_phase}'"
|
|
|
|
|
|
# Plan Identity Steps
|
|
|
|
|
|
@when('I try to create a plan identity with invalid ULID "{ulid}"')
|
|
def step_try_create_invalid_identity(context: Context, ulid: str) -> None:
|
|
"""Attempt to create a plan identity with invalid ULID."""
|
|
context.error = None
|
|
try:
|
|
context.plan_identity = PlanIdentity(plan_id=ulid)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
@when('I create a plan identity with ULID "{ulid}"')
|
|
def step_create_valid_identity(context: Context, ulid: str) -> None:
|
|
"""Create a plan identity with valid ULID."""
|
|
context.error = None
|
|
try:
|
|
context.plan_identity = PlanIdentity(plan_id=ulid)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
@then("the plan identity should be valid")
|
|
def step_check_identity_valid(context: Context) -> None:
|
|
"""Verify that the plan identity was created successfully."""
|
|
assert context.error is None, f"Expected no validation error, got: {context.error}"
|
|
assert context.plan_identity is not None, "Expected plan identity to exist"
|
|
|
|
|
|
@when('I create a plan identity with parent "{parent}" and root "{root}"')
|
|
def step_create_identity_with_hierarchy(
|
|
context: Context, parent: str, root: str
|
|
) -> None:
|
|
"""Create a plan identity with parent and root plan IDs."""
|
|
context.plan_identity = PlanIdentity(
|
|
plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAX",
|
|
parent_plan_id=parent,
|
|
root_plan_id=root,
|
|
)
|
|
|
|
|
|
@then('the parent plan ID should be "{expected}"')
|
|
def step_check_parent_plan_id(context: Context, expected: str) -> None:
|
|
"""Check the parent plan ID matches expected."""
|
|
assert context.plan_identity.parent_plan_id == expected, (
|
|
f"Expected parent '{expected}', got '{context.plan_identity.parent_plan_id}'"
|
|
)
|
|
|
|
|
|
@then('the root plan ID should be "{expected}"')
|
|
def step_check_root_plan_id(context: Context, expected: str) -> None:
|
|
"""Check the root plan ID matches expected."""
|
|
assert context.plan_identity.root_plan_id == expected, (
|
|
f"Expected root '{expected}', got '{context.plan_identity.root_plan_id}'"
|
|
)
|
|
|
|
|
|
# Phase Transition Steps
|
|
|
|
|
|
@then('transition from "{from_phase}" to "{to_phase}" should be valid')
|
|
def step_check_valid_transition(
|
|
context: Context, from_phase: str, to_phase: str
|
|
) -> None:
|
|
"""Check that a phase transition is valid."""
|
|
from_p = PlanPhase(from_phase)
|
|
to_p = PlanPhase(to_phase)
|
|
assert can_transition(from_p, to_p), (
|
|
f"Expected transition from {from_phase} to {to_phase} to be valid"
|
|
)
|
|
|
|
|
|
@then('transition from "{from_phase}" to "{to_phase}" should be invalid')
|
|
def step_check_invalid_transition(
|
|
context: Context, from_phase: str, to_phase: str
|
|
) -> None:
|
|
"""Check that a phase transition is invalid."""
|
|
from_p = PlanPhase(from_phase)
|
|
to_p = PlanPhase(to_phase)
|
|
assert not can_transition(from_p, to_p), (
|
|
f"Expected transition from {from_phase} to {to_phase} to be invalid"
|
|
)
|
|
|
|
|
|
# Invalid Plan Creation Steps
|
|
|
|
|
|
@when("I try to create a plan without description")
|
|
def step_try_create_plan_without_description(context: Context) -> None:
|
|
"""Attempt to create a plan without description."""
|
|
context.error = None
|
|
try:
|
|
context.plan = Plan.model_validate(
|
|
{
|
|
"identity": PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
"namespaced_name": NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
"action_name": "local/test-action",
|
|
"definition_of_done": None,
|
|
"phase": PlanPhase.STRATEGIZE,
|
|
"processing_state": ProcessingState.QUEUED,
|
|
"strategy_actor": None,
|
|
"execution_actor": None,
|
|
"created_by": None,
|
|
"reusable": True,
|
|
"read_only": False,
|
|
}
|
|
)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I try to create a plan with invalid phase value")
|
|
def step_try_create_plan_with_invalid_phase(context: Context) -> None:
|
|
"""Test that creating a plan with an invalid phase value raises ValidationError."""
|
|
context.error = None
|
|
try:
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
description="Test plan",
|
|
action_name="local/test-action",
|
|
definition_of_done=None,
|
|
phase="nonexistent", # Invalid phase value
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
created_by=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I try to create a plan with empty description")
|
|
def step_try_create_plan_empty_description(context: Context) -> None:
|
|
"""Attempt to create a plan with empty description."""
|
|
context.error = None
|
|
try:
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
description="",
|
|
action_name="local/test-action",
|
|
definition_of_done=None,
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
created_by=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
# Automation Profile Provenance Steps
|
|
|
|
|
|
@given('I create a plan with automation profile "{profile_name}"')
|
|
def step_create_plan_with_automation_profile(
|
|
context: Context, profile_name: str
|
|
) -> None:
|
|
"""Create a plan with an automation profile name set."""
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
description="Plan with automation profile",
|
|
action_name="local/test-action",
|
|
definition_of_done=None,
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
automation_profile=AutomationProfileRef(
|
|
profile_name=profile_name,
|
|
provenance=AutomationProfileProvenance.PLAN,
|
|
),
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
created_by=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
@given("I create a plan with automation profile and snapshot")
|
|
def step_create_plan_with_profile_snapshot(context: Context) -> None:
|
|
"""Create a plan with automation profile."""
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
description="Plan with profile snapshot",
|
|
action_name="local/test-action",
|
|
definition_of_done=None,
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
automation_profile=AutomationProfileRef(
|
|
profile_name="org/high-trust-profile",
|
|
provenance=AutomationProfileProvenance.ACTION,
|
|
),
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
created_by=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
@then('the plan automation profile name should be "{expected}"')
|
|
def step_check_automation_profile_name(context: Context, expected: str) -> None:
|
|
"""Check the plan automation profile name."""
|
|
assert context.plan.automation_profile is not None, "Expected profile to be set"
|
|
assert context.plan.automation_profile.profile_name == expected, (
|
|
f"Expected profile '{expected}', got '{context.plan.automation_profile.profile_name}'"
|
|
)
|
|
|
|
|
|
@then('the plan model automation level should be "{expected}"')
|
|
def step_check_automation_level(context: Context, expected: str) -> None:
|
|
"""Check the plan automation profile (automation_level removed)."""
|
|
# automation_level field was removed; verify via profile instead
|
|
if context.plan.automation_profile is not None:
|
|
actual = context.plan.automation_profile.profile_name
|
|
else:
|
|
actual = "manual"
|
|
# Map old level values to profile names for backward compat
|
|
level_to_profile = {
|
|
"manual": "manual",
|
|
"review_before_apply": "auto",
|
|
"full_automation": "full-auto",
|
|
}
|
|
expected_profile = level_to_profile.get(expected, expected)
|
|
assert actual == expected_profile, (
|
|
f"Expected profile '{expected_profile}', got '{actual}'"
|
|
)
|
|
|
|
|
|
@then('the plan effective profile snapshot should contain "{key}"')
|
|
def step_check_profile_snapshot_key(context: Context, key: str) -> None:
|
|
"""Check the automation profile has expected provenance."""
|
|
assert context.plan.automation_profile is not None, "Expected profile to be set"
|
|
# Profile snapshot is no longer separate; check provenance instead
|
|
assert context.plan.automation_profile.provenance is not None
|
|
|
|
|
|
@then("the automation profile should be immutable after creation")
|
|
def step_check_profile_immutable(context: Context) -> None:
|
|
"""Verify the profile exists and is set."""
|
|
assert context.plan.automation_profile is not None, "Expected profile to be set"
|
|
|
|
|
|
# Invariant Ordering Steps
|
|
|
|
|
|
@given("I create a plan with ordered invariants")
|
|
def step_create_plan_with_invariants(context: Context) -> None:
|
|
"""Create a plan with invariants in specific order."""
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="test-plan"
|
|
),
|
|
description="Plan with invariants",
|
|
action_name="local/test-action",
|
|
definition_of_done=None,
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
invariants=[
|
|
PlanInvariant(
|
|
text="Global rule",
|
|
source=InvariantSource.GLOBAL,
|
|
),
|
|
PlanInvariant(
|
|
text="Project rule",
|
|
source=InvariantSource.PROJECT,
|
|
),
|
|
PlanInvariant(
|
|
text="Plan rule",
|
|
source=InvariantSource.PLAN,
|
|
),
|
|
],
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
created_by=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
@then('the invariant at index {index:d} should have text "{expected}"')
|
|
def step_check_invariant_text(context: Context, index: int, expected: str) -> None:
|
|
"""Check invariant text at given index."""
|
|
invariants = context.plan.invariants
|
|
assert index < len(invariants), (
|
|
f"Index {index} out of range (have {len(invariants)} invariants)"
|
|
)
|
|
actual = invariants[index].text
|
|
assert actual == expected, (
|
|
f"Expected text '{expected}' at index {index}, got '{actual}'"
|
|
)
|
|
|
|
|
|
@then('the invariant at index {index:d} should have scope "{expected}"')
|
|
def step_check_invariant_scope(context: Context, index: int, expected: str) -> None:
|
|
"""Check invariant scope at given index."""
|
|
invariants = context.plan.invariants
|
|
assert index < len(invariants), (
|
|
f"Index {index} out of range (have {len(invariants)} invariants)"
|
|
)
|
|
actual = invariants[index].source.value
|
|
assert actual == expected, (
|
|
f"Expected scope '{expected}' at index {index}, got '{actual}'"
|
|
)
|
|
|
|
|
|
@when("I try to create a plan invariant with empty text")
|
|
def step_try_create_invariant_empty_text(context: Context) -> None:
|
|
"""Attempt to create an invariant with empty text."""
|
|
context.error = None
|
|
try:
|
|
PlanInvariant(text="", source=InvariantSource.GLOBAL)
|
|
except ValidationError as e:
|
|
context.error = e
|
|
|
|
|
|
# Fully Populated Plan for CLI Dict Testing
|
|
|
|
|
|
@given("I create a fully populated plan for cli dict testing")
|
|
def step_create_fully_populated_plan(context: Context) -> None:
|
|
"""Create a plan with all optional fields populated."""
|
|
context.plan = Plan(
|
|
identity=PlanIdentity(
|
|
plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
|
|
parent_plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAW",
|
|
root_plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAX",
|
|
attempt=3,
|
|
),
|
|
namespaced_name=NamespacedName(
|
|
server=None, namespace="local", name="full-plan"
|
|
),
|
|
description="Fully populated plan",
|
|
action_name="local/full-action",
|
|
definition_of_done="All features complete",
|
|
phase=PlanPhase.EXECUTE,
|
|
processing_state=ProcessingState.PROCESSING,
|
|
automation_profile=AutomationProfileRef(
|
|
profile_name="org/full-profile",
|
|
provenance=AutomationProfileProvenance.ACTION,
|
|
),
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
estimation_actor="openai/gpt-4-turbo",
|
|
invariant_actor="openai/gpt-4-turbo",
|
|
arguments={"target": "/src", "mode": "strict"},
|
|
arguments_order=["target", "mode"],
|
|
invariants=[
|
|
PlanInvariant(
|
|
text="Keep it safe",
|
|
source=InvariantSource.GLOBAL,
|
|
),
|
|
],
|
|
project_links=[
|
|
ProjectLink(project_name="local/proj"),
|
|
],
|
|
changeset_id="01ARZ3NDEKTSV4RRFFQ69G5FAY",
|
|
sandbox_refs=["sandbox-001"],
|
|
error_message="Partial failure in step 3",
|
|
error_details={"step": 3, "reason": "timeout"},
|
|
tags=["critical", "v2"],
|
|
created_by="test-user",
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|