cleveragents-core/features/steps/sandbox_merge_strategies_steps.py

"""Step definitions for sandbox merge strategy tests."""

import json
import os
import shutil
import tempfile
from unittest.mock import patch

from behave import given, then, when
from behave.runner import Context

from cleveragents.infrastructure.sandbox.merge import (
    GitMergeStrategy,
    JsonMergeStrategy,
    SequentialMergeStrategy,
)


def _assert_git_available() -> None:
    """Fail fast with a clear message when git is not on PATH."""
    assert shutil.which("git") is not None, (
        "git is not available on PATH. "
        "The @git_merge scenarios require git to be installed. "
        "Install git (e.g. apt-get install git) before running these tests."
    )


# ---------------------------------------------------------------------------
# Git three-way merge: setup
# ---------------------------------------------------------------------------


@given("a base document with three paragraphs")
def step_base_three_paragraphs(context: Context) -> None:
    """Create a base document with three paragraphs."""
    context.base = "line one\nline two\nline three\n"


@given("one side appends a paragraph at the end")
def step_ours_appends(context: Context) -> None:
    """Create the current-side edit that appends content."""
    context.ours = context.base + "line four from ours\n"


@given("the other side prepends a paragraph at the start")
def step_theirs_prepends(context: Context) -> None:
    """Create the incoming-side edit that prepends content."""
    context.theirs = "line zero from theirs\n" + context.base


@given("a base document with a single line")
def step_base_single_line(context: Context) -> None:
    """Create a base document with one line."""
    context.base = "original content\n"


@given('one side changes the line to "{text}"')
def step_one_side_changes(context: Context, text: str) -> None:
    """Set the current-side edit to a specific text."""
    if not hasattr(context, "ours") or context.ours is None:
        context.ours = text + "\n"
    else:
        context.theirs = text + "\n"


@given('the other side changes the line to "{text}"')
def step_other_side_changes(context: Context, text: str) -> None:
    """Set the incoming-side edit to a specific text."""
    context.theirs = text + "\n"


@given("all three merge inputs are empty")
def step_all_empty(context: Context) -> None:
    """Set all three merge inputs to empty strings."""
    context.base = ""
    context.ours = ""
    context.theirs = ""


@given("one side leaves the document unchanged")
def step_ours_unchanged(context: Context) -> None:
    """Keep the current side the same as the base."""
    context.ours = context.base


@given("the other side appends a new line")
def step_theirs_appends_line(context: Context) -> None:
    """Append a new line to the incoming side."""
    context.theirs = context.base + "appended line\n"


# ---------------------------------------------------------------------------
# Git three-way merge: actions
# ---------------------------------------------------------------------------


@when("the changes are merged using the git strategy")
def step_merge_git(context: Context) -> None:
    """Perform a merge using the GitMergeStrategy."""
    _assert_git_available()
    strategy = GitMergeStrategy()
    created_tmpdirs: list[str] = []
    real_mkdtemp = tempfile.mkdtemp

    def _tracked_mkdtemp(*args, **kwargs):
        path = real_mkdtemp(*args, **kwargs)
        created_tmpdirs.append(path)
        return path

    with patch(
        "cleveragents.infrastructure.sandbox.merge.tempfile.mkdtemp",
        side_effect=_tracked_mkdtemp,
    ):
        context.merge_result = strategy.merge(
            context.base, context.ours, context.theirs
        )
    context.ca_merge_tmpdirs = created_tmpdirs


@when("the changes are merged using the git strategy with git unavailable")
def step_merge_git_no_git(context: Context) -> None:
    """Perform a merge using GitMergeStrategy when git is not on PATH."""
    strategy = GitMergeStrategy()
    with patch(
        "cleveragents.infrastructure.sandbox.merge.subprocess.run",
        side_effect=FileNotFoundError("git not found"),
    ):
        context.merge_result = strategy.merge(
            context.base, context.ours, context.theirs
        )


# ---------------------------------------------------------------------------
# Git three-way merge: assertions
# ---------------------------------------------------------------------------


@then("the merge should succeed without conflicts")
def step_merge_success(context: Context) -> None:
    """Verify the merge succeeded with no conflicts."""
    assert context.merge_result.success is True, (
        f"Expected merge success, got success={context.merge_result.success}"
    )
    assert context.merge_result.has_conflicts is False, (
        f"Expected no conflicts, got has_conflicts={context.merge_result.has_conflicts}"
    )


@then("the merged document should contain content from both sides")
def step_merged_has_both(context: Context) -> None:
    """Verify the merged content includes contributions from both sides."""
    content = context.merge_result.content
    assert "line four from ours" in content, (
        "Merged content missing 'ours' contribution"
    )
    assert "line zero from theirs" in content, (
        "Merged content missing 'theirs' contribution"
    )


