113 lines
4.3 KiB
Python
113 lines
4.3 KiB
Python
"""Project scope and resource limits: real local paths, no broad subprocesses."""
|
|
from unittest.mock import Mock
|
|
|
|
from tools.file_operations import ShellFileOperations, SearchResult, ExecuteResult
|
|
from tools.environments.local import LocalEnvironment
|
|
|
|
|
|
def projects(tmp_path):
|
|
for name in ('one', 'two'):
|
|
project = tmp_path / name
|
|
project.mkdir()
|
|
(project / '.git').mkdir()
|
|
(project / 'source.py').write_text('needle\n')
|
|
return tmp_path / 'one'
|
|
|
|
|
|
def test_ambiguous_scope_starts_no_process(tmp_path):
|
|
projects(tmp_path)
|
|
ops = ShellFileOperations(LocalEnvironment(str(tmp_path)))
|
|
ops._exec = Mock(side_effect=AssertionError('must not start process'))
|
|
result = ops.search('needle', str(tmp_path))
|
|
assert 'scope' in result.error.lower()
|
|
assert 'one' in result.error and 'two' in result.error
|
|
ops._exec.assert_not_called()
|
|
|
|
|
|
def test_umbrella_uses_cwd_project_and_reports_it(tmp_path):
|
|
project = projects(tmp_path)
|
|
ops = ShellFileOperations(LocalEnvironment(str(project)))
|
|
ops._exec = Mock(return_value=ExecuteResult(stdout='exists', exit_code=0))
|
|
ops._search_content = Mock(return_value=SearchResult())
|
|
result = ops.search('needle', str(tmp_path))
|
|
assert ops._search_content.call_args.args[1] == str(project)
|
|
assert str(project) in result.warning
|
|
|
|
|
|
def test_broad_opt_in_and_explicit_file(tmp_path):
|
|
project = projects(tmp_path)
|
|
ops = ShellFileOperations(LocalEnvironment(str(project)))
|
|
ops._exec = Mock(return_value=ExecuteResult(stdout='exists', exit_code=0))
|
|
ops._search_content = Mock(return_value=SearchResult())
|
|
for root, broad in ((tmp_path, True), (project / 'source.py', False)):
|
|
assert not ops.search('needle', str(root), allow_broad=broad).error
|
|
assert ops._search_content.call_args.args[1] == str(root)
|
|
|
|
|
|
def test_rg_commands_bound_threads_and_prune(tmp_path):
|
|
project = projects(tmp_path)
|
|
ops = ShellFileOperations(LocalEnvironment(str(project)))
|
|
ops._exec = Mock(return_value=ExecuteResult(stdout='', exit_code=1))
|
|
ops._search_with_rg('needle', str(project), '*.py', 50, 0, 'content', 0, rg_executable='rg')
|
|
command = ops._exec.call_args.args[0]
|
|
assert '--threads 2' in command
|
|
assert 'node_modules' in command and '.git' in command
|
|
assert '*.py' in command
|
|
ops._search_files_rg('*.py', str(project), 50, 0, rg_executable='rg')
|
|
command = ops._exec.call_args.args[0]
|
|
assert '--threads 2' in command and 'node_modules' in command
|
|
|
|
|
|
def test_concurrent_searches_have_two_slots_and_release_after_failure(tmp_path):
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
project = projects(tmp_path)
|
|
lock = threading.Lock()
|
|
release = threading.Event()
|
|
entered = threading.Event()
|
|
active = peak = 0
|
|
|
|
def run_search(*args):
|
|
nonlocal active, peak
|
|
with lock:
|
|
active += 1
|
|
peak = max(peak, active)
|
|
if active == 2:
|
|
entered.set()
|
|
try:
|
|
assert release.wait(5)
|
|
return SearchResult()
|
|
finally:
|
|
with lock:
|
|
active -= 1
|
|
|
|
with ThreadPoolExecutor(max_workers=6) as pool:
|
|
pending = []
|
|
for _ in range(6):
|
|
ops = ShellFileOperations(LocalEnvironment(str(project)))
|
|
ops._search_scoped = run_search
|
|
pending.append(pool.submit(ops.search, 'needle'))
|
|
try:
|
|
assert entered.wait(3)
|
|
assert peak == 2
|
|
finally:
|
|
release.set()
|
|
assert all(not task.result().error for task in pending)
|
|
assert peak == 2
|
|
|
|
|
|
def test_real_rg_project_content_and_nested_glob(tmp_path):
|
|
project = projects(tmp_path)
|
|
(project / 'src').mkdir()
|
|
(project / 'src' / 'nested.py').write_text('needle\n')
|
|
(project / 'node_modules').mkdir()
|
|
(project / 'node_modules' / 'noise.py').write_text('needle\n')
|
|
ops = ShellFileOperations(LocalEnvironment(str(project)))
|
|
result = ops.search('needle', str(tmp_path), file_glob='*.py')
|
|
assert not result.error
|
|
assert len(result.matches) == 2
|
|
assert all('node_modules' not in item.path for item in result.matches)
|
|
result = ops.search('src/*.py', str(project), target='files')
|
|
assert not result.error
|
|
assert len(result.files) == 1 and 'nested.py' in result.files[0]
|