"""Step definitions for plan_lifecycle_service_coverage_boost.feature. Targets uncovered lines and branches in PlanLifecycleService: - Line 284: get_action / get_action_by_name case-insensitive lookup - Line 420: use_action PlanInvariant append from action.invariants - Line 643: execute_plan InvalidPhaseTransitionError - Lines 829-846: constrain_apply method (happy + wrong-phase) - Line 959: auto_progress final return (no condition matched) """ from behave import given, then, when from behave.runner import Context from cleveragents.application.services.plan_lifecycle_service import ( InvalidPhaseTransitionError, PlanLifecycleService, ) from cleveragents.config.settings import Settings from cleveragents.core.exceptions import PlanError from cleveragents.domain.models.core.plan import ( InvariantSource, PlanPhase, ProcessingState, ProjectLink, ) # ----------------------------------------------------------------- # Background # ----------------------------------------------------------------- @given("I have a fresh plan lifecycle service for coverage boost") def step_create_fresh_service(context: Context) -> None: """Create a clean PlanLifecycleService instance.""" Settings._instance = None settings = Settings() context.service = PlanLifecycleService(settings=settings) context.error = None # ----------------------------------------------------------------- # Helpers (not steps) # ----------------------------------------------------------------- def _create_action(context: Context, name: str, **kwargs): """Helper to create a basic action.""" defaults = { "name": name, "description": f"Action {name}", "definition_of_done": "Tests pass", "strategy_actor": "openai/gpt-4", "execution_actor": "openai/gpt-4", } defaults.update(kwargs) return context.service.create_action(**defaults) def _create_plan_in_phase(context: Context, target_phase: PlanPhase): """Create a plan and advance it to the given phase. Returns the plan. Stores it on context.plan. """ action = _create_action(context, f"local/cov-{id(context)}") plan = context.service.use_action( action_name=str(action.namespaced_name), project_links=[ProjectLink(project_name="proj-1")], ) pid = plan.identity.plan_id if target_phase in (PlanPhase.EXECUTE, PlanPhase.APPLY): context.service.start_strategize(pid) context.service.complete_strategize(pid) context.service.execute_plan(pid) if target_phase == PlanPhase.APPLY: context.service.start_execute(pid) context.service.complete_execute(pid) context.service.apply_plan(pid) context.plan = context.service.get_plan(pid) return context.plan # ================================================================= # Scenario: get_action_by_name case-insensitive lookup (line 284) # ================================================================= @given("an action stored under its normalised lowercase key") def step_store_action_under_normalised_key(context: Context) -> None: """Create an action normally — it is stored under its lowercase key.""" action = _create_action(context, "local/scan-target") context.expected_action = action @when("I look up the action with mixed-case input via get_action_by_name") def step_lookup_via_get_action_by_name(context: Context) -> None: """Call get_action_by_name with mixed-case input.""" context.found_action = context.service.get_action_by_name("Local/Scan-Target") @then("the action should be found via case-insensitive lookup") def step_verify_case_insensitive_found(context: Context) -> None: """Verify the action returned is the one we stored.""" assert context.found_action is not None assert context.found_action is context.expected_action assert str(context.found_action.namespaced_name) == "local/scan-target" # ================================================================= # Scenario: use_action copies action.invariants (line 420) # ================================================================= @given('an action with invariants "{inv1}" and "{inv2}"') def step_create_action_with_invariants(context: Context, inv1: str, inv2: str) -> None: """Create an action whose ``invariants`` list is non-empty.""" context.action = _create_action( context, "local/inv-action", invariants=[inv1, inv2], ) @when("I use that action to create a plan") def step_use_action_create_plan(context: Context) -> None: """Use the action to instantiate a plan.""" context.plan = context.service.use_action( action_name=str(context.action.namespaced_name), project_links=[ProjectLink(project_name="proj-inv")], ) @then("the plan should contain {count:d} action-sourced invariants") def step_check_action_sourced_invariant_count(context: Context, count: int) -> None: """Verify the plan has the expected number of ACTION-sourced invariants.""" action_invariants = [ inv for inv in context.plan.invariants if inv.source == InvariantSource.ACTION ] assert len(action_invariants) == count, ( f"Expected {count} ACTION-sourced invariants, got {len(action_invariants)}" ) @then('the invariant texts should include "{text}"') def step_check_invariant_text_present(context: Context, text: str) -> None: """Verify a specific invariant text is present.""" texts = [inv.text for inv in context.plan.invariants] assert text in texts, f"Expected '{text}' in invariant texts: {texts}" # ================================================================= # Scenario: execute_plan InvalidPhaseTransitionError (line 643) # ================================================================= @given("a plan that is already in execute phase") def step_plan_already_in_execute(context: Context) -> None: """Create a plan already in Execute/QUEUED phase.""" _create_plan_in_phase(context, PlanPhase.EXECUTE) @given("a plan that is already in apply phase") def step_plan_already_in_apply(context: Context) -> None: """Create a plan already in Apply/QUEUED phase.""" _create_plan_in_phase(context, PlanPhase.APPLY) @when("I try to execute the plan from execute phase") def step_try_execute_from_execute(context: Context) -> None: """Attempt execute_plan when plan is already in Execute phase.""" context.error = None try: context.service.execute_plan(context.plan.identity.plan_id) except InvalidPhaseTransitionError as e: context.error = e @when("I try to execute the plan from apply phase") def step_try_execute_from_apply(context: Context) -> None: """Attempt execute_plan when plan is in Apply phase.""" context.error = None try: context.service.execute_plan(context.plan.identity.plan_id) except InvalidPhaseTransitionError as e: context.error = e @then("an InvalidPhaseTransitionError should be raised for the coverage boost") def step_verify_invalid_transition_error(context: Context) -> None: """Verify that an InvalidPhaseTransitionError was raised.""" assert context.error is not None, ( "Expected InvalidPhaseTransitionError but none raised" ) assert isinstance(context.error, InvalidPhaseTransitionError), ( f"Expected InvalidPhaseTransitionError, got {type(context.error).__name__}" ) # ================================================================= # Scenario: constrain_apply happy path (lines 829-846) # ================================================================= @given("a plan in apply phase with processing state for coverage boost") def step_plan_in_apply_processing(context: Context) -> None: """Create a plan in Apply/PROCESSING state.""" _create_plan_in_phase(context, PlanPhase.APPLY) pid = context.plan.identity.plan_id context.service.start_apply(pid) context.plan = context.service.get_plan(pid) @when('I constrain the apply with reason "{reason}" for coverage boost') def step_constrain_apply(context: Context, reason: str) -> None: """Call constrain_apply on the plan.""" context.plan = context.service.constrain_apply( context.plan.identity.plan_id, reason ) @then('the plan processing state should be "{expected}" for coverage boost') def step_check_processing_state_cb(context: Context, expected: str) -> None: """Verify the plan processing state.""" actual = context.plan.processing_state.value assert actual == expected, f"Expected state '{expected}', got '{actual}'" @then('the plan error message should be "{expected}" for coverage boost') def step_check_error_message_cb(context: Context, expected: str) -> None: """Verify the plan error_message field.""" assert context.plan.error_message == expected, ( f"Expected error message '{expected}', got '{context.plan.error_message}'" ) # ================================================================= # Scenario: constrain_apply wrong phase (line 831-832) # ================================================================= @given("a plan in strategize phase for coverage boost") def step_plan_in_strategize_for_cb(context: Context) -> None: """Create a plan that is still in Strategize phase.""" action = _create_action(context, "local/strat-only") context.plan = context.service.use_action( action_name=str(action.namespaced_name), project_links=[ProjectLink(project_name="proj-strat")], ) @when('I try to constrain the apply with reason "{reason}"') def step_try_constrain_apply(context: Context, reason: str) -> None: """Attempt constrain_apply expecting it to fail.""" context.error = None try: context.service.constrain_apply(context.plan.identity.plan_id, reason) except PlanError as e: context.error = e @then("a PlanError should be raised for coverage boost") def step_verify_plan_error_cb(context: Context) -> None: """Verify a PlanError was raised.""" assert context.error is not None, "Expected PlanError but none raised" assert isinstance(context.error, PlanError), ( f"Expected PlanError, got {type(context.error).__name__}" ) # ================================================================= # Scenario: auto_progress final return (line 959) # ================================================================= @given("a plan in strategize queued state with full automation for coverage boost") def step_plan_strategize_queued_full_auto(context: Context) -> None: """Create a plan in Strategize/QUEUED with FULL_AUTOMATION. should_auto_progress will return False because the plan is QUEUED (not COMPLETE), so auto_progress hits the early return at line 935. To hit line 959 instead, we need should_auto_progress to return True but neither if-block to match. However, looking at the logic: should_auto_progress returns True only when phase==STRATEGIZE+COMPLETE or phase==EXECUTE+COMPLETE. If should_auto_progress is True, one of the two if-blocks in auto_progress *will* match — unless the plan state changed between the check and the if-blocks. The simplest way to reach line 959 is for should_auto_progress to return False, which makes auto_progress return at line 935. But line 935 is already covered. To truly hit line 959, we need should_auto_progress to return True but the plan to NOT match either if-condition by the time we reach them. We can do this by monkey-patching should_auto_progress. """ action = _create_action(context, "local/auto-prog-test") context.plan = context.service.use_action( action_name=str(action.namespaced_name), project_links=[ProjectLink(project_name="proj-auto")], ) # Plan is now Strategize/QUEUED — should_auto_progress returns False, # so we monkey-patch it to return True to force execution past line 935 # into the if-blocks, which won't match (QUEUED != COMPLETE). context.service.should_auto_progress = lambda plan: True @when("I call auto_progress on the plan for coverage boost") def step_call_auto_progress(context: Context) -> None: """Call auto_progress.""" context.plan = context.service.auto_progress(context.plan.identity.plan_id) @then("the plan should be returned unchanged in strategize queued state") def step_verify_plan_unchanged(context: Context) -> None: """Verify the plan was returned without phase/state changes.""" assert context.plan.phase == PlanPhase.STRATEGIZE, ( f"Expected STRATEGIZE, got {context.plan.phase}" ) assert context.plan.processing_state == ProcessingState.QUEUED, ( f"Expected QUEUED, got {context.plan.processing_state}" )