@then("the merge should report a conflict")
def step_merge_has_conflict(context: Context) -> None:
    """Verify the merge reported a conflict."""
    assert context.merge_result.has_conflicts is True, (
        "Expected conflicts but none were reported"
    )
    assert context.merge_result.success is False, (
        "Expected merge success=False when conflicts exist"
    )


@then("the merged output should contain conflict markers")
def step_merged_has_markers(context: Context) -> None:
    """Verify the merged output contains git conflict markers."""
    content = context.merge_result.content
    assert "<<<<<<<" in content, "Missing <<<<<<< conflict marker"
    assert ">>>>>>>" in content, "Missing >>>>>>> conflict marker"


@then("the conflict regions should be identified with line ranges")
def step_conflict_regions_identified(context: Context) -> None:
    """Verify conflict marker regions are reported as line ranges."""
    markers = context.merge_result.conflict_markers
    assert len(markers) > 0, "Expected at least one conflict region"
    for start, end in markers:
        assert isinstance(start, int) and isinstance(end, int), (
            f"Conflict region should be (int, int), got ({type(start)}, {type(end)})"
        )
        assert start <= end, f"Conflict region start ({start}) should be <= end ({end})"


@then("the merged content should be empty")
def step_merged_empty(context: Context) -> None:
    """Verify the merged content is empty."""
    assert context.merge_result.content == "", (
        f"Expected empty content, got {context.merge_result.content!r}"
    )


@then("no temporary merge files should remain on disk")
def step_no_temp_files(context: Context) -> None:
    """Verify the tmpdirs THIS scenario's merge created have been cleaned up.

    Checking the specific paths the merge created (rather than scanning the
    shared temp directory for any ``ca_merge_*`` entry) keeps the assertion
    robust under ``behave-parallel`` — a sibling scenario's in-flight tmpdir
    in the same shared ``/tmp`` is not our concern.
    """
    created = getattr(context, "ca_merge_tmpdirs", [])
    assert created, (
        "Expected the merge step to record at least one tracked tmpdir; got none."
    )
    leftover = [d for d in created if os.path.exists(d)]
    assert not leftover, f"Found leftover temp directories: {leftover}"


@then("the merged content should be the incoming side")
def step_merged_is_theirs(context: Context) -> None:
    """Verify the merged content is the incoming (theirs) side."""
    assert context.merge_result.content == context.theirs, (
        f"Expected theirs content, got {context.merge_result.content!r}"
    )


# ---------------------------------------------------------------------------
# Conflict marker detection
# ---------------------------------------------------------------------------


@given("a document with no conflict markers")
def step_doc_no_markers(context: Context) -> None:
    """Create a document with no conflict markers."""
    context.scan_content = "just some\nnormal text\nwithout markers\n"


@given("a document containing one conflict region")
def step_doc_one_conflict(context: Context) -> None:
    """Create a document with one conflict region."""
    context.scan_content = (
        "clean line\n"
        "<<<<<<< ours\n"
        "our version\n"
        "=======\n"
        "their version\n"
        ">>>>>>> theirs\n"
        "another clean line\n"
    )


@given("a document containing two conflict regions")
def step_doc_two_conflicts(context: Context) -> None:
    """Create a document with two conflict regions."""
    context.scan_content = (
        "<<<<<<< ours\n"
        "first ours\n"
        "=======\n"
        "first theirs\n"
        ">>>>>>> theirs\n"
        "middle line\n"
        "<<<<<<< ours\n"
        "second ours\n"
        "=======\n"
        "second theirs\n"
        ">>>>>>> theirs\n"
    )


@when("the document is scanned for conflict regions")
def step_scan_for_markers(context: Context) -> None:
    """Scan a document for conflict marker regions."""
    context.found_markers = GitMergeStrategy._find_conflict_markers(
        context.scan_content
    )


@then("no conflict regions should be found")
def step_no_markers_found(context: Context) -> None:
    """Verify no conflict regions were found."""
    assert len(context.found_markers) == 0, (
        f"Expected 0 conflict regions, found {len(context.found_markers)}"
    )


@then("exactly one conflict region should be found")
def step_one_marker_found(context: Context) -> None:
    """Verify exactly one conflict region was found."""
    assert len(context.found_markers) == 1, (
        f"Expected 1 conflict region, found {len(context.found_markers)}"
    )


@then("the conflict region should span the expected lines")
def step_marker_spans_expected(context: Context) -> None:
    """Verify the conflict region has valid start/end line numbers."""
    start, end = context.found_markers[0]
    assert start < end, f"Expected start ({start}) < end ({end})"
    # The <<<<<<< is on line 2 (1-indexed) and >>>>>>> is on line 6
    assert start == 2, f"Expected conflict to start at line 2, got {start}"
    assert end == 6, f"Expected conflict to end at line 6, got {end}"


