freemo
4ad51561fc
Cap time.sleep and asyncio.sleep globally at 10ms in before_all to eliminate retry/backoff waits (tenacity wait_fixed, retry_auto_debug exponential backoff). Save originals as time._original_sleep and asyncio._original_sleep for tests that need real wall-clock delays (CircuitBreaker recovery, debounce timers, validation timeouts). Enhance template-DB patch: add "db." prefix to _SCENARIO_DB_PREFIXES and check db_path.stat().st_size > 0 so 0-byte auto-created SQLite files receive the template copy instead of falling through to real migrations. Replace subprocess.run CLI invocations with typer.testing.CliRunner in module_coverage, main_coverage_complete, and coverage_extras step files, eliminating ~6s Python cold-start overhead per call. Switch plan_persistence and action_persistence to in-memory SQLite by default; only cross-restart scenarios use file-based via an explicit Given step. Replace MigrationRunner.run_migrations with Base.metadata.create_all in plan_service_steps for freshly created in-memory engines. Removed leftover debug comment in environment.py. Total tier runtime: 565s -> 21s (96% reduction), 20 features all under 5s behave-internal time. ISSUES CLOSED: #480
413 lines
14 KiB
Python
413 lines
14 KiB
Python
"""Step definitions for action persistence feature.
|
|
|
|
Tests ActionRepository CRUD, listing, archiving, argument/invariant
|
|
ordering, and terminal state storage for plans created from actions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.domain.models.core.action import (
|
|
Action,
|
|
ActionArgument,
|
|
ActionState,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
)
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
ActionRepository,
|
|
LifecyclePlanRepository,
|
|
)
|
|
|
|
_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
|
|
_ap_counter = 0
|
|
|
|
|
|
def _ap_ulid() -> str:
|
|
"""Generate a monotonically-increasing ULID for action persistence tests."""
|
|
global _ap_counter
|
|
_ap_counter += 1
|
|
n = _ap_counter
|
|
suffix = ""
|
|
for _ in range(8):
|
|
suffix = _CB32[n % 32] + suffix
|
|
n //= 32
|
|
return f"01HGZACT10QDYTR4AP{suffix}"
|
|
|
|
|
|
def _make_action(
|
|
name: str,
|
|
state: ActionState = ActionState.AVAILABLE,
|
|
arguments: list[ActionArgument] | None = None,
|
|
invariants: list[str] | None = None,
|
|
) -> Action:
|
|
"""Build an Action domain object for testing."""
|
|
return Action(
|
|
namespaced_name=NamespacedName.parse(name),
|
|
description="Persistence test action",
|
|
long_description="Detailed description",
|
|
definition_of_done="All tests pass",
|
|
strategy_actor="local/strategist",
|
|
execution_actor="local/executor",
|
|
review_actor="local/reviewer",
|
|
arguments=arguments or [],
|
|
invariants=invariants or [],
|
|
reusable=True,
|
|
read_only=False,
|
|
state=state,
|
|
created_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
updated_at=datetime(2026, 1, 1, tzinfo=UTC),
|
|
created_by="test-user",
|
|
tags=["test"],
|
|
)
|
|
|
|
|
|
def _ap_setup_db(context: Context) -> None:
|
|
"""Create an in-memory SQLite DB for action persistence tests."""
|
|
engine = create_engine("sqlite://", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
sm = sessionmaker(bind=engine)
|
|
session = sm()
|
|
context._ap_db_path = None
|
|
context._ap_engine = engine
|
|
context._ap_session = session
|
|
context._ap_session_factory = lambda: session
|
|
context._ap_action_repo = ActionRepository(
|
|
session_factory=context._ap_session_factory,
|
|
)
|
|
context._ap_plan_repo = LifecyclePlanRepository(
|
|
session_factory=context._ap_session_factory,
|
|
)
|
|
|
|
|
|
def _ap_teardown_db(context: Context) -> None:
|
|
"""Clean up session and engine."""
|
|
if hasattr(context, "_ap_session"):
|
|
context._ap_session.close()
|
|
if hasattr(context, "_ap_engine"):
|
|
context._ap_engine.dispose()
|
|
if getattr(context, "_ap_db_path", None):
|
|
Path(context._ap_db_path).unlink(missing_ok=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a fresh in-memory action persistence database")
|
|
def step_fresh_action_persistence_db(context: Context) -> None:
|
|
"""Set up a clean SQLite database for action persistence tests."""
|
|
_ap_setup_db(context)
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(lambda: _ap_teardown_db(context))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a new action with namespaced name "{name}"')
|
|
def step_new_action(context: Context, name: str) -> None:
|
|
"""Build a new Action domain object."""
|
|
context._ap_action = _make_action(name)
|
|
|
|
|
|
@given('a fully-populated action "{name}"')
|
|
def step_fully_populated_action(context: Context, name: str) -> None:
|
|
"""Build a fully-populated Action domain object."""
|
|
args = [
|
|
ActionArgument(
|
|
name="coverage",
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Target coverage percentage",
|
|
),
|
|
]
|
|
context._ap_action = _make_action(
|
|
name, arguments=args, invariants=["No regressions"]
|
|
)
|
|
|
|
|
|
@when("I persist the action via the action repository")
|
|
def step_persist_action(context: Context) -> None:
|
|
"""Persist the action through the repository."""
|
|
context._ap_action_repo.create(context._ap_action)
|
|
context._ap_session.commit()
|
|
|
|
|
|
@then('I can retrieve the action by name "{name}"')
|
|
def step_retrieve_action_by_name(context: Context, name: str) -> None:
|
|
"""Retrieve the action and store for further assertions."""
|
|
action = context._ap_action_repo.get_by_name(name)
|
|
assert action is not None, f"Action {name} not found"
|
|
context._ap_retrieved = action
|
|
|
|
|
|
@then('the retrieved action description should be "{desc}"')
|
|
def step_action_description(context: Context, desc: str) -> None:
|
|
"""Verify the action's description."""
|
|
assert context._ap_retrieved.description == desc
|
|
|
|
|
|
@then('the action strategy_actor should be "{actor}"')
|
|
def step_action_strategy_actor(context: Context, actor: str) -> None:
|
|
"""Verify the action's strategy_actor."""
|
|
assert context._ap_retrieved.strategy_actor == actor
|
|
|
|
|
|
@then('the action execution_actor should be "{actor}"')
|
|
def step_action_execution_actor(context: Context, actor: str) -> None:
|
|
"""Verify the action's execution_actor."""
|
|
assert context._ap_retrieved.execution_actor == actor
|
|
|
|
|
|
@then('the persisted action state should be "{state}"')
|
|
def step_action_state_value(context: Context, state: str) -> None:
|
|
"""Verify the persisted action's state."""
|
|
assert context._ap_retrieved.state == ActionState(state)
|
|
|
|
|
|
@then("the action reusable flag should be true")
|
|
def step_action_reusable_true(context: Context) -> None:
|
|
"""Verify the action's reusable flag."""
|
|
assert context._ap_retrieved.reusable is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List and archive scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("{count:d} persisted actions in available state")
|
|
def step_n_persisted_actions(context: Context, count: int) -> None:
|
|
"""Create N actions in available state."""
|
|
for i in range(count):
|
|
action = _make_action(f"local/bulk-action-{i}")
|
|
context._ap_action_repo.create(action)
|
|
context._ap_session.commit()
|
|
|
|
|
|
@given('a persisted action "{name}" in available state')
|
|
def step_persisted_action_available(context: Context, name: str) -> None:
|
|
"""Create and persist an available action."""
|
|
action = _make_action(name)
|
|
context._ap_action_repo.create(action)
|
|
context._ap_session.commit()
|
|
|
|
|
|
@when("I list available actions from the action repository")
|
|
def step_list_available_actions(context: Context) -> None:
|
|
"""List all available actions."""
|
|
context._ap_listed = context._ap_action_repo.list_available()
|
|
|
|
|
|
@when('I list actions in namespace "{namespace}" from the action repository')
|
|
def step_list_actions_namespace(context: Context, namespace: str) -> None:
|
|
"""List actions filtered by namespace."""
|
|
context._ap_listed = context._ap_action_repo.get_by_namespace(namespace)
|
|
|
|
|
|
@then("I should get {count:d} actions in the action list")
|
|
def step_action_list_count(context: Context, count: int) -> None:
|
|
"""Verify the number of actions returned."""
|
|
assert len(context._ap_listed) == count, (
|
|
f"Expected {count}, got {len(context._ap_listed)}"
|
|
)
|
|
|
|
|
|
@then("I should get {count:d} action in the action list")
|
|
def step_action_list_count_singular(context: Context, count: int) -> None:
|
|
"""Verify the number of actions returned (singular)."""
|
|
assert len(context._ap_listed) == count
|
|
|
|
|
|
@when('I archive the action "{name}" via the action repository')
|
|
def step_archive_action(context: Context, name: str) -> None:
|
|
"""Archive an action by updating its state."""
|
|
action = context._ap_action_repo.get_by_name(name)
|
|
assert action is not None
|
|
action.state = ActionState.ARCHIVED
|
|
action.updated_at = datetime.now(tz=UTC)
|
|
context._ap_action_repo.update(action)
|
|
context._ap_session.commit()
|
|
|
|
|
|
@then('the action "{name}" should have state "{state}"')
|
|
def step_action_has_state(context: Context, name: str, state: str) -> None:
|
|
"""Verify the action's state."""
|
|
action = context._ap_action_repo.get_by_name(name)
|
|
assert action is not None
|
|
assert action.state == ActionState(state)
|
|
|
|
|
|
@when('I delete the action "{name}" via the action repository')
|
|
def step_delete_action(context: Context, name: str) -> None:
|
|
"""Delete an action."""
|
|
result = context._ap_action_repo.delete(name)
|
|
assert result is True
|
|
context._ap_session.commit()
|
|
|
|
|
|
@then('the action "{name}" should not exist')
|
|
def step_action_not_exist(context: Context, name: str) -> None:
|
|
"""Verify the action no longer exists."""
|
|
action = context._ap_action_repo.get_by_name(name)
|
|
assert action is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Arguments ordering scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'an action "{name}" with arguments "{arg1}" at position 0 and "{arg2}" at position 1 and "{arg3}" at position 2'
|
|
)
|
|
def step_action_with_ordered_args(
|
|
context: Context, name: str, arg1: str, arg2: str, arg3: str
|
|
) -> None:
|
|
"""Create an action with ordered arguments."""
|
|
args = [
|
|
ActionArgument(
|
|
name=arg1,
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description=f"Arg {arg1}",
|
|
),
|
|
ActionArgument(
|
|
name=arg2,
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description=f"Arg {arg2}",
|
|
),
|
|
ActionArgument(
|
|
name=arg3,
|
|
arg_type=ArgumentType.BOOLEAN,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description=f"Arg {arg3}",
|
|
),
|
|
]
|
|
context._ap_action = _make_action(name, arguments=args)
|
|
|
|
|
|
@when('I retrieve the action "{name}" from the action repository')
|
|
def step_retrieve_action(context: Context, name: str) -> None:
|
|
"""Retrieve an action from the repo."""
|
|
context._ap_retrieved = context._ap_action_repo.get_by_name(name)
|
|
assert context._ap_retrieved is not None
|
|
|
|
|
|
@then("the persisted action should have {count:d} arguments")
|
|
def step_action_argument_count(context: Context, count: int) -> None:
|
|
"""Verify the persisted action argument count."""
|
|
assert len(context._ap_retrieved.arguments) == count
|
|
|
|
|
|
@then('the persisted argument at position {pos:d} should be "{name}"')
|
|
def step_argument_at_position(context: Context, pos: int, name: str) -> None:
|
|
"""Verify the argument at the given position."""
|
|
assert context._ap_retrieved.arguments[pos].name == name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Invariants ordering scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('an action "{name}" with invariants "{inv1}" and "{inv2}" and "{inv3}"')
|
|
def step_action_with_ordered_invariants(
|
|
context: Context, name: str, inv1: str, inv2: str, inv3: str
|
|
) -> None:
|
|
"""Create an action with ordered invariants."""
|
|
context._ap_action = _make_action(name, invariants=[inv1, inv2, inv3])
|
|
|
|
|
|
@then("the persisted action should have {count:d} invariants")
|
|
def step_action_invariant_count(context: Context, count: int) -> None:
|
|
"""Verify the invariant count."""
|
|
assert len(context._ap_retrieved.invariants) == count
|
|
|
|
|
|
@then('the persisted invariant at position {pos:d} should be "{text}"')
|
|
def step_invariant_at_position(context: Context, pos: int, text: str) -> None:
|
|
"""Verify the invariant text at the given position."""
|
|
assert context._ap_retrieved.invariants[pos] == text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action state persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a new action "{name}" in archived state')
|
|
def step_new_action_archived(context: Context, name: str) -> None:
|
|
"""Build a new action in archived state."""
|
|
context._ap_action = _make_action(name, state=ActionState.ARCHIVED)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Apply terminal state storage for plans from actions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a plan from action "{action_name}" in apply phase with state "{state}"')
|
|
def step_plan_from_action_terminal(
|
|
context: Context, action_name: str, state: str
|
|
) -> None:
|
|
"""Create a plan from the given action in apply phase with terminal state."""
|
|
plan_id = _ap_ulid()
|
|
now = datetime(2026, 3, 1, tzinfo=UTC)
|
|
plan = Plan(
|
|
identity=PlanIdentity(plan_id=plan_id, attempt=1),
|
|
namespaced_name=NamespacedName(namespace="local", name="terminal-plan"),
|
|
action_name=action_name,
|
|
description="Terminal state plan",
|
|
definition_of_done="Done",
|
|
phase=PlanPhase.APPLY,
|
|
processing_state=ProcessingState(state),
|
|
strategy_actor="local/s",
|
|
execution_actor="local/e",
|
|
timestamps=PlanTimestamps(created_at=now, updated_at=now),
|
|
created_by="test",
|
|
tags=[],
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
context._ap_plan_repo.create(plan)
|
|
context._ap_session.commit()
|
|
context._ap_terminal_plan_id = plan_id
|
|
|
|
|
|
@when("I retrieve the terminal plan from the persistence database")
|
|
def step_retrieve_terminal_plan(context: Context) -> None:
|
|
"""Retrieve the terminal plan."""
|
|
context._ap_terminal_plan = context._ap_plan_repo.get(context._ap_terminal_plan_id)
|
|
assert context._ap_terminal_plan is not None
|
|
|
|
|
|
@then('the terminal plan processing_state should be "{state}"')
|
|
def step_terminal_plan_state(context: Context, state: str) -> None:
|
|
"""Verify the terminal plan's processing state."""
|
|
assert context._ap_terminal_plan.processing_state == ProcessingState(state)
|