Files
rag-solution/services/rag/langchain/test/test_local_filesystem_adaptive_collection.py

53 lines
1.8 KiB
Python

import os
import unittest
from pathlib import Path
from helpers import LocalFilesystemAdaptiveCollection, LocalFilesystemAdaptiveFile
class TestLocalFilesystemAdaptiveCollection(unittest.TestCase):
def setUp(self):
self.samples_dir = Path(__file__).parent / "samples"
def test_iterate_non_recursive_returns_only_root_files(self):
collection = LocalFilesystemAdaptiveCollection(str(self.samples_dir))
files = list(collection.iterate(recursive=False))
file_names = sorted(Path(file.local_path).name for file in files)
self.assertEqual(file_names, ["root.txt"])
self.assertTrue(all(isinstance(file, LocalFilesystemAdaptiveFile) for file in files))
def test_iterate_recursive_returns_nested_files(self):
collection = LocalFilesystemAdaptiveCollection(str(self.samples_dir))
files = list(collection.iterate(recursive=True))
relative_paths = sorted(
str(Path(file.local_path).relative_to(self.samples_dir)) for file in files
)
self.assertEqual(
relative_paths,
["level1/first.md", "level1/level2/second.log", "root.txt"],
)
def test_work_with_file_locally_provides_existing_path(self):
target_path = self.samples_dir / "root.txt"
adaptive_file = LocalFilesystemAdaptiveFile(target_path.suffix, str(target_path))
observed = {}
def callback(path: str):
observed["path"] = path
with open(path, "r", encoding="utf-8") as handle:
observed["content"] = handle.read().strip()
adaptive_file.work_with_file_locally(callback)
self.assertEqual(observed["path"], str(target_path))
self.assertEqual(observed["content"], "root file")
if __name__ == "__main__":
unittest.main()