@then("exactly two conflict regions should be found")
def step_two_markers_found(context: Context) -> None:
    """Verify exactly two conflict regions were found."""
    assert len(context.found_markers) == 2, (
        f"Expected 2 conflict regions, found {len(context.found_markers)}"
    )


# ---------------------------------------------------------------------------
# Sequential last-write-wins merge
# ---------------------------------------------------------------------------


@given("a base document, a current version, and an incoming version")
def step_sequential_setup(context: Context) -> None:
    """Set up three documents for sequential merge."""
    context.base = "base content"
    context.ours = "our modified content"
    context.theirs = "their incoming content"


@given("a base document, a current version, and an empty incoming version")
def step_sequential_empty_theirs(context: Context) -> None:
    """Set up documents where the incoming version is empty."""
    context.base = "base content"
    context.ours = "our modified content"
    context.theirs = ""


@when("the changes are merged using the sequential strategy")
def step_merge_sequential(context: Context) -> None:
    """Perform a merge using the SequentialMergeStrategy."""
    strategy = SequentialMergeStrategy()
    context.merge_result = strategy.merge(context.base, context.ours, context.theirs)


@then("the merged content should be the incoming version")
def step_merged_is_theirs_sequential(context: Context) -> None:
    """Verify the merged content equals the incoming version."""
    assert context.merge_result.content == context.theirs, (
        f"Expected {context.theirs!r}, got {context.merge_result.content!r}"
    )


# ---------------------------------------------------------------------------
# JSON deep merge: setup
# ---------------------------------------------------------------------------


@given('a current JSON document with key "{key}"')
def step_json_ours_key(context: Context, key: str) -> None:
    """Create a current JSON document with one key."""
    context.json_ours = json.dumps({key: "value_from_ours"})


@given('an incoming JSON document with key "{key}"')
def step_json_theirs_key(context: Context, key: str) -> None:
    """Create an incoming JSON document with one key."""
    context.json_theirs = json.dumps({key: "value_from_theirs"})


@given('a current JSON document with nested object under "{key}"')
def step_json_ours_nested(context: Context, key: str) -> None:
    """Create a current JSON document with a nested object."""
    context.json_ours = json.dumps({key: {"setting_a": 1, "setting_b": 2}})


@given('an incoming JSON document with additional nested keys under "{key}"')
def step_json_theirs_nested(context: Context, key: str) -> None:
    """Create an incoming JSON document with additional nested keys."""
    context.json_theirs = json.dumps({key: {"setting_b": 99, "setting_c": 3}})


@given('a current JSON document with "{key}" set to {value:d}')
def step_json_ours_scalar(context: Context, key: str, value: int) -> None:
    """Create a current JSON document with a scalar key."""
    context.json_ours = json.dumps({key: value})


@given('an incoming JSON document with "{key}" set to {value:d}')
def step_json_theirs_scalar(context: Context, key: str, value: int) -> None:
    """Create an incoming JSON document with a scalar key."""
    context.json_theirs = json.dumps({key: value})


@given('a current JSON document with an array under "{key}"')
def step_json_ours_array(context: Context, key: str) -> None:
    """Create a current JSON document with an array."""
    context.json_ours = json.dumps({key: [1, 2, 3]})


@given('an incoming JSON document with a different array under "{key}"')
def step_json_theirs_array(context: Context, key: str) -> None:
    """Create an incoming JSON document with a different array."""
    context.json_theirs = json.dumps({key: [4, 5, 6]})


@given("a current document containing invalid JSON")
def step_json_ours_invalid(context: Context) -> None:
    """Create a current document with invalid JSON."""
    context.json_ours = "{ not valid json !!!"


@given("an empty current JSON document")
def step_json_ours_empty(context: Context) -> None:
    """Create an empty current JSON document."""
    context.json_ours = ""


@given("an empty incoming JSON document")
def step_json_theirs_empty(context: Context) -> None:
    """Create an empty incoming JSON document."""
    context.json_theirs = ""


@given('a current JSON document with "{key}" as a string')
def step_json_ours_string_field(context: Context, key: str) -> None:
    """Create a current JSON document with a string field."""
    context.json_ours = json.dumps({key: "text_value"})


@given('an incoming JSON document with "{key}" as a number')
def step_json_theirs_number_field(context: Context, key: str) -> None:
    """Create an incoming JSON document with a numeric field."""
    context.json_theirs = json.dumps({key: 42})


# ---------------------------------------------------------------------------
# JSON deep merge: actions
# ---------------------------------------------------------------------------


