From 4dc0cd1453ff50cd57c4aae8125616b6ad9dffce Mon Sep 17 00:00:00 2001 From: Michael Panchenko Date: Wed, 10 Jun 2026 18:52:58 +0300 Subject: [PATCH] Add ReplaceInFilesTool --- CHANGELOG.md | 1 + src/serena/tools/file_tools.py | 242 ++++++++++++++++++++++++++++++- src/serena/util/text_utils.py | 157 ++++++++++++++++++++ test/serena/test_serena_agent.py | 52 +++++++ test/serena/test_text_utils.py | 64 +++++++- 5 files changed, 514 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f26ea0c..c3795e43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,7 @@ Status of the `main` branch. Changes prior to the next official version change w list of projects was modified (triggering a save of the configuration with transient overrides applied) * Tools: + - New tool: `replace_in_files` - Add tool parameter alias support, adding `name_path` as an alias for `name_path_pattern` in `find_symbol` tools - Allow `query_project` tool to access read-only tools that are not enabled in the current configuration - Make tool call errors surface explicitly as errors at the MCP protocol level diff --git a/src/serena/tools/file_tools.py b/src/serena/tools/file_tools.py index 2b7f1a9b..dd61903a 100644 --- a/src/serena/tools/file_tools.py +++ b/src/serena/tools/file_tools.py @@ -14,7 +14,14 @@ from typing import Literal from serena.tools import SUCCESS_RESULT, EditedFileContext, EditingToolWithDiagnostics, Tool, ToolMarkerOptional from serena.util.file_system import scan_directory -from serena.util.text_utils import ContentReplacer, search_files +from serena.util.text_utils import ( + ContentReplacer, + MultiFileContentReplacer, + ReplacementOccurrence, + expand_braces, + glob_match, + search_files, +) class ReadFileTool(Tool): @@ -225,6 +232,239 @@ class ReplaceContentTool(EditingToolWithDiagnostics): return diagnostics_context.format_result(SUCCESS_RESULT) +class ReplaceInFilesTool(EditingToolWithDiagnostics): + """ + Replaces occurrences of a pattern across multiple files, with dry-run preview and per-occurrence selection. + """ + + def apply( + self, + needle: str, + repl: str, + mode: Literal["literal", "regex"], + relative_path: str = "", + paths_include_glob: str = "", + paths_exclude_glob: str = "", + dry_run: bool = False, + occurrence_ids: list[str] | None = None, + expected_count: int = -1, + max_answer_chars: int = -1, + ) -> str: + r""" + Replaces occurrences of a pattern across multiple files in ONE call. + + This is the preferred tool for repeated small edits (renames, import swaps, annotation changes, + path prefixes) spanning several files or many places in one file: one call with a SHORT pattern + replaces many single-file replacements with long disambiguating needles. + + Recommended protocol whenever there is ANY risk of unintended replacements: + 1. Call with dry_run=True: every prospective change is returned as a minimal line diff with an + occurrence id; nothing is modified. + 2. Call again with dry_run=False, passing the ids you want in occurrence_ids (omit it to apply + all). You pick the desired replacements from the list - no counting, no needle-crafting. + + For clearly unambiguous bulk replacements you may skip the dry run; pass expected_count as a + guard. If the actual number of matches differs, NOTHING is changed and the diff list is + returned, so a failed guard costs one call and gives you the dry-run output to select from. + + :param needle: the string (mode "literal") or regular expression (mode "regex"; Python `re` + syntax with DOTALL and MULTILINE) to search for + :param repl: the replacement string. In regex mode, backreferences to matched groups can be + specified as $!1, $!2, etc. + :param mode: either "literal" or "regex", specifying how `needle` is to be interpreted + :param relative_path: only consider this file or directory (default: the whole project) + :param paths_include_glob: optional glob (relative to the project root, e.g. "src/**/*.java") + restricting which files are considered + :param paths_exclude_glob: optional glob of files to exclude; takes precedence over the include glob + :param dry_run: if True, do not modify anything; return the prospective changes as a list of + diffs with occurrence ids + :param occurrence_ids: optional list of occurrence ids (obtained from a dry run) to which the + replacement is restricted; if any id is unknown or stale, NOTHING is changed. If omitted, + all occurrences are replaced. + :param expected_count: optional guard for calls without occurrence_ids: the number of + occurrences you expect to be replaced. If the actual count differs, nothing is changed and + the list of prospective changes is returned. -1 disables the guard. + :param max_answer_chars: if the output exceeds this many characters, a shortened version is + returned. -1 uses the configured default. + :return: in a dry run, the prospective changes; otherwise a summary of the applied replacements + """ + replacer = MultiFileContentReplacer(mode=mode) + files = self._collect_files(relative_path, paths_include_glob, paths_exclude_glob) + occurrences = replacer.find_occurrences(files, needle, repl) + contents = dict(files) + + if dry_run: + return self._render_listing(replacer, occurrences, contents, max_answer_chars, dry_run=True) + + if occurrence_ids is not None: + selected, problems = self._resolve_occurrence_ids(occurrence_ids, occurrences) + if problems: + problem_lines = "\n".join(f" {p}" for p in problems) + raise ValueError( + f"{len(problems)} of the given occurrence_ids could not be resolved - NO changes were applied:\n" + f"{problem_lines}\n" + "Re-run with dry_run=True to obtain current occurrence ids." + ) + if not selected: + raise ValueError("occurrence_ids is empty - pass at least one id from a dry run, or omit the parameter to replace all.") + return self._apply_occurrences(replacer, selected, contents, needle, repl) + + # blind apply (no ids) + if not occurrences: + raise ValueError( + "No occurrences of the pattern were found - NO changes were applied. " + "Check the mode (a literal needle containing regex metacharacters must use mode 'literal'; " + "wildcards require mode 'regex') and the path/glob restrictions, " + "or locate the content with search_for_pattern first." + ) + if expected_count >= 0 and len(occurrences) != expected_count: + listing = self._render_listing(replacer, occurrences, contents, max_answer_chars, dry_run=False) + raise ValueError( + f"expected_count={expected_count}, but the pattern matches {len(occurrences)} occurrence(s) - " + f"NO changes were applied. Review the prospective changes below; re-issue with the corrected " + f"expectation, a refined pattern, or occurrence_ids selecting the intended subset.\n{listing}" + ) + ambiguous = [o for o in occurrences if o.is_ambiguous] + if ambiguous: + listing = self._render_listing(replacer, occurrences, contents, max_answer_chars, dry_run=False) + raise ValueError( + f"{len(ambiguous)} occurrence(s) are ambiguous (the pattern matches again inside the matched text, " + f"indicating possible over-matching) - NO changes were applied. Review the prospective changes below " + f"and either refine the pattern or explicitly select occurrences via occurrence_ids.\n{listing}" + ) + return self._apply_occurrences(replacer, occurrences, contents, needle, repl) + + def _collect_files(self, relative_path: str, paths_include_glob: str, paths_exclude_glob: str) -> list[tuple[str, str]]: + """Collects (relative_path, content) pairs of the non-ignored files in scope, in sorted path order.""" + relative_path = relative_path.strip() + if relative_path: + self.project.validate_relative_path(relative_path, require_not_ignored=True) + abs_path = os.path.join(self.get_project_root(), relative_path) + if not os.path.exists(abs_path): + raise FileNotFoundError(f"Relative path {relative_path} does not exist.") + if os.path.isfile(abs_path): + rel_paths = [relative_path] + else: + _dirs, rel_paths = scan_directory( + path=abs_path, + recursive=True, + is_ignored_dir=self.project.is_ignored_path, + is_ignored_file=self.project.is_ignored_path, + relative_to=self.get_project_root(), + ) + include_patterns = expand_braces(paths_include_glob.strip()) if paths_include_glob.strip() else None + exclude_patterns = expand_braces(paths_exclude_glob.strip()) if paths_exclude_glob.strip() else None + files: list[tuple[str, str]] = [] + for path in sorted(rel_paths): + if include_patterns and not any(glob_match(p, path) for p in include_patterns): + continue + if exclude_patterns and any(glob_match(p, path) for p in exclude_patterns): + continue + try: + files.append((path, self.project.read_file(path))) + except Exception: + continue # skip unreadable (e.g. binary) files + return files + + def _render_listing( + self, + replacer: MultiFileContentReplacer, + occurrences: list[ReplacementOccurrence], + contents: dict[str, str], + max_answer_chars: int, + dry_run: bool, + ) -> str: + affected_files = sorted({o.relative_path for o in occurrences}) + header = f"Found {len(occurrences)} occurrence(s) in {len(affected_files)} file(s)." + if dry_run: + header += ( + " DRY RUN - no changes were applied.\n" + "Re-issue with dry_run=False to replace all of them, or additionally pass occurrence_ids " + "with the ids of the occurrences to replace." + ) + parts = [header] + for path in affected_files: + file_occurrences = [o for o in occurrences if o.relative_path == path] + parts.append(f"\n{path} ({len(file_occurrences)} occurrence(s)):") + for occ in file_occurrences: + parts.append(replacer.render_occurrence_diff(occ, contents[path])) + result = "\n".join(parts) + + def make_locations_only() -> str: + lines = [header] + [f" [{o.occurrence_id}] line {o.start_line}" for o in occurrences] + return "\n".join(lines) + + def make_per_file_counts() -> str: + counts = {path: sum(1 for o in occurrences if o.relative_path == path) for path in affected_files} + return f"{header}\nOccurrence counts per file:\n{self._to_json(counts)}" + + def make_summary() -> str: + return header + + return self._limit_length( + result, max_answer_chars, shortened_result_factories=[make_locations_only, make_per_file_counts, make_summary] + ) + + @staticmethod + def _resolve_occurrence_ids( + occurrence_ids: list[str], occurrences: list[ReplacementOccurrence] + ) -> tuple[list[ReplacementOccurrence], list[str]]: + """Resolves the requested ids against the current occurrences, diagnosing each failure.""" + occurrences_by_id = {o.occurrence_id: o for o in occurrences} + indices_by_path: dict[str, set[int]] = {} + for o in occurrences: + indices_by_path.setdefault(o.relative_path, set()).add(o.index_in_file) + selected: dict[str, ReplacementOccurrence] = {} + problems: list[str] = [] + for oid in occurrence_ids: + occurrence = occurrences_by_id.get(oid) + if occurrence is not None: + selected[oid] = occurrence + continue + id_match = MultiFileContentReplacer.OCCURRENCE_ID_REGEX.match(oid) + if id_match is None: + problems.append(f"{oid}: malformed id (expected ':@' as returned by a dry run)") + elif id_match.group("path") not in indices_by_path: + problems.append(f"{oid}: the pattern currently has no matches in this file") + elif int(id_match.group("index")) not in indices_by_path[id_match.group("path")]: + problems.append(f"{oid}: the file now has fewer matches than at dry-run time (content changed)") + else: + problems.append(f"{oid}: the matched text changed since the dry run (content changed)") + return list(selected.values()), problems + + def _apply_occurrences( + self, + replacer: MultiFileContentReplacer, + occurrences: list[ReplacementOccurrence], + contents: dict[str, str], + needle: str, + repl: str, + ) -> str: + occurrences_by_file: dict[str, list[ReplacementOccurrence]] = {} + for occ in occurrences: + occurrences_by_file.setdefault(occ.relative_path, []).append(occ) + with self.DiagnosticsContext(self, *occurrences_by_file.keys()) as diagnostics_context: + code_editor = self.create_code_editor() + for path, file_occurrences in occurrences_by_file.items(): + with EditedFileContext(path, code_editor) as context: + original_content = context.get_original_content() + if original_content != contents[path]: + # the editor's view differs from what was scanned (e.g. line-ending normalization); + # re-derive the occurrences from the authoritative content and re-validate by id + fresh_by_id = {o.occurrence_id: o for o in replacer.find_occurrences([(path, original_content)], needle, repl)} + try: + file_occurrences = [fresh_by_id[o.occurrence_id] for o in file_occurrences] + except KeyError as e: + raise ValueError( + f"The content of {path} changed while replacing (occurrence {e} no longer resolves); " + f"the file was NOT modified. Re-run with dry_run=True for current ids." + ) from e + context.set_updated_content(replacer.apply_to_content(original_content, file_occurrences)) + per_file = "\n".join(f" {path}: {len(occs)}" for path, occs in occurrences_by_file.items()) + summary = f"Replaced {len(occurrences)} occurrence(s) in {len(occurrences_by_file)} file(s):\n{per_file}" + return diagnostics_context.format_result(summary) + + class DeleteLinesTool(EditingToolWithDiagnostics, ToolMarkerOptional): """ Deletes a range of lines within a file. diff --git a/src/serena/util/text_utils.py b/src/serena/util/text_utils.py index 16bee9d9..aff92fdb 100644 --- a/src/serena/util/text_utils.py +++ b/src/serena/util/text_utils.py @@ -1,4 +1,5 @@ import fnmatch +import hashlib import logging import os import re @@ -528,6 +529,162 @@ class ContentReplacer: return updated_content +@dataclass +class ReplacementOccurrence: + """A single prospective replacement of a pattern match within one file.""" + + occurrence_id: str + """stable, content-anchored identifier: ':@'""" + relative_path: str + index_in_file: int + """0-based index of this match among the matches within its file (in position order)""" + start: int + """character offset of the match start within the file content""" + end: int + """character offset of the match end within the file content""" + matched_text: str + replacement: str + """the fully expanded replacement text (backreferences already resolved)""" + start_line: int + """0-based line number of the match start""" + end_line: int + """0-based line number of the match end""" + is_ambiguous: bool = False + """whether the pattern matches again within the matched text (possible over-match)""" + + +class MultiFileContentReplacer: + """ + Occurrence-level counterpart of :class:`ContentReplacer` operating on multiple files: + finds every match of a pattern across a set of file contents, assigns each occurrence a + stable content-anchored id, renders minimal line diffs for previewing, and computes the + updated content of a file for a selected subset of occurrences. + """ + + OCCURRENCE_ID_REGEX = re.compile(r"^(?P.+):(?P\d+)@(?P[0-9a-f]{6})$") + _DIGEST_LEN = 6 + + def __init__(self, mode: Literal["literal", "regex"], regex_multiline: bool = True): + """ + :param mode: whether the needle is a literal string ("literal") or a regular expression ("regex") + :param regex_multiline: whether to apply multi-line regex matching, enabling the flags re.DOTALL and re.MULTILINE + """ + if mode not in ("literal", "regex"): + raise ValueError(f"Invalid mode: '{mode}', expected 'literal' or 'regex'.") + self.mode = mode + self._flags = (re.MULTILINE | re.DOTALL) if regex_multiline else 0 + + def _compile(self, needle: str) -> re.Pattern: + return re.compile(re.escape(needle) if self.mode == "literal" else needle, flags=self._flags) + + @classmethod + def _digest(cls, matched_text: str) -> str: + return hashlib.sha1(matched_text.encode("utf-8")).hexdigest()[: cls._DIGEST_LEN] + + @classmethod + def make_occurrence_id(cls, relative_path: str, index_in_file: int, matched_text: str) -> str: + return f"{relative_path}:{index_in_file}@{cls._digest(matched_text)}" + + @staticmethod + def _expand_backreferences(match: re.Match, repl_template: str) -> str: + """Expands $!1, $!2, ... in the replacement template (same syntax as :class:`ContentReplacer`).""" + + def expand(m: re.Match) -> str: + group_value = match.group(int(m.group(1))) + return group_value if group_value is not None else m.group(0) + + return re.sub(r"\$!(\d+)", expand, repl_template) + + def find_occurrences(self, files: list[tuple[str, str]], needle: str, repl: str) -> list[ReplacementOccurrence]: + """ + Finds all matches of the needle in the given files. + + :param files: (relative_path, content) pairs; processed in the given order + :param needle: the search expression (literal string or regex, depending on the mode) + :param repl: the replacement template (may contain $!N backreferences in regex mode) + :return: occurrences in deterministic order (file order, then position within the file) + """ + pattern = self._compile(needle) + occurrences: list[ReplacementOccurrence] = [] + for relative_path, content in files: + for index_in_file, match in enumerate(pattern.finditer(content)): + matched_text = match.group(0) + replacement = self._expand_backreferences(match, repl) if self.mode == "regex" else repl + # same over-match heuristic as ContentReplacer: for a multi-line match, the pattern + # matching again within the matched text indicates the match may have swallowed + # more than intended + is_ambiguous = "\n" in matched_text and pattern.search(matched_text[1:]) is not None + occurrences.append( + ReplacementOccurrence( + occurrence_id=self.make_occurrence_id(relative_path, index_in_file, matched_text), + relative_path=relative_path, + index_in_file=index_in_file, + start=match.start(), + end=match.end(), + matched_text=matched_text, + replacement=replacement, + start_line=content.count("\n", 0, match.start()), + end_line=content.count("\n", 0, match.end()), + is_ambiguous=is_ambiguous, + ) + ) + return occurrences + + @staticmethod + def apply_to_content(content: str, occurrences: list[ReplacementOccurrence]) -> str: + """ + Applies the given occurrences (which must have been derived from exactly this content) + and returns the updated content. + """ + for occ in sorted(occurrences, key=lambda o: o.start, reverse=True): + assert content[occ.start : occ.end] == occ.matched_text, ( + f"Occurrence {occ.occurrence_id} does not match the content it is being applied to" + ) + content = content[: occ.start] + occ.replacement + content[occ.end :] + return content + + @staticmethod + def _format_block(block: str, prefix: str, max_lines: int, max_line_chars: int) -> list[str]: + lines = block.split("\n") + shown = lines[:max_lines] + result = [] + for line in shown: + if len(line) > max_line_chars: + line = line[:max_line_chars] + f"… (+{len(line) - max_line_chars} chars)" + result.append(f" {prefix} {line}") + if len(lines) > max_lines: + result.append(f" {prefix} … ({len(lines) - max_lines} more lines)") + return result + + def render_occurrence_diff( + self, occ: ReplacementOccurrence, content: str, max_lines_per_side: int = 6, max_line_chars: int = 200 + ) -> str: + """ + Renders a minimal line diff for the occurrence: the full lines spanned by the match, + before and after the replacement. + + :param occ: the occurrence (must have been derived from exactly this content) + :param content: the file content the occurrence was found in + :param max_lines_per_side: cap on the number of displayed lines per diff side + :param max_line_chars: cap on the number of displayed characters per line + :return: the rendered diff + """ + line_start = content.rfind("\n", 0, occ.start) + 1 + line_end = content.find("\n", occ.end) + if line_end == -1: + line_end = len(content) + old_block = content[line_start:line_end] + new_block = content[line_start : occ.start] + occ.replacement + content[occ.end : line_end] + location = f"line {occ.start_line}" if occ.start_line == occ.end_line else f"lines {occ.start_line}-{occ.end_line}" + header = f" [{occ.occurrence_id}] {location}" + if occ.is_ambiguous: + header += " (WARNING: the pattern matches again inside this match — possible over-match, verify the diff)" + diff_lines = [header] + diff_lines += self._format_block(old_block, "-", max_lines_per_side, max_line_chars) + diff_lines += self._format_block(new_block, "+", max_lines_per_side, max_line_chars) + return "\n".join(diff_lines) + + @dataclass class TextCoords: line: int diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 5ab27115..dd8a8407 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -26,6 +26,7 @@ from serena.tools import ( GetDiagnosticsForFileTool, InitialInstructionsTool, ReplaceContentTool, + ReplaceInFilesTool, ReplaceSymbolBodyTool, SafeDeleteSymbol, Tool, @@ -1171,6 +1172,57 @@ class TestSerenaAgent: new_content = read_project_file(serena_agent.get_active_project(), relative_path) assert repl in new_content + @pytest.mark.parametrize( + "serena_agent", + [ + pytest.param(Language.PYTHON, marks=get_pytest_markers(Language.PYTHON), id="python_replace_in_files"), + ], + indirect=["serena_agent"], + ) + def test_replace_in_files_dry_run_then_selective_apply(self, serena_agent: SerenaAgent): + """A dry run lists every occurrence as a diff with an id and modifies nothing; a follow-up call + restricted to one of the ids replaces exactly that occurrence. + """ + relative_path = os.path.join("test_repo", "models.py") + needle = "name: str | None = None" + repl = "name: str | None = MARKER_DEFAULT" + tool = serena_agent.get_tool(ReplaceInFilesTool) + with project_file_modification_context(serena_agent, relative_path): + original_content = read_project_file(serena_agent.get_active_project(), relative_path) + assert original_content.count(needle) >= 2 + + listing = tool.apply(needle=needle, repl=repl, mode="literal", relative_path=relative_path, dry_run=True) + assert "DRY RUN" in listing + occurrence_ids = re.findall(r"\[([^\[\]]+:\d+@[0-9a-f]{6})\]", listing) + assert len(occurrence_ids) == original_content.count(needle) + assert read_project_file(serena_agent.get_active_project(), relative_path) == original_content + + result = tool.apply(needle=needle, repl=repl, mode="literal", relative_path=relative_path, occurrence_ids=[occurrence_ids[0]]) + assert "Replaced 1 occurrence(s) in 1 file(s)" in result + new_content = read_project_file(serena_agent.get_active_project(), relative_path) + assert new_content.count("MARKER_DEFAULT") == 1 + + @pytest.mark.parametrize( + "serena_agent", + [ + pytest.param(Language.PYTHON, marks=get_pytest_markers(Language.PYTHON), id="python_replace_in_files_guard"), + ], + indirect=["serena_agent"], + ) + def test_replace_in_files_expected_count_guard_returns_listing(self, serena_agent: SerenaAgent): + """A blind call with a wrong expected_count must change nothing and return the prospective + changes (the failed guard doubles as a dry run). + """ + relative_path = os.path.join("test_repo", "models.py") + needle = "name: str | None = None" + tool = serena_agent.get_tool(ReplaceInFilesTool) + with project_file_modification_context(serena_agent, relative_path): + original_content = read_project_file(serena_agent.get_active_project(), relative_path) + with pytest.raises(ValueError, match="NO changes were applied") as exc_info: + tool.apply(needle=needle, repl="X", mode="literal", relative_path=relative_path, expected_count=1) + assert re.search(r"\[[^\[\]]+:\d+@[0-9a-f]{6}\]", str(exc_info.value)) # the listing with ids is included + assert read_project_file(serena_agent.get_active_project(), relative_path) == original_content + @pytest.mark.parametrize( "serena_agent", [ diff --git a/test/serena/test_text_utils.py b/test/serena/test_text_utils.py index e89ccf71..e6cd4539 100644 --- a/test/serena/test_text_utils.py +++ b/test/serena/test_text_utils.py @@ -2,7 +2,7 @@ import re import pytest -from serena.util.text_utils import LineType, search_files, search_text +from serena.util.text_utils import LineType, MultiFileContentReplacer, search_files, search_text class TestSearchText: @@ -575,3 +575,65 @@ class TestExpandBraces: from serena.util.text_utils import expand_braces assert sorted(expand_braces(pattern)) == sorted(expected) + + +class TestMultiFileContentReplacer: + FILES = [ + ("a/first.py", "import old_pkg\n\nvalue = old_pkg.compute()\n"), + ("b/second.py", "import old_pkg\nother = 1\n"), + ] + + def test_find_occurrences_order_ids_and_lines(self): + replacer = MultiFileContentReplacer(mode="literal") + occurrences = replacer.find_occurrences(self.FILES, "old_pkg", "new_pkg") + assert [o.relative_path for o in occurrences] == ["a/first.py", "a/first.py", "b/second.py"] + assert [o.index_in_file for o in occurrences] == [0, 1, 0] + assert [o.start_line for o in occurrences] == [0, 2, 0] + for o in occurrences: + assert o.matched_text == "old_pkg" + assert o.replacement == "new_pkg" + assert MultiFileContentReplacer.OCCURRENCE_ID_REGEX.match(o.occurrence_id) + # ids are content-anchored: same matched text at the same index yields the same id across calls + again = replacer.find_occurrences(self.FILES, "old_pkg", "new_pkg") + assert [o.occurrence_id for o in again] == [o.occurrence_id for o in occurrences] + + def test_regex_backreference_expansion(self): + replacer = MultiFileContentReplacer(mode="regex") + files = [("f.txt", "name=alpha\nname=beta\n")] + occurrences = replacer.find_occurrences(files, r"name=(\w+)", "id=$!1") + assert [o.replacement for o in occurrences] == ["id=alpha", "id=beta"] + + def test_apply_to_content_selected_subset(self): + replacer = MultiFileContentReplacer(mode="literal") + path, content = self.FILES[0] + occurrences = [o for o in replacer.find_occurrences(self.FILES, "old_pkg", "new_pkg") if o.relative_path == path] + updated = replacer.apply_to_content(content, occurrences[1:]) # only the second occurrence + assert updated == "import old_pkg\n\nvalue = new_pkg.compute()\n" + updated_all = replacer.apply_to_content(content, occurrences) + assert updated_all == "import new_pkg\n\nvalue = new_pkg.compute()\n" + + def test_ambiguous_multiline_match_is_flagged(self): + replacer = MultiFileContentReplacer(mode="regex") + files = [("f.txt", "start A\nstart B\nend\n")] + occurrences = replacer.find_occurrences(files, r"start.*?end", "X") + assert len(occurrences) == 1 + assert occurrences[0].is_ambiguous + + def test_render_occurrence_diff_minimal_lines(self): + replacer = MultiFileContentReplacer(mode="literal") + path, content = self.FILES[0] + occ = replacer.find_occurrences([(path, content)], "old_pkg.compute()", "new_pkg.compute_all()")[0] + diff = replacer.render_occurrence_diff(occ, content) + assert occ.occurrence_id in diff + assert "line 2" in diff + assert " - value = old_pkg.compute()" in diff + assert " + value = new_pkg.compute_all()" in diff + # only the affected line is shown, not the whole file + assert "import old_pkg" not in diff + + def test_apply_to_content_rejects_drifted_occurrence(self): + replacer = MultiFileContentReplacer(mode="literal") + path, content = self.FILES[0] + occ = replacer.find_occurrences([(path, content)], "old_pkg", "new_pkg")[0] + with pytest.raises(AssertionError): + replacer.apply_to_content("completely different content", [occ])