wehub-resource-sync
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
"""Regression tests for leann watch sync scope (#345)."""
|
|
|
|
from leann.cli import LeannCLI
|
|
from leann.sync import FileSynchronizer
|
|
|
|
|
|
def test_resolve_sync_scope_keeps_loose_files_separate(tmp_path):
|
|
repo = tmp_path / "repo"
|
|
src = repo / "src"
|
|
src.mkdir(parents=True)
|
|
readme = repo / "README.md"
|
|
readme.write_text("# hello", encoding="utf-8")
|
|
|
|
cli = LeannCLI()
|
|
directories, files = cli._resolve_sync_scope([str(src), str(readme)])
|
|
|
|
assert directories == [str(src.resolve())]
|
|
assert files == [str(readme.resolve())]
|
|
|
|
|
|
def test_watch_scope_does_not_scan_sibling_media(tmp_path):
|
|
repo = tmp_path / "repo"
|
|
src = repo / "src"
|
|
assets = repo / "assets"
|
|
src.mkdir(parents=True)
|
|
assets.mkdir()
|
|
readme = repo / "README.md"
|
|
readme.write_text("# hello", encoding="utf-8")
|
|
(src / "main.py").write_text("print('ok')", encoding="utf-8")
|
|
(assets / "icon.png").write_bytes(b"png")
|
|
|
|
synchronizers = [
|
|
FileSynchronizer(
|
|
root_dir=str(src),
|
|
include_extensions=[".py", ".md"],
|
|
snapshot_path=str(tmp_path / "sync_src.pickle"),
|
|
auto_load=False,
|
|
),
|
|
FileSynchronizer(
|
|
explicit_files=[str(readme.resolve())],
|
|
include_extensions=[".py", ".md"],
|
|
snapshot_path=str(tmp_path / "sync_readme.pickle"),
|
|
auto_load=False,
|
|
),
|
|
]
|
|
|
|
hashed_paths: set[str] = set()
|
|
for fs in synchronizers:
|
|
hashed_paths.update(fs.generate_file_hashes().keys())
|
|
|
|
assert str((src / "main.py").resolve()) in hashed_paths
|
|
assert str(readme.resolve()) in hashed_paths
|
|
assert str((assets / "icon.png").resolve()) not in hashed_paths
|
|
|
|
|
|
def test_mixed_txt_and_bin_directory_skips_bin_without_crash(tmp_path):
|
|
"""Same dir with .txt and .bin: hash only text, ignore binary (review #377)."""
|
|
docs = tmp_path / "docs"
|
|
docs.mkdir()
|
|
txt = docs / "notes.txt"
|
|
bin_file = docs / "payload.bin"
|
|
txt.write_text("hello", encoding="utf-8")
|
|
bin_file.write_bytes(bytes(range(256)))
|
|
|
|
fs = FileSynchronizer(
|
|
root_dir=str(docs),
|
|
include_extensions=[".txt"],
|
|
snapshot_path=str(tmp_path / "sync.pickle"),
|
|
auto_load=False,
|
|
)
|
|
|
|
hashes = fs.generate_file_hashes()
|
|
assert set(hashes.keys()) == {str(txt.resolve())}
|
|
assert str(bin_file.resolve()) not in hashes
|
|
|
|
fs.create_snapshot()
|
|
fs2 = FileSynchronizer(
|
|
root_dir=str(docs),
|
|
include_extensions=[".txt"],
|
|
snapshot_path=str(tmp_path / "sync.pickle"),
|
|
)
|
|
added, removed, modified = fs2.detect_changes()
|
|
assert not added and not removed and not modified
|