@when("the documents are merged using the JSON strategy")
def step_merge_json_default(context: Context) -> None:
    """Perform a merge using the JsonMergeStrategy with default settings."""
    strategy = JsonMergeStrategy()
    context.merge_result = strategy.merge("", context.json_ours, context.json_theirs)


@when("the documents are merged using the JSON strategy in replace mode")
def step_merge_json_replace(context: Context) -> None:
    """Perform a merge using the JsonMergeStrategy in replace array mode."""
    strategy = JsonMergeStrategy(array_mode="replace")
    context.merge_result = strategy.merge("", context.json_ours, context.json_theirs)


@when("the documents are merged using the JSON strategy in concat mode")
def step_merge_json_concat(context: Context) -> None:
    """Perform a merge using the JsonMergeStrategy in concat array mode."""
    strategy = JsonMergeStrategy(array_mode="concat")
    context.merge_result = strategy.merge("", context.json_ours, context.json_theirs)


@when("the JSON strategy is configured with an invalid array mode")
def step_json_invalid_mode(context: Context) -> None:
    """Attempt to configure the JsonMergeStrategy with an invalid array mode."""
    context.config_error = None
    try:
        JsonMergeStrategy(array_mode="invalid_mode")
    except ValueError as exc:
        context.config_error = exc


# ---------------------------------------------------------------------------
# JSON deep merge: assertions
# ---------------------------------------------------------------------------


@then("the merged JSON should contain both keys")
def step_json_has_both_keys(context: Context) -> None:
    """Verify the merged JSON contains keys from both sides."""
    merged = json.loads(context.merge_result.content)
    assert "alpha" in merged, "Missing key 'alpha' from ours"
    assert "beta" in merged, "Missing key 'beta' from theirs"


@then('the merged JSON "{key}" should contain keys from both sides')
def step_json_nested_has_both(context: Context, key: str) -> None:
    """Verify a nested merged object contains keys from both sides."""
    merged = json.loads(context.merge_result.content)
    nested = merged[key]
    assert "setting_a" in nested, "Missing 'setting_a' from ours"
    assert "setting_b" in nested, "Missing 'setting_b' (should be from theirs)"
    assert "setting_c" in nested, "Missing 'setting_c' from theirs"
    # theirs wins for shared keys
    assert nested["setting_b"] == 99, (
        f"Expected setting_b=99 (theirs wins), got {nested['setting_b']}"
    )


@then('the merged JSON "{key}" should be the incoming value')
def step_json_scalar_theirs(context: Context, key: str) -> None:
    """Verify a scalar value in the merged JSON is the incoming value."""
    merged = json.loads(context.merge_result.content)
    assert merged[key] == 2, f"Expected {key}=2, got {merged[key]}"


@then('the merged JSON "{key}" should be the incoming array')
def step_json_array_replaced(context: Context, key: str) -> None:
    """Verify the array was replaced by the incoming side."""
    merged = json.loads(context.merge_result.content)
    assert merged[key] == [4, 5, 6], f"Expected [4, 5, 6], got {merged[key]}"


@then('the merged JSON "{key}" should contain elements from both arrays')
def step_json_array_concat(context: Context, key: str) -> None:
    """Verify the array contains elements from both sides."""
    merged = json.loads(context.merge_result.content)
    assert merged[key] == [1, 2, 3, 4, 5, 6], (
        f"Expected [1, 2, 3, 4, 5, 6], got {merged[key]}"
    )


@then("the merge should fail")
def step_merge_failed(context: Context) -> None:
    """Verify the merge failed."""
    assert context.merge_result.success is False, "Expected merge to fail"


@then("the failed merge content should fall back to the incoming text")
def step_failed_merge_fallback(context: Context) -> None:
    """Verify the failed merge content falls back to the incoming text."""
    assert context.merge_result.content == context.json_theirs, (
        f"Expected fallback to theirs, got {context.merge_result.content!r}"
    )


@then('the merged JSON should contain key "{key}"')
def step_json_has_key(context: Context, key: str) -> None:
    """Verify the merged JSON contains a specific key."""
    merged = json.loads(context.merge_result.content)
    assert key in merged, f"Missing key '{key}' in merged JSON"


@then("a configuration error should be raised")
def step_config_error_raised(context: Context) -> None:
    """Verify a configuration error was raised."""
    assert context.config_error is not None, "Expected a ValueError to be raised"
    assert isinstance(context.config_error, ValueError), (
        f"Expected ValueError, got {type(context.config_error)}"
    )


@then('the merged JSON "{key}" should be the incoming number value')
def step_json_type_mismatch_theirs(context: Context, key: str) -> None:
    """Verify a type-mismatched field uses the incoming value."""
    merged = json.loads(context.merge_result.content)
    assert merged[key] == 42, f"Expected {key}=42, got {merged[key]}"