tessl/pypi-langchain-tests

Standard tests for LangChain implementations

—

Pending

Overview

Eval results

Files

Configuration and Utilities

Name: tessl/pypi-langchain-tests
Author: tessl

VCR (Video Cassette Recorder) integration for HTTP call recording/playback, pytest fixtures for test isolation, Pydantic utilities, and custom serialization components that provide the testing infrastructure for the langchain-tests framework.

Capabilities

VCR Configuration

VCR integration enables recording and replaying HTTP interactions for consistent, offline testing of API-dependent components.

from langchain_tests.conftest import CustomSerializer, CustomPersister

class CustomSerializer:
    """Custom VCR cassette serializer using YAML and gzip compression."""
    
    def serialize(self, cassette_dict: dict) -> bytes:
        """
        Convert cassette dictionary to compressed YAML format.
        
        Args:
            cassette_dict: VCR cassette data structure
            
        Returns:
            bytes: Compressed YAML representation
        """
    
    def deserialize(self, data: bytes) -> dict:
        """
        Decompress and convert YAML data back to cassette dictionary.
        
        Args:
            data: Compressed YAML bytes
            
        Returns:
            dict: Reconstructed cassette data structure
        """

class CustomPersister:
    """Custom VCR persister using CustomSerializer for efficient storage."""
    
    def load_cassette(self, cassette_path: str, serializer) -> dict:
        """
        Load cassette from file using the custom serializer.
        
        Args:
            cassette_path: Path to cassette file
            serializer: Serializer instance for data conversion
            
        Returns:
            dict: Loaded cassette data
        """
    
    def save_cassette(self, cassette_path: str, cassette_dict: dict, serializer) -> None:
        """
        Save cassette to file using the custom serializer.
        
        Args:
            cassette_path: Path where cassette should be saved
            cassette_dict: Cassette data to save
            serializer: Serializer instance for data conversion
        """

Usage Example

import vcr
from langchain_tests.conftest import CustomSerializer, CustomPersister

# Configure VCR with custom serialization
my_vcr = vcr.VCR(
    serializer_name='custom',
    persister=CustomPersister(),
    cassette_library_dir='tests/cassettes'
)

# Register the custom serializer
my_vcr.register_serializer('custom', CustomSerializer())

# Use in tests
@my_vcr.use_cassette('my_api_test.yaml')
def test_api_call():
    # API calls will be recorded/replayed
    response = my_api_client.get_data()
    assert response.status_code == 200

Pytest Fixtures

Global pytest fixtures for VCR configuration and test setup.

import pytest

@pytest.fixture
def _base_vcr_config() -> dict:
    """
    Base VCR configuration with default settings.
    
    Returns:
        dict: Base VCR configuration parameters
    """

@pytest.fixture
def vcr_config(_base_vcr_config: dict) -> dict:
    """
    VCR configuration fixture that can be customized by test classes.
    
    Args:
        _base_vcr_config: Base configuration from _base_vcr_config fixture
        
    Returns:
        dict: Complete VCR configuration for test execution
    """

VCR Configuration Options

# Default VCR configuration
_BASE_VCR_CONFIG = {
    'serializer_name': 'custom',
    'persister': CustomPersister(),
    'decode_compressed_response': True,
    'record_mode': 'once',
    'match_on': ['method', 'scheme', 'host', 'port', 'path', 'query'],
    'filter_headers': _BASE_FILTER_HEADERS,
    'filter_query_parameters': ['api_key', 'access_token'],
    'filter_post_data_parameters': ['password', 'secret']
}

Header Filtering

Configuration for filtering sensitive headers from VCR cassettes.

_BASE_FILTER_HEADERS = [
    'authorization',
    'x-api-key',
    'x-auth-token',
    'cookie',
    'set-cookie',
    'x-session-id',
    'x-request-id'
]

Custom Header Filtering

# In your test class
@pytest.fixture
def vcr_config(self, _base_vcr_config):
    config = _base_vcr_config.copy()
    config['filter_headers'] = [
        *_BASE_FILTER_HEADERS,
        'x-custom-auth',
        'x-tenant-id'
    ]
    return config

Pydantic Utilities

Utilities for handling different Pydantic versions and compatibility.

from langchain_tests.utils.pydantic import get_pydantic_major_version, PYDANTIC_MAJOR_VERSION

def get_pydantic_major_version() -> int:
    """
    Detect the major version of Pydantic installed.
    
    Returns:
        int: Major version number (1 or 2)
    """

PYDANTIC_MAJOR_VERSION: int
"""Global constant containing the detected Pydantic major version."""

Pydantic Version Compatibility

from langchain_tests.utils.pydantic import PYDANTIC_MAJOR_VERSION

if PYDANTIC_MAJOR_VERSION == 1:
    # Use Pydantic v1 API
    from pydantic import BaseModel
    
    class MyModel(BaseModel):
        name: str
        
        class Config:
            extra = "forbid"

else:
    # Use Pydantic v2 API
    from pydantic import BaseModel, ConfigDict
    
    class MyModel(BaseModel):
        model_config = ConfigDict(extra="forbid")
        name: str

Test Model Generation

Utilities for generating test Pydantic models for structured output testing.

from langchain_tests.unit_tests.chat_models import (
    generate_schema_pydantic,
    generate_schema_pydantic_v1_from_2,
    TEST_PYDANTIC_MODELS
)

