test_filesystem.py — langchain Source File

Source Code

import tempfile
from collections.abc import Generator
from pathlib import Path

import pytest
from langchain_core.stores import InvalidKeyException

from langchain_classic.storage.file_system import LocalFileStore


@pytest.fixture
def file_store() -> Generator[LocalFileStore, None, None]:
    # Create a temporary directory for testing
    with tempfile.TemporaryDirectory() as temp_dir:
        # Instantiate the LocalFileStore with the temporary directory as the root path
        store = LocalFileStore(temp_dir)
        yield store


def test_mset_and_mget(file_store: LocalFileStore) -> None:
    # Set values for keys
    key_value_pairs = [("key1", b"value1"), ("key2", b"value2")]
    file_store.mset(key_value_pairs)

    # Get values for keys
    values = file_store.mget(["key1", "key2"])

    # Assert that the retrieved values match the original values
    assert values == [b"value1", b"value2"]


@pytest.mark.parametrize(
    ("chmod_dir_s", "chmod_file_s"),
    [("777", "666"), ("770", "660"), ("700", "600")],
)
def test_mset_chmod(chmod_dir_s: str, chmod_file_s: str) -> None:
    chmod_dir = int(chmod_dir_s, base=8)
    chmod_file = int(chmod_file_s, base=8)

    # Create a temporary directory for testing
    with tempfile.TemporaryDirectory() as temp_dir:
        # Instantiate the LocalFileStore with a directory inside the temporary directory
        # as the root path
        file_store = LocalFileStore(
            Path(temp_dir) / "store_dir",
            chmod_dir=chmod_dir,
            chmod_file=chmod_file,
        )

        # Set values for keys
        key_value_pairs = [("key1", b"value1"), ("key2", b"value2")]
        file_store.mset(key_value_pairs)

        # verify the permissions are set correctly
        # (test only the standard user/group/other bits)
        dir_path = file_store.root_path
        file_path = file_store.root_path / "key1"
        assert (dir_path.stat().st_mode & 0o777) == chmod_dir
        assert (file_path.stat().st_mode & 0o777) == chmod_file

// ... (99 more lines)