mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-11 03:13:51 +00:00
Add file system facade (fs); create_text_file moves there from edit
Introduce FsApi, exposing operations on the project's files as units: read_file, create_text_file, list_dir, find_file and search_for_pattern. The boundary to the edit facade, which modifies content within existing files, is stated in the facade description. Results expose their data to code (FileContent, DirectoryListing, PatternMatches) and render as before; the pattern search's shortening ladder moves from the tool into PatternMatchesRenderer. The file tools delegate to the API via FsApiMixin. The fs facade is always part of the REPL.
This commit is contained in:
1 parent
c0e74c7d7e
commit
5f129628d9
7 files changed
+446
-202
No files matched your search
+6
-1
@@ -48,6 +48,7 @@ from serena.memories.memory_manager import MemoryManager
|
||||
from serena.project import Project
|
||||
from serena.prompt_factory import SerenaPromptFactory
|
||||
from serena.repl.api.edit_api import EditApi
|
||||
from serena.repl.api.fs_api import FsApi
|
||||
from serena.repl.api.jb_api import JetBrainsApi
|
||||
from serena.repl.api.lsp_api import LspApi
|
||||
from serena.repl.api.mem_api import MemoryApi
|
||||
@@ -1185,7 +1186,11 @@ class SerenaAgent:
|
||||
api_scope.exclude_editing()
|
||||
|
||||
# gather facades
|
||||
facades = [Facade.from_api(EditApi(self), api_scope), Facade.from_api(MemoryApi(self), api_scope)]
|
||||
facades = [
|
||||
Facade.from_api(FsApi(self), api_scope),
|
||||
Facade.from_api(EditApi(self), api_scope),
|
||||
Facade.from_api(MemoryApi(self), api_scope),
|
||||
]
|
||||
if self._language_backend.is_lsp():
|
||||
facades.append(Facade.from_api(LspApi(self), api_scope))
|
||||
elif self._language_backend.is_jetbrains():
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
The implementation of editing operations, which are independent of the language backend.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Literal
|
||||
|
||||
from serena.code_editor import EditedFileContext
|
||||
@@ -69,34 +68,6 @@ class EditApi(FacadeApi):
|
||||
|
||||
# file-level operations
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def create_text_file(self, relative_path: str, content: str) -> str:
|
||||
"""
|
||||
Writes a new file or overwrites an existing file with the given content.
|
||||
|
||||
:param relative_path: the relative path to the file to create
|
||||
:param content: the (appropriately encoded) content to write to the file
|
||||
:return: a message indicating success
|
||||
"""
|
||||
project = self._get_project()
|
||||
project_root = Path(project.project_root)
|
||||
abs_path = (project_root / relative_path).resolve()
|
||||
will_overwrite_existing = abs_path.exists()
|
||||
|
||||
# validate the destination path
|
||||
if will_overwrite_existing:
|
||||
project.validate_relative_path(relative_path)
|
||||
else:
|
||||
assert abs_path.is_relative_to(project_root), f"Cannot create file outside of the project directory, got {relative_path=}"
|
||||
|
||||
# write the file
|
||||
abs_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
abs_path.write_text(content, encoding=project.project_config.encoding, newline=project.line_ending.newline_str)
|
||||
answer = f"File created: {relative_path}."
|
||||
if will_overwrite_existing:
|
||||
answer += " Overwrote existing file."
|
||||
return answer
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def replace_content(
|
||||
self,
|
||||
|
||||
@@ -0,0 +1,320 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""
|
||||
The implementation of operations on the project's files.
|
||||
"""
|
||||
|
||||
import os
|
||||
from collections import defaultdict
|
||||
from fnmatch import fnmatch
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from serena.util.file_system import scan_directory
|
||||
from serena.util.text_utils import MatchedConsecutiveLines
|
||||
from solidlsp.ls_utils import TextUtils
|
||||
|
||||
from ..facade import FacadeApi, facade_method
|
||||
from ..representable import Renderer, RepresentableViaRenderer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from serena.agent import SerenaAgent
|
||||
|
||||
|
||||
class FileContent(RepresentableViaRenderer):
|
||||
"""
|
||||
The content of a file (or of a range of its lines): `text` (the joined lines) and `lines`.
|
||||
"""
|
||||
|
||||
def __init__(self, lines: list[str], renderer: "FileContentRenderer"):
|
||||
"""
|
||||
:param lines: the lines (without line breaks)
|
||||
:param renderer: the renderer to use for representing the content
|
||||
"""
|
||||
super().__init__(renderer)
|
||||
self.lines = lines
|
||||
|
||||
@property
|
||||
def text(self) -> str:
|
||||
return "\n".join(self.lines)
|
||||
|
||||
|
||||
class FileContentRenderer(Renderer[FileContent]):
|
||||
def render(self, obj: FileContent) -> str:
|
||||
return self._limit_length(obj.text)
|
||||
|
||||
|
||||
class DirectoryListing(RepresentableViaRenderer):
|
||||
"""
|
||||
The entries of a directory: `dirs` and `files` (relative paths).
|
||||
"""
|
||||
|
||||
def __init__(self, dirs: list[str], files: list[str], renderer: "DirectoryListingRenderer"):
|
||||
"""
|
||||
:param dirs: the relative paths of the directories
|
||||
:param files: the relative paths of the files
|
||||
:param renderer: the renderer to use for representing the listing
|
||||
"""
|
||||
super().__init__(renderer)
|
||||
self.dirs = dirs
|
||||
self.files = files
|
||||
|
||||
|
||||
class DirectoryListingRenderer(Renderer[DirectoryListing]):
|
||||
def render(self, obj: DirectoryListing) -> str:
|
||||
return self._limit_length(self._to_json({"dirs": obj.dirs, "files": obj.files}))
|
||||
|
||||
|
||||
class PatternMatches(RepresentableViaRenderer):
|
||||
"""
|
||||
The matches of a pattern search: `matches`, each a `MatchedConsecutiveLines` with `source_file_path`,
|
||||
`matched_lines` (each with `line_number` and `line_content`) and `to_display_string()`.
|
||||
"""
|
||||
|
||||
def __init__(self, matches: list[MatchedConsecutiveLines], renderer: "PatternMatchesRenderer"):
|
||||
"""
|
||||
:param matches: the matches
|
||||
:param renderer: the renderer to use for representing the matches
|
||||
"""
|
||||
super().__init__(renderer)
|
||||
self.matches = matches
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.matches)
|
||||
|
||||
def matches_by_file_(self) -> dict[str, list[MatchedConsecutiveLines]]:
|
||||
result: defaultdict[str, list[MatchedConsecutiveLines]] = defaultdict(list)
|
||||
for match in self.matches:
|
||||
assert match.source_file_path is not None
|
||||
result[match.source_file_path].append(match)
|
||||
return result
|
||||
|
||||
|
||||
class PatternMatchesRenderer(Renderer[PatternMatches]):
|
||||
"""
|
||||
Renders matches as a mapping from file paths to matched line blocks (with context), falling back to progressively
|
||||
shorter representations (first lines, truncated first lines, line numbers, per-file counts, a summary) if the
|
||||
length limit is exceeded.
|
||||
"""
|
||||
|
||||
_TEXT_TRUNCATE = 60
|
||||
|
||||
def render(self, obj: PatternMatches) -> str:
|
||||
matches_by_file = obj.matches_by_file_()
|
||||
file_to_matches = {path: [m.to_display_string() for m in matches] for path, matches in matches_by_file.items()}
|
||||
|
||||
# capture lightweight match data for shortening before serialization
|
||||
match_lines_by_file = {
|
||||
path: [{"line": m.matched_lines[0].line_number, "text": m.matched_lines[0].line_content.strip()} for m in matches]
|
||||
for path, matches in matches_by_file.items()
|
||||
}
|
||||
|
||||
# shortened result closures, from least to most aggressive shortening
|
||||
def render_first_lines(truncate: bool) -> str:
|
||||
"""Render each match's first line, either in full or truncated to a fixed length."""
|
||||
|
||||
def entry_text(text: str) -> str:
|
||||
if truncate and len(text) > self._TEXT_TRUNCATE:
|
||||
return text[: self._TEXT_TRUNCATE] + "..."
|
||||
return text
|
||||
|
||||
compact = {
|
||||
path: [{"line": m["line"], "text": entry_text(str(m["text"]))} for m in lines]
|
||||
for path, lines in match_lines_by_file.items()
|
||||
}
|
||||
if truncate:
|
||||
header = (
|
||||
f"Matched lines (text over {self._TEXT_TRUNCATE} chars is truncated, marked with a trailing '...'); "
|
||||
"use read_file with the line numbers for full content:"
|
||||
)
|
||||
else:
|
||||
header = "Matched lines per file; use read_file with the line numbers for surrounding context:"
|
||||
return f"{header}\n{self._to_json(compact)}"
|
||||
|
||||
def make_first_lines_full() -> str:
|
||||
return render_first_lines(truncate=False)
|
||||
|
||||
def make_first_lines_truncated() -> str:
|
||||
return render_first_lines(truncate=True)
|
||||
|
||||
def make_line_numbers_only() -> str:
|
||||
numbers = {path: [m["line"] for m in lines] for path, lines in match_lines_by_file.items()}
|
||||
return f"Match lines per file:\n{self._to_json(numbers)}"
|
||||
|
||||
def make_per_file_counts() -> str:
|
||||
counts = {path: len(lines) for path, lines in match_lines_by_file.items()}
|
||||
return f"Match counts per file:\n{self._to_json(counts)}"
|
||||
|
||||
def make_summary() -> str:
|
||||
return f"Found {len(obj)} matches in {len(match_lines_by_file)} files."
|
||||
|
||||
return self._limit_length(
|
||||
self._to_json(file_to_matches),
|
||||
shortened_result_factories=[
|
||||
make_first_lines_full,
|
||||
make_first_lines_truncated,
|
||||
make_line_numbers_only,
|
||||
make_per_file_counts,
|
||||
make_summary,
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
class FsApi(FacadeApi):
|
||||
def __init__(self, agent: "SerenaAgent") -> None:
|
||||
super().__init__(agent, name="fs", description="the project's files as units (as opposed to their content, see `edit`)")
|
||||
|
||||
@facade_method()
|
||||
def read_file(self, relative_path: str, start_line: int = 0, end_line: int | None = None, max_answer_chars: int = -1) -> FileContent:
|
||||
"""
|
||||
Reads the given file or a range of its lines.
|
||||
|
||||
:param relative_path: the relative path to the file to read
|
||||
:param start_line: the 0-based index of the first line to be retrieved, negative values count from the end of the file.
|
||||
:param end_line: the 0-based index of the last line to be retrieved (inclusive). If None, read until the end of the file.
|
||||
:param max_answer_chars: if the rendered content is longer than this number of characters, no content is rendered;
|
||||
-1 for the configured default. Don't adjust unless there is really no other way to get the content required for the task.
|
||||
:return: the content
|
||||
"""
|
||||
project = self._get_project()
|
||||
project.validate_relative_path(relative_path)
|
||||
|
||||
# read lines, using the same (LSP-compliant) notion of line breaks as the line-based editing operations
|
||||
lines = TextUtils.split_lines(project.read_file(relative_path))
|
||||
lines = lines[start_line:] if end_line is None else lines[start_line : end_line + 1]
|
||||
return FileContent(lines, FileContentRenderer(self._agent, max_answer_chars))
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def create_text_file(self, relative_path: str, content: str) -> str:
|
||||
"""
|
||||
Writes a new file or overwrites an existing file with the given content.
|
||||
|
||||
:param relative_path: the relative path to the file to create
|
||||
:param content: the (appropriately encoded) content to write to the file
|
||||
:return: a message indicating success
|
||||
"""
|
||||
project = self._get_project()
|
||||
project_root = Path(project.project_root)
|
||||
abs_path = (project_root / relative_path).resolve()
|
||||
will_overwrite_existing = abs_path.exists()
|
||||
|
||||
# validate the destination path
|
||||
if will_overwrite_existing:
|
||||
project.validate_relative_path(relative_path)
|
||||
else:
|
||||
assert abs_path.is_relative_to(project_root), f"Cannot create file outside of the project directory, got {relative_path=}"
|
||||
|
||||
# write the file
|
||||
abs_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
abs_path.write_text(content, encoding=project.project_config.encoding, newline=project.line_ending.newline_str)
|
||||
answer = f"File created: {relative_path}."
|
||||
if will_overwrite_existing:
|
||||
answer += " Overwrote existing file."
|
||||
return answer
|
||||
|
||||
@facade_method()
|
||||
def list_dir(
|
||||
self, relative_path: str, recursive: bool, skip_ignored_files: bool = False, max_answer_chars: int = -1
|
||||
) -> DirectoryListing:
|
||||
"""
|
||||
Lists files and directories in the given directory (optionally with recursion).
|
||||
|
||||
:param relative_path: the relative path to the directory to list; pass "." to scan the project root
|
||||
:param recursive: whether to scan subdirectories recursively
|
||||
:param skip_ignored_files: whether to skip files and directories that are ignored
|
||||
:param max_answer_chars: if the rendered listing is longer than this number of characters, no content is rendered;
|
||||
-1 for the configured default. Don't adjust unless there is really no other way to get the content required for the task.
|
||||
:return: the listing
|
||||
"""
|
||||
project = self._get_project()
|
||||
if not project.relative_path_exists(relative_path):
|
||||
raise FileNotFoundError(f"Directory not found: {relative_path} (check if the path is correct relative to the project root)")
|
||||
project.validate_relative_path(relative_path)
|
||||
|
||||
is_ignored_path_fn = project.get_is_ignored_path_fn(relative_path, skip_ignored_files)
|
||||
dirs, files = scan_directory(
|
||||
os.path.join(project.project_root, relative_path),
|
||||
relative_to=project.project_root,
|
||||
recursive=recursive,
|
||||
is_ignored_dir=is_ignored_path_fn,
|
||||
is_ignored_file=is_ignored_path_fn,
|
||||
)
|
||||
return DirectoryListing(dirs, files, DirectoryListingRenderer(self._agent, max_answer_chars))
|
||||
|
||||
@facade_method()
|
||||
def find_file(self, file_mask: str, relative_path: str) -> list[str]:
|
||||
"""
|
||||
Finds files matching the given file mask within the given relative path.
|
||||
|
||||
:param file_mask: the filename or file mask (using the wildcards * or ?) to search for
|
||||
:param relative_path: the relative path to the directory to search in; pass "." to scan the project root
|
||||
:return: the relative paths of the matching files
|
||||
"""
|
||||
project = self._get_project()
|
||||
project.validate_relative_path(relative_path)
|
||||
|
||||
is_ignored_path_fn = project.get_is_ignored_path_fn(relative_path, skip_ignored_paths=False)
|
||||
|
||||
# find the files by ignoring everything that doesn't match
|
||||
def is_ignored_file(abs_path: str) -> bool:
|
||||
if is_ignored_path_fn(abs_path):
|
||||
return True
|
||||
return not fnmatch(os.path.basename(abs_path), file_mask)
|
||||
|
||||
_dirs, files = scan_directory(
|
||||
path=os.path.join(project.project_root, relative_path),
|
||||
recursive=True,
|
||||
is_ignored_dir=is_ignored_path_fn,
|
||||
is_ignored_file=is_ignored_file,
|
||||
relative_to=project.project_root,
|
||||
)
|
||||
return files
|
||||
|
||||
@facade_method()
|
||||
def search_for_pattern(
|
||||
self,
|
||||
substring_pattern: str,
|
||||
context_lines_before: int = 0,
|
||||
context_lines_after: int = 0,
|
||||
paths_include_glob: str = "",
|
||||
paths_exclude_glob: str = "",
|
||||
relative_path: str = "",
|
||||
restrict_search_to_code_files: bool = False,
|
||||
skip_ignored_files: bool = True,
|
||||
multiline: bool = True,
|
||||
max_answer_chars: int = -1,
|
||||
) -> PatternMatches:
|
||||
"""
|
||||
Searches for a regex pattern across project files, returning whole matched lines (plus optional context).
|
||||
Prefer symbolic operations if you know which symbols you are looking for!
|
||||
|
||||
:param substring_pattern: regular expression to search for.
|
||||
:param context_lines_before: number of context lines to include before each match.
|
||||
:param context_lines_after: number of context lines to include after each match.
|
||||
:param paths_include_glob: optional glob (relative to project root, e.g. ``"src/**/*.ts"``) restricting which files are searched.
|
||||
:param paths_exclude_glob: optional glob to exclude files; takes precedence over `paths_include_glob`.
|
||||
:param relative_path: restricts the search to this file or subdirectory of the project root
|
||||
:param restrict_search_to_code_files: whether to search only (non-ignored) files containing analyzable code symbols
|
||||
(useful when looking for class/method definitions); otherwise also search non-code files.
|
||||
:param skip_ignored_files: whether to skip ignored sub-paths (default: True)
|
||||
:param multiline: whether to apply multi-line matching (default: True), enabling the flags re.DOTALL and re.MULTILINE
|
||||
:param max_answer_chars: if the rendered output exceeds this many characters, a progressively shortened summary is
|
||||
rendered instead. -1 for the configured default.
|
||||
:return: the matches, rendered as a mapping from file paths to matched consecutive lines (0-based line numbers)
|
||||
"""
|
||||
project = self._get_project()
|
||||
relative_path = relative_path.strip()
|
||||
if relative_path:
|
||||
project.validate_relative_path(relative_path)
|
||||
|
||||
matches = project.search_project_files_for_pattern(
|
||||
pattern=substring_pattern,
|
||||
relative_path=relative_path,
|
||||
context_lines_before=context_lines_before,
|
||||
context_lines_after=context_lines_after,
|
||||
paths_include_glob=paths_include_glob.strip(),
|
||||
paths_exclude_glob=paths_exclude_glob.strip(),
|
||||
multiline=multiline,
|
||||
code_files_only=restrict_search_to_code_files,
|
||||
skip_ignored_files=skip_ignored_files,
|
||||
)
|
||||
return PatternMatches(matches, PatternMatchesRenderer(self._agent, max_answer_chars))
|
||||
+37
-155
@@ -7,15 +7,11 @@ File and file system-related tools, specifically for
|
||||
"""
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import os
|
||||
from collections import defaultdict
|
||||
from fnmatch import fnmatch
|
||||
from typing import Literal, cast
|
||||
|
||||
from serena.repl.api.edit_api import EditApi
|
||||
from serena.repl.api.fs_api import FsApi
|
||||
from serena.tools import EditingToolWithDiagnostics, Tool, ToolMarkerOptional
|
||||
from serena.util.file_system import scan_directory
|
||||
from solidlsp.ls_utils import TextUtils
|
||||
|
||||
|
||||
class EditApiMixin:
|
||||
@@ -28,7 +24,17 @@ class EditApiMixin:
|
||||
return EditApi(tool.agent)
|
||||
|
||||
|
||||
class ReadFileTool(Tool):
|
||||
class FsApiMixin:
|
||||
"""
|
||||
Mixin for tools which delegate to the file system API
|
||||
"""
|
||||
|
||||
def _api(self) -> FsApi:
|
||||
tool = cast(Tool, cast(object, self))
|
||||
return FsApi(tool.agent)
|
||||
|
||||
|
||||
class ReadFileTool(Tool, FsApiMixin):
|
||||
"""
|
||||
Reads a file within the project directory.
|
||||
"""
|
||||
@@ -45,22 +51,10 @@ class ReadFileTool(Tool):
|
||||
required for the task.
|
||||
:return: the full text of the file at the given relative path
|
||||
"""
|
||||
self.project.validate_relative_path(relative_path)
|
||||
|
||||
# read lines, using the same (LSP-compliant) notion of line breaks as the line-based editing tools
|
||||
result = self.project.read_file(relative_path)
|
||||
result_lines = TextUtils.split_lines(result)
|
||||
|
||||
if end_line is None:
|
||||
result_lines = result_lines[start_line:]
|
||||
else:
|
||||
result_lines = result_lines[start_line : end_line + 1]
|
||||
result = "\n".join(result_lines)
|
||||
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
return self._api().read_file(relative_path, start_line, end_line, max_answer_chars).represent()
|
||||
|
||||
|
||||
class CreateTextFileTool(EditingToolWithDiagnostics, EditApiMixin):
|
||||
class CreateTextFileTool(EditingToolWithDiagnostics, FsApiMixin):
|
||||
"""
|
||||
Creates/overwrites a file in the project directory.
|
||||
"""
|
||||
@@ -77,7 +71,7 @@ class CreateTextFileTool(EditingToolWithDiagnostics, EditApiMixin):
|
||||
return diagnostics_context.format_result(self._api().create_text_file(relative_path, content))
|
||||
|
||||
|
||||
class ListDirTool(Tool):
|
||||
class ListDirTool(Tool, FsApiMixin):
|
||||
"""
|
||||
Lists files and directories in the given directory (optionally with recursion).
|
||||
"""
|
||||
@@ -94,31 +88,13 @@ class ListDirTool(Tool):
|
||||
Don't adjust unless there is really no other way to get the content required for the task.
|
||||
:return: a JSON object with the names of directories and files within the given directory
|
||||
"""
|
||||
# Check if the directory exists before validation
|
||||
if not self.project.relative_path_exists(relative_path):
|
||||
error_info = {
|
||||
"error": f"Directory not found: {relative_path}",
|
||||
"project_root": self.get_project_root(),
|
||||
"hint": "Check if the path is correct relative to the project root",
|
||||
}
|
||||
return self._to_json(error_info)
|
||||
|
||||
self.project.validate_relative_path(relative_path)
|
||||
|
||||
is_ignored_path_fn = self.project.get_is_ignored_path_fn(relative_path, skip_ignored_files)
|
||||
dirs, files = scan_directory(
|
||||
os.path.join(self.get_project_root(), relative_path),
|
||||
relative_to=self.get_project_root(),
|
||||
recursive=recursive,
|
||||
is_ignored_dir=is_ignored_path_fn,
|
||||
is_ignored_file=is_ignored_path_fn,
|
||||
)
|
||||
|
||||
result = self._to_json({"dirs": dirs, "files": files})
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
try:
|
||||
return self._api().list_dir(relative_path, recursive, skip_ignored_files, max_answer_chars).represent()
|
||||
except FileNotFoundError as e:
|
||||
return self._to_json({"error": str(e), "project_root": self.get_project_root()})
|
||||
|
||||
|
||||
class FindFileTool(Tool):
|
||||
class FindFileTool(Tool, FsApiMixin):
|
||||
"""
|
||||
Finds files in the given relative paths
|
||||
"""
|
||||
@@ -132,28 +108,7 @@ class FindFileTool(Tool):
|
||||
:param skip_ignored_files: whether to skip ignored files/directories
|
||||
:return: a JSON object with the list of matching files
|
||||
"""
|
||||
self.project.validate_relative_path(relative_path)
|
||||
|
||||
is_ignored_path_fn = self.project.get_is_ignored_path_fn(relative_path, skip_ignored_paths=False)
|
||||
dir_to_scan = os.path.join(self.get_project_root(), relative_path)
|
||||
|
||||
# find the files by ignoring everything that doesn't match
|
||||
def is_ignored_file(abs_path: str) -> bool:
|
||||
if is_ignored_path_fn(abs_path):
|
||||
return True
|
||||
filename = os.path.basename(abs_path)
|
||||
return not fnmatch(filename, file_mask)
|
||||
|
||||
_dirs, files = scan_directory(
|
||||
path=dir_to_scan,
|
||||
recursive=True,
|
||||
is_ignored_dir=is_ignored_path_fn,
|
||||
is_ignored_file=is_ignored_file,
|
||||
relative_to=self.get_project_root(),
|
||||
)
|
||||
|
||||
result = self._to_json({"files": files})
|
||||
return result
|
||||
return self._to_json({"files": self._api().find_file(file_mask, relative_path)})
|
||||
|
||||
|
||||
class ReplaceContentTool(EditingToolWithDiagnostics, EditApiMixin):
|
||||
@@ -349,7 +304,7 @@ class InsertAtLineTool(EditingToolWithDiagnostics, ToolMarkerOptional, EditApiMi
|
||||
return diagnostics_context.format_result(self._api().insert_at_line(relative_path, line, content))
|
||||
|
||||
|
||||
class SearchForPatternTool(Tool):
|
||||
class SearchForPatternTool(Tool, FsApiMixin):
|
||||
def apply(
|
||||
self,
|
||||
substring_pattern: str,
|
||||
@@ -381,92 +336,19 @@ class SearchForPatternTool(Tool):
|
||||
``-1`` uses the configured default.
|
||||
:return: A mapping from file paths to matched consecutive lines (0-based line numbers).
|
||||
"""
|
||||
relative_path = relative_path.strip()
|
||||
if relative_path:
|
||||
self.project.validate_relative_path(relative_path)
|
||||
|
||||
matches = self.project.search_project_files_for_pattern(
|
||||
pattern=substring_pattern,
|
||||
relative_path=relative_path,
|
||||
context_lines_before=context_lines_before,
|
||||
context_lines_after=context_lines_after,
|
||||
paths_include_glob=paths_include_glob.strip(),
|
||||
paths_exclude_glob=paths_exclude_glob.strip(),
|
||||
multiline=multiline,
|
||||
code_files_only=restrict_search_to_code_files,
|
||||
skip_ignored_files=skip_ignored_files,
|
||||
return (
|
||||
self._api()
|
||||
.search_for_pattern(
|
||||
substring_pattern,
|
||||
context_lines_before=context_lines_before,
|
||||
context_lines_after=context_lines_after,
|
||||
paths_include_glob=paths_include_glob,
|
||||
paths_exclude_glob=paths_exclude_glob,
|
||||
relative_path=relative_path,
|
||||
restrict_search_to_code_files=restrict_search_to_code_files,
|
||||
skip_ignored_files=skip_ignored_files,
|
||||
multiline=multiline,
|
||||
max_answer_chars=max_answer_chars,
|
||||
)
|
||||
.represent()
|
||||
)
|
||||
|
||||
# group matches by file
|
||||
file_to_matches: dict[str, list[str]] = defaultdict(list)
|
||||
for match in matches:
|
||||
assert match.source_file_path is not None
|
||||
file_to_matches[match.source_file_path].append(match.to_display_string())
|
||||
|
||||
# capture lightweight match data for shortening before serialization
|
||||
match_lines_by_file: dict[str, list[dict[str, int | str]]] = defaultdict(list)
|
||||
for match in matches:
|
||||
assert match.source_file_path is not None
|
||||
first = match.matched_lines[0]
|
||||
match_lines_by_file[match.source_file_path].append({"line": first.line_number, "text": first.line_content.strip()})
|
||||
|
||||
# shortened result closures, from least to most aggressive shortening
|
||||
_TEXT_TRUNCATE = 60
|
||||
|
||||
def render_first_lines(truncate: bool) -> str:
|
||||
"""Render each match's first line, either in full or truncated to a fixed length."""
|
||||
|
||||
def entry_text(text: str) -> str:
|
||||
if truncate and len(text) > _TEXT_TRUNCATE:
|
||||
return text[:_TEXT_TRUNCATE] + "..."
|
||||
return text
|
||||
|
||||
compact = {
|
||||
path: [{"line": m["line"], "text": entry_text(str(m["text"]))} for m in lines]
|
||||
for path, lines in match_lines_by_file.items()
|
||||
}
|
||||
if truncate:
|
||||
header = (
|
||||
f"Matched lines (text over {_TEXT_TRUNCATE} chars is truncated, marked with a trailing '...'); "
|
||||
"use read_file with the line numbers for full content:"
|
||||
)
|
||||
else:
|
||||
header = "Matched lines per file; use read_file with the line numbers for surrounding context:"
|
||||
return f"{header}\n{self._to_json(compact)}"
|
||||
|
||||
def make_first_lines_full() -> str:
|
||||
"""Match locations with each match's full first line."""
|
||||
return render_first_lines(truncate=False)
|
||||
|
||||
def make_first_lines_truncated() -> str:
|
||||
"""Match locations with each match's first line truncated to a fixed length."""
|
||||
return render_first_lines(truncate=True)
|
||||
|
||||
def make_line_numbers_only() -> str:
|
||||
"""Match locations as bare line numbers (no text)."""
|
||||
numbers = {path: [m["line"] for m in lines] for path, lines in match_lines_by_file.items()}
|
||||
return f"Match lines per file:\n{self._to_json(numbers)}"
|
||||
|
||||
def make_per_file_counts() -> str:
|
||||
counts = {path: len(lines) for path, lines in match_lines_by_file.items()}
|
||||
return f"Match counts per file:\n{self._to_json(counts)}"
|
||||
|
||||
def make_summary() -> str:
|
||||
return f"Found {len(matches)} matches in {len(match_lines_by_file)} files."
|
||||
|
||||
result = self._to_json(file_to_matches)
|
||||
return self._limit_length(
|
||||
result,
|
||||
max_answer_chars,
|
||||
shortened_result_factories=[
|
||||
make_first_lines_full,
|
||||
make_first_lines_truncated,
|
||||
make_line_numbers_only,
|
||||
make_per_file_counts,
|
||||
make_summary,
|
||||
],
|
||||
)
|
||||
|
||||
"""
|
||||
Performs a search for a pattern in the project.
|
||||
"""
|
||||
@@ -30,7 +30,6 @@ def api(project: Project) -> EditApi:
|
||||
|
||||
def test_facade_exposes_editing_operations(api: EditApi) -> None:
|
||||
default_methods = {
|
||||
"create_text_file",
|
||||
"replace_content",
|
||||
"replace_in_files",
|
||||
"replace_symbol_body",
|
||||
@@ -47,18 +46,6 @@ def test_facade_exposes_editing_operations(api: EditApi) -> None:
|
||||
assert all(facade.get_method(name).info.can_edit for name in default_methods | optional_methods)
|
||||
|
||||
|
||||
def test_create_text_file(api: EditApi, project: Project) -> None:
|
||||
result = api.create_text_file("sub/new.txt", "hello\n")
|
||||
assert "new.txt" in result
|
||||
assert (Path(project.project_root) / "sub" / "new.txt").read_text(encoding="utf-8") == "hello\n"
|
||||
|
||||
result = api.create_text_file("sub/new.txt", "changed\n")
|
||||
assert "Overwrote" in result
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
api.create_text_file("../outside.txt", "nope")
|
||||
|
||||
|
||||
def test_replace_in_files_dry_run_returns_inspectable_preview(api: EditApi, project: Project) -> None:
|
||||
preview = api.replace_in_files("foo", "bar", mode="literal", dry_run=True)
|
||||
assert isinstance(preview, ReplacementPreview)
|
||||
|
||||
@@ -15,10 +15,8 @@ def read_file_tool(tmp_path: Path) -> ReadFileTool:
|
||||
project = Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
tool = ReadFileTool(agent)
|
||||
# bypass the length limit, which would otherwise depend on the agent configuration
|
||||
tool._limit_length = lambda result, max_answer_chars, shortened_result_factories=None: result # type: ignore[method-assign]
|
||||
return tool
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return ReadFileTool(agent)
|
||||
|
||||
|
||||
def _deleted_by_delete_lines(content: str, line: int) -> str:
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
Tests for the file system facade API.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.fs_api import FsApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project(tmp_path: Path) -> Project:
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "src" / "a.py").write_text("x = foo(1)\ny = foo(2)\nz = 3\n", encoding="utf-8")
|
||||
(tmp_path / "src" / "b.txt").write_text("foo in text\n", encoding="utf-8")
|
||||
(tmp_path / "README.md").write_text("# readme\n", encoding="utf-8")
|
||||
return Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(project: Project) -> FsApi:
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return FsApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_file_operations(api: FsApi) -> None:
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "fs"
|
||||
assert set(facade.enabled_method_names) == {"read_file", "create_text_file", "list_dir", "find_file", "search_for_pattern"}
|
||||
assert {name for name in facade.enabled_method_names if facade.get_method(name).info.can_edit} == {"create_text_file"}
|
||||
|
||||
|
||||
def test_read_file(api: FsApi) -> None:
|
||||
content = api.read_file("src/a.py")
|
||||
assert content.lines == ["x = foo(1)", "y = foo(2)", "z = 3", ""]
|
||||
assert content.represent() == content.text
|
||||
|
||||
assert api.read_file("src/a.py", start_line=1, end_line=1).text == "y = foo(2)"
|
||||
assert api.read_file("src/a.py", start_line=-2).lines == ["z = 3", ""]
|
||||
|
||||
|
||||
def test_create_text_file(api: FsApi, project: Project) -> None:
|
||||
result = api.create_text_file("sub/new.txt", "hello\n")
|
||||
assert "new.txt" in result
|
||||
assert (Path(project.project_root) / "sub" / "new.txt").read_text(encoding="utf-8") == "hello\n"
|
||||
|
||||
result = api.create_text_file("sub/new.txt", "changed\n")
|
||||
assert "Overwrote" in result
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
api.create_text_file("../outside.txt", "nope")
|
||||
|
||||
|
||||
def test_list_dir_and_find_file(api: FsApi) -> None:
|
||||
listing = api.list_dir(".", recursive=True)
|
||||
assert "src" in listing.dirs
|
||||
assert {"src/a.py", "src/b.txt", "README.md"} <= {f.replace("\\", "/") for f in listing.files}
|
||||
assert '"dirs"' in listing.represent() and '"files"' in listing.represent()
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
api.list_dir("missing", recursive=False)
|
||||
|
||||
assert [f.replace("\\", "/") for f in api.find_file("*.py", ".")] == ["src/a.py"]
|
||||
|
||||
|
||||
def test_search_for_pattern(api: FsApi) -> None:
|
||||
matches = api.search_for_pattern("foo", relative_path="src")
|
||||
assert len(matches) == 3
|
||||
assert {m.source_file_path.replace("\\", "/") for m in matches.matches} == {"src/a.py", "src/b.txt"}
|
||||
|
||||
# restricting to code files excludes the text file; the rendering maps files to matched lines
|
||||
code_matches = api.search_for_pattern("foo", restrict_search_to_code_files=True)
|
||||
assert all(m.source_file_path.endswith("a.py") for m in code_matches.matches)
|
||||
assert "foo(1)" in code_matches.represent()
|
||||
Reference in new issue
Block a user