tessl/pypi-langchain-tests
Standard tests for LangChain implementations
—
Pending
key-value-stores.mddocs/
Key-Value Store Testing
Generic testing suites for key-value store implementations with comprehensive CRUD operations, bulk operations, and both synchronous and asynchronous support. These tests ensure that key-value stores meet LangChain's storage interface requirements.
Capabilities
Synchronous Key-Value Store Testing
Comprehensive testing suite for synchronous key-value store implementations.
from typing import Generic, TypeVar, List
from langchain_tests.integration_tests import BaseStoreSyncTests
V = TypeVar('V')
class BaseStoreSyncTests(BaseStandardTests, Generic[V]):
"""Synchronous key-value store testing suite."""
# Required fixtures
@pytest.fixture
def kv_store(self):
"""Empty key-value store instance for testing. Must be implemented by test class."""
@pytest.fixture
def three_values(self) -> List[V]:
"""Three example values for testing. Must be implemented by test class."""
# Fixture validation
def test_three_values(self) -> None:
"""Validate that the three_values fixture provides correct test data."""
# Store state tests
def test_kv_store_is_empty(self) -> None:
"""Verify that the key-value store starts empty."""
def test_store_still_empty(self) -> None:
"""Verify that the store is properly cleaned up after tests."""
# Basic CRUD operations
def test_set_and_get_values(self) -> None:
"""Test setting and getting values from the store."""
def test_delete_values(self) -> None:
"""Test deleting individual values from the store."""
def test_delete_bulk_values(self) -> None:
"""Test bulk deletion of multiple values."""
def test_delete_missing_keys(self) -> None:
"""Test deletion behavior when keys don't exist."""
# Idempotency and consistency tests
def test_set_values_is_idempotent(self) -> None:
"""Test that setting the same key multiple times is idempotent."""
def test_get_can_get_same_value(self) -> None:
"""Test that getting the same key returns consistent values."""
def test_overwrite_values_by_key(self) -> None:
"""Test overwriting existing values with new ones."""
# Key iteration
def test_yield_keys(self) -> None:
"""Test iterating over all keys in the store."""Usage Example
from typing import List
import pytest
from langchain_tests.integration_tests import BaseStoreSyncTests
from my_integration import MyKeyValueStore
class TestMyKeyValueStore(BaseStoreSyncTests[str]):
@pytest.fixture
def kv_store(self):
# Create a fresh store instance for each test
store = MyKeyValueStore(
connection_url="redis://localhost:6379/1",
namespace="test_kv"
)
yield store
# Cleanup after test
store.clear()
@pytest.fixture
def three_values(self) -> List[str]:
return ["value1", "value2", "value3"]Asynchronous Key-Value Store Testing
Comprehensive testing suite for asynchronous key-value store implementations.
from typing import Generic, TypeVar, List
from langchain_tests.integration_tests import BaseStoreAsyncTests
V = TypeVar('V')
class BaseStoreAsyncTests(BaseStandardTests, Generic[V]):
"""Asynchronous key-value store testing suite."""
# Required fixtures
@pytest.fixture
async def kv_store(self):
"""Empty async key-value store instance for testing. Must be implemented by test class."""
@pytest.fixture
def three_values(self) -> List[V]:
"""Three example values for testing. Must be implemented by test class."""
# Async fixture validation
async def test_three_values(self) -> None:
"""Validate that the three_values fixture provides correct test data."""
# Async store state tests
async def test_kv_store_is_empty(self) -> None:
"""Verify that the async key-value store starts empty."""
async def test_store_still_empty(self) -> None:
"""Verify that the async store is properly cleaned up after tests."""
# Async CRUD operations
async def test_set_and_get_values(self) -> None:
"""Test async setting and getting values from the store."""
async def test_delete_values(self) -> None:
"""Test async deletion of individual values from the store."""
async def test_delete_bulk_values(self) -> None:
"""Test async bulk deletion of multiple values."""
async def test_delete_missing_keys(self) -> None:
"""Test async deletion behavior when keys don't exist."""
# Async idempotency and consistency tests
async def test_set_values_is_idempotent(self) -> None:
"""Test that async setting the same key multiple times is idempotent."""
async def test_get_can_get_same_value(self) -> None:
"""Test that async getting the same key returns consistent values."""
async def test_overwrite_values_by_key(self) -> None:
"""Test async overwriting existing values with new ones."""
# Async key iteration
async def test_yield_keys(self) -> None:
"""Test async iteration over all keys in the store."""Usage Example
from typing import List, Dict, Any
import pytest
from langchain_tests.integration_tests import BaseStoreAsyncTests
from my_integration import MyAsyncKeyValueStore
class TestMyAsyncKeyValueStore(BaseStoreAsyncTests[Dict[str, Any]]):
@pytest.fixture
async def kv_store(self):
# Create a fresh async store instance for each test
store = await MyAsyncKeyValueStore.create(
connection_url="redis://localhost:6379/1",
namespace="test_async_kv"
)
yield store
# Cleanup after test
await store.clear()
await store.close()
@pytest.fixture
def three_values(self) -> List[Dict[str, Any]]:
return [
{"id": 1, "name": "Alice", "role": "admin"},
{"id": 2, "name": "Bob", "role": "user"},
{"id": 3, "name": "Charlie", "role": "guest"}
]Generic Type Support
The key-value store testing framework supports generic types, allowing you to test stores with any value type:
String Values
class TestStringStore(BaseStoreSyncTests[str]):
@pytest.fixture
def three_values(self) -> List[str]:
return ["hello", "world", "test"]JSON/Dictionary Values
class TestJSONStore(BaseStoreSyncTests[Dict[str, Any]]):
@pytest.fixture
def three_values(self) -> List[Dict[str, Any]]:
return [
{"key": "value1", "count": 1},
{"key": "value2", "count": 2},
{"key": "value3", "count": 3}
]Custom Object Values
from dataclasses import dataclass
@dataclass
class CustomObject:
name: str
value: int
tags: List[str]
class TestCustomObjectStore(BaseStoreSyncTests[CustomObject]):
@pytest.fixture
def three_values(self) -> List[CustomObject]:
return [
CustomObject("obj1", 100, ["tag1", "tag2"]),
CustomObject("obj2", 200, ["tag2", "tag3"]),
CustomObject("obj3", 300, ["tag1", "tag3"])
]Key Generation Patterns
Key-value store implementations typically use string keys. The testing framework validates various key patterns:
Simple String Keys
def test_simple_string_keys(self):
"""Test with simple alphanumeric keys."""
keys = ["key1", "key2", "key3"]UUID Keys
import uuid
def test_uuid_keys(self):
"""Test with UUID-based keys."""
keys = [str(uuid.uuid4()) for _ in range(3)]Hierarchical Keys
def test_hierarchical_keys(self):
"""Test with hierarchical key structures."""
keys = ["user:123:profile", "user:123:settings", "user:456:profile"]Bulk Operations
The testing framework validates bulk operations for performance:
Bulk Set Operations
def test_bulk_set_operations(self) -> None:
"""Test setting multiple key-value pairs in bulk."""
def test_bulk_get_operations(self) -> None:
"""Test getting multiple values in bulk."""
def test_bulk_delete_operations(self) -> None:
"""Test deleting multiple keys in bulk."""Error Handling
Key-value store tests verify proper error handling:
Connection Errors
def test_connection_error_handling(self) -> None:
"""Test behavior when store backend is unavailable."""
def test_timeout_handling(self) -> None:
"""Test handling of operation timeouts."""Serialization Errors
def test_serialization_error_handling(self) -> None:
"""Test handling of values that cannot be serialized."""
def test_deserialization_error_handling(self) -> None:
"""Test handling of corrupted data during retrieval."""Key Validation
def test_invalid_key_handling(self) -> None:
"""Test handling of invalid key formats."""
def test_empty_key_handling(self) -> None:
"""Test handling of empty or null keys."""Performance Testing
Key-value store tests include performance benchmarks:
Latency Tests
def test_get_latency(self) -> None:
"""Benchmark get operation latency."""
def test_set_latency(self) -> None:
"""Benchmark set operation latency."""
def test_delete_latency(self) -> None:
"""Benchmark delete operation latency."""Throughput Tests
def test_bulk_operation_throughput(self) -> None:
"""Benchmark bulk operation throughput."""
def test_concurrent_access_performance(self) -> None:
"""Test performance under concurrent access."""Memory Management
Tests for memory-efficient implementations:
def test_memory_usage_patterns(self) -> None:
"""Test memory usage with large datasets."""
def test_garbage_collection_behavior(self) -> None:
"""Test proper cleanup of resources."""Concurrency Testing
For thread-safe implementations:
Thread Safety
def test_concurrent_reads(self) -> None:
"""Test concurrent read operations."""
def test_concurrent_writes(self) -> None:
"""Test concurrent write operations."""
def test_read_write_consistency(self) -> None:
"""Test consistency between concurrent reads and writes."""Lock-Free Operations
def test_lock_free_operations(self) -> None:
"""Test lock-free operation patterns."""
def test_atomic_operations(self) -> None:
"""Test atomic read-modify-write operations."""Persistence and Durability
For persistent store implementations:
Data Persistence
def test_data_persistence(self) -> None:
"""Test that data survives store restart."""
def test_crash_recovery(self) -> None:
"""Test recovery from unexpected shutdown."""Backup and Restore
def test_backup_functionality(self) -> None:
"""Test data backup capabilities."""
def test_restore_functionality(self) -> None:
"""Test data restore capabilities."""Namespace and Isolation
For multi-tenant implementations:
Namespace Separation
def test_namespace_isolation(self) -> None:
"""Test that different namespaces are isolated."""
def test_namespace_cleanup(self) -> None:
"""Test proper cleanup of namespace data."""Configuration Testing
Validate that stores respect configuration parameters:
Connection Parameters
def test_connection_string_parsing(self) -> None:
"""Test parsing of connection strings."""
def test_connection_pool_management(self) -> None:
"""Test connection pool behavior."""Serialization Options
def test_json_serialization(self) -> None:
"""Test JSON serialization of values."""
def test_pickle_serialization(self) -> None:
"""Test pickle serialization of Python objects."""
def test_custom_serialization(self) -> None:
"""Test custom serialization schemes."""The key-value store testing framework provides comprehensive validation of storage implementations, ensuring they meet LangChain's requirements for reliability, performance, and consistency across both synchronous and asynchronous usage patterns.
Install with Tessl CLI
npx tessl i tessl/pypi-langchain-tests