From 139fb44facbf74d19fd6dd2ebb1f36b35f14b150 Mon Sep 17 00:00:00 2001 From: meeks Date: Sun, 19 Jul 2026 12:05:28 +0200 Subject: [PATCH] fix: FileTools uses local workspace before API for file content (#6.2) --- gitea/tools/file_tools.py | 43 ++++++++++++++++++++++++++++++++++++++- main.py | 4 ++-- tests/test_file_tools.py | 28 +++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 3 deletions(-) diff --git a/gitea/tools/file_tools.py b/gitea/tools/file_tools.py index a2be5c9..ffb9ce8 100644 --- a/gitea/tools/file_tools.py +++ b/gitea/tools/file_tools.py @@ -1,3 +1,5 @@ +import os +from pathlib import Path from typing import Any from gitea.client import GiteaClient @@ -5,8 +7,9 @@ from gitea.client import GiteaClient class FileTools: """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._repo_path: str | None = repo_path def _paginate_lines( self, @@ -35,6 +38,15 @@ class FileTools: ) 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( self, owner: str, @@ -45,10 +57,22 @@ class FileTools: ) -> str: """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: offset: 1-indexed line to start from (default 1). 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: content = self._client.files.get_file_content(owner, repo, path) raw: str = "\n".join(content) if isinstance(content, list) else content @@ -67,11 +91,28 @@ class FileTools: ) -> str: """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: ref: Branch, tag, or commit SHA (default 'master'). offset: 1-indexed line to start from (default 1). 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: content = self._client.files.get_file_content(owner, repo, path, ref) raw: str = "\n".join(content) if isinstance(content, list) else content diff --git a/main.py b/main.py index a483e71..718d7c3 100644 --- a/main.py +++ b/main.py @@ -11,7 +11,7 @@ from gitea.tools.issue_tools import IssueTools from gitea.tools.pr_tools import PRTools from gitea.tools.file_tools import FileTools 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 import json @@ -73,7 +73,7 @@ async def main() -> None: issue_tools: IssueTools = IssueTools(client) pr_tools: PRTools = PRTools(client) - file_tools: FileTools = FileTools(client) + file_tools: FileTools = FileTools(client, GITEA_REPOS_ROOT) git_tools: GitTools = GitTools(client) model_name: str = AGENT_MODEL_ID diff --git a/tests/test_file_tools.py b/tests/test_file_tools.py index 1f3d3e3..20fa4b8 100644 --- a/tests/test_file_tools.py +++ b/tests/test_file_tools.py @@ -124,3 +124,31 @@ def test_update_file_failure() -> None: "owner", "repo", "path/to/file", "msg", "content", "branch" ) 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" + )