"""Step definitions for correction_flows.feature. Exercises CorrectionService revert/append flows, impact analysis, dry-run reports, error handling, and model validation. """ from __future__ import annotations from behave import given, then, when from pydantic import ValidationError as PydanticValidationError from cleveragents.application.services.correction_service import CorrectionService from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError from cleveragents.domain.models.core.correction import ( CorrectionImpact, CorrectionMode, ) # ------------------------------------------------------------------- # Helpers # ------------------------------------------------------------------- def _build_chain_tree(root: str, count: int) -> dict[str, list[str]]: """Build a linear chain tree with *count* total nodes starting at *root*.""" tree: dict[str, list[str]] = {} current = root for i in range(1, count): child = f"{root}_child{i}" tree[current] = [child] current = child return tree # ------------------------------------------------------------------- # Given steps # ------------------------------------------------------------------- @given("a correction flow service") def step_create_service(context): context.service = CorrectionService() context.correction_id = None context.decision_tree = None context.result = None context.report = None context.impact = None context.error = None context.corrections_list = None @given( 'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in revert mode' ) def step_create_revert_request(context, plan_id, decision_id): req = context.service.request_correction( plan_id=plan_id, target_decision_id=decision_id, mode=CorrectionMode.REVERT, ) context.correction_id = req.correction_id @given( 'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in append mode' ) def step_create_append_request(context, plan_id, decision_id): req = context.service.request_correction( plan_id=plan_id, target_decision_id=decision_id, mode=CorrectionMode.APPEND, ) context.correction_id = req.correction_id @given( 'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in append mode with guidance "{guidance}"' ) def step_create_append_with_guidance(context, plan_id, decision_id, guidance): req = context.service.request_correction( plan_id=plan_id, target_decision_id=decision_id, mode=CorrectionMode.APPEND, guidance=guidance, ) context.correction_id = req.correction_id @given('a decision tree where "{parent}" has no children') def step_tree_no_children(context, parent): context.decision_tree = {} @given('a simple decision tree where "{parent}" has children "{children}"') def step_tree_with_children(context, parent, children): tree = getattr(context, "decision_tree", None) or {} tree[parent] = [c.strip() for c in children.split(",")] context.decision_tree = tree @given( 'a multi-parent decision tree with "{p1}" children "{c1}" and "{p2}" children "{c2}"' ) def step_tree_two_parents(context, p1, c1, p2, c2): tree = getattr(context, "decision_tree", None) or {} tree[p1] = [c.strip() for c in c1.split(",")] tree[p2] = [c.strip() for c in c2.split(",")] context.decision_tree = tree @given('a decision tree with {count:d} nodes rooted at "{root}"') def step_tree_with_n_nodes(context, count, root): context.decision_tree = _build_chain_tree(root, count) @given("an empty decision tree") def step_empty_tree(context): context.decision_tree = {} @given( 'I execute the revert correction with a decision tree where "{parent}" has no children' ) def step_execute_revert_inline(context, parent): context.decision_tree = {} context.result = context.service.execute_revert( context.correction_id, context.decision_tree ) @given("I cancel the correction flow") def step_cancel_given(context): context.service.cancel_correction(context.correction_id) # ------------------------------------------------------------------- # When steps # ------------------------------------------------------------------- @when("I execute the revert correction") def step_execute_revert(context): context.result = context.service.execute_revert( context.correction_id, context.decision_tree ) @when( 'I execute the revert correction with a decision tree where "{parent}" has no children' ) def step_when_execute_revert_inline(context, parent): context.decision_tree = {} context.result = context.service.execute_revert( context.correction_id, context.decision_tree ) @when("I execute the append correction") def step_execute_append(context): context.result = context.service.execute_append(context.correction_id) @when("I analyze the impact") def step_analyze_impact(context): context.impact = context.service.analyze_impact( context.correction_id, context.decision_tree ) @when("I generate a dry-run report") def step_generate_dry_run(context): context.report = context.service.generate_dry_run_report( context.correction_id, context.decision_tree ) @when('I try to get correction flow "{cid}"') def step_try_get_correction(context, cid): try: context.service.get_correction(cid) context.error = None except ResourceNotFoundError as exc: context.error = exc @when("I try to execute the revert correction again") def step_try_execute_revert_again(context): try: context.service.execute_revert(context.correction_id, {}) context.error = None except ValidationError as exc: context.error = exc @when("I try to cancel the correction flow") def step_try_cancel(context): try: context.service.cancel_correction(context.correction_id) context.error = None except ValidationError as exc: context.error = exc @when("I cancel the correction flow") def step_cancel_correction(context): context.service.cancel_correction(context.correction_id) @when('I list correction flows for plan "{plan_id}"') def step_list_corrections(context, plan_id): context.corrections_list = context.service.list_corrections(plan_id=plan_id) @when("I try to create a correction with empty plan_id") def step_create_empty_plan_id(context): try: context.service.request_correction( plan_id="", target_decision_id="D1", mode=CorrectionMode.REVERT, ) context.error = None except ValidationError as exc: context.error = exc @when("I try to create a correction with empty target_decision_id") def step_create_empty_target(context): try: context.service.request_correction( plan_id="P1", target_decision_id="", mode=CorrectionMode.REVERT, ) context.error = None except ValidationError as exc: context.error = exc @when('I try to create a correction impact with risk level "{level}"') def step_create_bad_risk_level(context, level): try: CorrectionImpact(risk_level=level) context.error = None except PydanticValidationError as exc: context.error = exc @when("I execute correction via dispatch") def step_execute_dispatch(context): context.result = context.service.execute_correction( context.correction_id, context.decision_tree ) # ------------------------------------------------------------------- # Then steps # ------------------------------------------------------------------- @then('the correction flow result status should be "{status}"') def step_result_status(context, status): assert context.result is not None, "No result produced" assert context.result.status == status, ( f"Expected status '{status}', got '{context.result.status}'" ) @then('the reverted decisions should contain "{decision_id}"') def step_reverted_contains(context, decision_id): assert decision_id in context.result.reverted_decisions, ( f"'{decision_id}' not in reverted: {context.result.reverted_decisions}" ) @then('the archived artifacts should include "{artifact}"') def step_archived_artifact(context, artifact): assert artifact in context.result.archived_artifacts, ( f"'{artifact}' not in archived: {context.result.archived_artifacts}" ) @then('the correction flow status should be "{status}"') def step_correction_status(context, status): req = context.service.get_correction(context.correction_id) assert req.status == status, ( f"Expected correction status '{status}', got '{req.status}'" ) @then("the result should have a spawned child plan id") def step_has_spawned_child(context): assert context.result.spawned_child_plan_id is not None, ( "Expected spawned_child_plan_id, got None" ) @then('the correction flow guidance should be "{guidance}"') def step_guidance(context, guidance): req = context.service.get_correction(context.correction_id) assert req.guidance == guidance, ( f"Expected guidance '{guidance}', got '{req.guidance}'" ) @then("the result new decisions should not be empty") def step_new_decisions_not_empty(context): assert len(context.result.new_decisions) > 0, "new_decisions is empty" @then('the affected decisions should be "{expected}"') def step_affected_decisions(context, expected): expected_list = [d.strip() for d in expected.split(",")] assert context.impact.affected_decisions == expected_list, ( f"Expected {expected_list}, got {context.impact.affected_decisions}" ) @then('the risk level should be "{level}"') def step_risk_level(context, level): assert context.impact.risk_level == level, ( f"Expected risk '{level}', got '{context.impact.risk_level}'" ) @then('the report mode should be "{mode}"') def step_report_mode(context, mode): assert context.report.mode == mode, ( f"Expected report mode '{mode}', got '{context.report.mode}'" ) @then('the report decisions to invalidate should contain "{decision_id}"') def step_report_invalidate_contains(context, decision_id): assert decision_id in context.report.decisions_to_invalidate, ( f"'{decision_id}' not in invalidate list" ) @then("the report estimated recompute time should be greater than 0") def step_report_recompute_positive(context): assert context.report.estimated_recompute_time_seconds > 0 @then('the report warnings should contain "{fragment}"') def step_report_warning_fragment(context, fragment): joined = " ".join(context.report.warnings) assert fragment in joined, ( f"'{fragment}' not found in warnings: {context.report.warnings}" ) @then("the report decisions to invalidate should be empty") def step_report_invalidate_empty(context): assert len(context.report.decisions_to_invalidate) == 0, ( f"Expected empty invalidate list, got {context.report.decisions_to_invalidate}" ) @then("a correction flow resource not found error should be raised") def step_resource_not_found(context): assert isinstance(context.error, ResourceNotFoundError), ( f"Expected ResourceNotFoundError, got {type(context.error)}" ) @then("a validation error should be raised for status") def step_validation_error_status(context): assert isinstance(context.error, ValidationError), ( f"Expected ValidationError, got {type(context.error)}" ) @then("a validation error should be raised for empty field") def step_validation_error_empty(context): assert isinstance(context.error, ValidationError), ( f"Expected ValidationError, got {type(context.error)}" ) @then("a correction flow pydantic validation error should be raised") def step_pydantic_error(context): assert isinstance(context.error, PydanticValidationError), ( f"Expected PydanticValidationError, got {type(context.error)}" ) @then("the correction list should have {count:d} items") def step_list_count(context, count): assert len(context.corrections_list) == count, ( f"Expected {count} corrections, got {len(context.corrections_list)}" ) @then("the attempt list should have {count:d} entries") def step_attempt_count(context, count): attempts = context.service.list_attempts(context.correction_id) assert len(attempts) == count, f"Expected {count} attempts, got {len(attempts)}" @then("the first attempt should be successful") def step_first_attempt_success(context): attempts = context.service.list_attempts(context.correction_id) assert attempts[0].success is True, "First attempt was not successful" @then("the report should have a correction_id") def step_report_has_cid(context): assert context.report.correction_id is not None @then("the report should have an impact object") def step_report_has_impact(context): assert context.report.impact is not None @then("the report should have a warnings list") def step_report_has_warnings(context): assert isinstance(context.report.warnings, list)