"""Step definitions for LifecyclePlanRepository persistence coverage. Targets ``LifecyclePlanRepository`` in ``src/cleveragents/infrastructure/database/repositories.py``. """ from __future__ import annotations from datetime import datetime from typing import Any from behave import given, then, when from behave.runner import Context from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from cleveragents.core.exceptions import DatabaseError from cleveragents.domain.models.core.action import Action, ActionState from cleveragents.domain.models.core.plan import ( InvariantSource, NamespacedName, PlanIdentity, PlanInvariant, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) from cleveragents.domain.models.core.plan import Plan as V3Plan from cleveragents.infrastructure.database.models import ( Base, PlanArgumentModel, PlanInvariantModel, PlanProjectModel, ) from cleveragents.infrastructure.database.repositories import ( ActionRepository, DuplicatePlanError, LifecyclePlanRepository, ) # Crockford base32 alphabet for ULIDs _CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" _PLAN_REPO_ULID_COUNTER = 100 def _next_plan_repo_ulid() -> str: """Return a unique, valid ULID string for each call.""" global _PLAN_REPO_ULID_COUNTER _PLAN_REPO_ULID_COUNTER += 1 n = _PLAN_REPO_ULID_COUNTER suffix = "" for _ in range(8): suffix = _CB32[n % 32] + suffix n //= 32 return f"01HGZ6FE0AQDYTR4BX{suffix}" def _ensure_action(ctx: Context, action_name: str = "local/test-action") -> None: """Ensure the test action exists in the database.""" try: actions_created = ctx._plan_repo_actions_created # type: ignore[attr-defined] except (AttributeError, KeyError): actions_created: set[str] = set() ctx._plan_repo_actions_created = actions_created # type: ignore[attr-defined] if action_name in actions_created: return parts = action_name.split("/", 1) namespace = parts[0] if len(parts) == 2 else "local" short_name = parts[1] if len(parts) == 2 else parts[0] action = Action( namespaced_name=NamespacedName(namespace=namespace, name=short_name), description=f"Test action {short_name}", long_description=None, definition_of_done=f"Verify {short_name} completes", strategy_actor="local/strategist", execution_actor="local/executor", estimation_actor=None, review_actor=None, arguments=[], reusable=True, read_only=False, state=ActionState("available"), created_at=datetime.now(), updated_at=datetime.now(), created_by=None, tags=[], ) action_repo = ActionRepository(session_factory=ctx.db_session_factory) action_repo.create(action) ctx.db_session.commit() ctx._plan_repo_actions_created.add(action_name) def _make_plan( plan_id: str, phase: str = "action", processing_state: str = "queued", action_name: str = "local/test-action", description: str = "Test plan description", project_links: list[ProjectLink] | None = None, invariants: list[PlanInvariant] | None = None, arguments: dict[str, Any] | None = None, arguments_order: list[str] | None = None, ) -> V3Plan: """Create a minimal valid v3 Plan domain object.""" now = datetime.now() return V3Plan( identity=PlanIdentity(plan_id=plan_id, attempt=1), namespaced_name=NamespacedName(namespace="local", name="test-plan"), action_name=action_name, description=description, definition_of_done="All tests pass", phase=PlanPhase(phase), processing_state=ProcessingState(processing_state), strategy_actor="local/strategist", execution_actor="local/executor", project_links=project_links or [], invariants=invariants or [], arguments=arguments or {}, arguments_order=arguments_order or list((arguments or {}).keys()), timestamps=PlanTimestamps(created_at=now, updated_at=now), created_by="test-user", tags=[], reusable=True, read_only=False, ) # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("a clean in-memory database for plan repository tests") def step_clean_db_plan_repo(context: Context) -> None: """Create a fresh in-memory SQLite database with all tables.""" engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) context.db_engine = engine session = sessionmaker(bind=engine)() context.db_session = session context.db_session_factory = lambda: session @given("a lifecycle plan repository backed by the database") def step_plan_repo(context: Context) -> None: """Instantiate a LifecyclePlanRepository using the session factory.""" context.plan_repo = LifecyclePlanRepository( session_factory=context.db_session_factory, ) context.plan_error = None context.plan_result = None context.retrieved_plan = None # --------------------------------------------------------------------------- # Create plans # --------------------------------------------------------------------------- @given('a valid plan domain object with id "{plan_id}"') def step_make_plan(context: Context, plan_id: str) -> None: """Build a Plan domain object with the given ULID.""" _ensure_action(context) context.plan = _make_plan(plan_id=plan_id) @given("the plan is saved through the plan repository") def step_save_plan_given(context: Context) -> None: """Persist the plan (as a Given step).""" try: context.plan_repo.create(context.plan) context.db_session.commit() except Exception as exc: context.plan_error = exc @when("the plan is saved through the plan repository") def step_save_plan(context: Context) -> None: """Persist the plan and capture any errors.""" try: context.plan_repo.create(context.plan) context.db_session.commit() except Exception as exc: context.plan_error = exc @when('a duplicate plan with id "{plan_id}" is saved') def step_save_duplicate_plan(context: Context, plan_id: str) -> None: """Attempt to persist a second plan with the same ID.""" dup = _make_plan(plan_id=plan_id, description="Duplicate plan") try: context.plan_repo.create(dup) context.db_session.commit() except (DuplicatePlanError, DatabaseError) as exc: context.plan_error = exc except Exception as exc: cause = exc.__cause__ or exc context.plan_error = cause # --------------------------------------------------------------------------- # Retrieve plans # --------------------------------------------------------------------------- @then('the plan retrieved by id "{plan_id}" should exist') def step_get_plan_by_id_exists(context: Context, plan_id: str) -> None: context.retrieved_plan = context.plan_repo.get(plan_id) assert context.retrieved_plan is not None, f"Plan {plan_id} not found" @then('the plan retrieved by id "{plan_id}" should not exist') def step_get_plan_by_id_not_exists(context: Context, plan_id: str) -> None: result = context.plan_repo.get(plan_id) assert result is None, f"Plan {plan_id} should not exist but was found" @then('the retrieved plan description should be "{desc}"') def step_check_plan_description(context: Context, desc: str) -> None: assert context.retrieved_plan.description == desc, ( f"Expected '{desc}', got '{context.retrieved_plan.description}'" ) @then('the plan retrieved by name "{name}" should exist') def step_get_plan_by_name(context: Context, name: str) -> None: result = context.plan_repo.get_by_name(name) assert result is not None, f"Plan with name {name} not found" @when('the plan is retrieved by id "{plan_id}"') def step_when_get_plan_by_id(context: Context, plan_id: str) -> None: context.plan_result = context.plan_repo.get(plan_id) @then("the plan retrieve result should be None") def step_check_plan_none(context: Context) -> None: assert context.plan_result is None, "Expected None but got a plan" # --------------------------------------------------------------------------- # Update plans # --------------------------------------------------------------------------- @when('the plan phase is updated to "{phase}" with state "{state}"') def step_update_plan(context: Context, phase: str, state: str) -> None: plan = context.plan_repo.get(context.plan.identity.plan_id) plan.phase = PlanPhase(phase) plan.processing_state = ProcessingState(state) plan.timestamps.updated_at = datetime.now() context.plan_repo.update(plan) context.db_session.commit() context.retrieved_plan = context.plan_repo.get(context.plan.identity.plan_id) @then('the retrieved plan should have phase "{phase}"') def step_check_plan_phase(context: Context, phase: str) -> None: # If retrieved_plan is not set yet, fetch it from the repo if context.retrieved_plan is None: context.retrieved_plan = context.plan_repo.get(context.plan.identity.plan_id) actual = ( context.retrieved_plan.phase.value if hasattr(context.retrieved_plan.phase, "value") else context.retrieved_plan.phase ) assert actual == phase, f"Expected phase '{phase}', got '{actual}'" @then('the retrieved plan should have processing state "{state}"') def step_check_plan_state(context: Context, state: str) -> None: # If retrieved_plan is not set yet, fetch it from the repo if context.retrieved_plan is None: context.retrieved_plan = context.plan_repo.get(context.plan.identity.plan_id) actual = ( context.retrieved_plan.processing_state.value if hasattr(context.retrieved_plan.processing_state, "value") else context.retrieved_plan.processing_state ) assert actual == state, f"Expected state '{state}', got '{actual}'" # --------------------------------------------------------------------------- # List filters # --------------------------------------------------------------------------- @given('{n:d} plans exist in phase "{phase1}" and {m:d} plans in phase "{phase2}"') def step_create_plans_by_phase( context: Context, n: int, phase1: str, m: int, phase2: str ) -> None: _ensure_action(context) for _ in range(n): plan = _make_plan(plan_id=_next_plan_repo_ulid(), phase=phase1) context.plan_repo.create(plan) for _ in range(m): plan = _make_plan(plan_id=_next_plan_repo_ulid(), phase=phase2) context.plan_repo.create(plan) context.db_session.commit() @given('{n:d} plans exist in state "{state1}" and {m:d} plan in state "{state2}"') def step_create_plans_by_state( context: Context, n: int, state1: str, m: int, state2: str ) -> None: _ensure_action(context) for _ in range(n): plan = _make_plan(plan_id=_next_plan_repo_ulid(), processing_state=state1) context.plan_repo.create(plan) for _ in range(m): plan = _make_plan(plan_id=_next_plan_repo_ulid(), processing_state=state2) context.plan_repo.create(plan) context.db_session.commit() @given( '{n:d} plans referencing action "{action1}" and {m:d} plan referencing "{action2}"' ) def step_create_plans_by_action( context: Context, n: int, action1: str, m: int, action2: str ) -> None: _ensure_action(context, action1) _ensure_action(context, action2) for _ in range(n): plan = _make_plan(plan_id=_next_plan_repo_ulid(), action_name=action1) context.plan_repo.create(plan) for _ in range(m): plan = _make_plan(plan_id=_next_plan_repo_ulid(), action_name=action2) context.plan_repo.create(plan) context.db_session.commit() @given('a plan linked to project "{project}" and another with no project link') def step_create_plans_with_project(context: Context, project: str) -> None: _ensure_action(context) plan_with = _make_plan( plan_id=_next_plan_repo_ulid(), project_links=[ProjectLink(project_name=project)], ) context.plan_repo.create(plan_with) plan_without = _make_plan(plan_id=_next_plan_repo_ulid()) context.plan_repo.create(plan_without) context.db_session.commit() @when('plans are listed with phase filter "{phase}"') def step_list_by_phase(context: Context, phase: str) -> None: context.plan_list_result = context.plan_repo.list_plans(phase=phase) @when('plans are listed with processing state filter "{state}"') def step_list_by_state(context: Context, state: str) -> None: context.plan_list_result = context.plan_repo.list_plans(processing_state=state) @when('plans are listed with action name filter "{action}"') def step_list_by_action(context: Context, action: str) -> None: context.plan_list_result = context.plan_repo.list_plans(action_name=action) @when('plans are listed with project name filter "{project}"') def step_list_by_project(context: Context, project: str) -> None: context.plan_list_result = context.plan_repo.list_plans(project_name=project) @then("the plan list should contain {n:d} plans") def step_check_plan_list_count_plural(context: Context, n: int) -> None: actual = len(context.plan_list_result) assert actual == n, f"Expected {n} plans, got {actual}" @then("the plan list should contain {n:d} plan") def step_check_plan_list_count_singular(context: Context, n: int) -> None: actual = len(context.plan_list_result) assert actual == n, f"Expected {n} plan(s), got {actual}" # --------------------------------------------------------------------------- # Delete # --------------------------------------------------------------------------- @given('a valid plan domain object with id "{plan_id}" and child rows') def step_make_plan_with_children(context: Context, plan_id: str) -> None: _ensure_action(context) context.plan = _make_plan( plan_id=plan_id, project_links=[ProjectLink(project_name="local/child-proj")], invariants=[PlanInvariant(text="Must pass tests", source=InvariantSource.PLAN)], arguments={"key": "value"}, arguments_order=["key"], ) @when('the plan with id "{plan_id}" is deleted') def step_delete_plan(context: Context, plan_id: str) -> None: context.plan_repo.delete(plan_id) context.db_session.commit() @then('the child rows for plan "{plan_id}" should be gone') def step_check_children_gone(context: Context, plan_id: str) -> None: session: Session = context.db_session projects = session.query(PlanProjectModel).filter_by(plan_id=plan_id).count() arguments = session.query(PlanArgumentModel).filter_by(plan_id=plan_id).count() invariants = session.query(PlanInvariantModel).filter_by(plan_id=plan_id).count() assert projects == 0, f"Expected 0 project links, got {projects}" assert arguments == 0, f"Expected 0 arguments, got {arguments}" assert invariants == 0, f"Expected 0 invariants, got {invariants}" # --------------------------------------------------------------------------- # Error handling # --------------------------------------------------------------------------- @then("the plan repository should not raise an error") def step_no_plan_error(context: Context) -> None: assert context.plan_error is None, f"Unexpected error: {context.plan_error}" @then('a DuplicatePlanError should be raised mentioning "{plan_id}"') def step_check_duplicate_plan_error(context: Context, plan_id: str) -> None: assert context.plan_error is not None, "Expected DuplicatePlanError but no error" assert isinstance(context.plan_error, DuplicatePlanError), ( f"Expected DuplicatePlanError, got {type(context.plan_error).__name__}" ) assert plan_id in str(context.plan_error), ( f"Error message should mention '{plan_id}': {context.plan_error}" ) # --------------------------------------------------------------------------- # Phase and state persistence # --------------------------------------------------------------------------- @given( 'a plan domain object in phase "{phase}" with state "{state}" and id "{plan_id}"' ) def step_make_plan_phase_state( context: Context, phase: str, state: str, plan_id: str ) -> None: _ensure_action(context) context.plan = _make_plan(plan_id=plan_id, phase=phase, processing_state=state) # --------------------------------------------------------------------------- # Child row ordering # --------------------------------------------------------------------------- @given('a plan with ordered children and id "{plan_id}"') def step_make_plan_ordered_children(context: Context, plan_id: str) -> None: _ensure_action(context) context.plan = _make_plan( plan_id=plan_id, project_links=[ ProjectLink(project_name="local/proj-a"), ProjectLink(project_name="local/proj-b"), ], invariants=[ PlanInvariant(text="first", source=InvariantSource.PLAN), PlanInvariant(text="second", source=InvariantSource.ACTION), PlanInvariant(text="third", source=InvariantSource.PROJECT), ], arguments={"alpha": "a", "beta": "b", "gamma": "g"}, arguments_order=["alpha", "beta", "gamma"], ) @then('the retrieved plan should have arguments in order "{expected}"') def step_check_arguments_order(context: Context, expected: str) -> None: plan = context.plan_repo.get(context.plan.identity.plan_id) expected_list = expected.split(",") assert plan.arguments_order == expected_list, ( f"Expected arguments order {expected_list}, got {plan.arguments_order}" ) @then('the retrieved plan should have invariants in order "{expected}"') def step_check_invariants_order(context: Context, expected: str) -> None: plan = context.plan_repo.get(context.plan.identity.plan_id) expected_list = expected.split(",") actual = [inv.text for inv in plan.invariants] assert actual == expected_list, f"Expected invariants {expected_list}, got {actual}" @then('the retrieved plan should have project links including "{project}"') def step_check_project_links(context: Context, project: str) -> None: plan = context.plan_repo.get(context.plan.identity.plan_id) names = [pl.project_name for pl in plan.project_links] assert project in names, f"Expected project '{project}' in links, got {names}"