420 lines
13 KiB
Python
420 lines
13 KiB
Python
"""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)
|