fix: FileTools uses local workspace before API for file content (#6.2)

This commit is contained in:
meeks
2026-07-19 12:05:28 +02:00
parent 21eefd9824
commit 139fb44fac
3 changed files with 72 additions and 3 deletions
+42 -1
View File
@@ -1,3 +1,5 @@
import os
from pathlib import Path
from typing import Any from typing import Any
from gitea.client import GiteaClient from gitea.client import GiteaClient
@@ -5,8 +7,9 @@ from gitea.client import GiteaClient
class FileTools: class FileTools:
"""Tools for Gitea file/content operations.""" """Tools for Gitea file/content operations."""
def __init__(self, client: GiteaClient) -> None: def __init__(self, client: GiteaClient, repo_path: str | None = None) -> None:
self._client = client self._client = client
self._repo_path: str | None = repo_path
def _paginate_lines( def _paginate_lines(
self, self,
@@ -35,6 +38,15 @@ class FileTools:
) )
return result return result
def _resolve_local_path(self, owner: str, repo: str, path: str) -> str | None:
"""Resolve owner/repo/path to a local filesystem path if the repo is cloned."""
if not self._repo_path:
return None
local_repo: str = os.path.join(self._repo_path, owner, repo)
if os.path.isdir(local_repo):
return os.path.join(local_repo, path)
return None
def get_file_content( def get_file_content(
self, self,
owner: str, owner: str,
@@ -45,10 +57,22 @@ class FileTools:
) -> str: ) -> str:
"""Get the content of a file from a Gitea repository with line paging. """Get the content of a file from a Gitea repository with line paging.
Checks the local workspace first if repo_path is configured, falling
back to the remote API when the file is not available locally.
Args: Args:
offset: 1-indexed line to start from (default 1). offset: 1-indexed line to start from (default 1).
limit: Maximum number of lines to return (default 250). limit: Maximum number of lines to return (default 250).
""" """
local_path: str | None = self._resolve_local_path(owner, repo, path)
if local_path and os.path.isfile(local_path):
try:
with open(local_path, 'r', encoding='utf-8', errors='replace') as f:
raw: str = f.read()
return self._paginate_lines(raw, offset, limit)
except Exception:
pass
try: try:
content = self._client.files.get_file_content(owner, repo, path) content = self._client.files.get_file_content(owner, repo, path)
raw: str = "\n".join(content) if isinstance(content, list) else content raw: str = "\n".join(content) if isinstance(content, list) else content
@@ -67,11 +91,28 @@ class FileTools:
) -> str: ) -> str:
"""Get file content at a specific git ref with line paging. """Get file content at a specific git ref with line paging.
Checks the local workspace first using ``git show`` if the repo is
cloned locally, falling back to the remote API.
Args: Args:
ref: Branch, tag, or commit SHA (default 'master'). ref: Branch, tag, or commit SHA (default 'master').
offset: 1-indexed line to start from (default 1). offset: 1-indexed line to start from (default 1).
limit: Maximum number of lines to return (default 250). limit: Maximum number of lines to return (default 250).
""" """
if self._repo_path:
local_repo: str = os.path.join(self._repo_path, owner, repo)
if os.path.isdir(local_repo):
try:
import subprocess
result = subprocess.run(
["git", "-C", local_repo, "show", f"{ref}:{path}"],
capture_output=True, text=True, timeout=15,
)
if result.returncode == 0:
return self._paginate_lines(result.stdout, offset, limit)
except Exception:
pass
try: try:
content = self._client.files.get_file_content(owner, repo, path, ref) content = self._client.files.get_file_content(owner, repo, path, ref)
raw: str = "\n".join(content) if isinstance(content, list) else content raw: str = "\n".join(content) if isinstance(content, list) else content
+2 -2
View File
@@ -11,7 +11,7 @@ from gitea.tools.issue_tools import IssueTools
from gitea.tools.pr_tools import PRTools from gitea.tools.pr_tools import PRTools
from gitea.tools.file_tools import FileTools from gitea.tools.file_tools import FileTools
from gitea.tools.git_tools import GitTools from gitea.tools.git_tools import GitTools
from gitea.config import AGENT_MODEL_ID, AGENT_MAX_RETRIES from gitea.config import AGENT_MODEL_ID, AGENT_MAX_RETRIES, GITEA_REPOS_ROOT
from core.orchestrator import AgentOrchestrator from core.orchestrator import AgentOrchestrator
import json import json
@@ -73,7 +73,7 @@ async def main() -> None:
issue_tools: IssueTools = IssueTools(client) issue_tools: IssueTools = IssueTools(client)
pr_tools: PRTools = PRTools(client) pr_tools: PRTools = PRTools(client)
file_tools: FileTools = FileTools(client) file_tools: FileTools = FileTools(client, GITEA_REPOS_ROOT)
git_tools: GitTools = GitTools(client) git_tools: GitTools = GitTools(client)
model_name: str = AGENT_MODEL_ID model_name: str = AGENT_MODEL_ID
+28
View File
@@ -124,3 +124,31 @@ def test_update_file_failure() -> None:
"owner", "repo", "path/to/file", "msg", "content", "branch" "owner", "repo", "path/to/file", "msg", "content", "branch"
) )
assert "Error updating file: API Error" in res assert "Error updating file: API Error" in res
def test_get_file_content_uses_local_file_when_available(tmp_path: str) -> None:
import os
mock_client = _create_mock_client()
repo_dir = tmp_path / "owner" / "repo"
repo_dir.mkdir(parents=True)
file_path = repo_dir / "path" / "to" / "file"
file_path.parent.mkdir(parents=True)
file_path.write_text("local file content")
file_tools = FileTools(mock_client, str(tmp_path))
res = file_tools.get_file_content("owner", "repo", "path/to/file")
assert res == "1: local file content"
mock_client.files.get_file_content.assert_not_called()
def test_get_file_content_falls_back_to_api_when_no_local(tmp_path: str) -> None:
mock_client = _create_mock_client()
mock_client.files.get_file_content.return_value = "api content"
file_tools = FileTools(mock_client, str(tmp_path))
res = file_tools.get_file_content("owner", "repo", "path/to/file")
assert res == "1: api content"
mock_client.files.get_file_content.assert_called_once_with(
"owner", "repo", "path/to/file"
)