CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-localstack-core

The core library and runtime of LocalStack - a comprehensive AWS cloud emulator for local development and testing.

—

Pending
Overview
Eval results
Files

utils-testing.mddocs/

Utilities and Testing

LocalStack provides comprehensive utility modules for file operations, networking, async operations, string manipulation, time handling, and collections, along with an integrated testing framework with pytest fixtures, snapshot testing, and AWS-specific test utilities.

Capabilities

File Operations

File system utilities for cross-platform file handling with proper error handling and encoding support.

def save_file(file_path: str, content: str, append: bool = False, permissions: int = None) -> None:
    """
    Save content to file with automatic directory creation.
    
    Args:
        file_path: Target file path
        content: Content to write
        append: Whether to append to existing file
        
    Location: localstack.utils.files
    """

def load_file(file_path: str, default=None, mode: str = None) -> str:
    """
    Load file content with optional default and mode specification.
    
    Args:
        file_path: Source file path
        default: Default value if file doesn't exist
        mode: File open mode ('r', 'rb', etc.)
        
    Returns:
        File content as string or default value
        
    Location: localstack.utils.files
    """

def chmod_r(path: str, mode: int) -> None:
    """
    Recursively change file permissions.
    
    Args:
        path: Directory or file path
        mode: Permission mode (e.g., 0o755)
        
    Location: localstack.utils.files
    """

def mkdir(path: str) -> None:
    """
    Create directory with parent directories as needed.
    
    Args:
        path: Directory path to create
        
    Location: localstack.utils.files
    """

def rm_rf(path: str) -> None:
    """
    Recursively remove directory or file (equivalent to rm -rf).
    
    Args:
        path: Path to remove
        
    Location: localstack.utils.files
    """

Async Utilities

Utilities for bridging synchronous and asynchronous code execution.

async def run_sync(func: callable, *args, **kwargs):
    """
    Run synchronous function in async context using thread pool.
    
    Args:
        func: Synchronous function to execute
        *args: Positional arguments for function
        **kwargs: Keyword arguments for function
        
    Returns:
        Function result
        
    Location: localstack.utils.asyncio
    """

def run_async(coro, loop=None):
    """
    Run async coroutine in synchronous context.
    
    Args:
        coro: Coroutine to execute
        loop: Optional event loop (creates new if None)
        
    Returns:
        Coroutine result
        
    Location: localstack.utils.asyncio
    """

Network Utilities

Network connectivity and port management utilities for service communication.

def get_free_tcp_port() -> int:
    """
    Get an available TCP port on the local system.
    
    Returns:
        Available port number
        
    Location: localstack.utils.net
    """

def is_port_open(
    port_or_url: str | int, 
    http_path: str = None, 
    expect_success: bool = True,
    protocols: str | list[str] = None,
    quiet: bool = True
) -> bool:
    """
    Check if port is open and optionally test HTTP endpoint.
    
    Args:
        port_or_url: Port number or full URL to test
        http_path: Optional HTTP path for endpoint testing
        expect_success: Whether to expect successful HTTP response
        
    Returns:
        True if port is open and responsive
        
    Location: localstack.utils.net
    """

def wait_for_port_open(
    port: int, 
    http_path: str = None,
    expect_success: bool = True,
    retries: int = 10, 
    sleep_time: float = 0.5
) -> bool:
    """
    Wait for port to become available with configurable timeout.
    
    Args:
        port: Port number to wait for
        timeout: Maximum time to wait in seconds
        sleep_time: Sleep interval between checks
        
    Returns:
        True if port became available, False if timeout
        
    Location: localstack.utils.net
    """

String and Formatting

String manipulation, encoding, and ID generation utilities.

def to_str(obj, encoding: str = None, errors: str = "strict") -> str:
    """
    Convert object to string with encoding handling.
    
    Args:
        obj: Object to convert
        encoding: Character encoding (default: utf-8)
        errors: Error handling strategy
        
    Returns:
        String representation
        
    Location: localstack.utils.strings
    """

def to_bytes(obj, encoding: str = None) -> bytes:
    """
    Convert object to bytes with encoding.
    
    Args:
        obj: Object to convert
        encoding: Character encoding (default: utf-8)
        
    Returns:
        Bytes representation
        
    Location: localstack.utils.strings
    """

def short_uid(length: int = 8) -> str:
    """
    Generate short unique identifier.
    
    Args:
        length: Length of generated ID
        
    Returns:
        Random alphanumeric string
        
    Location: localstack.utils.strings
    """

def long_uid() -> str:
    """
    Generate long unique identifier (UUID format).
    
    Returns:
        UUID-style unique identifier
        
    Location: localstack.utils.strings
    """

def md5(text: str) -> str:
    """
    Generate MD5 hash of text.
    
    Args:
        text: Text to hash
        
    Returns:
        MD5 hash as hexadecimal string
        
    Location: localstack.utils.strings
    """

def base64_to_hex(b64_string: str) -> str:
    """
    Convert base64 string to hexadecimal representation.
    
    Args:
        b64_string: Base64 encoded string
        
    Returns:
        Hexadecimal representation
        
    Location: localstack.utils.strings
    """

