feat(plans): implement configurable merge strategy for three-way merge
CI / lint (pull_request) Failing after 18s
CI / push-validation (pull_request) Successful in 24s
CI / helm (pull_request) Successful in 24s
CI / quality (pull_request) Successful in 33s
CI / build (pull_request) Successful in 42s
CI / typecheck (pull_request) Failing after 51s
CI / security (pull_request) Failing after 1m16s
CI / coverage (pull_request) Has been skipped
CI / e2e_tests (pull_request) Successful in 4m55s
CI / integration_tests (pull_request) Successful in 6m52s
CI / unit_tests (pull_request) Failing after 8m10s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 1s
CI / lint (pull_request) Failing after 18s
CI / push-validation (pull_request) Successful in 24s
CI / helm (pull_request) Successful in 24s
CI / quality (pull_request) Successful in 33s
CI / build (pull_request) Successful in 42s
CI / typecheck (pull_request) Failing after 51s
CI / security (pull_request) Failing after 1m16s
CI / coverage (pull_request) Has been skipped
CI / e2e_tests (pull_request) Successful in 4m55s
CI / integration_tests (pull_request) Successful in 6m52s
CI / unit_tests (pull_request) Failing after 8m10s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 1s
Implemented a configurable merge strategy framework for three-way merges. - New module: src/cleveragents/domain/models/core/merge_strategy.py - MergeStrategy enum with options: prefer-parent, prefer-subplan, manual - Validation utilities and helper to determine auto-resolution capability for a given strategy - New module: src/cleveragents/domain/models/core/merge_strategy_service.py - MergeConflict class to represent conflicts - MergeStrategyService to apply merge strategies - Support for resolving conflicts based on the configured strategy - New BDD feature: features/plan_merge_strategy.feature - Tests for all three merge strategies - Tests for conflict resolution - Tests for error handling - New step definitions: features/steps/plan_merge_strategy_steps.py - Step implementations for BDD tests ISSUES CLOSED: #9559
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
Feature: Plan Merge Strategy Configuration
|
||||
As a team lead
|
||||
I want to configure how plan conflicts are resolved
|
||||
So that my team can choose the appropriate conflict resolution behavior
|
||||
|
||||
Background:
|
||||
Given a merge strategy service is initialized
|
||||
|
||||
Scenario: Prefer-parent strategy auto-resolves conflicts in favor of parent
|
||||
Given the merge strategy is set to "prefer-parent"
|
||||
When a conflict occurs with parent value "parent_data" and subplan value "subplan_data"
|
||||
Then the conflict should be auto-resolved with value "parent_data"
|
||||
And the strategy should support auto-resolution
|
||||
|
||||
Scenario: Prefer-subplan strategy auto-resolves conflicts in favor of subplan
|
||||
Given the merge strategy is set to "prefer-subplan"
|
||||
When a conflict occurs with parent value "parent_data" and subplan value "subplan_data"
|
||||
Then the conflict should be auto-resolved with value "subplan_data"
|
||||
And the strategy should support auto-resolution
|
||||
|
||||
Scenario: Manual strategy surfaces conflicts for user resolution
|
||||
Given the merge strategy is set to "manual"
|
||||
When a conflict occurs with parent value "parent_data" and subplan value "subplan_data"
|
||||
Then the conflict should require manual resolution
|
||||
And the strategy should not support auto-resolution
|
||||
|
||||
Scenario: Multiple conflicts are resolved according to strategy
|
||||
Given the merge strategy is set to "prefer-parent"
|
||||
When multiple conflicts occur:
|
||||
| path | parent_value | subplan_value |
|
||||
| field1 | value1 | value2 |
|
||||
| field2 | value3 | value4 |
|
||||
Then all conflicts should be resolved with parent values
|
||||
|
||||
Scenario: Manual strategy raises error when attempting auto-resolution
|
||||
Given the merge strategy is set to "manual"
|
||||
When attempting to auto-resolve conflicts
|
||||
Then an error should be raised indicating manual resolution is required
|
||||
|
||||
Scenario: Merge strategy can be created from string value
|
||||
When a merge strategy is created from string "prefer-parent"
|
||||
Then the strategy should be "prefer-parent"
|
||||
And the strategy string representation should be "prefer-parent"
|
||||
|
||||
Scenario: Invalid merge strategy string raises error
|
||||
When attempting to create a merge strategy from invalid string "invalid-strategy"
|
||||
Then an error should be raised with message containing "Invalid merge strategy"
|
||||
|
||||
Scenario: No conflicts means no resolution needed
|
||||
Given the merge strategy is set to "prefer-parent"
|
||||
When there are no conflicts
|
||||
Then the resolution should return an empty result
|
||||
And no conflicts should be detected
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Step definitions for plan merge strategy feature tests."""
|
||||
|
||||
from behave import given, when, then
|
||||
from cleveragents.domain.models.core.merge_strategy import MergeStrategy
|
||||
from cleveragents.domain.models.core.merge_strategy_service import (
|
||||
MergeConflict,
|
||||
MergeStrategyService,
|
||||
)
|
||||
|
||||
|
||||
@given("a merge strategy service is initialized")
|
||||
def step_initialize_service(context):
|
||||
"""Initialize a merge strategy service."""
|
||||
context.service = MergeStrategyService()
|
||||
context.conflicts = []
|
||||
context.resolved = {}
|
||||
context.error = None
|
||||
|
||||
|
||||
@given('the merge strategy is set to "{strategy}"')
|
||||
def step_set_merge_strategy(context, strategy):
|
||||
"""Set the merge strategy."""
|
||||
context.service = MergeStrategyService(strategy=MergeStrategy.from_string(strategy))
|
||||
|
||||
|
||||
@when(
|
||||
'a conflict occurs with parent value "{parent_value}" and subplan value "{subplan_value}"'
|
||||
)
|
||||
def step_create_conflict(context, parent_value, subplan_value):
|
||||
"""Create a conflict."""
|
||||
conflict = MergeConflict(
|
||||
path="test_field", parent_value=parent_value, subplan_value=subplan_value
|
||||
)
|
||||
context.conflicts = [conflict]
|
||||
|
||||
|
||||
@then('the conflict should be auto-resolved with value "{expected_value}"')
|
||||
def step_verify_auto_resolution(context, expected_value):
|
||||
"""Verify that the conflict was auto-resolved with the expected value."""
|
||||
try:
|
||||
context.resolved = context.service.resolve_conflicts(context.conflicts)
|
||||
assert "test_field" in context.resolved
|
||||
assert context.resolved["test_field"] == expected_value
|
||||
except ValueError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then("the strategy should support auto-resolution")
|
||||
def step_verify_auto_resolve_support(context):
|
||||
"""Verify that the strategy supports auto-resolution."""
|
||||
assert context.service.can_auto_resolve()
|
||||
|
||||
|
||||
@then("the conflict should require manual resolution")
|
||||
def step_verify_manual_resolution(context):
|
||||
"""Verify that the conflict requires manual resolution."""
|
||||
try:
|
||||
context.service.resolve_conflicts(context.conflicts)
|
||||
assert False, "Expected ValueError for manual strategy"
|
||||
except ValueError as e:
|
||||
context.error = e
|
||||
assert "manual" in str(e).lower()
|
||||
|
||||
|
||||
@then("the strategy should not support auto-resolution")
|
||||
def step_verify_no_auto_resolve_support(context):
|
||||
"""Verify that the strategy does not support auto-resolution."""
|
||||
assert not context.service.can_auto_resolve()
|
||||
|
||||
|
||||
@when("multiple conflicts occur:")
|
||||
def step_create_multiple_conflicts(context):
|
||||
"""Create multiple conflicts from table."""
|
||||
context.conflicts = []
|
||||
for row in context.table:
|
||||
conflict = MergeConflict(
|
||||
path=row["path"],
|
||||
parent_value=row["parent_value"],
|
||||
subplan_value=row["subplan_value"],
|
||||
)
|
||||
context.conflicts.append(conflict)
|
||||
|
||||
|
||||
@then("all conflicts should be resolved with parent values")
|
||||
def step_verify_all_parent_resolution(context):
|
||||
"""Verify that all conflicts are resolved with parent values."""
|
||||
context.resolved = context.service.resolve_conflicts(context.conflicts)
|
||||
for conflict in context.conflicts:
|
||||
assert conflict.path in context.resolved
|
||||
assert context.resolved[conflict.path] == conflict.parent_value
|
||||
|
||||
|
||||
@when("attempting to auto-resolve conflicts")
|
||||
def step_attempt_auto_resolve(context):
|
||||
"""Attempt to auto-resolve conflicts."""
|
||||
try:
|
||||
context.service.resolve_conflicts(context.conflicts)
|
||||
context.error = None
|
||||
except ValueError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then("an error should be raised indicating manual resolution is required")
|
||||
def step_verify_manual_resolution_error(context):
|
||||
"""Verify that an error was raised for manual resolution."""
|
||||
assert context.error is not None
|
||||
assert "manual" in str(context.error).lower()
|
||||
|
||||
|
||||
@when('a merge strategy is created from string "{strategy_str}"')
|
||||
def step_create_strategy_from_string(context, strategy_str):
|
||||
"""Create a merge strategy from a string."""
|
||||
try:
|
||||
context.strategy = MergeStrategy.from_string(strategy_str)
|
||||
context.error = None
|
||||
except ValueError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then('the strategy should be "{expected_strategy}"')
|
||||
def step_verify_strategy_value(context, expected_strategy):
|
||||
"""Verify the strategy value."""
|
||||
assert context.strategy.value == expected_strategy
|
||||
|
||||
|
||||
@then('the strategy string representation should be "{expected_str}"')
|
||||
def step_verify_strategy_string(context, expected_str):
|
||||
"""Verify the strategy string representation."""
|
||||
assert str(context.strategy) == expected_str
|
||||
|
||||
|
||||
@when('attempting to create a merge strategy from invalid string "{invalid_str}"')
|
||||
def step_create_invalid_strategy(context, invalid_str):
|
||||
"""Attempt to create a strategy from an invalid string."""
|
||||
try:
|
||||
MergeStrategy.from_string(invalid_str)
|
||||
context.error = None
|
||||
except ValueError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@then('an error should be raised with message containing "{message_part}"')
|
||||
def step_verify_error_message(context, message_part):
|
||||
"""Verify that an error was raised with the expected message."""
|
||||
assert context.error is not None
|
||||
assert message_part in str(context.error)
|
||||
|
||||
|
||||
@when("there are no conflicts")
|
||||
def step_no_conflicts(context):
|
||||
"""Set up a scenario with no conflicts."""
|
||||
context.conflicts = []
|
||||
|
||||
|
||||
@then("the resolution should return an empty result")
|
||||
def step_verify_empty_resolution(context):
|
||||
"""Verify that resolution returns an empty result."""
|
||||
context.resolved = context.service.resolve_conflicts(context.conflicts)
|
||||
assert context.resolved == {}
|
||||
|
||||
|
||||
@then("no conflicts should be detected")
|
||||
def step_verify_no_conflicts_detected(context):
|
||||
"""Verify that no conflicts are detected."""
|
||||
assert not context.service.has_conflicts(context.conflicts)
|
||||
@@ -0,0 +1,70 @@
|
||||
"""Merge strategy configuration for plan three-way merges.
|
||||
|
||||
This module provides configurable merge strategies for resolving conflicts
|
||||
when merging parent plans with subplans.
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
from typing import Literal
|
||||
|
||||
|
||||
class MergeStrategy(str, Enum):
|
||||
"""Enumeration of available merge strategies for plan merging.
|
||||
|
||||
Attributes:
|
||||
PREFER_PARENT: Auto-resolve conflicts by choosing parent values
|
||||
PREFER_SUBPLAN: Auto-resolve conflicts by choosing subplan values
|
||||
MANUAL: Surface conflicts for user resolution
|
||||
"""
|
||||
|
||||
PREFER_PARENT: Literal["prefer-parent"] = "prefer-parent"
|
||||
"""Auto-resolve conflicts in favor of parent plan."""
|
||||
|
||||
PREFER_SUBPLAN: Literal["prefer-subplan"] = "prefer-subplan"
|
||||
"""Auto-resolve conflicts in favor of subplan."""
|
||||
|
||||
MANUAL: Literal["manual"] = "manual"
|
||||
"""Surface conflicts for manual user resolution."""
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""Return the string representation of the merge strategy."""
|
||||
return self.value
|
||||
|
||||
@classmethod
|
||||
def from_string(cls, value: str) -> "MergeStrategy":
|
||||
"""Create a MergeStrategy from a string value.
|
||||
|
||||
Args:
|
||||
value: The string representation of the merge strategy
|
||||
|
||||
Returns:
|
||||
The corresponding MergeStrategy enum value
|
||||
|
||||
Raises:
|
||||
ValueError: If the value is not a valid merge strategy
|
||||
"""
|
||||
try:
|
||||
return cls(value)
|
||||
except ValueError:
|
||||
valid_values = ", ".join(str(s) for s in cls)
|
||||
raise ValueError(
|
||||
f"Invalid merge strategy '{value}'. "
|
||||
f"Valid options are: {valid_values}"
|
||||
) from None
|
||||
|
||||
def is_auto_resolve(self) -> bool:
|
||||
"""Check if this strategy auto-resolves conflicts.
|
||||
|
||||
Returns:
|
||||
True if the strategy automatically resolves conflicts,
|
||||
False if conflicts need manual resolution
|
||||
"""
|
||||
return self in (MergeStrategy.PREFER_PARENT, MergeStrategy.PREFER_SUBPLAN)
|
||||
|
||||
def is_manual(self) -> bool:
|
||||
"""Check if this strategy requires manual conflict resolution.
|
||||
|
||||
Returns:
|
||||
True if conflicts need manual resolution, False otherwise
|
||||
"""
|
||||
return self == MergeStrategy.MANUAL
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Service for handling plan merge strategy operations.
|
||||
|
||||
This module provides the MergeStrategyService for applying merge strategies
|
||||
to resolve conflicts during plan merging.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from .merge_strategy import MergeStrategy
|
||||
|
||||
|
||||
class MergeConflict:
|
||||
"""Represents a conflict during plan merge.
|
||||
|
||||
Attributes:
|
||||
path: The path to the conflicting field
|
||||
parent_value: The value from the parent plan
|
||||
subplan_value: The value from the subplan
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, path: str, parent_value: Any, subplan_value: Any
|
||||
) -> None:
|
||||
"""Initialize a merge conflict.
|
||||
|
||||
Args:
|
||||
path: The path to the conflicting field
|
||||
parent_value: The value from the parent plan
|
||||
subplan_value: The value from the subplan
|
||||
"""
|
||||
self.path = path
|
||||
self.parent_value = parent_value
|
||||
self.subplan_value = subplan_value
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Return string representation of the conflict."""
|
||||
return (
|
||||
f"MergeConflict(path={self.path!r}, "
|
||||
f"parent_value={self.parent_value!r}, "
|
||||
f"subplan_value={self.subplan_value!r})"
|
||||
)
|
||||
|
||||
|
||||
class MergeStrategyService:
|
||||
"""Service for applying merge strategies to resolve conflicts.
|
||||
|
||||
This service handles the resolution of conflicts that occur during
|
||||
three-way merges of parent plans with subplans.
|
||||
"""
|
||||
|
||||
def __init__(self, strategy: MergeStrategy = MergeStrategy.MANUAL) -> None:
|
||||
"""Initialize the merge strategy service.
|
||||
|
||||
Args:
|
||||
strategy: The merge strategy to use (default: MANUAL)
|
||||
"""
|
||||
self.strategy = strategy
|
||||
|
||||
def resolve_conflicts(
|
||||
self, conflicts: list[MergeConflict]
|
||||
) -> Dict[str, Any]:
|
||||
"""Resolve conflicts using the configured merge strategy.
|
||||
|
||||
Args:
|
||||
conflicts: List of conflicts to resolve
|
||||
|
||||
Returns:
|
||||
Dictionary mapping conflict paths to resolved values
|
||||
|
||||
Raises:
|
||||
ValueError: If strategy is MANUAL and conflicts exist
|
||||
"""
|
||||
if not conflicts:
|
||||
return {}
|
||||
|
||||
if self.strategy.is_manual():
|
||||
raise ValueError(
|
||||
f"Cannot auto-resolve {len(conflicts)} conflicts with "
|
||||
f"MANUAL merge strategy. Conflicts must be resolved manually."
|
||||
)
|
||||
|
||||
resolved: Dict[str, Any] = {}
|
||||
for conflict in conflicts:
|
||||
if self.strategy == MergeStrategy.PREFER_PARENT:
|
||||
resolved[conflict.path] = conflict.parent_value
|
||||
elif self.strategy == MergeStrategy.PREFER_SUBPLAN:
|
||||
resolved[conflict.path] = conflict.subplan_value
|
||||
|
||||
return resolved
|
||||
|
||||
def has_conflicts(self, conflicts: list[MergeConflict]) -> bool:
|
||||
"""Check if there are any conflicts.
|
||||
|
||||
Args:
|
||||
conflicts: List of conflicts to check
|
||||
|
||||
Returns:
|
||||
True if there are conflicts, False otherwise
|
||||
"""
|
||||
return len(conflicts) > 0
|
||||
|
||||
def can_auto_resolve(self) -> bool:
|
||||
"""Check if the current strategy can auto-resolve conflicts.
|
||||
|
||||
Returns:
|
||||
True if the strategy can auto-resolve, False otherwise
|
||||
"""
|
||||
return self.strategy.is_auto_resolve()
|
||||
Reference in New Issue
Block a user