forked from HAL9000/cleveragents-core
freemo
dd3770f930
Remove three top-level fields from CorrectionDryRunReport that duplicated data already present in the embedded CorrectionImpact object: - excluded_decisions → now accessed via report.impact.excluded_decisions - rollback_tier_depth → now accessed via report.impact.rollback_tier_depth - child_plans_to_rollback → now accessed via report.impact.affected_child_plans The redundant fields created a divergence risk: both models are frozen=True Pydantic models, so post-construction mutation is impossible, but nothing prevented constructing an instance where the top-level copies disagreed with the impact sub-object. Approach: Option B (nest-only) — remove the duplicated top-level fields and update all consumers to use the impact object's fields instead. Closes #1087 Co-authored-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me> Co-committed-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
679 lines
25 KiB
Python
679 lines
25 KiB
Python
"""Step definitions for correction subtree isolation tests.
|
|
|
|
All step names are prefixed with ``subtree isolat`` to avoid
|
|
``AmbiguousStep`` conflicts with existing correction steps.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.application.services.correction_service import CorrectionService
|
|
from cleveragents.core.exceptions import ValidationError
|
|
from cleveragents.domain.models.core.correction import CorrectionMode, CorrectionStatus
|
|
|
|
_PLAN_ID = "01SUBTREEPLAN000000000000001"
|
|
|
|
|
|
def _parse_children(text: str) -> list[str]:
|
|
"""Split a comma-separated children string into a list."""
|
|
return [c.strip() for c in text.split(",") if c.strip()]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a subtree isolation correction service")
|
|
def step_subtree_service(context: Context) -> None:
|
|
"""Set up a fresh CorrectionService for subtree tests."""
|
|
context.subtree_svc = CorrectionService()
|
|
context.subtree_tree = {}
|
|
context.subtree_impact = None
|
|
context.subtree_report = None
|
|
context.subtree_tier = None
|
|
context.subtree_isolation_valid = None
|
|
|
|
|
|
@given('a subtree isolation decision tree with root "{root}" and children')
|
|
def step_subtree_tree(context: Context, root: str) -> None:
|
|
"""Build the decision tree from the table."""
|
|
tree: dict[str, list[str]] = {}
|
|
for row in context.table:
|
|
parent = row["parent"]
|
|
children = _parse_children(row["children"])
|
|
tree[parent] = children
|
|
context.subtree_tree = tree
|
|
|
|
|
|
@given("a subtree isolation deep tree with depth {depth:d}")
|
|
def step_subtree_deep_tree(context: Context, depth: int) -> None:
|
|
"""Build a linear chain tree with given depth."""
|
|
tree: dict[str, list[str]] = {}
|
|
for i in range(depth):
|
|
parent = f"depth_{i}" if i > 0 else "root"
|
|
child = f"depth_{i + 1}"
|
|
tree[parent] = [child]
|
|
context.subtree_tree = tree
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I subtree isolate target decision "{decision_id}" with mode "{mode}"')
|
|
def step_subtree_target(context: Context, decision_id: str, mode: str) -> None:
|
|
"""Create a correction request targeting the given decision."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
correction_mode = CorrectionMode(mode)
|
|
req = svc.request_correction(
|
|
plan_id=_PLAN_ID,
|
|
target_decision_id=decision_id,
|
|
mode=correction_mode,
|
|
)
|
|
context.subtree_correction_id = req.correction_id
|
|
|
|
|
|
@when("I subtree isolate analyze the impact")
|
|
def step_subtree_analyze(context: Context) -> None:
|
|
"""Analyze impact for the current correction."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
impact = svc.analyze_impact(
|
|
context.subtree_correction_id,
|
|
context.subtree_tree,
|
|
)
|
|
context.subtree_impact = impact
|
|
|
|
|
|
@when("I subtree isolate generate a dry-run report")
|
|
def step_subtree_dry_run(context: Context) -> None:
|
|
"""Generate a dry-run report for the current correction."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
report = svc.generate_dry_run_report(
|
|
context.subtree_correction_id,
|
|
context.subtree_tree,
|
|
)
|
|
context.subtree_report = report
|
|
context.subtree_impact = report.impact
|
|
|
|
|
|
@when('I subtree isolate compute rollback tier for "{decision_id}"')
|
|
def step_subtree_compute_tier(context: Context, decision_id: str) -> None:
|
|
"""Compute the rollback tier for a decision."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
context.subtree_tier = svc.compute_rollback_tier(
|
|
decision_id, _PLAN_ID, context.subtree_tree
|
|
)
|
|
|
|
|
|
@when('I subtree isolate validate isolation for target "{decision_id}"')
|
|
def step_subtree_validate(context: Context, decision_id: str) -> None:
|
|
"""Validate subtree isolation for a target decision."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
context.subtree_isolation_valid = svc.validate_subtree_isolation(
|
|
decision_id, context.subtree_tree
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps: affected / excluded decisions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the subtree isolate affected decisions should be "{expected}"')
|
|
def step_subtree_affected_exact(context: Context, expected: str) -> None:
|
|
"""Check that affected decisions match exactly."""
|
|
impact = context.subtree_impact
|
|
expected_list = _parse_children(expected)
|
|
assert impact.affected_decisions == expected_list, (
|
|
f"Expected affected={expected_list}, got {impact.affected_decisions}"
|
|
)
|
|
|
|
|
|
@then('the subtree isolate affected decisions should contain "{decision_id}"')
|
|
def step_subtree_affected_contains(context: Context, decision_id: str) -> None:
|
|
"""Check that a specific decision is in the affected set."""
|
|
impact = context.subtree_impact
|
|
assert decision_id in impact.affected_decisions, (
|
|
f"Expected '{decision_id}' in affected={impact.affected_decisions}"
|
|
)
|
|
|
|
|
|
@then('the subtree isolate excluded decisions should contain "{decision_id}"')
|
|
def step_subtree_excluded_contains(context: Context, decision_id: str) -> None:
|
|
"""Check that a specific decision is in the excluded set."""
|
|
impact = context.subtree_impact
|
|
assert decision_id in impact.excluded_decisions, (
|
|
f"Expected '{decision_id}' in excluded={impact.excluded_decisions}"
|
|
)
|
|
|
|
|
|
@then("the subtree isolate excluded decisions should be empty")
|
|
def step_subtree_excluded_empty(context: Context) -> None:
|
|
"""Check that excluded decisions is empty."""
|
|
impact = context.subtree_impact
|
|
assert len(impact.excluded_decisions) == 0, (
|
|
f"Expected empty excluded, got {impact.excluded_decisions}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps: rollback tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the subtree isolate rollback tier depth should be {expected:d}")
|
|
def step_subtree_tier_depth(context: Context, expected: int) -> None:
|
|
"""Check the rollback tier depth from impact analysis."""
|
|
impact = context.subtree_impact
|
|
assert impact.rollback_tier_depth == expected, (
|
|
f"Expected tier_depth={expected}, got {impact.rollback_tier_depth}"
|
|
)
|
|
|
|
|
|
@then("the subtree isolate rollback tier should be {expected:d}")
|
|
def step_subtree_tier_value(context: Context, expected: int) -> None:
|
|
"""Check the rollback tier from direct computation."""
|
|
assert context.subtree_tier == expected, (
|
|
f"Expected tier={expected}, got {context.subtree_tier}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps: dry-run report
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the subtree isolate dry-run warnings should contain "{text}"')
|
|
def step_subtree_warnings_contain(context: Context, text: str) -> None:
|
|
"""Check that dry-run warnings contain specific text."""
|
|
report = context.subtree_report
|
|
all_warnings = " ".join(report.warnings)
|
|
assert text in all_warnings, f"Expected '{text}' in warnings: {report.warnings}"
|
|
|
|
|
|
@then("the subtree isolate dry-run report should include excluded decisions")
|
|
def step_subtree_report_has_excluded(context: Context) -> None:
|
|
"""Check that the dry-run report includes excluded decisions via impact."""
|
|
report = context.subtree_report
|
|
assert len(report.impact.excluded_decisions) > 0, (
|
|
f"Expected non-empty excluded_decisions, got {report.impact.excluded_decisions}"
|
|
)
|
|
|
|
|
|
@then("the subtree isolate dry-run report rollback tier depth should be {expected:d}")
|
|
def step_subtree_report_tier(context: Context, expected: int) -> None:
|
|
"""Check the rollback tier depth in the dry-run report via impact."""
|
|
report = context.subtree_report
|
|
assert report.impact.rollback_tier_depth == expected, (
|
|
f"Expected report tier_depth={expected}, got {report.impact.rollback_tier_depth}"
|
|
)
|
|
|
|
|
|
@then(
|
|
'the subtree isolate dry-run report excluded decisions should contain "{decision_id}"'
|
|
)
|
|
def step_subtree_report_excluded_contains(context: Context, decision_id: str) -> None:
|
|
"""Check the dry-run report's excluded decisions list via impact."""
|
|
report = context.subtree_report
|
|
assert decision_id in report.impact.excluded_decisions, (
|
|
f"Expected '{decision_id}' in report excluded={report.impact.excluded_decisions}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps: subtree isolation validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the subtree isolate isolation should be valid")
|
|
def step_subtree_isolation_valid(context: Context) -> None:
|
|
"""Check that subtree isolation validation passed."""
|
|
assert context.subtree_isolation_valid is True, (
|
|
"Expected isolation to be valid, but it was not"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edge-case steps for additional coverage
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a subtree isolation empty tree")
|
|
def step_subtree_empty_tree(context: Context) -> None:
|
|
"""Set the decision tree to empty for edge-case testing."""
|
|
context.subtree_tree = {}
|
|
|
|
|
|
@when('I subtree isolate check parent of "{decision_id}"')
|
|
def step_subtree_check_parent(context: Context, decision_id: str) -> None:
|
|
"""Check the parent of a decision in the tree."""
|
|
parent = CorrectionService._find_parent(decision_id, context.subtree_tree)
|
|
context.subtree_parent_result = parent
|
|
|
|
|
|
@then("the subtree isolate parent should be none")
|
|
def step_subtree_parent_none(context: Context) -> None:
|
|
"""Verify that the parent lookup returned None."""
|
|
assert context.subtree_parent_result is None, (
|
|
f"Expected None, got {context.subtree_parent_result}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Influence DAG steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a subtree isolate influence edge from "{src}" to "{dst}"')
|
|
def step_subtree_influence_edge(context: Context, src: str, dst: str) -> None:
|
|
"""Set up a single influence edge."""
|
|
context.subtree_influence_edges = {src: [dst]}
|
|
|
|
|
|
@given('a subtree isolate influence cycle from "{src}" to "{mid}" back to "{back}"')
|
|
def step_subtree_influence_cycle(
|
|
context: Context,
|
|
src: str,
|
|
mid: str,
|
|
back: str,
|
|
) -> None:
|
|
"""Set up cyclic influence edges."""
|
|
context.subtree_influence_edges = {src: [mid], mid: [back]}
|
|
|
|
|
|
@when("I subtree isolate analyze the impact with influence edges")
|
|
def step_subtree_analyze_with_influence(context: Context) -> None:
|
|
"""Analyze impact using both tree and influence edges."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
impact = svc.analyze_impact(
|
|
context.subtree_correction_id,
|
|
context.subtree_tree,
|
|
influence_edges=context.subtree_influence_edges,
|
|
)
|
|
context.subtree_impact = impact
|
|
|
|
|
|
@when(
|
|
'I subtree isolate validate isolation for target "{decision_id}"'
|
|
" with influence edges"
|
|
)
|
|
def step_subtree_validate_with_influence(
|
|
context: Context,
|
|
decision_id: str,
|
|
) -> None:
|
|
"""Validate subtree isolation with influence edges."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
context.subtree_isolation_valid = svc.validate_subtree_isolation(
|
|
decision_id,
|
|
context.subtree_tree,
|
|
influence_edges=context.subtree_influence_edges,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Append mode steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I subtree isolate execute append correction")
|
|
def step_subtree_execute_append(context: Context) -> None:
|
|
"""Execute an append correction."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
result = svc.execute_append(context.subtree_correction_id)
|
|
context.subtree_append_result = result
|
|
|
|
|
|
@then("the subtree isolate append result should have a spawned child plan")
|
|
def step_subtree_append_has_child_plan(context: Context) -> None:
|
|
"""Check that the append result has a spawned child plan."""
|
|
result = context.subtree_append_result
|
|
assert result.spawned_child_plan_id is not None, (
|
|
"Expected spawned_child_plan_id, got None"
|
|
)
|
|
|
|
|
|
@then('the subtree isolate append result status should be "{status}"')
|
|
def step_subtree_append_status(context: Context, status: str) -> None:
|
|
"""Check the append result status."""
|
|
result = context.subtree_append_result
|
|
assert result.status.value == status, (
|
|
f"Expected status='{status}', got '{result.status.value}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dry-run enforcement steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I subtree isolate target decision "{decision_id}" with mode "{mode}" and dry_run'
|
|
)
|
|
def step_subtree_target_dry_run(
|
|
context: Context,
|
|
decision_id: str,
|
|
mode: str,
|
|
) -> None:
|
|
"""Create a dry-run correction request."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
correction_mode = CorrectionMode(mode)
|
|
req = svc.request_correction(
|
|
plan_id=_PLAN_ID,
|
|
target_decision_id=decision_id,
|
|
mode=correction_mode,
|
|
dry_run=True,
|
|
)
|
|
context.subtree_correction_id = req.correction_id
|
|
|
|
|
|
@then("subtree isolate executing the correction should raise a dry-run error")
|
|
def step_subtree_execute_raises_dry_run(context: Context) -> None:
|
|
"""Verify that executing a dry-run correction raises ValidationError."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
raised = False
|
|
try:
|
|
svc.execute_correction(context.subtree_correction_id, context.subtree_tree)
|
|
except ValidationError as exc:
|
|
raised = True
|
|
assert "dry-run" in str(exc).lower(), (
|
|
f"Expected 'dry-run' in error message, got: {exc}"
|
|
)
|
|
assert raised, "Expected ValidationError for dry-run execution, but none was raised"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Negative isolation validation steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a subtree isolation tree where child structurally reaches root")
|
|
def step_subtree_cyclic_tree(context: Context) -> None:
|
|
"""Build a tree where a child has the root as structural child."""
|
|
tree: dict[str, list[str]] = {}
|
|
for row in context.table:
|
|
parent = row["parent"]
|
|
children = _parse_children(row["children"])
|
|
tree[parent] = children
|
|
context.subtree_tree = tree
|
|
|
|
|
|
@then("the subtree isolate isolation should be invalid")
|
|
def step_subtree_isolation_invalid(context: Context) -> None:
|
|
"""Check that subtree isolation validation failed."""
|
|
assert context.subtree_isolation_valid is False, (
|
|
"Expected isolation to be invalid, but it was valid"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Execute revert (non-dry-run) steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I subtree isolate execute revert correction")
|
|
def step_subtree_execute_revert(context: Context) -> None:
|
|
"""Execute a revert correction."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
result = svc.execute_revert(
|
|
context.subtree_correction_id,
|
|
context.subtree_tree,
|
|
)
|
|
context.subtree_revert_result = result
|
|
|
|
|
|
@then('the subtree isolate revert result status should be "{status}"')
|
|
def step_subtree_revert_status(context: Context, status: str) -> None:
|
|
"""Check the revert result status."""
|
|
result = context.subtree_revert_result
|
|
assert result.status.value == status, (
|
|
f"Expected status='{status}', got '{result.status.value}'"
|
|
)
|
|
|
|
|
|
@then(
|
|
'the subtree isolate revert result should contain reverted decisions "{expected}"'
|
|
)
|
|
def step_subtree_revert_decisions(context: Context, expected: str) -> None:
|
|
"""Check the reverted decisions list."""
|
|
result = context.subtree_revert_result
|
|
expected_list = sorted(_parse_children(expected))
|
|
actual_list = sorted(result.reverted_decisions)
|
|
assert actual_list == expected_list, (
|
|
f"Expected reverted={expected_list}, got {actual_list}"
|
|
)
|
|
|
|
|
|
@then("the subtree isolate revert result should have archived artifacts")
|
|
def step_subtree_revert_has_artifacts(context: Context) -> None:
|
|
"""Check that archived artifacts is non-empty."""
|
|
result = context.subtree_revert_result
|
|
assert len(result.archived_artifacts) > 0, (
|
|
f"Expected non-empty archived_artifacts, got {result.archived_artifacts}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status guard enforcement steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("subtree isolate executing the correction again should raise a status error")
|
|
def step_subtree_execute_again_raises(context: Context) -> None:
|
|
"""Verify re-executing an applied correction raises ValidationError."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
raised = False
|
|
try:
|
|
svc.execute_correction(context.subtree_correction_id, context.subtree_tree)
|
|
except ValidationError:
|
|
raised = True
|
|
assert raised, "Expected ValidationError when re-executing an applied correction"
|
|
|
|
|
|
@when("I subtree isolate cancel the correction")
|
|
def step_subtree_cancel(context: Context) -> None:
|
|
"""Cancel the current correction."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
svc.cancel_correction(context.subtree_correction_id)
|
|
|
|
|
|
@then("subtree isolate executing the correction should raise a status error")
|
|
def step_subtree_execute_cancelled_raises(context: Context) -> None:
|
|
"""Verify executing a cancelled correction raises ValidationError."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
raised = False
|
|
try:
|
|
svc.execute_correction(context.subtree_correction_id, context.subtree_tree)
|
|
except ValidationError:
|
|
raised = True
|
|
assert raised, "Expected ValidationError when executing a cancelled correction"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mode mismatch enforcement steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("subtree isolate calling execute_revert should raise a mode error")
|
|
def step_subtree_revert_on_append_raises(context: Context) -> None:
|
|
"""Verify calling execute_revert on an append correction raises."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
raised = False
|
|
try:
|
|
svc.execute_revert(context.subtree_correction_id, context.subtree_tree)
|
|
except ValidationError as exc:
|
|
raised = True
|
|
assert "REVERT" in str(exc), f"Expected 'REVERT' in error, got: {exc}"
|
|
assert raised, "Expected ValidationError for mode mismatch"
|
|
|
|
|
|
@then("subtree isolate calling execute_append should raise a mode error")
|
|
def step_subtree_append_on_revert_raises(context: Context) -> None:
|
|
"""Verify calling execute_append on a revert correction raises."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
raised = False
|
|
try:
|
|
svc.execute_append(context.subtree_correction_id)
|
|
except ValidationError as exc:
|
|
raised = True
|
|
assert "APPEND" in str(exc), f"Expected 'APPEND' in error, got: {exc}"
|
|
assert raised, "Expected ValidationError for mode mismatch"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Single-node tree steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a subtree isolation single node tree")
|
|
def step_subtree_single_node_tree(context: Context) -> None:
|
|
"""Set a tree with only a root node and no children."""
|
|
context.subtree_tree = {"root": []}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Terminal state guard steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("subtree isolate re-analyzing the impact should raise a terminal state error")
|
|
def step_subtree_reanalyze_raises(context: Context) -> None:
|
|
"""Verify re-analyzing an applied correction raises ValidationError."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
raised = False
|
|
try:
|
|
svc.analyze_impact(context.subtree_correction_id, context.subtree_tree)
|
|
except ValidationError as exc:
|
|
raised = True
|
|
assert "terminal" in str(exc).lower(), (
|
|
f"Expected 'terminal' in error, got: {exc}"
|
|
)
|
|
assert raised, "Expected ValidationError when re-analyzing in terminal state"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exact-match affected count steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the subtree isolate affected decisions count should be {expected:d}")
|
|
def step_subtree_affected_count(context: Context, expected: int) -> None:
|
|
"""Check the exact count of affected decisions."""
|
|
impact = context.subtree_impact
|
|
assert len(impact.affected_decisions) == expected, (
|
|
f"Expected {expected} affected decisions, "
|
|
f"got {len(impact.affected_decisions)}: {impact.affected_decisions}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Convergent (diamond) DAG topology steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'a subtree isolate diamond influence where "{src1}" and "{src2}" both target "{dst}"'
|
|
)
|
|
def step_subtree_diamond_influence(
|
|
context: Context,
|
|
src1: str,
|
|
src2: str,
|
|
dst: str,
|
|
) -> None:
|
|
"""Set up a convergent (diamond) influence DAG."""
|
|
context.subtree_influence_edges = {src1: [dst], src2: [dst]}
|
|
|
|
|
|
@then(
|
|
'the subtree isolate affected decision "{decision_id}" should appear exactly once'
|
|
)
|
|
def step_subtree_affected_appears_once(context: Context, decision_id: str) -> None:
|
|
"""Verify a decision appears exactly once in the affected list."""
|
|
impact = context.subtree_impact
|
|
count = impact.affected_decisions.count(decision_id)
|
|
assert count == 1, (
|
|
f"Expected '{decision_id}' exactly once in affected, "
|
|
f"found {count} times: {impact.affected_decisions}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dry-run exception recovery steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I subtree isolate generate a dry-run report that fails internally")
|
|
def step_subtree_dry_run_fails(context: Context) -> None:
|
|
"""Generate a dry-run report where analyze_impact raises internally."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
try:
|
|
with patch.object(
|
|
CorrectionService,
|
|
"_compute_affected_subtree",
|
|
side_effect=RuntimeError("simulated BFS failure"),
|
|
):
|
|
svc.generate_dry_run_report(
|
|
context.subtree_correction_id,
|
|
context.subtree_tree,
|
|
)
|
|
except RuntimeError:
|
|
pass # Expected — the exception propagates from analyze_impact
|
|
|
|
|
|
@then('the subtree isolate correction status should be "{expected_status}"')
|
|
def step_subtree_correction_status(context: Context, expected_status: str) -> None:
|
|
"""Verify the correction request's current status."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
request = svc.get_correction(context.subtree_correction_id)
|
|
expected = CorrectionStatus(expected_status)
|
|
assert request.status == expected, (
|
|
f"Expected status='{expected}', got '{request.status}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Execute revert with influence edges steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I subtree isolate execute revert correction with influence edges")
|
|
def step_subtree_execute_revert_with_influence(context: Context) -> None:
|
|
"""Execute a revert correction with influence edges."""
|
|
svc: CorrectionService = context.subtree_svc
|
|
result = svc.execute_revert(
|
|
context.subtree_correction_id,
|
|
context.subtree_tree,
|
|
influence_edges=context.subtree_influence_edges,
|
|
)
|
|
context.subtree_revert_result = result
|
|
|
|
|
|
@then(
|
|
'the subtree isolate revert result should contain reverted decision "{decision_id}"'
|
|
)
|
|
def step_subtree_revert_contains_decision(
|
|
context: Context,
|
|
decision_id: str,
|
|
) -> None:
|
|
"""Check that a specific decision is in the reverted set."""
|
|
result = context.subtree_revert_result
|
|
assert decision_id in result.reverted_decisions, (
|
|
f"Expected '{decision_id}' in reverted={result.reverted_decisions}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DAG-only nodes in excluded set steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a subtree isolate influence edge from "{src}" to "{dst}" only in DAG')
|
|
def step_subtree_dag_only_influence(context: Context, src: str, dst: str) -> None:
|
|
"""Set up influence edges with nodes that are NOT in the structural tree."""
|
|
context.subtree_influence_edges = {src: [dst]}
|