Time and Scheduling

Time manipulation and task scheduling utilities with timezone handling.

def now_utc() -> datetime:
    """
    Get current UTC datetime.
    
    Returns:
        Current datetime in UTC timezone
        
    Location: localstack.utils.time
    """

def timestamp_millis(dt: datetime = None) -> int:
    """
    Get timestamp in milliseconds.
    
    Args:
        dt: Datetime object (defaults to current time)
        
    Returns:
        Timestamp in milliseconds since epoch
        
    Location: localstack.utils.time
    """

def parse_timestamp(ts: str | int | float, format: str = None) -> datetime:
    """
    Parse timestamp string or number to datetime.
    
    Args:
        ts: Timestamp to parse
        format: Optional format string for parsing
        
    Returns:
        Parsed datetime object
        
    Location: localstack.utils.time
    """

class Scheduler:
    """
    Task scheduler for delayed and periodic execution.
    
    Location: localstack.utils.scheduler
    """
    
    def schedule(
        self, 
        func: callable, 
        delay: float, 
        period: float = None
    ) -> object:
        """
        Schedule function execution.
        
        Args:
            func: Function to execute
            delay: Initial delay in seconds
            period: Repeat period in seconds (None for one-time)
            
        Returns:
            Scheduled task handle
        """

Collections

Collection utilities and enhanced dictionary types for data manipulation.

class AttributeDict(dict):
    """
    Dictionary with attribute-style access to keys.
    
    Supports both dict['key'] and dict.key access patterns.
    
    Location: localstack.utils.collections
    """
    
    def __getattr__(self, key: str):
        """Access dictionary key as attribute"""
        
    def __setattr__(self, key: str, value):
        """Set dictionary key as attribute"""

def select_attributes(obj, attrs: list[str]) -> dict[str, any]:
    """
    Extract specific attributes from object into dictionary.
    
    Args:
        obj: Source object
        attrs: List of attribute names to extract
        
    Returns:
        Dictionary with selected attributes
        
    Location: localstack.utils.collections
    """

def ensure_list(obj) -> list:
    """
    Ensure object is a list, wrapping single items.
    
    Args:
        obj: Object to convert to list
        
    Returns:
        List containing obj or obj itself if already a list
        
    Location: localstack.utils.collections
    """

Testing Framework

Test Configuration

Configuration management for LocalStack testing environments.

class TestConfig:
    """
    Test-specific configuration management.
    
    Provides isolated configuration for test environments
    with service-specific settings and cleanup.
    
    Location: localstack.testing.config
    """
    
    def setup_test_environment(self) -> None:
        """Setup isolated test environment"""
        
    def cleanup_test_environment(self) -> None:
        """Cleanup test resources and state"""

AWS Testing Utilities

Specialized utilities for testing AWS service integrations and functionality.

def create_test_client(service_name: str, region: str = "us-east-1") -> object:
    """
    Create AWS client configured for LocalStack testing.
    
    Args:
        service_name: AWS service name (e.g., 's3', 'lambda')
        region: AWS region for client
        
    Returns:
        Boto3 client configured for LocalStack endpoint
        
    Location: localstack.testing.aws (convenience wrapper)
    """

def wait_for_service_startup(service_name: str, timeout: int = 30) -> bool:
    """
    Wait for AWS service to become available in LocalStack.
    
    Args:
        service_name: Service to wait for
        timeout: Maximum wait time in seconds
        
    Returns:
        True if service became available
        
    Location: localstack.testing.aws
    """

def cleanup_aws_resources(services: list[str] = None) -> None:
    """
    Clean up AWS resources after tests.
    
    Args:
        services: List of services to clean (None for all)
        
    Location: localstack.testing.aws
    """

Pytest Integration

Pytest fixtures and markers for LocalStack testing integration.

# Pytest fixtures for LocalStack testing
@pytest.fixture(scope="session")
def localstack():
    """
    Session-scoped LocalStack instance for tests.
    
    Starts LocalStack before tests and stops after session ends.
    
    Location: localstack.testing.pytest
    """

@pytest.fixture
def aws_client():
    """
    Factory fixture for creating AWS clients.
    
    Returns function that creates AWS clients configured for LocalStack.
    
    Location: localstack.testing.pytest
    """

@pytest.fixture
def s3_client():
    """
    S3 client configured for LocalStack.
    
    Location: localstack.testing.pytest
    """

@pytest.fixture  
def lambda_client():
    """
    Lambda client configured for LocalStack.
    
    Location: localstack.testing.pytest
    """

# Test markers
def pytest_configure(config):
    """
    Configure pytest markers for LocalStack tests.
    
    Markers:
        - aws: Tests that require AWS services
        - slow: Long-running tests
        - integration: Integration tests
        
    Location: localstack.testing.pytest
    """

Scenario Testing

Multi-service test scenarios for complex integration testing.