def generate_schema_pydantic():
    """
    Generate a Pydantic model for testing structured output.
    
    Returns:
        BaseModel: A test Pydantic model with various field types
    """

def generate_schema_pydantic_v1_from_2():
    """
    Generate a Pydantic V1 model from V2 for compatibility testing.
    
    Returns:
        BaseModel: A Pydantic V1 compatible model
    """

TEST_PYDANTIC_MODELS: List
"""List of pre-defined Pydantic models for comprehensive testing."""

Test Model Examples

# Generated test models include various field types
class GeneratedTestModel(BaseModel):
    # Basic types
    name: str
    age: int
    score: float
    active: bool
    
    # Optional fields
    description: Optional[str] = None
    
    # Collections
    tags: List[str]
    metadata: Dict[str, Any]
    
    # Nested models
    address: Address
    
    # Enums
    status: StatusEnum
    
    # Date/time fields
    created_at: datetime
    updated_at: Optional[datetime] = None

Test Constants

Pre-defined constants used throughout the testing framework.

EMBEDDING_SIZE = 6
"""Standard embedding dimension for vector store tests."""

_BASE_FILTER_HEADERS = [
    'authorization',
    'x-api-key',
    'x-auth-token',
    'cookie',
    'set-cookie'
]
"""Default headers to filter from VCR cassettes."""

Environment Variable Management

Utilities for managing test environment variables and configuration.

def get_test_env_var(var_name: str, default: str = None) -> str:
    """
    Get environment variable with test-specific prefix.
    
    Args:
        var_name: Base variable name
        default: Default value if not found
        
    Returns:
        str: Environment variable value
    """

def set_test_env_vars(env_vars: Dict[str, str]) -> None:
    """
    Set multiple test environment variables.
    
    Args:
        env_vars: Dictionary of variable names and values
    """

Environment Variable Patterns

# Test environment variable naming
TEST_API_KEY = get_test_env_var('API_KEY')
TEST_MODEL_NAME = get_test_env_var('MODEL_NAME', 'test-model')
TEST_BASE_URL = get_test_env_var('BASE_URL', 'https://api.test.com')

# Setting test environment
set_test_env_vars({
    'LANGCHAIN_TEST_API_KEY': 'test-key-123',
    'LANGCHAIN_TEST_MODEL': 'gpt-3.5-turbo',
    'LANGCHAIN_TEST_TIMEOUT': '30'
})

Fixture Utilities

Helper utilities for creating and managing pytest fixtures.

def create_model_fixture(model_class, params: dict):
    """
    Create a model fixture factory.
    
    Args:
        model_class: The model class to instantiate
        params: Parameters for model initialization
        
    Returns:
        callable: Pytest fixture function
    """

def create_temp_directory_fixture(prefix: str = 'langchain_test_'):
    """
    Create a temporary directory fixture.
    
    Args:
        prefix: Prefix for temporary directory name
        
    Returns:
        callable: Pytest fixture function that yields temp directory path
    """

Test Data Factories

Factories for generating consistent test data across different test suites.

class TestDataFactory:
    """Factory for generating consistent test data."""
    
    @staticmethod
    def create_sample_documents(count: int = 3) -> List[Document]:
        """Create sample documents for testing."""
    
    @staticmethod
    def create_sample_messages(count: int = 2) -> List[BaseMessage]:
        """Create sample messages for chat testing."""
    
    @staticmethod
    def create_sample_tools(count: int = 2) -> List[BaseTool]:
        """Create sample tools for tool testing."""
    
    @staticmethod
    def create_sample_embeddings(dimension: int = 6) -> List[List[float]]:
        """Create deterministic sample embeddings."""

Test Data Generation

Utilities for generating consistent test data and managing test state.

def get_test_documents(count: int = 3) -> List[Document]:
    """
    Generate standard test documents for testing.
    
    Args:
        count: Number of test documents to generate
        
    Returns:
        List[Document]: Generated test documents with metadata
    """

def get_test_messages(count: int = 2) -> List[BaseMessage]:
    """
    Generate standard test messages for chat testing.
    
    Args:
        count: Number of test messages to generate
        
    Returns:
        List[BaseMessage]: Generated test messages
    """

Async Test Utilities

Utilities for async test execution and management.

import asyncio

async def run_async_test_with_timeout(coro, timeout: float = 30.0):
    """
    Run async test with timeout.
    
    Args:
        coro: Coroutine to execute
        timeout: Timeout in seconds
        
    Returns:
        Any: Result of coroutine execution
    """

def async_test_fixture(async_func):
    """
    Decorator to convert async function to sync fixture.
    
    Args:
        async_func: Async function to wrap
        
    Returns:
        callable: Sync fixture function
    """

Performance Monitoring

Utilities for monitoring test performance and resource usage.

class PerformanceMonitor:
    """Monitor performance metrics during test execution."""
    
    def __init__(self):
        self.metrics = {}
    
    def start_timing(self, operation: str) -> None:
        """Start timing an operation."""
    
    def end_timing(self, operation: str) -> float:
        """End timing and return duration."""
    
    def record_memory_usage(self, label: str) -> None:
        """Record current memory usage."""
    
    def get_metrics(self) -> Dict[str, Any]:
        """Get all recorded metrics."""

The configuration and utilities module provides the essential infrastructure for reliable, consistent testing across all LangChain integration test suites, including HTTP recording/replay, environment management, data generation, and performance monitoring capabilities.

Install with Tessl CLI