Files

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 plan 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 plan 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 plan 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 plan 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,
)