class TestScenario:
    """
    Base class for multi-service test scenarios.
    
    Provides framework for testing complex interactions
    between multiple AWS services.
    
    Location: localstack.testing.scenario
    """
    
    def setup_scenario(self) -> dict[str, object]:
        """
        Setup scenario with required services and resources.
        
        Returns:
            Dictionary of initialized service clients
        """
        
    def run_scenario(self) -> dict[str, any]:
        """
        Execute the test scenario steps.
        
        Returns:
            Scenario execution results
        """
        
    def cleanup_scenario(self) -> None:
        """Clean up scenario resources"""

def run_integration_test(
    services: list[str], 
    scenario_func: callable
) -> dict[str, any]:
    """
    Run integration test across multiple services.
    
    Args:
        services: List of AWS services to initialize
        scenario_func: Test scenario function
        
    Returns:
        Test execution results
        
    Location: localstack.testing.scenario
    """

Snapshot Testing

Test snapshot management for consistent test output validation.

class SnapshotManager:
    """
    Manages test snapshots for consistent output validation.
    
    Location: localstack.testing.snapshots
    """
    
    def capture_snapshot(
        self, 
        name: str, 
        data: dict | list | str
    ) -> None:
        """
        Capture test output as named snapshot.
        
        Args:
            name: Snapshot name/identifier
            data: Data to snapshot
        """
        
    def compare_snapshot(
        self, 
        name: str, 
        data: dict | list | str
    ) -> bool:
        """
        Compare data against existing snapshot.
        
        Args:
            name: Snapshot name to compare against
            data: Current data to validate
            
        Returns:
            True if data matches snapshot
        """
        
    def update_snapshots(self, pattern: str = None) -> None:
        """
        Update snapshots with current test output.
        
        Args:
            pattern: Optional pattern to filter snapshots
        """

@pytest.fixture
def snapshot():
    """
    Pytest fixture providing snapshot functionality.
    
    Location: localstack.testing.snapshots
    """

Usage Examples

File Operations Example

from localstack.utils.files import save_file, load_file, mkdir, rm_rf

# Create directory and save configuration
mkdir("/tmp/my-app/config")
save_file("/tmp/my-app/config/settings.json", '{"debug": true}')

# Load and modify configuration
config = load_file("/tmp/my-app/config/settings.json", default="{}")
print(f"Config: {config}")

# Cleanup
rm_rf("/tmp/my-app")

Network Utilities Example

from localstack.utils.net import get_free_tcp_port, is_port_open, wait_for_port_open

# Get available port for test server
port = get_free_tcp_port()
print(f"Using port: {port}")

# Wait for LocalStack to be ready
if wait_for_port_open(4566, timeout=30):
    print("LocalStack is ready")

# Test service endpoint
if is_port_open("localhost:4566", http_path="/_localstack/health"):
    print("LocalStack health endpoint is responsive")

Testing Integration Example

import pytest
from localstack.aws.connect import connect_to

@pytest.mark.aws
def test_s3_lambda_integration(localstack):
    """Test S3 and Lambda service integration"""
    
    # Create AWS clients using factory
    s3 = connect_to().s3
    lambda_client = connect_to().lambda_
    
    # Create S3 bucket
    s3.create_bucket(Bucket="test-bucket")
    
    # Upload Lambda function code
    with open("lambda_function.zip", "rb") as f:
        lambda_client.create_function(
            FunctionName="test-function",
            Runtime="python3.9",
            Role="arn:aws:iam::123456789012:role/lambda-role",
            Handler="index.handler",
            Code={"ZipFile": f.read()}
        )
    
    # Test S3 event trigger
    s3.put_object(Bucket="test-bucket", Key="test.txt", Body=b"Hello")
    
    # Verify function was triggered
    # ... test assertions

Async Utilities Example

import asyncio
from localstack.utils.asyncio import run_sync, run_async

async def async_operation():
    await asyncio.sleep(1)
    return "async result"

def sync_operation():
    time.sleep(1) 
    return "sync result"

# Run async operation in sync context
result = run_async(async_operation())
print(result)  # "async result"

# Run sync operation in async context
async def main():
    result = await run_sync(sync_operation)
    print(result)  # "sync result"

asyncio.run(main())

Snapshot Testing Example

import pytest

@pytest.mark.aws
def test_api_response(snapshot, s3_client):
    """Test API response matches expected snapshot"""
    
    # Create bucket and object
    s3_client.create_bucket(Bucket="test-bucket")
    s3_client.put_object(Bucket="test-bucket", Key="test.txt", Body=b"data")
    
    # Get response
    response = s3_client.list_objects_v2(Bucket="test-bucket")
    
    # Compare against snapshot (excluding timestamps)
    filtered_response = {
        k: v for k, v in response.items() 
        if k not in ["ResponseMetadata"]
    }
    
    snapshot.match("list_objects_response", filtered_response)

Install with Tessl CLI

npx tessl i tessl/pypi-localstack-core

docs

aws-services.md

cli-commands.md

configuration.md

index.md

plugins-extensions.md

utils-testing.md

tile.json