4fu
65832db6d8
* feat(python): migrate Python bindings from ctypes to PyO3 - Add native PyO3 extension in crates/fff-python - Replace ctypes wrapper with maturin-built package in packages/fff-python - Expose FileFinder, search/glob/directory/mixed/grep APIs and result types - Use PyPI package name fff-python (import name remains fff) - Update workspace Cargo.toml/Cargo.lock and .gitignore for Python artifacts * ci: add Python CI workflow and release wheel builds - Add .github/workflows/python.yml to test bindings on Ubuntu/macOS/Windows - Extend release.yaml with Python wheel builds (x86_64/aarch64) and sdist - Add optional PyPI publish job using trusted publishing * fix(python): address review feedback and release CI Rust bindings: - Release GIL during heavy search/grep operations via py.allow_threads - Add MixedFileItem/MixedDirItem::from_core to avoid double-cloning - Return PyDict directly from health_check and drop serde_json dependency - Call destroy() in __exit__ so the context manager releases resources - Add mode parameter to multi_grep for parity with grep - Preserve cache budget overrides across reindex() - Rename combo_boost param to match FuzzySearchOptions field Docs/tests: - Update Python test for dict-returning health_check - Add Python bindings section to main README CI: - Fix pypi-publish job to depend on build-python/build-python-sdist instead of release, making the workflow_dispatch checkbox functional * fix(python): sync release versions and expand tests * refactor(python): split lib.rs and improve Pythonic API - Split crates/fff-python/src/lib.rs into modules: - types.rs: all pyclass result types - finder.rs: FileFinder implementation - conversions.rs: From/core conversions - Make API more Pythonic: - FileFinder now accepts pathlib.Path / os.PathLike for base_path and reindex - Add close() alias for destroy() - grep/multi_grep now raise FFFException for invalid modes - GrepResult gains has_more property and next_cursor() method - Add type stubs: - packages/fff-python/src/fff/__init__.pyi - packages/fff-python/src/fff/py.typed - Add __repr__ implementations for all exposed pyclasses - Expand Python tests for pathlib, close(), reprs, invalid mode, and cursor pagination * fix(python): align type stubs and runtime API * refactor(python): polish binding API and GIL handling Make the Python binding API more idiomatic before merge: replace getter-style methods with properties, keep close() as the single explicit shutdown API, add container semantics for result objects, and tighten type stubs with Literal/Sequence/PathLike support. Also reduce Rust binding duplication with shared option/result helpers and release the Python GIL around blocking filesystem, git, and query-history operations. * feat(python): async wait_for_scan with blocking variant * fix(python): consistent frecency type and combo defaults matching node * fix(python): health_check defaults to indexed path and reports cwd errors; expand readme * chore(python): rename pypi distribution to fff-search * chore(python): align version to 0.9.4 for unified release * chore(release): bump python version with sed instead of inline python * refactor(python): use From trait for core type conversions * chore: do not run full prebuild of python wheels on PR --------- Co-authored-by: Dmitriy Kovalenko <dmtr.kovalenko@outlook.com>
392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""Tests for fff Python bindings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.metadata as metadata
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import fff
|
|
from fff import FFFException, FileFinder, GrepCursor, MixedDirItem, MixedFileItem
|
|
|
|
|
|
def rel(path: str) -> str:
|
|
return path.replace("\\", "/")
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_dir() -> str:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "src").mkdir()
|
|
(root / "docs").mkdir()
|
|
(root / "src" / "main.py").write_text(
|
|
"from utils import helper\n\n"
|
|
"def main():\n"
|
|
" value = helper()\n"
|
|
" return value\n"
|
|
)
|
|
(root / "src" / "utils.py").write_text(
|
|
"def helper():\n"
|
|
" return 'alpha'\n"
|
|
)
|
|
(root / "src" / "profile.ts").write_text(
|
|
"class Profile {}\n"
|
|
"function renderProfile() { return new Profile(); }\n"
|
|
)
|
|
(root / "docs" / "guide.txt").write_text(
|
|
"alpha line before\n"
|
|
"needle target\n"
|
|
"omega line after\n"
|
|
)
|
|
(root / "README.md").write_text("# Sample project\n")
|
|
yield str(root)
|
|
|
|
|
|
def test_imports_and_package_version() -> None:
|
|
assert fff.__version__ == metadata.version("fff-search")
|
|
assert GrepCursor(12).offset == 12
|
|
assert "GrepCursor" in fff.__all__
|
|
|
|
|
|
def test_pathlib_base_path(sample_dir: str) -> None:
|
|
with FileFinder(Path(sample_dir), watch=False, enable_content_indexing=False) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
assert finder.closed is False
|
|
assert finder.base_path is not None
|
|
assert finder.scan_progress.scanned_files_count >= 1
|
|
result = finder.search("main")
|
|
assert result.total_matched >= 1
|
|
|
|
|
|
async def test_wait_for_scan_async(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as finder:
|
|
assert await finder.wait_for_scan(timeout_ms=5000) is True
|
|
assert finder.is_scanning() is False
|
|
result = finder.search("main")
|
|
assert result.total_matched >= 1
|
|
|
|
|
|
async def test_wait_for_scan_async_does_not_block_loop(sample_dir: str) -> None:
|
|
import asyncio
|
|
|
|
ticks = 0
|
|
|
|
async def ticker() -> None:
|
|
nonlocal ticks
|
|
while True:
|
|
ticks += 1
|
|
await asyncio.sleep(0.01)
|
|
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=True) as finder:
|
|
background = asyncio.ensure_future(ticker())
|
|
try:
|
|
assert await finder.wait_for_scan(timeout_ms=5000) is True
|
|
finally:
|
|
background.cancel()
|
|
# the loop kept running other tasks while we awaited the scan
|
|
assert ticks > 0
|
|
|
|
|
|
async def test_wait_for_scan_blocking_and_async_agree(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000) is True
|
|
# already finished, so the async wait resolves immediately to True
|
|
assert await finder.wait_for_scan(timeout_ms=5000) is True
|
|
|
|
|
|
def test_keyword_only_options_and_cursor_constructor(sample_dir: str) -> None:
|
|
with pytest.raises(TypeError):
|
|
GrepCursor()
|
|
|
|
with pytest.raises(TypeError):
|
|
FileFinder(Path(sample_dir), None)
|
|
|
|
with FileFinder(Path(sample_dir), watch=False, enable_content_indexing=True) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
with pytest.raises(TypeError):
|
|
finder.search("main", None)
|
|
with pytest.raises(TypeError):
|
|
finder.glob("*.py", None)
|
|
with pytest.raises(TypeError):
|
|
finder.directory_search("src", None)
|
|
with pytest.raises(TypeError):
|
|
finder.mixed_search("src", None)
|
|
with pytest.raises(TypeError):
|
|
finder.grep("needle", "plain")
|
|
with pytest.raises(TypeError):
|
|
finder.multi_grep(["needle"], None)
|
|
|
|
|
|
def test_close_and_context_manager(sample_dir: str) -> None:
|
|
finder = FileFinder(sample_dir, watch=False, enable_content_indexing=False)
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
assert finder.closed is False
|
|
assert finder.base_path is not None
|
|
finder.close()
|
|
assert finder.closed is True
|
|
|
|
with pytest.raises(FFFException, match="File picker not initialized"):
|
|
finder.search("main")
|
|
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as ctx_finder:
|
|
assert ctx_finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
assert ctx_finder.closed is False
|
|
|
|
assert ctx_finder.closed is True
|
|
with pytest.raises(FFFException, match="File picker not initialized"):
|
|
ctx_finder.search("main")
|
|
|
|
fresh = FileFinder(sample_dir, watch=False, enable_content_indexing=False)
|
|
assert fresh.wait_for_scan_blocking(timeout_ms=5000)
|
|
fresh.close()
|
|
assert fresh.closed is True
|
|
with pytest.raises(FFFException, match="File picker not initialized"):
|
|
fresh.search("main")
|
|
|
|
|
|
def test_reprs(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
assert repr(finder).startswith("FileFinder(")
|
|
result = finder.search("main")
|
|
assert repr(result).startswith("SearchResult(")
|
|
assert repr(result.items[0]).startswith("FileItem(")
|
|
assert repr(result.scores[0]).startswith("Score(")
|
|
|
|
grep_result = finder.grep("needle")
|
|
assert repr(grep_result).startswith("GrepResult(")
|
|
assert repr(grep_result.items[0]).startswith("GrepMatch(")
|
|
assert repr(grep_result.items[0].match_ranges[0]).startswith("MatchRange(")
|
|
|
|
dir_result = finder.directory_search("src")
|
|
assert repr(dir_result).startswith("DirSearchResult(")
|
|
assert repr(dir_result.items[0]).startswith("DirItem(")
|
|
|
|
mixed = finder.mixed_search("src", page_size=10)
|
|
assert repr(mixed).startswith("MixedSearchResult(")
|
|
|
|
cursor = GrepCursor(42)
|
|
assert repr(cursor) == "GrepCursor(offset=42)"
|
|
|
|
progress = finder.scan_progress
|
|
assert repr(progress).startswith("ScanProgress(")
|
|
|
|
|
|
def test_file_search_scores_and_pagination(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
result = finder.search("main", page_size=1)
|
|
assert result.total_matched >= 1
|
|
assert len(result) == len(result.items) == 1
|
|
assert bool(result) is True
|
|
assert any("main.py" in rel(item.relative_path) for item in result.items)
|
|
|
|
score = result.scores[0]
|
|
assert isinstance(score.total, int)
|
|
assert isinstance(score.exact_match, bool)
|
|
assert isinstance(score.match_type, str)
|
|
|
|
second_page = finder.search("", page_index=1, page_size=1)
|
|
assert len(second_page) == len(second_page.items) == 1
|
|
|
|
empty = finder.search("definitely_no_such_file_xyz")
|
|
assert len(empty) == 0
|
|
assert bool(empty) is False
|
|
|
|
|
|
def test_glob_variants(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
py_files = finder.glob("*.py")
|
|
assert {Path(rel(item.relative_path)).name for item in py_files.items} == {
|
|
"main.py",
|
|
"utils.py",
|
|
}
|
|
|
|
src_files = finder.glob("src/*.py")
|
|
assert src_files.total_matched == 2
|
|
|
|
md_files = finder.glob("*.md")
|
|
assert md_files.total_matched == 1
|
|
assert rel(md_files.items[0].relative_path) == "README.md"
|
|
|
|
|
|
def test_directory_and_mixed_search(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=False) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
dirs = finder.directory_search("src")
|
|
assert dirs.total_matched >= 1
|
|
assert len(dirs) == len(dirs.items)
|
|
assert bool(dirs) is True
|
|
assert any(rel(item.relative_path).startswith("src") for item in dirs.items)
|
|
|
|
mixed = finder.mixed_search("src", page_size=10)
|
|
assert mixed.total_matched >= 3
|
|
assert len(mixed) == len(mixed.items)
|
|
assert bool(mixed) is True
|
|
assert any(isinstance(item, MixedDirItem) for item in mixed.items)
|
|
assert any(isinstance(item, MixedFileItem) for item in mixed.items)
|
|
|
|
# max_access_frecency is the same field on both dir item types, so the
|
|
# values must agree for a shared directory regardless of which search
|
|
# produced them.
|
|
dir_frecency = {
|
|
rel(item.relative_path): item.max_access_frecency for item in dirs.items
|
|
}
|
|
for item in mixed.items:
|
|
if isinstance(item, MixedDirItem):
|
|
path = rel(item.relative_path)
|
|
if path in dir_frecency:
|
|
assert item.max_access_frecency == dir_frecency[path]
|
|
|
|
|
|
def test_grep_plain_regex_fuzzy_and_context(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=True) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
plain = finder.grep("needle", before_context=1, after_context=1)
|
|
assert plain.total_matched == 1
|
|
assert len(plain) == len(plain.items) == 1
|
|
assert bool(plain) is True
|
|
match = plain.items[0]
|
|
assert rel(match.relative_path) == "docs/guide.txt"
|
|
assert match.line_content == "needle target"
|
|
assert match.context_before == ["alpha line before"]
|
|
assert match.context_after == ["omega line after"]
|
|
assert [(r.start, r.end) for r in match.match_ranges] == [(0, 6)]
|
|
|
|
regex = finder.grep(r"def \w+", mode="regex")
|
|
assert regex.total_matched == 2
|
|
assert {Path(rel(m.relative_path)).name for m in regex.items} == {
|
|
"main.py",
|
|
"utils.py",
|
|
}
|
|
|
|
fuzzy = finder.grep("df mn", mode="fuzzy")
|
|
assert fuzzy.total_matched >= 1
|
|
assert any(rel(m.relative_path) == "src/main.py" for m in fuzzy.items)
|
|
assert any(m.fuzzy_score is not None for m in fuzzy.items)
|
|
|
|
invalid = finder.grep("[", mode="regex")
|
|
assert invalid.regex_fallback_error is not None
|
|
|
|
|
|
def test_grep_invalid_mode_raises(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=True) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
with pytest.raises(FFFException, match="invalid grep mode"):
|
|
finder.grep("needle", mode="typo")
|
|
with pytest.raises(FFFException, match="invalid grep mode"):
|
|
finder.multi_grep(["needle"], mode="typo")
|
|
|
|
|
|
def test_grep_cursor_paginates_by_file(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=True) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
first = finder.grep("def", page_limit=1)
|
|
assert first.total_matched >= 1
|
|
assert first.next_file_offset > 0
|
|
assert first.has_more is True
|
|
assert first.next_cursor() is not None
|
|
assert first.next_cursor().offset == first.next_file_offset
|
|
|
|
second = finder.grep("def", cursor=GrepCursor(first.next_file_offset), page_limit=1)
|
|
assert second.total_matched >= 1
|
|
|
|
first_paths = {rel(m.relative_path) for m in first.items}
|
|
second_paths = {rel(m.relative_path) for m in second.items}
|
|
assert first_paths.isdisjoint(second_paths)
|
|
|
|
exhausted = finder.grep("nonexistent_xyz")
|
|
assert exhausted.has_more is False
|
|
assert exhausted.next_cursor() is None
|
|
assert len(exhausted) == 0
|
|
assert bool(exhausted) is False
|
|
|
|
|
|
def test_multi_grep_and_error_handling(sample_dir: str) -> None:
|
|
with FileFinder(sample_dir, watch=False, enable_content_indexing=True) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
result = finder.multi_grep(("def main", "def helper"))
|
|
assert result.total_matched == 2
|
|
assert {Path(rel(m.relative_path)).name for m in result.items} == {
|
|
"main.py",
|
|
"utils.py",
|
|
}
|
|
|
|
with pytest.raises(FFFException, match="patterns must not be empty"):
|
|
finder.multi_grep([])
|
|
|
|
|
|
def test_query_history_persists(sample_dir: str, tmp_path: Path) -> None:
|
|
history_db = tmp_path / "history"
|
|
selected_file = Path(sample_dir) / "src" / "main.py"
|
|
|
|
with FileFinder(
|
|
sample_dir,
|
|
history_db_path=history_db,
|
|
watch=False,
|
|
enable_content_indexing=False,
|
|
) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
assert finder.track_query("main", selected_file)
|
|
assert finder.get_historical_query(0) == "main"
|
|
|
|
with FileFinder(
|
|
sample_dir,
|
|
history_db_path=history_db,
|
|
watch=False,
|
|
enable_content_indexing=False,
|
|
) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
assert finder.get_historical_query(0) == "main"
|
|
|
|
|
|
def test_reindex_and_health_check(sample_dir: str, tmp_path: Path) -> None:
|
|
other = tmp_path / "other-project"
|
|
other.mkdir()
|
|
(other / "other.py").write_text("def other():\n return 42\n")
|
|
|
|
frecency_db = tmp_path / "frecency"
|
|
history_db = tmp_path / "history"
|
|
with FileFinder(
|
|
sample_dir,
|
|
frecency_db_path=frecency_db,
|
|
history_db_path=history_db,
|
|
watch=False,
|
|
enable_content_indexing=False,
|
|
) as finder:
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
|
|
health = finder.health_check(Path(sample_dir))
|
|
assert health["file_picker"]["initialized"] is True
|
|
assert health["frecency"]["initialized"] is True
|
|
assert health["query_tracker"]["initialized"] is True
|
|
|
|
# With no explicit path, the check inspects the indexed base path
|
|
# rather than the process cwd.
|
|
default_health = finder.health_check()
|
|
assert default_health["file_picker"]["base_path"] is not None
|
|
assert "error" not in default_health.get("git", {}) or default_health["git"][
|
|
"error"
|
|
] != "could not determine current directory"
|
|
|
|
finder.reindex(str(other))
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
result = finder.search("other")
|
|
assert result.total_matched == 1
|
|
assert rel(result.items[0].relative_path) == "other.py"
|
|
|
|
finder.reindex(Path(other))
|
|
assert finder.wait_for_scan_blocking(timeout_ms=5000)
|
|
result2 = finder.search("other")
|
|
assert result2.total_matched == 1
|