diff --git a/src/serena/tools/file_tools.py b/src/serena/tools/file_tools.py index ea52e9eb..fc3d3184 100644 --- a/src/serena/tools/file_tools.py +++ b/src/serena/tools/file_tools.py @@ -9,6 +9,7 @@ File and file system-related tools, specifically for import os import re from collections import defaultdict +from collections.abc import Callable from fnmatch import fnmatch from pathlib import Path from typing import Literal @@ -202,6 +203,45 @@ class ReplaceContentTool(Tool, ToolMarkerCanEdit): relative_path, needle, repl, mode=mode, allow_multiple_occurrences=allow_multiple_occurrences, require_not_ignored=True ) + @staticmethod + def _create_replacement_function(regex_pattern: str, repl_template: str, regex_flags: int) -> Callable[[re.Match], str]: + """ + Creates a replacement function that validates for ambiguity and handles backreferences. + + :param regex_pattern: The regex pattern being used for matching + :param repl_template: The replacement template with $!1, $!2, etc. for backreferences + :param regex_flags: The flags to use when searching (e.g., re.DOTALL | re.MULTILINE) + :return: A function suitable for use with re.sub() or re.subn() + """ + + def validate_and_replace(match: re.Match) -> str: + matched_text = match.group(0) + + # For multi-line match, check if the same pattern matches again within the already-matched text, + # rendering the match ambiguous. Typical pattern in the code: + # + # When matching + # .*? + # this will match the entire span above, while only the suffix may have been intended. + # (See test case for a practical example.) + # To detect this, we check if the same pattern matches again within the matched text, + if "\n" in matched_text and re.search(regex_pattern, matched_text[1:], flags=regex_flags): + raise ValueError( + "Match is ambiguous: the search pattern matches multiple overlapping occurrences. " + "Please revise the search pattern to be more specific to avoid ambiguity." + ) + + # Handle backreferences: replace $!1, $!2, etc. with actual matched groups + def expand_backreference(m: re.Match) -> str: + group_num = int(m.group(1)) + group_value = match.group(group_num) + return group_value if group_value is not None else m.group(0) + + result = re.sub(r"\$!(\d+)", expand_backreference, repl_template) + return result + + return validate_and_replace + def replace_content( self, relative_path: str, @@ -226,19 +266,22 @@ class ReplaceContentTool(Tool, ToolMarkerCanEdit): else: raise ValueError(f"Invalid mode: '{mode}', expected 'literal' or 'regex'.") + regex_flags = re.DOTALL | re.MULTILINE + # escape backslashes in repl repl = repl.replace("\\", "\\\\") - # convert $!1, $!2, etc. to \1, \2, etc. for backreferences - repl = re.sub(r"\$!(\d+)", "\\\\1", repl) + + # create replacement function with validation and backreference handling + repl_fn = self._create_replacement_function(regex, repl, regex_flags=regex_flags) # perform replacement - updated_content, n = re.subn(regex, repl, original_content, flags=re.DOTALL | re.MULTILINE) + updated_content, n = re.subn(regex, repl_fn, original_content, flags=regex_flags) if n == 0: - return f"Error: No matches of search expression found in file '{relative_path}'." + raise ValueError(f"Error: No matches of search expression found in file '{relative_path}'.") if not allow_multiple_occurrences and n > 1: - return ( - f"Error: Expression matches {n} occurrences in file '{relative_path}'. " + raise ValueError( + f"Expression matches {n} occurrences in file '{relative_path}'. " "Please revise the expression to be more specific or enable allow_multiple_occurrences if this is expected." ) context.set_updated_content(updated_content) diff --git a/test/resources/repos/typescript/test_repo/ws_manager.js b/test/resources/repos/typescript/test_repo/ws_manager.js new file mode 100644 index 00000000..5262cc58 --- /dev/null +++ b/test/resources/repos/typescript/test_repo/ws_manager.js @@ -0,0 +1,73 @@ +/** + * Dummy WebSocket manager class for testing ambiguous regex matching. + */ +class WebSocketManager { + constructor() { + this.ws = null; + this.statusElement = document.getElementById("status"); + } + + /** + * Connects to the WebSocket server. + */ + connectToServer() { + if (this.ws?.readyState === WebSocket.OPEN) { + this.updateConnectionStatus("Already connected", true); + return; + } + + try { + this.ws = new WebSocket("ws://localhost:4402"); + this.updateConnectionStatus("Connecting...", false); + + this.ws.onopen = () => { + console.log("Connected to server"); + this.updateConnectionStatus("Connected", true); + }; + + this.ws.onmessage = (event) => { + console.log("Received message:", event.data); + try { + const data = JSON.parse(event.data); + this.handleMessage(data); + } catch (error) { + console.error("Failed to parse message:", error); + this.updateConnectionStatus("Parse error", false); + } + }; + + this.ws.onclose = (event) => { + console.log("Connection closed"); + const message = event.reason || undefined; + this.updateConnectionStatus("Disconnected", false, message); + this.ws = null; + }; + + this.ws.onerror = (error) => { + console.error("WebSocket error:", error); + this.updateConnectionStatus("Connection error", false); + }; + } catch (error) { + console.error("Failed to connect to server:", error); + this.updateConnectionStatus("Connection failed", false); + } + } + + /** + * Updates the connection status display. + */ + updateConnectionStatus(status, isConnected, message) { + if (this.statusElement) { + const text = message ? `${status}: ${message}` : status; + this.statusElement.textContent = text; + this.statusElement.style.color = isConnected ? "green" : "red"; + } + } + + /** + * Handles incoming messages. + */ + handleMessage(data) { + console.log("Handling:", data); + } +} \ No newline at end of file diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 441cdf53..011386c1 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -3,13 +3,14 @@ import logging import os import time from collections.abc import Iterator +from contextlib import contextmanager import pytest from serena.agent import SerenaAgent from serena.config.serena_config import ProjectConfig, RegisteredProject, SerenaConfig from serena.project import Project -from serena.tools import FindReferencingSymbolsTool, FindSymbolTool, ReplaceSymbolBodyTool +from serena.tools import SUCCESS_RESULT, FindReferencingSymbolsTool, FindSymbolTool, ReplaceContentTool, ReplaceSymbolBodyTool from solidlsp.ls_config import Language from test.conftest import get_repo_path, language_tests_enabled from test.solidlsp import clojure as clj @@ -56,6 +57,24 @@ def serena_config(): return config +@contextmanager +def project_file_modification_context(serena_agent: SerenaAgent, relative_path: str) -> Iterator[None]: + """Context manager to modify a project file and revert the changes after use.""" + project = serena_agent.get_active_project() + file_path = os.path.join(project.project_root, relative_path) + + # Read the original content + with open(file_path, encoding=project.project_config.encoding) as f: + original_content = f.read() + + try: + yield + finally: + # Revert to the original content + with open(file_path, "w", encoding=project.project_config.encoding) as f: + f.write(original_content) + + @pytest.fixture def serena_agent(request: pytest.FixtureRequest, serena_config) -> Iterator[SerenaAgent]: language = Language(request.param) @@ -70,7 +89,8 @@ def serena_agent(request: pytest.FixtureRequest, serena_config) -> Iterator[Sere agent.execute_task(lambda: None) yield agent - # explicitly delete to free resources + + # explicitly shut down to free resources agent.shutdown(timeout=5) @@ -355,3 +375,52 @@ class TestSerenaAgent: replace_symbol_body_tool = serena_agent.get_tool(ReplaceSymbolBodyTool) with pytest.raises(ValueError, match=match_text): replace_symbol_body_tool.apply(name_path=name_path, relative_path=relative_path, body="") + + @pytest.mark.parametrize( + "serena_agent", + [ + pytest.param( + Language.TYPESCRIPT, + marks=pytest.mark.typescript, + ), + ], + indirect=["serena_agent"], + ) + def test_replace_content_regex(self, serena_agent: SerenaAgent): + """ + Tests a regex-based content replacement that has a unique match + """ + relative_path = "ws_manager.js" + with project_file_modification_context(serena_agent, relative_path): + replace_content_tool = serena_agent.get_tool(ReplaceContentTool) + result = replace_content_tool.apply( + needle=r'catch \(error\) \{\s*console.error\("Failed to connect.*?\}', + repl='catch(error) { console.log("Never mind"); }', + relative_path=relative_path, + mode="regex", + ) + assert result == SUCCESS_RESULT + + @pytest.mark.parametrize( + "serena_agent", + [ + pytest.param( + Language.TYPESCRIPT, + marks=pytest.mark.typescript, + ), + ], + indirect=["serena_agent"], + ) + def test_replace_content_regex_ambiguous(self, serena_agent: SerenaAgent): + """ + Tests that an ambiguous replacement where there is a larger match that internally contains + a smaller match triggers an exception + """ + replace_content_tool = serena_agent.get_tool(ReplaceContentTool) + with pytest.raises(ValueError, match="ambiguous"): + replace_content_tool.apply( + needle=r'catch \(error\) \{.*?this\.updateConnectionStatus\("Connection failed", false\);.*?\}', + repl='catch(error) { console.log("Never mind"); }', + relative_path="ws_manager.js", + mode="regex", + )