fix(planconfig): preserve legacy merge conflict serialization

This commit is contained in:
2026-06-16 17:51:32 -04:00
committed by Forgejo
parent 77ba5091d1
commit d54028fa6d
3 changed files with 120 additions and 71 deletions
@@ -38,14 +38,14 @@ def step_set_ancestor_data(context):
context.ancestor[field_name] = value
@given("parent plan modifies field \"{field}\" to {value}")
@given('parent plan modifies field "{field}" to {value}')
def step_parent_modifies_field(context, field, value):
"""Set parent plan field value."""
value = _parse_value(value)
context.parent[field] = value
@given("subplan modifies field \"{field}\" to {value}")
@given('subplan modifies field "{field}" to {value}')
def step_subplan_modifies_field(context, field, value):
"""Set subplan field value."""
value = _parse_value(value)
@@ -64,27 +64,27 @@ def _parse_value(value):
return value
@given("parent plan does not modify field \"{field}\"")
@given('parent plan does not modify field "{field}"')
def step_parent_does_not_modify(context, field):
"""Ensure parent doesn't modify field (keep ancestor value)."""
if field in context.ancestor:
context.parent[field] = context.ancestor[field]
@given("subplan does not modify field \"{field}\"")
@given('subplan does not modify field "{field}"')
def step_subplan_does_not_modify(context, field):
"""Ensure subplan doesn't modify field (keep ancestor value)."""
if field in context.ancestor:
context.subplan[field] = context.ancestor[field]
@given("parent plan deletes field \"{field}\"")
@given('parent plan deletes field "{field}"')
def step_parent_deletes_field(context, field):
"""Mark field as deleted in parent (set to None)."""
context.parent[field] = None
@given("subplan deletes field \"{field}\"")
@given('subplan deletes field "{field}"')
def step_subplan_deletes_field(context, field):
"""Mark field as deleted in subplan (set to None)."""
context.subplan[field] = None
@@ -108,7 +108,7 @@ def step_empty_ancestor(context):
context.ancestor = {}
@given("parent plan has field \"{field}\" with value \"{value}\"")
@given('parent plan has field "{field}" with value "{value}"')
def step_parent_has_field(context, field, value):
"""Set parent field value."""
if value == "None":
@@ -118,7 +118,7 @@ def step_parent_has_field(context, field, value):
context.parent[field] = value
@given("subplan has field \"{field}\" with value \"{value}\"")
@given('subplan has field "{field}" with value "{value}"')
def step_subplan_has_field(context, field, value):
"""Set subplan field value."""
if value == "None":
@@ -128,7 +128,7 @@ def step_subplan_has_field(context, field, value):
context.subplan[field] = value
@given("ancestor plan has field \"{field}\" with value \"{value}\"")
@given('ancestor plan has field "{field}" with value "{value}"')
def step_ancestor_has_field(context, field, value):
"""Set ancestor field value."""
if value == "None":
@@ -152,7 +152,7 @@ def step_detect_conflicts(context):
)
@when("I resolve conflict for field \"{field}\" using \"{resolution}\"")
@when('I resolve conflict for field "{field}" using "{resolution}"')
def step_resolve_conflict(context, field, resolution):
"""Resolve a conflict."""
resolution_enum = ConflictResolution[resolution.upper()]
@@ -165,7 +165,7 @@ def step_auto_resolve(context):
context.report = context.detector.auto_resolve_conflicts(context.report)
@when("I try to add a conflict for field \"{field}\" twice")
@when('I try to add a conflict for field "{field}" twice')
def step_try_add_duplicate(context, field):
"""Try to add duplicate conflict."""
conflict = ConflictContext(
@@ -183,7 +183,7 @@ def step_try_add_duplicate(context, field):
context.error = str(e)
@when("I try to resolve conflict for field \"{field}\"")
@when('I try to resolve conflict for field "{field}"')
def step_try_resolve_nonexistent(context, field):
"""Try to resolve non-existent conflict."""
try:
@@ -193,7 +193,7 @@ def step_try_resolve_nonexistent(context, field):
context.error = str(e)
@when("I try to get resolved value for field \"{field}\"")
@when('I try to get resolved value for field "{field}"')
def step_try_get_unresolved_value(context, field):
"""Try to get value for unresolved conflict."""
try:
@@ -206,9 +206,9 @@ def step_try_get_unresolved_value(context, field):
@then("no plan merge conflicts should be detected")
def step_no_conflicts(context):
"""Assert no conflicts detected."""
assert (
context.report.conflict_count == 0
), f"Expected 0 conflicts, got {context.report.conflict_count}"
assert context.report.conflict_count == 0, (
f"Expected 0 conflicts, got {context.report.conflict_count}"
)
@then("{count} plan merge conflict should be detected")
@@ -216,19 +216,19 @@ def step_no_conflicts(context):
def step_conflict_count(context, count):
"""Assert conflict count."""
count = int(count)
assert (
context.report.conflict_count == count
), f"Expected {count} conflicts, got {context.report.conflict_count}"
assert context.report.conflict_count == count, (
f"Expected {count} conflicts, got {context.report.conflict_count}"
)
@then("the conflict should have type \"{conflict_type}\"")
@then('the conflict should have type "{conflict_type}"')
def step_conflict_type(context, conflict_type):
"""Assert conflict type."""
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
assert (
conflict.conflict_type.value == conflict_type
), f"Expected {conflict_type}, got {conflict.conflict_type.value}"
assert conflict.conflict_type.presentation_value == conflict_type, (
f"Expected {conflict_type}, got {conflict.conflict_type.presentation_value}"
)
@then("the conflict should have ancestor value {value}")
@@ -237,9 +237,9 @@ def step_conflict_ancestor_value(context, value):
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
value = _parse_value(value)
assert (
conflict.ancestor_value == value
), f"Expected {value}, got {conflict.ancestor_value}"
assert conflict.ancestor_value == value, (
f"Expected {value}, got {conflict.ancestor_value}"
)
@then("the conflict should have parent value {value}")
@@ -248,9 +248,9 @@ def step_conflict_parent_value(context, value):
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
value = _parse_value(value)
assert (
conflict.parent_value == value
), f"Expected {value}, got {conflict.parent_value}"
assert conflict.parent_value == value, (
f"Expected {value}, got {conflict.parent_value}"
)
@then("the conflict should have subplan value {value}")
@@ -259,39 +259,39 @@ def step_conflict_subplan_value(context, value):
assert len(context.report.conflicts) > 0, "No conflicts detected"
conflict = context.report.conflicts[0]
value = _parse_value(value)
assert (
conflict.subplan_value == value
), f"Expected {value}, got {conflict.subplan_value}"
assert conflict.subplan_value == value, (
f"Expected {value}, got {conflict.subplan_value}"
)
@then("conflict for field \"{field}\" should have type \"{conflict_type}\"")
@then('conflict for field "{field}" should have type "{conflict_type}"')
def step_field_conflict_type(context, field, conflict_type):
"""Assert conflict type for specific field."""
conflict = next(
(c for c in context.report.conflicts if c.field_path == field), None
)
assert conflict is not None, f"No conflict found for field '{field}'"
assert (
conflict.conflict_type.value == conflict_type
), f"Expected {conflict_type}, got {conflict.conflict_type.value}"
assert conflict.conflict_type.presentation_value == conflict_type, (
f"Expected {conflict_type}, got {conflict.conflict_type.presentation_value}"
)
@then("the conflict report should have plan_id set")
def step_report_has_plan_id(context):
"""Assert report has plan_id."""
assert context.report.plan_id is not None, "plan_id not set"
assert (
context.report.plan_id == "plan1"
), f"Expected 'plan1', got {context.report.plan_id}"
assert context.report.plan_id == "plan1", (
f"Expected 'plan1', got {context.report.plan_id}"
)
@then("the conflict report should have subplan_id set")
def step_report_has_subplan_id(context):
"""Assert report has subplan_id."""
assert context.report.subplan_id is not None, "subplan_id not set"
assert (
context.report.subplan_id == "subplan1"
), f"Expected 'subplan1', got {context.report.subplan_id}"
assert context.report.subplan_id == "subplan1", (
f"Expected 'subplan1', got {context.report.subplan_id}"
)
@then("the conflict report should have has_conflicts as true")
@@ -304,12 +304,12 @@ def step_report_has_conflicts(context):
def step_report_conflict_count(context, count):
"""Assert report conflict_count."""
count = int(count)
assert (
context.report.conflict_count == count
), f"Expected {count}, got {context.report.conflict_count}"
assert context.report.conflict_count == count, (
f"Expected {count}, got {context.report.conflict_count}"
)
@then("the resolved value for field \"{field}\" should be {value}")
@then('the resolved value for field "{field}" should be {value}')
def step_resolved_value(context, field, value):
"""Assert resolved value."""
value = _parse_value(value)
@@ -317,7 +317,7 @@ def step_resolved_value(context, field, value):
assert resolved == value, f"Expected {value}, got {resolved}"
@then("conflict for field \"{field}\" should be resolved with \"{resolution}\"")
@then('conflict for field "{field}" should be resolved with "{resolution}"')
def step_conflict_resolved_with(context, field, resolution):
"""Assert conflict is resolved with specific strategy."""
assert field in context.report.resolutions, f"Field '{field}' not resolved"
@@ -326,12 +326,12 @@ def step_conflict_resolved_with(context, field, resolution):
assert actual == expected, f"Expected {expected}, got {actual}"
@then("conflict for field \"{field}\" should not be resolved")
@then('conflict for field "{field}" should not be resolved')
def step_conflict_not_resolved(context, field):
"""Assert conflict is not resolved."""
assert (
field not in context.report.resolutions
), f"Field '{field}' should not be resolved"
assert field not in context.report.resolutions, (
f"Field '{field}' should not be resolved"
)
@then("unresolved conflicts count should be {count}")
@@ -350,10 +350,10 @@ def step_resolved_count(context, count):
assert actual == count, f"Expected {count} resolved, got {actual}"
@then("a merge conflict error should be raised with message containing \"{message}\"")
@then('a merge conflict error should be raised with message containing "{message}"')
def step_error_raised(context, message):
"""Assert error was raised with message."""
assert context.error is not None, "Expected error but none was raised"
assert (
message in context.error
), f"Expected message containing '{message}', got '{context.error}'"
assert message in context.error, (
f"Expected message containing '{message}', got '{context.error}'"
)
@@ -13,16 +13,21 @@ from cleveragents.domain.models.base import DomainBaseModel
class ConflictType(StrEnum):
"""Types of conflicts that can occur during a three-way merge."""
MODIFY_MODIFY = "modify_modify"
MODIFY_DELETE = "modify_delete"
DELETE_MODIFY = "delete_modify"
DELETE_DELETE = "delete_delete"
ADD_ADD = "add_add"
MODIFY_MODIFY = "MODIFY_MODIFY"
MODIFY_DELETE = "MODIFY_DELETE"
DELETE_MODIFY = "DELETE_MODIFY"
DELETE_DELETE = "DELETE_DELETE"
ADD_ADD = "ADD_ADD"
def __str__(self) -> str:
"""Render legacy summaries with the enum member name."""
return self.name
@property
def presentation_value(self) -> str:
"""Return the lowercase presentation value used by aggregate reports."""
return self.value.lower()
class ConflictResolution(StrEnum):
"""Resolution strategies for conflicts."""
@@ -120,9 +125,7 @@ class ConflictReport(DomainBaseModel):
)
self.conflicts.append(conflict)
def resolve_conflict(
self, field_path: str, resolution: ConflictResolution
) -> None:
def resolve_conflict(self, field_path: str, resolution: ConflictResolution) -> None:
"""Record a resolution strategy for a conflict."""
if not any(c.field_path == field_path for c in self.conflicts):
raise ValueError(f"No conflict found for field '{field_path}'")
@@ -133,9 +136,7 @@ class ConflictReport(DomainBaseModel):
if field_path not in self.resolutions:
raise ValueError(f"Field '{field_path}' is not resolved")
conflict = next(
(c for c in self.conflicts if c.field_path == field_path), None
)
conflict = next((c for c in self.conflicts if c.field_path == field_path), None)
if conflict is None:
raise ValueError(f"No conflict found for field '{field_path}'")
@@ -210,8 +211,9 @@ class ThreeWayMergeConflictDetector:
resolved = conflict.resolved_value
if resolved is not None:
merged[key] = resolved
auto_resolved[key] = resolved
legacy_resolved = _legacy_stringify(resolved)
merged[key] = legacy_resolved
auto_resolved[key] = legacy_resolved
elif conflict.conflict_type == ConflictType.DELETE_DELETE:
continue
else:
@@ -435,14 +437,20 @@ class ThreeWayMergeConflictDetector:
conflict_type=conflict_type,
context=ConflictContext(
field_path=key,
ancestor_value=ancestor_val,
parent_value=parent_val,
subplan_value=subplan_val,
ancestor_value=_legacy_stringify(ancestor_val),
parent_value=_legacy_stringify(parent_val),
subplan_value=_legacy_stringify(subplan_val),
conflict_type=conflict_type,
),
)
def _legacy_stringify(value: Any) -> str | None:
if value is None:
return None
return str(value)
def _format_value(value: Any) -> str:
if value is None:
return "(absent)"
@@ -0,0 +1,41 @@
from cleveragents.domain.models.planconfig import ThreeWayMergeConflictDetector
def test_detect_serializes_conflict_type_with_legacy_uppercase_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 200},
subplan={"timeout": 300},
).detect()
dumped = result.model_dump(mode="json")
assert dumped["conflicts"][0]["conflict_type"] == "MODIFY_MODIFY"
assert dumped["conflicts"][0]["context"]["conflict_type"] == "MODIFY_MODIFY"
def test_detect_auto_resolved_timeout_uses_legacy_string_value() -> None:
result = ThreeWayMergeConflictDetector(
ancestor={"timeout": 100},
parent={"timeout": 500},
subplan={},
).detect()
assert result.auto_resolved["timeout"] == "500"
assert result.model_dump(mode="json")["auto_resolved"]["timeout"] == "500"
def test_detect_conflicts_keeps_native_aggregate_values() -> None:
report = ThreeWayMergeConflictDetector.detect_conflicts(
plan_id="plan1",
subplan_id="subplan1",
ancestor={"timeout": 100},
parent={"timeout": 500},
subplan={"timeout": 600},
)
conflict = report.conflicts[0]
assert conflict.ancestor_value == 100
assert conflict.parent_value == 500
assert conflict.subplan_value == 600
assert conflict.conflict_type.presentation_value == "modify_modify"