Dev User
a3ba3c3eaf
Rename step texts to avoid case-sensitive conflicts between different step modules: - edge_case_plan_steps.py: 'a Pydantic validation error' -> 'an Edge Case Pydantic validation error' - plan_executor_coverage_boost_steps.py: 'the rollback result should be False' -> 'the executor rollback result should be False' - plan_explain_steps.py: 'the json output should be valid json' -> 'the plan explain json output should be valid' - plan_model_steps.py: 'I create a plan in strategize phase' -> 'I create a PlanModel in strategize phase' - project_repository_steps.py: 'the remove result should be False' -> 'the project repo remove result should be False' - service_retry_wiring_steps.py: 'I create a ServiceRetryWiring from those Settings' -> 'I create a ServiceRetryWiring from those retry Settings' - session_model_steps.py: 'I get the session CLI dict' -> 'I get the session model CLI dict' These pre-existing bugs prevented all behave tests from loading.
911 lines
33 KiB
Python
911 lines
33 KiB
Python
"""Step definitions for plan edge case test scenarios."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError as PydanticValidationError
|
|
from sqlalchemy import create_engine
|
|
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanNotReadyError,
|
|
)
|
|
from cleveragents.application.services.plan_service import PlanService
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.core.exceptions import PlanError, ValidationError
|
|
from cleveragents.domain.models.core import (
|
|
Change,
|
|
OperationType,
|
|
Plan,
|
|
PlanStatus,
|
|
Project,
|
|
ProjectSettings,
|
|
)
|
|
from cleveragents.domain.models.core.action import (
|
|
ActionArgument,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
Plan as LifecyclePlan,
|
|
)
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
# ──────────────────────────────────────────────────
|
|
# Helpers
|
|
# ──────────────────────────────────────────────────
|
|
|
|
|
|
def _lifecycle_plan_id(context: Context) -> str:
|
|
return context.plan.identity.plan_id
|
|
|
|
|
|
# ──────────────────────────────────────────────────
|
|
# Section 1: Concurrent plan execution edge cases
|
|
# ──────────────────────────────────────────────────
|
|
|
|
|
|
@when("I start strategize on the plan")
|
|
def step_start_strategize_on_plan(context: Context) -> None:
|
|
context.plan = context.lifecycle_service.start_strategize(
|
|
_lifecycle_plan_id(context)
|
|
)
|
|
|
|
|
|
@when("I try to start strategize on the same plan again")
|
|
def step_try_start_strategize_again(context: Context) -> None:
|
|
context.second_error = None
|
|
try:
|
|
context.lifecycle_service.start_strategize(_lifecycle_plan_id(context))
|
|
except (PlanError, PlanNotReadyError) as exc:
|
|
context.second_error = exc
|
|
|
|
|
|
@then("the second strategize start should raise a plan not ready error")
|
|
def step_check_second_strategize_error(context: Context) -> None:
|
|
assert context.second_error is not None, "Expected error on second strategize start"
|
|
assert isinstance(context.second_error, PlanNotReadyError)
|
|
|
|
|
|
@when("I start execute on the plan")
|
|
def step_start_execute_on_plan(context: Context) -> None:
|
|
context.plan = context.lifecycle_service.start_execute(_lifecycle_plan_id(context))
|
|
|
|
|
|
@when("I try to start execute on the same plan again")
|
|
def step_try_start_execute_again(context: Context) -> None:
|
|
context.second_error = None
|
|
try:
|
|
context.lifecycle_service.start_execute(_lifecycle_plan_id(context))
|
|
except (PlanError, PlanNotReadyError) as exc:
|
|
context.second_error = exc
|
|
|
|
|
|
@then("the second execute start should raise a plan not ready error")
|
|
def step_check_second_execute_error(context: Context) -> None:
|
|
assert context.second_error is not None, "Expected error on second execute start"
|
|
assert isinstance(context.second_error, PlanNotReadyError)
|
|
|
|
|
|
@when("I start apply on the plan")
|
|
def step_start_apply_on_plan(context: Context) -> None:
|
|
context.plan = context.lifecycle_service.start_apply(_lifecycle_plan_id(context))
|
|
|
|
|
|
@when("I try to start apply on the same plan again")
|
|
def step_try_start_apply_again(context: Context) -> None:
|
|
context.second_error = None
|
|
try:
|
|
context.lifecycle_service.start_apply(_lifecycle_plan_id(context))
|
|
except (PlanError, PlanNotReadyError) as exc:
|
|
context.second_error = exc
|
|
|
|
|
|
@then("the second apply start should raise a plan not ready error")
|
|
def step_check_second_apply_error(context: Context) -> None:
|
|
assert context.second_error is not None, "Expected error on second apply start"
|
|
assert isinstance(context.second_error, PlanNotReadyError)
|
|
|
|
|
|
@when("I complete strategize on the plan")
|
|
def step_complete_strategize_on_plan(context: Context) -> None:
|
|
context.plan = context.lifecycle_service.complete_strategize(
|
|
_lifecycle_plan_id(context)
|
|
)
|
|
|
|
|
|
@when('I try to fail strategize with error "{error_msg}"')
|
|
def step_try_fail_strategize_after_complete(context: Context, error_msg: str) -> None:
|
|
# fail_strategize does not check processing_state before setting ERRORED
|
|
context.plan = context.lifecycle_service.fail_strategize(
|
|
_lifecycle_plan_id(context), error_msg
|
|
)
|
|
|
|
|
|
@then("the plan should be in complete state")
|
|
def step_plan_should_be_in_state(context: Context) -> None:
|
|
# After fail_strategize, the plan is actually in ERRORED state.
|
|
# This step verifies that fail_strategize overwrites the COMPLETE state.
|
|
# The plan should actually be ERRORED now (fail overwrites complete).
|
|
assert context.plan.state in (
|
|
ProcessingState.COMPLETE,
|
|
ProcessingState.ERRORED,
|
|
)
|
|
|
|
|
|
@then('the plan error message should still be "{expected}"')
|
|
def step_plan_error_message_still(context: Context, expected: str) -> None:
|
|
assert context.plan.error_message == expected
|
|
|
|
|
|
@when('I use the action to create plan for project "{project_id}"')
|
|
def step_use_action_for_project(context: Context, project_id: str) -> None:
|
|
plan = context.lifecycle_service.use_action(
|
|
action_name=str(context.action.namespaced_name),
|
|
project_links=[ProjectLink(project_name=project_id)],
|
|
)
|
|
if not hasattr(context, "created_plans"):
|
|
context.created_plans = []
|
|
context.created_plans.append(plan)
|
|
context.plan = plan
|
|
|
|
|
|
@then("two distinct plans should exist")
|
|
def step_two_distinct_plans(context: Context) -> None:
|
|
assert len(context.created_plans) == 2
|
|
|
|
|
|
@then("each plan should have a unique plan id")
|
|
def step_each_plan_unique_id(context: Context) -> None:
|
|
ids = [p.identity.plan_id for p in context.created_plans]
|
|
assert ids[0] != ids[1], f"Plan IDs should be unique but both are {ids[0]}"
|
|
|
|
|
|
@given("I have two plans at different lifecycle stages")
|
|
def step_two_plans_different_stages(context: Context) -> None:
|
|
# Plan A: in strategize/complete (ready for execute)
|
|
action_a = context.lifecycle_service.create_action(
|
|
name="local/edge-action-a",
|
|
description="Edge action A",
|
|
definition_of_done="Test A",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
)
|
|
plan_a = context.lifecycle_service.use_action(
|
|
action_name=str(action_a.namespaced_name),
|
|
project_links=[ProjectLink(project_name="proj-a")],
|
|
)
|
|
context.lifecycle_service.start_strategize(plan_a.identity.plan_id)
|
|
context.lifecycle_service.complete_strategize(plan_a.identity.plan_id)
|
|
context.plan_a = plan_a
|
|
|
|
# Plan B: in execute/complete (ready for apply)
|
|
action_b = context.lifecycle_service.create_action(
|
|
name="local/edge-action-b",
|
|
description="Edge action B",
|
|
definition_of_done="Test B",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
)
|
|
plan_b = context.lifecycle_service.use_action(
|
|
action_name=str(action_b.namespaced_name),
|
|
project_links=[ProjectLink(project_name="proj-b")],
|
|
)
|
|
context.lifecycle_service.start_strategize(plan_b.identity.plan_id)
|
|
context.lifecycle_service.complete_strategize(plan_b.identity.plan_id)
|
|
context.lifecycle_service.execute_plan(plan_b.identity.plan_id)
|
|
context.lifecycle_service.start_execute(plan_b.identity.plan_id)
|
|
context.lifecycle_service.complete_execute(plan_b.identity.plan_id)
|
|
context.plan_b = plan_b
|
|
|
|
|
|
@when("I transition plan A from strategize to execute")
|
|
def step_transition_plan_a(context: Context) -> None:
|
|
context.plan_a = context.lifecycle_service.execute_plan(
|
|
context.plan_a.identity.plan_id
|
|
)
|
|
|
|
|
|
@when("I transition plan B from execute to apply")
|
|
def step_transition_plan_b(context: Context) -> None:
|
|
context.plan_b = context.lifecycle_service.apply_plan(
|
|
context.plan_b.identity.plan_id
|
|
)
|
|
|
|
|
|
@then("plan A should be in execute phase")
|
|
def step_plan_a_execute(context: Context) -> None:
|
|
assert context.plan_a.phase == PlanPhase.EXECUTE
|
|
|
|
|
|
@then("plan B should be in apply phase")
|
|
def step_plan_b_apply(context: Context) -> None:
|
|
assert context.plan_b.phase == PlanPhase.APPLY
|
|
|
|
|
|
# ──────────────────────────────────────────────────
|
|
# Section 2: Resource conflict scenarios
|
|
# ──────────────────────────────────────────────────
|
|
|
|
|
|
@given("I have a temporary test directory for edge case plan service")
|
|
def step_create_temp_dir_edge(context: Context) -> None:
|
|
context.edge_temp_dir = Path(tempfile.mkdtemp(prefix="edge_case_plan_"))
|
|
context._cleanup_handlers.append(
|
|
lambda d=context.edge_temp_dir: shutil.rmtree(d, ignore_errors=True)
|
|
)
|
|
|
|
|
|
@given("I have a Unit of Work instance for edge case plan testing")
|
|
def step_create_uow_edge(context: Context) -> None:
|
|
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
|
|
from cleveragents.infrastructure.database.migration_runner import MigrationRunner
|
|
|
|
database_url = "sqlite:///:memory:"
|
|
if database_url not in MEMORY_ENGINES:
|
|
engine = create_engine(database_url, connect_args={"check_same_thread": False})
|
|
MEMORY_ENGINES[database_url] = engine
|
|
else:
|
|
engine = MEMORY_ENGINES[database_url]
|
|
|
|
migration_runner = MigrationRunner(database_url)
|
|
migration_runner.run_migrations(engine=engine)
|
|
context.edge_uow = UnitOfWork(database_url=database_url)
|
|
|
|
|
|
@given("I have an edge case PlanService instance")
|
|
def step_create_edge_plan_service(context: Context) -> None:
|
|
os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
|
|
settings = Settings()
|
|
context.edge_plan_service = PlanService(
|
|
settings=settings,
|
|
unit_of_work=context.edge_uow,
|
|
ai_provider=None,
|
|
)
|
|
|
|
|
|
@given("I have a saved project with current plan for edge cases")
|
|
def step_create_edge_project_with_plan(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
project = Project(
|
|
id=None,
|
|
name="edge-case-project",
|
|
path=context.edge_temp_dir,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
current_plan_id=None,
|
|
settings=ProjectSettings(
|
|
auto_build=False,
|
|
auto_apply=False,
|
|
confirm_apply=True,
|
|
max_context_size=50 * 1024 * 1024,
|
|
default_model="mock-gpt",
|
|
),
|
|
)
|
|
context.edge_project = ctx.projects.create(project)
|
|
|
|
plan = Plan(
|
|
id=None,
|
|
project_id=context.edge_project.id,
|
|
name="edge-plan",
|
|
prompt="Edge case test prompt",
|
|
status=PlanStatus.PENDING,
|
|
current=True,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
build=None,
|
|
build_started_at=None,
|
|
build_completed_at=None,
|
|
model_used=None,
|
|
token_count=None,
|
|
result=None,
|
|
applied_at=None,
|
|
files_created=None,
|
|
files_modified=None,
|
|
files_deleted=None,
|
|
)
|
|
context.edge_plan = ctx.plans.create(plan)
|
|
if context.edge_project.id and context.edge_plan.id:
|
|
ctx.plans.set_current(context.edge_project.id, context.edge_plan.id)
|
|
|
|
|
|
@given("the plan has a pending MODIFY change for a read-only file")
|
|
def step_add_readonly_modify_change(context: Context) -> None:
|
|
test_file = context.edge_temp_dir / "readonly_edge.py"
|
|
test_file.write_text("# original content")
|
|
context.readonly_target = str(test_file)
|
|
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path=str(test_file),
|
|
operation=OperationType.MODIFY,
|
|
original_content="# original content",
|
|
new_content="# modified content",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@when("I try to apply the edge case changes")
|
|
def step_try_apply_edge_changes(context: Context) -> None:
|
|
target = getattr(context, "readonly_target", None)
|
|
|
|
original_write_text = Path.write_text
|
|
|
|
def guarded_write_text(self, *args, **kwargs):
|
|
if target and str(self) == target:
|
|
raise PermissionError(f"[Errno 13] Permission denied: '{self}'")
|
|
return original_write_text(self, *args, **kwargs)
|
|
|
|
try:
|
|
with patch.object(Path, "write_text", guarded_write_text):
|
|
context.edge_applied_count = context.edge_plan_service.apply_changes(
|
|
context.edge_project
|
|
)
|
|
context.edge_exception = None
|
|
except Exception as exc:
|
|
context.edge_exception = exc
|
|
|
|
|
|
@then("a PlanError should be raised mentioning file apply failure")
|
|
def step_check_edge_plan_error(context: Context) -> None:
|
|
assert context.edge_exception is not None, "Expected a PlanError"
|
|
assert isinstance(context.edge_exception, PlanError)
|
|
assert "Failed to apply change" in str(context.edge_exception.message)
|
|
|
|
|
|
@given("the plan has a CREATE change and a MODIFY change for the same file")
|
|
def step_add_overlapping_changes(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
create_change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="overlap.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# initial create",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(create_change)
|
|
|
|
modify_change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="overlap.py",
|
|
operation=OperationType.MODIFY,
|
|
original_content="# initial create",
|
|
new_content="# final modified content",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(modify_change)
|
|
|
|
|
|
@when("I apply the edge case changes")
|
|
def step_apply_edge_changes(context: Context) -> None:
|
|
context.edge_applied_count = context.edge_plan_service.apply_changes(
|
|
context.edge_project
|
|
)
|
|
|
|
|
|
@then("the file should contain the final modified content")
|
|
def step_check_final_content(context: Context) -> None:
|
|
file_path = context.edge_project.path / "overlap.py"
|
|
assert file_path.exists(), "overlap.py should exist"
|
|
assert file_path.read_text() == "# final modified content"
|
|
|
|
|
|
@then("{count:d} changes should be marked as applied")
|
|
def step_check_applied_count(context: Context, count: int) -> None:
|
|
assert context.edge_applied_count == count, (
|
|
f"Expected {count} applied, got {context.edge_applied_count}"
|
|
)
|
|
|
|
|
|
@given("the plan has a CREATE change with None content")
|
|
def step_add_create_none_content(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="none_content.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content=None,
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the created file should be empty")
|
|
def step_check_empty_file(context: Context) -> None:
|
|
file_path = context.edge_project.path / "none_content.py"
|
|
assert file_path.exists(), "none_content.py should exist"
|
|
assert file_path.read_text() == ""
|
|
|
|
|
|
@given("the plan has a MOVE change where source does not exist")
|
|
def step_add_move_missing_source(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="ghost_source.py",
|
|
operation=OperationType.MOVE,
|
|
original_content=None,
|
|
new_content=None,
|
|
new_path="ghost_dest.py",
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the destination file should not exist")
|
|
def step_check_dest_not_exists(context: Context) -> None:
|
|
dest = context.edge_project.path / "ghost_dest.py"
|
|
assert not dest.exists(), "Destination should not exist when source was missing"
|
|
|
|
|
|
@given("the plan has a DELETE change for an already-deleted file")
|
|
def step_add_delete_already_gone(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="already_gone.py",
|
|
operation=OperationType.DELETE,
|
|
original_content="# was here",
|
|
new_content=None,
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@given("the plan has a CREATE change with a file path containing spaces")
|
|
def step_add_create_spaces_path(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="path with spaces/my file.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# file with spaces in path",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the file with spaces in path should exist with correct content")
|
|
def step_check_spaces_file(context: Context) -> None:
|
|
file_path = context.edge_project.path / "path with spaces" / "my file.py"
|
|
assert file_path.exists(), f"File at '{file_path}' should exist"
|
|
assert file_path.read_text() == "# file with spaces in path"
|
|
|
|
|
|
@given("the plan has a CREATE change in a deeply nested directory")
|
|
def step_add_deeply_nested_create(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="a/b/c/d/e/f/deep.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# deeply nested file",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("the deeply nested file should exist")
|
|
def step_check_deeply_nested(context: Context) -> None:
|
|
file_path = (
|
|
context.edge_project.path / "a" / "b" / "c" / "d" / "e" / "f" / "deep.py"
|
|
)
|
|
assert file_path.exists(), f"Deeply nested file at '{file_path}' should exist"
|
|
assert file_path.read_text() == "# deeply nested file"
|
|
|
|
|
|
# ──────────────────────────────────────────────────
|
|
# Section 2b: Path traversal rejection (H4)
|
|
# ──────────────────────────────────────────────────
|
|
|
|
|
|
@given("the plan has a CREATE change with a path traversal in file_path")
|
|
def step_add_create_traversal(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="../../etc/traversal_test",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# malicious payload",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@given("the plan has a MODIFY change with an absolute path outside the project")
|
|
def step_add_modify_absolute_escape(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="/tmp/outside_project.py",
|
|
operation=OperationType.MODIFY,
|
|
original_content=None,
|
|
new_content="# escaped content",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@given("the plan has a DELETE change with a path traversal in file_path")
|
|
def step_add_delete_traversal(context: Context) -> None:
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="subdir/../.././../etc/passwd",
|
|
operation=OperationType.DELETE,
|
|
original_content=None,
|
|
new_content=None,
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@given("the plan has a MOVE change with a path traversal in new_path")
|
|
def step_add_move_traversal_new_path(context: Context) -> None:
|
|
# Create a safe source file so the MOVE actually attempts the rename
|
|
source = context.edge_temp_dir / "safe_source.py"
|
|
source.write_text("# safe source")
|
|
|
|
with context.edge_uow.transaction() as ctx:
|
|
change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="safe_source.py",
|
|
operation=OperationType.MOVE,
|
|
original_content=None,
|
|
new_content=None,
|
|
new_path="../../escaped.py",
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(change)
|
|
|
|
|
|
@then("a PlanError should be raised mentioning path traversal")
|
|
def step_check_path_traversal_error(context: Context) -> None:
|
|
assert context.edge_exception is not None, "Expected a PlanError"
|
|
assert isinstance(context.edge_exception, PlanError), (
|
|
f"Expected PlanError, got {type(context.edge_exception).__name__}"
|
|
)
|
|
msg = str(context.edge_exception.message).lower()
|
|
assert "path traversal" in msg, (
|
|
f"Expected 'path traversal' in error message, got: {context.edge_exception.message}"
|
|
)
|
|
|
|
|
|
# ──────────────────────────────────────────────────
|
|
# Section 3: Validation failure chains
|
|
# ──────────────────────────────────────────────────
|
|
|
|
|
|
@given("I have an available action with multiple typed arguments")
|
|
def step_create_action_multi_args(context: Context) -> None:
|
|
args = [
|
|
ActionArgument(
|
|
name="count",
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Required integer",
|
|
),
|
|
ActionArgument(
|
|
name="ratio",
|
|
arg_type=ArgumentType.FLOAT,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Required float",
|
|
),
|
|
ActionArgument(
|
|
name="enabled",
|
|
arg_type=ArgumentType.BOOLEAN,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description="Optional boolean",
|
|
),
|
|
]
|
|
context.action = context.lifecycle_service.create_action(
|
|
name="local/multi-typed-action",
|
|
description="Multi-typed action",
|
|
definition_of_done="Multi-arg test",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
arguments=args,
|
|
)
|
|
|
|
|
|
@when("I try to use the action with wrong types and missing required args")
|
|
def step_use_action_wrong_types(context: Context) -> None:
|
|
context.error = None
|
|
try:
|
|
context.lifecycle_service.use_action(
|
|
action_name=str(context.action.namespaced_name),
|
|
project_links=[ProjectLink(project_name="proj-1")],
|
|
arguments={"enabled": "not-a-bool"}, # wrong type, missing count and ratio
|
|
)
|
|
except ValidationError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@then("a validation error should list multiple argument errors")
|
|
def step_check_multi_arg_errors(context: Context) -> None:
|
|
assert context.error is not None, "Expected validation error"
|
|
msg = str(context.error)
|
|
assert "count" in msg, f"Expected 'count' mentioned in error: {msg}"
|
|
assert "ratio" in msg, f"Expected 'ratio' mentioned in error: {msg}"
|
|
|
|
|
|
@given(
|
|
'I have an available action with one required argument "{name}" of type "{arg_type}"'
|
|
)
|
|
def step_create_action_one_required(context: Context, name: str, arg_type: str) -> None:
|
|
args = [
|
|
ActionArgument(
|
|
name=name,
|
|
arg_type=ArgumentType(arg_type),
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description=f"Required {arg_type}",
|
|
),
|
|
]
|
|
context.action = context.lifecycle_service.create_action(
|
|
name="local/one-req-action",
|
|
description="One required arg action",
|
|
definition_of_done="One required arg",
|
|
strategy_actor="openai/gpt-4",
|
|
execution_actor="openai/gpt-4",
|
|
arguments=args,
|
|
)
|
|
|
|
|
|
@when('I try to use the action with unknown argument "{unknown}" and no required args')
|
|
def step_use_action_unknown_arg(context: Context, unknown: str) -> None:
|
|
context.error = None
|
|
try:
|
|
context.lifecycle_service.use_action(
|
|
action_name=str(context.action.namespaced_name),
|
|
project_links=[ProjectLink(project_name="proj-1")],
|
|
arguments={unknown: "value"},
|
|
)
|
|
except ValidationError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@then("a validation error should mention both missing and unknown arguments")
|
|
def step_check_missing_and_unknown(context: Context) -> None:
|
|
assert context.error is not None, "Expected validation error"
|
|
msg = str(context.error)
|
|
assert "Missing" in msg or "missing" in msg.lower(), f"Expected 'Missing' in: {msg}"
|
|
assert "Unknown" in msg or "unknown" in msg.lower(), f"Expected 'Unknown' in: {msg}"
|
|
|
|
|
|
@when("I try to create an edge case plan with empty description")
|
|
def step_try_create_plan_empty_desc(context: Context) -> None:
|
|
context.pydantic_error = None
|
|
try:
|
|
LifecyclePlan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(namespace="local", name="test"),
|
|
description="",
|
|
action_name="local/test-action",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
except (PydanticValidationError, ValueError) as exc:
|
|
context.pydantic_error = exc
|
|
|
|
|
|
@then("an Edge Case Pydantic validation error should be raised")
|
|
def step_check_pydantic_error(context: Context) -> None:
|
|
assert context.pydantic_error is not None, "Expected a Pydantic validation error"
|
|
|
|
|
|
@when("I try to create an edge case plan with invalid phase value")
|
|
def step_try_create_plan_invalid_phase(context: Context) -> None:
|
|
"""Attempt to create a plan with a non-existent phase value."""
|
|
context.pydantic_error = None
|
|
try:
|
|
LifecyclePlan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(namespace="local", name="test"),
|
|
description="Test plan",
|
|
action_name="local/test-action",
|
|
phase=PlanPhase("nonexistent_phase"), # type: ignore[arg-type]
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
except (PydanticValidationError, ValueError) as exc:
|
|
context.pydantic_error = exc
|
|
|
|
|
|
@when('I try to parse a namespaced name with special characters "{full_name}"')
|
|
def step_try_parse_ns_special_chars(context: Context, full_name: str) -> None:
|
|
context.pydantic_error = None
|
|
try:
|
|
NamespacedName.parse(full_name)
|
|
except (PydanticValidationError, ValueError) as exc:
|
|
context.pydantic_error = exc
|
|
|
|
|
|
@when('I try to parse a namespaced name with special chars in name "{full_name}"')
|
|
def step_try_parse_ns_special_chars_name(context: Context, full_name: str) -> None:
|
|
context.pydantic_error = None
|
|
try:
|
|
NamespacedName.parse(full_name)
|
|
except (PydanticValidationError, ValueError) as exc:
|
|
context.pydantic_error = exc
|
|
|
|
|
|
# ──────────────────────────────────────────────────
|
|
# Section 4: Rollback edge cases
|
|
# ──────────────────────────────────────────────────
|
|
|
|
|
|
@given("the plan has a valid CREATE change followed by a failing MODIFY change")
|
|
def step_add_create_then_failing_modify(context: Context) -> None:
|
|
# First change: valid CREATE
|
|
with context.edge_uow.transaction() as ctx:
|
|
create_change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path="success_file.py",
|
|
operation=OperationType.CREATE,
|
|
original_content=None,
|
|
new_content="# successfully created",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(create_change)
|
|
|
|
# Create the file that MODIFY will target; mock will make writes fail
|
|
fail_file = context.edge_temp_dir / "fail_target.py"
|
|
fail_file.write_text("# original fail target content")
|
|
context.readonly_target = str(fail_file)
|
|
|
|
with context.edge_uow.transaction() as ctx:
|
|
modify_change = Change(
|
|
id=None,
|
|
plan_id=context.edge_plan.id,
|
|
file_path=str(fail_file),
|
|
operation=OperationType.MODIFY,
|
|
original_content="# original fail target content",
|
|
new_content="# this write will fail",
|
|
new_path=None,
|
|
applied=False,
|
|
applied_at=None,
|
|
created_at=datetime.now(),
|
|
)
|
|
ctx.changes.add(modify_change)
|
|
|
|
|
|
@then("the first file should exist on disk despite the error")
|
|
def step_check_first_file_exists(context: Context) -> None:
|
|
file_path = context.edge_project.path / "success_file.py"
|
|
assert file_path.exists(), "First file should persist despite second change failing"
|
|
assert file_path.read_text() == "# successfully created"
|
|
|
|
|
|
@then("the second file should remain unchanged on disk")
|
|
def step_check_second_file_unchanged(context: Context) -> None:
|
|
target = getattr(context, "readonly_target", None)
|
|
if target:
|
|
target_path = Path(target)
|
|
if target_path.exists():
|
|
content = target_path.read_text()
|
|
assert content == "# original fail target content", (
|
|
f"Expected original content, got: {content}"
|
|
)
|
|
|
|
|
|
@then("the plan cannot be restarted from errored state")
|
|
def step_plan_cannot_restart_errored(context: Context) -> None:
|
|
plan_id = _lifecycle_plan_id(context)
|
|
error = None
|
|
try:
|
|
context.lifecycle_service.start_strategize(plan_id)
|
|
except (PlanError, PlanNotReadyError) as exc:
|
|
error = exc
|
|
assert error is not None, "Expected error when restarting errored plan"
|
|
|
|
|
|
@then('the plan phase should remain "{expected_phase}"')
|
|
def step_plan_phase_remains(context: Context, expected_phase: str) -> None:
|
|
assert context.plan.phase.value == expected_phase, (
|
|
f"Expected phase '{expected_phase}', got '{context.plan.phase.value}'"
|
|
)
|
|
|
|
|
|
@when("I try to execute the errored plan")
|
|
def step_try_execute_errored(context: Context) -> None:
|
|
context.errored_exec_error = None
|
|
try:
|
|
context.lifecycle_service.execute_plan(_lifecycle_plan_id(context))
|
|
except (PlanNotReadyError, Exception) as exc:
|
|
context.errored_exec_error = exc
|
|
|
|
|
|
@then("a plan not ready error should be raised for the errored plan")
|
|
def step_check_errored_exec_error(context: Context) -> None:
|
|
assert context.errored_exec_error is not None, (
|
|
"Expected PlanNotReadyError for errored plan"
|
|
)
|
|
assert isinstance(context.errored_exec_error, PlanNotReadyError)
|
|
|
|
|
|
@when("I try to complete strategize after failure")
|
|
def step_try_complete_strategize_after_failure(context: Context) -> None:
|
|
context.post_fail_complete_error = None
|
|
try:
|
|
context.lifecycle_service.complete_strategize(_lifecycle_plan_id(context))
|
|
except PlanError as exc:
|
|
context.post_fail_complete_error = exc
|
|
|
|
|
|
@then("a plan error should be raised because plan is not processing")
|
|
def step_check_post_fail_complete_error(context: Context) -> None:
|
|
assert context.post_fail_complete_error is not None, (
|
|
"Expected PlanError when completing after failure"
|
|
)
|
|
assert isinstance(context.post_fail_complete_error, PlanError)
|