BaseStoreAsyncTests Class — langchain Architecture

Architecture documentation for the BaseStoreAsyncTests class in base_store.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  5af59884_0fa7_4a22_d092_189a00ad0d87["BaseStoreAsyncTests"]
  2d95e59f_5168_1366_c8c2_e21e157938b5["BaseStandardTests"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|extends| 2d95e59f_5168_1366_c8c2_e21e157938b5
  ab32a486_8161_ae81_5956_d15d97c6646a["base_store.py"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|defined in| ab32a486_8161_ae81_5956_d15d97c6646a
  6b7eed82_d349_ebeb_53b2_d502383ed185["kv_store()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| 6b7eed82_d349_ebeb_53b2_d502383ed185
  3a469d10_51d7_af73_4f11_9969448f2a41["three_values()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| 3a469d10_51d7_af73_4f11_9969448f2a41
  a27e2c2f_2ce4_4123_b634_468385bb04cc["test_three_values()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| a27e2c2f_2ce4_4123_b634_468385bb04cc
  287cc22a_b73e_a3e2_499a_a7171838ddc1["test_kv_store_is_empty()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| 287cc22a_b73e_a3e2_499a_a7171838ddc1
  5326bbfe_7009_fd5a_4298_6cf93886bb8a["test_set_and_get_values()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| 5326bbfe_7009_fd5a_4298_6cf93886bb8a
  8ce49bc5_fed2_ae7b_89b4_3078368b9d52["test_store_still_empty()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| 8ce49bc5_fed2_ae7b_89b4_3078368b9d52
  f5c957d5_a608_0e89_9e0c_7829b42c014d["test_delete_values()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| f5c957d5_a608_0e89_9e0c_7829b42c014d
  b12428ed_ba8f_75e9_658c_b4aca51c5d80["test_delete_bulk_values()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| b12428ed_ba8f_75e9_658c_b4aca51c5d80
  733a72dd_fe46_77b3_2d6d_ba351cd2ddd1["test_delete_missing_keys()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| 733a72dd_fe46_77b3_2d6d_ba351cd2ddd1
  a083b51f_fb22_761f_14a6_72ec26752d90["test_set_values_is_idempotent()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| a083b51f_fb22_761f_14a6_72ec26752d90
  c34f3e59_68ac_201b_0054_5c873d3ec706["test_get_can_get_same_value()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| c34f3e59_68ac_201b_0054_5c873d3ec706
  c4e5c97c_a6d5_3f97_56f9_8993ddf4da86["test_overwrite_values_by_key()"]
  5af59884_0fa7_4a22_d092_189a00ad0d87 -->|method| c4e5c97c_a6d5_3f97_56f9_8993ddf4da86

Relationship Graph

Source Code

libs/standard-tests/langchain_tests/integration_tests/base_store.py lines 165–315

class BaseStoreAsyncTests(BaseStandardTests, Generic[V]):
    """Test suite for checking the key-value API of a `BaseStore`.

    This test suite verifies the basic key-value API of a `BaseStore`.

    The test suite is designed for synchronous key-value stores.

    Implementers should subclass this test suite and provide a fixture
    that returns an empty key-value store for each test.
    """

    @abstractmethod
    @pytest.fixture
    async def kv_store(self) -> BaseStore[str, V]:
        """Get the key-value store class to test.

        The returned key-value store should be EMPTY.
        """

    @abstractmethod
    @pytest.fixture
    def three_values(self) -> tuple[V, V, V]:
        """Three example values that will be used in the tests."""

    async def test_three_values(self, three_values: tuple[V, V, V]) -> None:
        """Test that the fixture provides three values."""
        assert isinstance(three_values, tuple)
        assert len(three_values) == 3

    async def test_kv_store_is_empty(self, kv_store: BaseStore[str, V]) -> None:
        """Test that the key-value store is empty."""
        keys = ["foo", "bar", "buzz"]
        assert await kv_store.amget(keys) == [None, None, None]

    async def test_set_and_get_values(
        self,
        kv_store: BaseStore[str, V],
        three_values: tuple[V, V, V],
    ) -> None:
        """Test setting and getting values in the key-value store."""
        foo = three_values[0]
        bar = three_values[1]
        key_value_pairs = [("foo", foo), ("bar", bar)]
        await kv_store.amset(key_value_pairs)
        assert await kv_store.amget(["foo", "bar"]) == [foo, bar]

    async def test_store_still_empty(self, kv_store: BaseStore[str, V]) -> None:
        """Test that the store is still empty.

        This test should follow a test that sets values.

        This just verifies that the fixture is set up properly to be empty
        after each test.
        """
        keys = ["foo"]
        assert await kv_store.amget(keys) == [None]

    async def test_delete_values(
        self,
        kv_store: BaseStore[str, V],
        three_values: tuple[V, V, V],
    ) -> None:
        """Test deleting values from the key-value store."""
        foo = three_values[0]
        bar = three_values[1]
        key_value_pairs = [("foo", foo), ("bar", bar)]
        await kv_store.amset(key_value_pairs)
        await kv_store.amdelete(["foo"])
        assert await kv_store.amget(["foo", "bar"]) == [None, bar]

    async def test_delete_bulk_values(
        self,
        kv_store: BaseStore[str, V],
        three_values: tuple[V, V, V],
    ) -> None:
        """Test that we can delete several values at once."""
        foo, bar, buz = three_values
        key_values = [("foo", foo), ("bar", bar), ("buz", buz)]
        await kv_store.amset(key_values)
        await kv_store.amdelete(["foo", "buz"])
        assert await kv_store.amget(["foo", "bar", "buz"]) == [None, bar, None]