Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 60 additions & 2 deletions tests/tools/test_file_operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
normalize_read_pagination,
normalize_search_pagination,
)
from tools.environments.local import LocalEnvironment
from tools.environments.ssh import SSHEnvironment


# =========================================================================
Expand Down Expand Up @@ -230,8 +232,8 @@ def test_error(self):

@pytest.fixture()
def mock_env():
"""Create a mock terminal environment."""
env = MagicMock()
"""Create a mock local terminal environment."""
env = MagicMock(spec=LocalEnvironment)
env.cwd = "/tmp/test"
env.execute.return_value = {"output": "", "returncode": 0}
return env
Expand Down Expand Up @@ -836,6 +838,62 @@ def test_msys_path_translated_back_to_native(self, mock_env, monkeypatch):
out = ops._escape_native_tool_arg("/c/Users/alice/project")
assert out == "'C:/Users/alice/project'"

def test_remote_posix_path_is_not_converted_on_windows_controller(self, monkeypatch):
"""A Windows controller must preserve paths sent to an SSH backend."""
import tools.environments.local as local_mod

monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
remote_env = MagicMock(spec=SSHEnvironment)
remote_env.cwd = "/c/remote/project"
commands = []

def side_effect(command, **kwargs):
commands.append(command)
if "test -e" in command:
return {"output": "exists", "returncode": 0}
if "command -v" in command:
return {"output": "yes", "returncode": 0}
return {"output": "", "returncode": 0}

remote_env.execute.side_effect = side_effect
ops = ShellFileOperations(remote_env)
result = ops.search("needle", path="/c/remote/project", target="files")

assert result.error is None
rg_commands = [c for c in commands if c.startswith("rg ")]
assert rg_commands, f"no rg command captured in: {commands}"
assert all("'/c/remote/project'" in c for c in rg_commands)
assert all("'C:/remote/project'" not in c for c in rg_commands)

@pytest.mark.windows_only
def test_local_windows_search_keeps_native_path_spellings(self, monkeypatch):
"""The #84378 local boundary remains native for both spellings."""
import tools.environments.local as local_mod

monkeypatch.setattr(local_mod, "_IS_WINDOWS", True)
local_env = MagicMock(spec=LocalEnvironment)
local_env.cwd = "/tmp/test"
commands = []

def side_effect(command, **kwargs):
commands.append(command)
if "test -e" in command:
return {"output": "exists", "returncode": 0}
if "command -v" in command:
return {"output": "yes", "returncode": 0}
return {"output": "", "returncode": 0}

local_env.execute.side_effect = side_effect
ops = ShellFileOperations(local_env)
for path in (r"C:\Users\fixture\target", "C:/Users/fixture/target"):
commands.clear()
result = ops.search("needle", path=path, target="files")
assert result.error is None
rg_commands = [c for c in commands if c.startswith("rg ")]
assert rg_commands, f"no rg command captured in: {commands}"
assert any("'C:/Users/fixture/target'" in c for c in rg_commands)
assert all("'/c/Users/fixture/target'" not in c for c in rg_commands)

def test_posix_path_untouched_on_windows(self, mock_env, monkeypatch):
"""Multi-segment POSIX paths (/home/x, /tmp/y) are not drive paths."""
import tools.environments.local as local_mod
Expand Down
55 changes: 55 additions & 0 deletions tests/tools/test_file_tools_live.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@


import os
import shutil
import sys
from pathlib import Path

Expand Down Expand Up @@ -213,6 +214,60 @@ def test_content_search_finds_exact_match(self, ops, populated_dir):
_assert_clean(m.path)


class TestWindowsNativeRgLiveSearch:
"""Real local Windows rg search regression for the #84378 path boundary."""

@pytest.mark.windows_only
def test_real_local_search_uses_native_temp_path(self, tmp_path):
"""Exercise LocalEnvironment -> rg through public search operations."""
if shutil.which("rg") is None:
pytest.skip("native rg is not available on PATH")

real_env = LocalEnvironment(cwd=str(tmp_path), timeout=15)
version = real_env.execute("rg --version")
if version["returncode"] != 0:
pytest.skip("rg is not executable through the local Windows shell")

(tmp_path / "native-rg-live.txt").write_text(
"NATIVE_RG_LIVE_TOKEN\n", encoding="utf-8"
)
(tmp_path / "other.txt").write_text("unrelated content\n", encoding="utf-8")

class RecordingLocalEnvironment(LocalEnvironment):
"""Test-only delegating recorder; production still uses LocalEnvironment."""

def __init__(self, delegate):
self.delegate = delegate
self.cwd = delegate.cwd
self.outputs = []

def execute(self, command, **kwargs):
result = self.delegate.execute(command, **kwargs)
self.outputs.append(result.get("output", ""))
return result

recording_env = RecordingLocalEnvironment(real_env)
ops = ShellFileOperations(recording_env, cwd=str(tmp_path))

content = ops.search(
"NATIVE_RG_LIVE_TOKEN", path=str(tmp_path), target="content"
)
files = ops.search(
"native-rg-live.txt", path=str(tmp_path), target="files"
)
zero = ops.search(
"NATIVE_RG_LIVE_DEFINITELY_ABSENT", path=str(tmp_path), target="content"
)

assert content.error is None
assert content.total_count >= 1
assert any("NATIVE_RG_LIVE_TOKEN" in m.content for m in content.matches)
assert files.error is None
assert any("native-rg-live.txt" in p for p in files.files)
assert zero.error is None
assert zero.total_count == 0
assert all("os error 3" not in output.lower() for output in recording_env.outputs)

def test_search_output_has_zero_noise(self, ops, populated_dir):
"""Dedicated noise check: search must return only real content."""
result = ops.search("func", str(populated_dir), target="content")
Expand Down
14 changes: 10 additions & 4 deletions tools/file_operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -1166,11 +1166,17 @@ def _escape_native_tool_arg(self, arg: str) -> str:

On non-Windows hosts this is exactly ``_escape_shell_arg``.
"""
from tools.environments.local import _IS_WINDOWS, _msys_to_windows_path

if _IS_WINDOWS and arg:
from tools.environments.local import LocalEnvironment, _IS_WINDOWS, _msys_to_windows_path

# The controller's OS is not the execution domain. A Windows Hermes
# process can drive SSH/Docker/other POSIX backends, where ``/c/...``
# is a remote POSIX path and must not be reverse-converted. Only the
# local backend invokes a native Windows binary against the host
# filesystem, so keep the #84378 conversion at that boundary.
if _IS_WINDOWS and isinstance(self.env, LocalEnvironment) and arg:
arg = _msys_to_windows_path(arg).replace("\\", "/")
return "'" + arg.replace("'", "'\"'\"'") + "'"
return "'" + arg.replace("'", "'\"'\"'") + "'"
return self._escape_shell_arg(arg)

def _atomic_write(self, path: str, content: str) -> "ExecuteResult":
"""Write ``content`` to ``path`` atomically via temp-file + rename.
Expand Down