Index

provide.testkit.chaos

Chaos testing utilities for property-based testing with Hypothesis.

This module provides reusable Hypothesis strategies and fixtures for chaos engineering in tests. It enables systematic exploration of edge cases, race conditions, and failure scenarios that are difficult to test with traditional methods.

Key Features

• Property-based testing strategies for common chaos patterns
• Time manipulation and clock skew simulation
• Concurrency and race condition triggers
• I/O failure injection patterns
• Reusable pytest fixtures for chaos testing

Example

from hypothesis import given
from provide.testkit.chaos import chaos_timings, failure_patterns

@given(
    timing=chaos_timings(),
    failures=failure_patterns()
)
async def test_with_chaos(timing, failures):
    # Your chaos test here
    pass

Functions

async_event_patterns

async_event_patterns(
    draw: DrawFn, max_events: int = 50
) -> list[dict[str, Any]]

Generate async event patterns for coroutine testing.

Creates patterns of async events, delays, and scheduling scenarios.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_events	int	Maximum number of events to generate	50

Returns:

Type	Description
list[dict[str, Any]]	List of event dictionaries

Example

@given(events=async_event_patterns())
async def test_async_handling(events):
    for event in events:
        if event['type'] == 'delay':
            await asyncio.sleep(event['duration'])
        elif event['type'] == 'task':
            await asyncio.create_task(event['coro']())

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def async_event_patterns(
    draw: DrawFn,
    max_events: int = 50,
) -> list[dict[str, Any]]:
    """Generate async event patterns for coroutine testing.

    Creates patterns of async events, delays, and scheduling scenarios.

    Args:
        draw: Hypothesis draw function
        max_events: Maximum number of events to generate

    Returns:
        List of event dictionaries

    Example:
        ```python
        @given(events=async_event_patterns())
        async def test_async_handling(events):
            for event in events:
                if event['type'] == 'delay':
                    await asyncio.sleep(event['duration'])
                elif event['type'] == 'task':
                    await asyncio.create_task(event['coro']())
        ```
    """
    num_events = draw(st.integers(min_value=1, max_value=max_events))

    events = []
    for _ in range(num_events):
        event_type = draw(st.sampled_from(["delay", "immediate", "cancel", "timeout"]))

        event: dict[str, Any] = {"type": event_type}

        if event_type == "delay":
            event["duration"] = draw(st.floats(min_value=0.0, max_value=1.0))
        elif event_type == "timeout":
            event["timeout"] = draw(st.floats(min_value=0.01, max_value=2.0))
        elif event_type == "cancel":
            event["after_delay"] = draw(st.floats(min_value=0.0, max_value=0.5))

        events.append(event)

    return events

buffer_overflow_patterns

buffer_overflow_patterns(
    draw: DrawFn, max_buffer_size: int = 1024 * 1024
) -> dict[str, Any]

Generate buffer overflow scenarios.

Creates situations where data exceeds buffer capacity.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_buffer_size	int	Maximum buffer size in bytes	1024 * 1024

Returns:

Type	Description
dict[str, Any]	Dictionary containing buffer overflow configuration

Example

@given(config=buffer_overflow_patterns())
def test_buffer_handling(config):
    buffer = bytearray(config['buffer_size'])
    # Attempt to write config['data_size'] bytes

Source code in provide/testkit/chaos/io_strategies.py

@composite
def buffer_overflow_patterns(
    draw: DrawFn,
    max_buffer_size: int = 1024 * 1024,  # 1MB
) -> dict[str, Any]:
    """Generate buffer overflow scenarios.

    Creates situations where data exceeds buffer capacity.

    Args:
        draw: Hypothesis draw function
        max_buffer_size: Maximum buffer size in bytes

    Returns:
        Dictionary containing buffer overflow configuration

    Example:
        ```python
        @given(config=buffer_overflow_patterns())
        def test_buffer_handling(config):
            buffer = bytearray(config['buffer_size'])
            # Attempt to write config['data_size'] bytes
        ```
    """
    buffer_size = draw(st.integers(min_value=64, max_value=max_buffer_size))

    # Data size might exceed buffer
    overflow = draw(st.booleans())
    if overflow:
        data_size = draw(st.integers(min_value=buffer_size + 1, max_value=buffer_size * 2))
    else:
        data_size = draw(st.integers(min_value=0, max_value=buffer_size))

    return {
        "buffer_size": buffer_size,
        "data_size": data_size,
        "will_overflow": data_size > buffer_size,
        "overflow_bytes": max(0, data_size - buffer_size),
        "chunk_size": draw(st.integers(min_value=1, max_value=min(buffer_size, 8192))),
    }

chaos_timings

chaos_timings(
    draw: DrawFn,
    min_value: float = 0.001,
    max_value: float = 10.0,
    allow_zero: bool = False,
) -> float

Generate chaotic timing values for unpredictable delays.

Useful for testing timeout handling, race conditions, and time-sensitive code.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_value	float	Minimum timing value in seconds	0.001
max_value	float	Maximum timing value in seconds	10.0
allow_zero	bool	Whether to allow zero delay	False

Returns:

Type	Description
float	A floating-point timing value in seconds

Example

@given(delay=chaos_timings())
async def test_with_delay(delay):
    await asyncio.sleep(delay)

Source code in provide/testkit/chaos/strategies.py

@composite
def chaos_timings(
    draw: DrawFn,
    min_value: float = 0.001,
    max_value: float = 10.0,
    allow_zero: bool = False,
) -> float:
    """Generate chaotic timing values for unpredictable delays.

    Useful for testing timeout handling, race conditions, and time-sensitive code.

    Args:
        draw: Hypothesis draw function
        min_value: Minimum timing value in seconds
        max_value: Maximum timing value in seconds
        allow_zero: Whether to allow zero delay

    Returns:
        A floating-point timing value in seconds

    Example:
        ```python
        @given(delay=chaos_timings())
        async def test_with_delay(delay):
            await asyncio.sleep(delay)
        ```
    """
    min_val = 0.0 if allow_zero else min_value
    return draw(st.floats(min_value=min_val, max_value=max_value, allow_nan=False, allow_infinity=False))

clock_skew

clock_skew(
    draw: DrawFn, max_skew: float = 300.0
) -> dict[str, Any]

Generate clock skew scenarios for distributed system testing.

Simulates clock drift, NTP sync issues, and timezone problems.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_skew	float	Maximum clock skew in seconds	300.0

Returns:

Type	Description
dict[str, Any]	Dictionary containing skew configuration

Example

@given(skew=clock_skew())
def test_with_clock_skew(skew):
    # Simulate clock skew between components
    pass

Source code in provide/testkit/chaos/time_strategies.py

@composite
def clock_skew(
    draw: DrawFn,
    max_skew: float = 300.0,
) -> dict[str, Any]:
    """Generate clock skew scenarios for distributed system testing.

    Simulates clock drift, NTP sync issues, and timezone problems.

    Args:
        draw: Hypothesis draw function
        max_skew: Maximum clock skew in seconds

    Returns:
        Dictionary containing skew configuration

    Example:
        ```python
        @given(skew=clock_skew())
        def test_with_clock_skew(skew):
            # Simulate clock skew between components
            pass
        ```
    """
    return {
        "skew_seconds": draw(st.floats(min_value=-max_skew, max_value=max_skew)),
        "drift_rate": draw(st.floats(min_value=0.0001, max_value=0.01)),  # seconds per second
        "has_backwards_jump": draw(st.booleans()),
        "sync_interval": draw(st.floats(min_value=1.0, max_value=3600.0)),
    }

deadline_scenarios

deadline_scenarios(
    draw: DrawFn,
    min_deadline: float = 0.1,
    max_deadline: float = 10.0,
) -> dict[str, Any]

Generate deadline-based test scenarios.

Creates scenarios with operations that may or may not meet their deadlines.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_deadline	float	Minimum deadline in seconds	0.1
max_deadline	float	Maximum deadline in seconds	10.0

Returns:

Type	Description
dict[str, Any]	Dictionary containing deadline scenario configuration

Example

@given(scenario=deadline_scenarios())
async def test_deadline_handling(scenario):
    deadline = scenario['deadline']
    work_duration = scenario['work_duration']
    # Test if deadline handling works correctly

Source code in provide/testkit/chaos/time_strategies.py

@composite
def deadline_scenarios(
    draw: DrawFn,
    min_deadline: float = 0.1,
    max_deadline: float = 10.0,
) -> dict[str, Any]:
    """Generate deadline-based test scenarios.

    Creates scenarios with operations that may or may not meet their deadlines.

    Args:
        draw: Hypothesis draw function
        min_deadline: Minimum deadline in seconds
        max_deadline: Maximum deadline in seconds

    Returns:
        Dictionary containing deadline scenario configuration

    Example:
        ```python
        @given(scenario=deadline_scenarios())
        async def test_deadline_handling(scenario):
            deadline = scenario['deadline']
            work_duration = scenario['work_duration']
            # Test if deadline handling works correctly
        ```
    """
    deadline = draw(st.floats(min_value=min_deadline, max_value=max_deadline))

    # Generate work duration that may exceed deadline
    exceeds_deadline = draw(st.booleans())
    if exceeds_deadline:
        work_duration = draw(st.floats(min_value=deadline, max_value=deadline * 2))
    else:
        work_duration = draw(st.floats(min_value=0.0, max_value=deadline * 0.9))

    return {
        "deadline": deadline,
        "work_duration": work_duration,
        "exceeds_deadline": exceeds_deadline,
        "grace_period": draw(st.floats(min_value=0.0, max_value=1.0)),
    }

deadlock_scenarios

deadlock_scenarios(
    draw: DrawFn, num_resources: int = 5
) -> dict[str, Any]

Generate resource lock patterns that may cause deadlocks.

Creates scenarios where circular dependencies between locks might occur.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
num_resources	int	Number of lockable resources	5

Returns:

Type	Description
dict[str, Any]	Dictionary containing deadlock scenario configuration

Example

@given(scenario=deadlock_scenarios())
def test_deadlock_prevention(scenario):
    # Attempt to acquire locks in the pattern
    # System should prevent deadlock
    pass

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def deadlock_scenarios(
    draw: DrawFn,
    num_resources: int = 5,
) -> dict[str, Any]:
    """Generate resource lock patterns that may cause deadlocks.

    Creates scenarios where circular dependencies between locks might occur.

    Args:
        draw: Hypothesis draw function
        num_resources: Number of lockable resources

    Returns:
        Dictionary containing deadlock scenario configuration

    Example:
        ```python
        @given(scenario=deadlock_scenarios())
        def test_deadlock_prevention(scenario):
            # Attempt to acquire locks in the pattern
            # System should prevent deadlock
            pass
        ```
    """
    num_threads = draw(st.integers(min_value=2, max_value=10))

    # Each thread gets a sequence of resources to lock
    lock_sequences = []
    for _ in range(num_threads):
        num_locks = draw(st.integers(min_value=1, max_value=num_resources))
        # Generate a permutation of resource IDs
        sequence = draw(
            st.lists(
                st.integers(min_value=0, max_value=num_resources - 1),
                min_size=num_locks,
                max_size=num_locks,
                unique=True,
            )
        )
        lock_sequences.append(sequence)

    return {
        "num_resources": num_resources,
        "num_threads": num_threads,
        "lock_sequences": lock_sequences,
        "has_timeout": draw(st.booleans()),
        "timeout": draw(st.floats(min_value=0.1, max_value=5.0)) if draw(st.booleans()) else None,
    }

disk_full_scenarios

disk_full_scenarios(draw: DrawFn) -> dict[str, Any]

Generate disk space exhaustion scenarios.

Simulates running out of disk space during operations.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing disk full scenario

Example

@given(scenario=disk_full_scenarios())
def test_disk_full_handling(scenario):
    # Simulate disk full at specific point
    pass

Source code in provide/testkit/chaos/io_strategies.py

@composite
def disk_full_scenarios(
    draw: DrawFn,
) -> dict[str, Any]:
    """Generate disk space exhaustion scenarios.

    Simulates running out of disk space during operations.

    Args:
        draw: Hypothesis draw function

    Returns:
        Dictionary containing disk full scenario

    Example:
        ```python
        @given(scenario=disk_full_scenarios())
        def test_disk_full_handling(scenario):
            # Simulate disk full at specific point
            pass
        ```
    """
    total_space = draw(st.integers(min_value=1024, max_value=1024 * 1024 * 100))  # Up to 100MB
    used_space = draw(st.integers(min_value=0, max_value=total_space))

    return {
        "total_space": total_space,
        "used_space": used_space,
        "available_space": total_space - used_space,
        "fills_at_byte": draw(st.integers(min_value=0, max_value=max(1, total_space - used_space))),
        "operation_size": draw(st.integers(min_value=1024, max_value=total_space)),
    }

edge_values

edge_values(draw: DrawFn, value_type: type = int) -> Any

Generate boundary and edge-case values for a given type.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
value_type	type	Type to generate edge values for	int

Returns:

Type	Description
Any	Edge-case value of the specified type

Example

@given(value=edge_values(value_type=int))
def test_integer_handling(value):
    result = process_int(value)

Source code in provide/testkit/chaos/strategies.py

@composite
def edge_values(  # type: ignore[misc]
    draw: DrawFn,
    value_type: type = int,
) -> Any:
    """Generate boundary and edge-case values for a given type.

    Args:
        draw: Hypothesis draw function
        value_type: Type to generate edge values for

    Returns:
        Edge-case value of the specified type

    Example:
        ```python
        @given(value=edge_values(value_type=int))
        def test_integer_handling(value):
            result = process_int(value)
        ```
    """
    if value_type is int:
        edges = [
            st.just(0),
            st.just(1),
            st.just(-1),
            st.just(sys.maxsize),
            st.just(-sys.maxsize - 1),
            st.just(2**31 - 1),  # INT32_MAX
            st.just(-(2**31)),  # INT32_MIN
            st.just(2**63 - 1),  # INT64_MAX
            st.just(-(2**63)),  # INT64_MIN
        ]
        return draw(st.one_of(*edges))

    elif value_type is float:
        edges = [
            st.just(0.0),
            st.just(-0.0),
            st.just(1.0),
            st.just(-1.0),
            st.just(float("inf")),
            st.just(float("-inf")),
            st.just(float("nan")),
            st.just(sys.float_info.min),
            st.just(sys.float_info.max),
            st.just(sys.float_info.epsilon),
        ]
        return draw(st.one_of(*edges))

    elif value_type is str:
        edges = [
            st.just(""),
            st.just(" "),
            st.just("\n"),
            st.just("\t"),
            st.just("\x00"),
            st.just("0"),
            st.just("-1"),
            st.just("null"),
            st.just("None"),
            st.just("undefined"),
        ]
        return draw(st.one_of(*edges))

    else:
        # Generic approach for other types
        return draw(st.from_type(value_type))

failure_patterns

failure_patterns(
    draw: DrawFn,
    max_failures: int = 10,
    exception_types: (
        tuple[type[Exception], ...] | None
    ) = None,
) -> list[tuple[int, type[Exception]]]

Generate patterns of when and how failures should occur.

Returns a list of (iteration_number, exception_type) tuples for failure injection.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_failures	int	Maximum number of failures to generate	10
exception_types	tuple[type[Exception], ...] | None	Specific exception types to use (defaults to common types)	None

Returns:

Type	Description
list[tuple[int, type[Exception]]]	List of (when_to_fail, exception_class) tuples

Example

@given(failures=failure_patterns())
def test_with_failures(failures):
    for i, result in enumerate(operations):
        if any(i == when and exc for when, exc in failures):
            raise next(exc for when, exc in failures if when == i)

Source code in provide/testkit/chaos/strategies.py

@composite
def failure_patterns(
    draw: DrawFn,
    max_failures: int = 10,
    exception_types: tuple[type[Exception], ...] | None = None,
) -> list[tuple[int, type[Exception]]]:
    """Generate patterns of when and how failures should occur.

    Returns a list of (iteration_number, exception_type) tuples for failure injection.

    Args:
        draw: Hypothesis draw function
        max_failures: Maximum number of failures to generate
        exception_types: Specific exception types to use (defaults to common types)

    Returns:
        List of (when_to_fail, exception_class) tuples

    Example:
        ```python
        @given(failures=failure_patterns())
        def test_with_failures(failures):
            for i, result in enumerate(operations):
                if any(i == when and exc for when, exc in failures):
                    raise next(exc for when, exc in failures if when == i)
        ```
    """
    exc_types = exception_types or COMMON_EXCEPTIONS
    num_failures = draw(st.integers(min_value=0, max_value=max_failures))

    return [
        (
            draw(st.integers(min_value=0, max_value=100)),
            draw(st.sampled_from(exc_types)),
        )
        for _ in range(num_failures)
    ]

file_corruption_patterns

file_corruption_patterns(draw: DrawFn) -> dict[str, Any]

Generate file corruption scenarios.

Simulates various types of file corruption for testing recovery.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing corruption configuration

Example

@given(corruption=file_corruption_patterns())
def test_corruption_recovery(tmp_path, corruption):
    # Create and corrupt a file
    pass

Source code in provide/testkit/chaos/io_strategies.py

@composite
def file_corruption_patterns(
    draw: DrawFn,
) -> dict[str, Any]:
    """Generate file corruption scenarios.

    Simulates various types of file corruption for testing recovery.

    Args:
        draw: Hypothesis draw function

    Returns:
        Dictionary containing corruption configuration

    Example:
        ```python
        @given(corruption=file_corruption_patterns())
        def test_corruption_recovery(tmp_path, corruption):
            # Create and corrupt a file
            pass
        ```
    """
    corruption_type = draw(
        st.sampled_from(
            [
                "truncated",  # File cut short
                "random_bytes",  # Random corruption
                "header_corrupt",  # Header/magic bytes corrupted
                "encoding_error",  # Invalid encoding
                "checksum_mismatch",  # Bad checksum
            ]
        )
    )

    config: dict[str, Any] = {
        "type": corruption_type,
    }

    if corruption_type == "truncated":
        config["truncate_at_percent"] = draw(st.floats(min_value=0.0, max_value=1.0))
    elif corruption_type == "random_bytes":
        config["corrupt_percent"] = draw(st.floats(min_value=0.01, max_value=0.5))
        config["num_corruptions"] = draw(st.integers(min_value=1, max_value=100))
    elif corruption_type == "header_corrupt":
        config["header_bytes_to_corrupt"] = draw(st.integers(min_value=1, max_value=64))

    return config

file_sizes

file_sizes(
    draw: DrawFn,
    min_size: int = 0,
    max_size: int = 10 * 1024 * 1024,
    include_huge: bool = False,
) -> int

Generate realistic file size distributions.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_size	int	Minimum file size in bytes	0
max_size	int	Maximum file size in bytes	10 * 1024 * 1024
include_huge	bool	Include very large file sizes (GB range)	False

Returns:

Type	Description
int	File size in bytes

Example

@given(size=file_sizes())
def test_file_handling(tmp_path, size):
    file_path = tmp_path / "test.dat"
    file_path.write_bytes(b"x" * size)

Source code in provide/testkit/chaos/io_strategies.py

@composite
def file_sizes(  # type: ignore[misc]
    draw: DrawFn,
    min_size: int = 0,
    max_size: int = 10 * 1024 * 1024,  # 10MB default
    include_huge: bool = False,
) -> int:
    """Generate realistic file size distributions.

    Args:
        draw: Hypothesis draw function
        min_size: Minimum file size in bytes
        max_size: Maximum file size in bytes
        include_huge: Include very large file sizes (GB range)

    Returns:
        File size in bytes

    Example:
        ```python
        @given(size=file_sizes())
        def test_file_handling(tmp_path, size):
            file_path = tmp_path / "test.dat"
            file_path.write_bytes(b"x" * size)
        ```
    """
    sizes = [
        st.just(0),  # Empty file
        st.integers(min_value=1, max_value=1024),  # Tiny files
        st.integers(min_value=1024, max_value=1024 * 1024),  # KB range
        st.integers(min_value=min_size, max_value=max_size),  # Normal range
    ]

    if include_huge:
        sizes.append(st.integers(min_value=100 * 1024 * 1024, max_value=1024 * 1024 * 1024))  # 100MB - 1GB

    return draw(st.one_of(*sizes))

jitter_patterns

jitter_patterns(
    draw: DrawFn,
    base_interval: float = 1.0,
    max_jitter_percent: float = 50.0,
) -> list[float]

Generate network-like timing jitter patterns.

Simulates variable latency and timing uncertainty.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
base_interval	float	Base time interval in seconds	1.0
max_jitter_percent	float	Maximum jitter as percentage of base interval	50.0

Returns:

Type	Description
list[float]	List of intervals with jitter applied

Example

@given(intervals=jitter_patterns(base_interval=0.1))
async def test_with_jitter(intervals):
    for interval in intervals:
        await asyncio.sleep(interval)
        await send_packet()

Source code in provide/testkit/chaos/time_strategies.py

@composite
def jitter_patterns(
    draw: DrawFn,
    base_interval: float = 1.0,
    max_jitter_percent: float = 50.0,
) -> list[float]:
    """Generate network-like timing jitter patterns.

    Simulates variable latency and timing uncertainty.

    Args:
        draw: Hypothesis draw function
        base_interval: Base time interval in seconds
        max_jitter_percent: Maximum jitter as percentage of base interval

    Returns:
        List of intervals with jitter applied

    Example:
        ```python
        @given(intervals=jitter_patterns(base_interval=0.1))
        async def test_with_jitter(intervals):
            for interval in intervals:
                await asyncio.sleep(interval)
                await send_packet()
        ```
    """
    num_intervals = draw(st.integers(min_value=1, max_value=100))
    jitter_range = base_interval * (max_jitter_percent / 100.0)

    return [
        base_interval + draw(st.floats(min_value=-jitter_range, max_value=jitter_range))
        for _ in range(num_intervals)
    ]

lock_contention_patterns

lock_contention_patterns(
    draw: DrawFn,
    num_locks: int = 5,
    num_operations: int = 20,
) -> dict[str, Any]

Generate lock contention patterns for testing synchronization.

Creates scenarios with varying levels of lock contention.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
num_locks	int	Number of available locks	5
num_operations	int	Number of operations competing for locks	20

Returns:

Type	Description
dict[str, Any]	Dictionary containing lock contention configuration

Example

@given(pattern=lock_contention_patterns())
async def test_lock_contention(pattern):
    locks = [asyncio.Lock() for _ in range(pattern['num_locks'])]
    # Execute operations with varying contention
    pass

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def lock_contention_patterns(
    draw: DrawFn,
    num_locks: int = 5,
    num_operations: int = 20,
) -> dict[str, Any]:
    """Generate lock contention patterns for testing synchronization.

    Creates scenarios with varying levels of lock contention.

    Args:
        draw: Hypothesis draw function
        num_locks: Number of available locks
        num_operations: Number of operations competing for locks

    Returns:
        Dictionary containing lock contention configuration

    Example:
        ```python
        @given(pattern=lock_contention_patterns())
        async def test_lock_contention(pattern):
            locks = [asyncio.Lock() for _ in range(pattern['num_locks'])]
            # Execute operations with varying contention
            pass
        ```
    """
    # Generate lock access patterns
    operations: list[dict[str, Any]] = []
    for _ in range(num_operations):
        # Which lock(s) does this operation need?
        num_locks_needed = draw(st.integers(min_value=1, max_value=min(3, num_locks)))
        locks_needed = draw(
            st.lists(
                st.integers(min_value=0, max_value=num_locks - 1),
                min_size=num_locks_needed,
                max_size=num_locks_needed,
                unique=True,
            )
        )

        # How long to hold the lock?
        hold_duration = draw(st.floats(min_value=0.001, max_value=0.5))

        operations.append(
            {
                "locks_needed": sorted(locks_needed),  # Always acquire in order to prevent deadlock
                "hold_duration": hold_duration,
                "operation_id": len(operations),
            }
        )

    return {
        "num_locks": num_locks,
        "operations": operations,
        "concurrent_workers": draw(st.integers(min_value=2, max_value=20)),
    }

lock_file_scenarios

lock_file_scenarios(draw: DrawFn) -> dict[str, Any]

Generate file lock conflict scenarios.

Creates situations with competing file locks and stale locks.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing lock scenario configuration

Example

@given(scenario=lock_file_scenarios())
def test_file_locking(tmp_path, scenario):
    # Test file locking behavior
    pass

Source code in provide/testkit/chaos/io_strategies.py

@composite
def lock_file_scenarios(
    draw: DrawFn,
) -> dict[str, Any]:
    """Generate file lock conflict scenarios.

    Creates situations with competing file locks and stale locks.

    Args:
        draw: Hypothesis draw function

    Returns:
        Dictionary containing lock scenario configuration

    Example:
        ```python
        @given(scenario=lock_file_scenarios())
        def test_file_locking(tmp_path, scenario):
            # Test file locking behavior
            pass
        ```
    """
    return {
        "num_processes": draw(st.integers(min_value=2, max_value=20)),
        "lock_duration": draw(st.floats(min_value=0.01, max_value=5.0)),
        "has_stale_lock": draw(st.booleans()),
        "stale_lock_age": draw(st.floats(min_value=1.0, max_value=3600.0)),
        "timeout": draw(st.floats(min_value=0.1, max_value=30.0)),
        "check_interval": draw(st.floats(min_value=0.001, max_value=1.0)),
        "corrupted_lock_file": draw(st.booleans()),
        "lock_content_type": draw(st.sampled_from(["json", "plain_text", "binary", "empty"])),
    }

malformed_inputs

malformed_inputs(
    draw: DrawFn,
    include_huge: bool = True,
    include_empty: bool = True,
    include_special: bool = True,
) -> Any

Generate edge-case inputs for robust testing.

Creates various malformed, extreme, or unusual inputs that often expose bugs.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
include_huge	bool	Include very large strings/data	True
include_empty	bool	Include empty/None values	True
include_special	bool	Include special numeric values (NaN, inf)	True

Returns:

Type	Description
Any	Various malformed input values

Example

@given(data=malformed_inputs())
def test_parser(data):
    result = parse(data)  # Should handle gracefully

Source code in provide/testkit/chaos/strategies.py

@composite
def malformed_inputs(  # type: ignore[misc]
    draw: DrawFn,
    include_huge: bool = True,
    include_empty: bool = True,
    include_special: bool = True,
) -> Any:
    """Generate edge-case inputs for robust testing.

    Creates various malformed, extreme, or unusual inputs that often expose bugs.

    Args:
        draw: Hypothesis draw function
        include_huge: Include very large strings/data
        include_empty: Include empty/None values
        include_special: Include special numeric values (NaN, inf)

    Returns:
        Various malformed input values

    Example:
        ```python
        @given(data=malformed_inputs())
        def test_parser(data):
            result = parse(data)  # Should handle gracefully
        ```
    """
    strategies = []

    # Text inputs
    strategies.append(st.text(min_size=0, max_size=1000))
    if include_huge:
        # Use 8KB as max to stay within Hypothesis BUFFER_SIZE limits
        strategies.append(st.text(min_size=5000, max_size=8000))

    # Binary inputs
    strategies.append(st.binary(min_size=0, max_size=1000))
    if include_huge:
        # Use 8KB as max to stay within Hypothesis BUFFER_SIZE limits
        strategies.append(st.binary(min_size=5000, max_size=8000))

    # Numeric inputs
    strategies.append(st.integers())
    strategies.append(st.floats(allow_nan=include_special, allow_infinity=include_special))

    # Empty/None values
    if include_empty:
        strategies.extend([st.just(None), st.just(""), st.just(b""), st.just([])])

    # Lists and dicts
    strategies.append(st.lists(st.integers(), min_size=0, max_size=100))
    strategies.append(st.dictionaries(st.text(), st.integers(), min_size=0, max_size=100))

    return draw(st.one_of(*strategies))

network_error_patterns

network_error_patterns(
    draw: DrawFn,
) -> list[dict[str, Any]]

Generate network failure patterns.

Creates sequences of network errors and recovery scenarios.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
list[dict[str, Any]]	List of network error events

Example

@given(errors=network_error_patterns())
async def test_network_resilience(errors):
    for error_event in errors:
        if error_event['type'] == 'timeout':
            # Simulate timeout
            pass

Source code in provide/testkit/chaos/io_strategies.py

@composite
def network_error_patterns(
    draw: DrawFn,
) -> list[dict[str, Any]]:
    """Generate network failure patterns.

    Creates sequences of network errors and recovery scenarios.

    Args:
        draw: Hypothesis draw function

    Returns:
        List of network error events

    Example:
        ```python
        @given(errors=network_error_patterns())
        async def test_network_resilience(errors):
            for error_event in errors:
                if error_event['type'] == 'timeout':
                    # Simulate timeout
                    pass
        ```
    """
    num_events = draw(st.integers(min_value=1, max_value=20))

    error_types = [
        "timeout",
        "connection_refused",
        "connection_reset",
        "dns_failure",
        "ssl_error",
        "partial_response",
        "slow_response",
    ]

    events = []
    for _ in range(num_events):
        error_type = draw(st.sampled_from(error_types))

        event: dict[str, Any] = {
            "type": error_type,
            "at_byte": draw(st.integers(min_value=0, max_value=10000)),
        }

        if error_type == "timeout":
            event["timeout_after"] = draw(st.floats(min_value=0.1, max_value=30.0))
        elif error_type == "slow_response":
            event["bytes_per_second"] = draw(st.integers(min_value=100, max_value=10000))
        elif error_type == "partial_response":
            event["bytes_received"] = draw(st.integers(min_value=0, max_value=10000))
            event["expected_bytes"] = draw(st.integers(min_value=event["bytes_received"], max_value=100000))

        events.append(event)

    return events

path_traversal_patterns

path_traversal_patterns(draw: DrawFn) -> str

Generate path traversal attack patterns.

Creates potentially malicious path patterns for security testing.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
str	A path string that might contain traversal attempts

Example

@given(path=path_traversal_patterns())
def test_path_validation(path):
    # Ensure path traversal is blocked
    sanitized = sanitize_path(path)

Source code in provide/testkit/chaos/io_strategies.py

@composite
def path_traversal_patterns(
    draw: DrawFn,
) -> str:
    """Generate path traversal attack patterns.

    Creates potentially malicious path patterns for security testing.

    Args:
        draw: Hypothesis draw function

    Returns:
        A path string that might contain traversal attempts

    Example:
        ```python
        @given(path=path_traversal_patterns())
        def test_path_validation(path):
            # Ensure path traversal is blocked
            sanitized = sanitize_path(path)
        ```
    """
    patterns = [
        "../../../etc/passwd",
        "..\\..\\..\\windows\\system32",
        "./././../../../",
        "/etc/passwd",
        "C:\\Windows\\System32\\config\\sam",
        "./../.../../",
        "....//....//....//",
        "%2e%2e%2f%2e%2e%2f",  # URL encoded
        "..%252f..%252f",  # Double encoded
    ]

    # Mix safe and unsafe paths
    if draw(st.booleans()):
        return draw(st.sampled_from(patterns))
    else:
        # Safe relative path
        parts = draw(
            st.lists(
                st.text(alphabet=st.characters(min_codepoint=97, max_codepoint=122), min_size=1, max_size=10),
                min_size=1,
                max_size=5,
            )
        )
        return str(Path(*parts))

permission_patterns

permission_patterns(draw: DrawFn) -> dict[str, Any]

Generate file permission scenarios.

Creates various permission configurations for testing access control.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing permission configuration

Example

@given(perms=permission_patterns())
def test_file_permissions(tmp_path, perms):
    file_path = tmp_path / "test.txt"
    file_path.touch()
    os.chmod(file_path, perms['mode'])

Source code in provide/testkit/chaos/io_strategies.py

@composite
def permission_patterns(
    draw: DrawFn,
) -> dict[str, Any]:
    """Generate file permission scenarios.

    Creates various permission configurations for testing access control.

    Args:
        draw: Hypothesis draw function

    Returns:
        Dictionary containing permission configuration

    Example:
        ```python
        @given(perms=permission_patterns())
        def test_file_permissions(tmp_path, perms):
            file_path = tmp_path / "test.txt"
            file_path.touch()
            os.chmod(file_path, perms['mode'])
        ```
    """
    # Common permission modes
    common_modes = [
        0o000,  # No permissions
        0o400,  # Read only (owner)
        0o600,  # Read/write (owner)
        0o644,  # Read/write (owner), read (others)
        0o755,  # All (owner), read/execute (others)
        0o777,  # All permissions
    ]

    mode = draw(st.sampled_from(common_modes))

    return {
        "mode": mode,
        "readable": (mode & 0o400) != 0,
        "writable": (mode & 0o200) != 0,
        "executable": (mode & 0o100) != 0,
        "change_during_test": draw(st.booleans()),
    }

pid_recycling_scenarios

pid_recycling_scenarios(draw: DrawFn) -> dict[str, Any]

Generate PID recycling attack scenarios.

Creates scenarios where PIDs might be reused, testing protection mechanisms.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing PID recycling scenario

Example

@given(scenario=pid_recycling_scenarios())
def test_pid_recycling_protection(scenario):
    # Test that system detects recycled PIDs correctly
    pass

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def pid_recycling_scenarios(
    draw: DrawFn,
) -> dict[str, Any]:
    """Generate PID recycling attack scenarios.

    Creates scenarios where PIDs might be reused, testing protection mechanisms.

    Args:
        draw: Hypothesis draw function

    Returns:
        Dictionary containing PID recycling scenario

    Example:
        ```python
        @given(scenario=pid_recycling_scenarios())
        def test_pid_recycling_protection(scenario):
            # Test that system detects recycled PIDs correctly
            pass
        ```
    """
    # Simulate a PID that gets recycled
    original_pid = draw(st.integers(min_value=1, max_value=65535))
    recycled_pid = original_pid  # Same PID, different process

    # Process start times (seconds since epoch)
    original_start_time = draw(st.floats(min_value=1000000000, max_value=2000000000))

    # Recycled process starts after original
    time_gap = draw(st.floats(min_value=0.1, max_value=3600.0))
    recycled_start_time = original_start_time + time_gap

    # Tolerance for time comparison
    time_tolerance = draw(st.floats(min_value=0.0, max_value=2.0))

    return {
        "original_pid": original_pid,
        "recycled_pid": recycled_pid,
        "original_start_time": original_start_time,
        "recycled_start_time": recycled_start_time,
        "time_tolerance": time_tolerance,
        "should_detect_recycling": abs(recycled_start_time - original_start_time) > time_tolerance,
    }

process_pool_patterns

process_pool_patterns(
    draw: DrawFn,
    max_workers: int = 10,
    max_tasks: int = 100,
) -> dict[str, Any]

Generate process pool execution patterns.

Creates scenarios for testing process pool behavior under various conditions.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_workers	int	Maximum number of worker processes	10
max_tasks	int	Maximum number of tasks to submit	100

Returns:

Type	Description
dict[str, Any]	Dictionary containing pool configuration

Example

@given(config=process_pool_patterns())
def test_process_pool(config):
    with ProcessPoolExecutor(max_workers=config['workers']) as pool:
        futures = [pool.submit(task) for task in config['tasks']]

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def process_pool_patterns(
    draw: DrawFn,
    max_workers: int = 10,
    max_tasks: int = 100,
) -> dict[str, Any]:
    """Generate process pool execution patterns.

    Creates scenarios for testing process pool behavior under various conditions.

    Args:
        draw: Hypothesis draw function
        max_workers: Maximum number of worker processes
        max_tasks: Maximum number of tasks to submit

    Returns:
        Dictionary containing pool configuration

    Example:
        ```python
        @given(config=process_pool_patterns())
        def test_process_pool(config):
            with ProcessPoolExecutor(max_workers=config['workers']) as pool:
                futures = [pool.submit(task) for task in config['tasks']]
        ```
    """
    num_workers = draw(st.integers(min_value=1, max_value=max_workers))
    num_tasks = draw(st.integers(min_value=1, max_value=max_tasks))

    task_patterns = draw(
        st.sampled_from(
            [
                "uniform",  # All tasks similar duration
                "mixed",  # Mix of fast and slow
                "bursty",  # Some tasks much slower
            ]
        )
    )

    return {
        "workers": num_workers,
        "num_tasks": num_tasks,
        "task_pattern": task_patterns,
        "timeout": draw(st.one_of(st.none(), st.floats(min_value=1.0, max_value=30.0))),
        "max_tasks_per_child": draw(st.one_of(st.none(), st.integers(min_value=1, max_value=50))),
    }

race_condition_triggers

race_condition_triggers(
    draw: DrawFn,
    num_operations: int = 10,
    max_delay: float = 0.1,
) -> list[tuple[int, float]]

Generate timing patterns designed to trigger race conditions.

Returns operation timings that maximize the chance of exposing race conditions.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
num_operations	int	Number of concurrent operations	10
max_delay	float	Maximum delay between operations in seconds	0.1

Returns:

Type	Description
list[tuple[int, float]]	List of (operation_id, delay_before_execution) tuples

Example

@given(timings=race_condition_triggers())
async def test_race_condition(timings):
    tasks = []
    for op_id, delay in timings:
        await asyncio.sleep(delay)
        tasks.append(asyncio.create_task(operation(op_id)))
    await asyncio.gather(*tasks)

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def race_condition_triggers(
    draw: DrawFn,
    num_operations: int = 10,
    max_delay: float = 0.1,
) -> list[tuple[int, float]]:
    """Generate timing patterns designed to trigger race conditions.

    Returns operation timings that maximize the chance of exposing race conditions.

    Args:
        draw: Hypothesis draw function
        num_operations: Number of concurrent operations
        max_delay: Maximum delay between operations in seconds

    Returns:
        List of (operation_id, delay_before_execution) tuples

    Example:
        ```python
        @given(timings=race_condition_triggers())
        async def test_race_condition(timings):
            tasks = []
            for op_id, delay in timings:
                await asyncio.sleep(delay)
                tasks.append(asyncio.create_task(operation(op_id)))
            await asyncio.gather(*tasks)
        ```
    """
    return [
        (
            i,
            draw(st.floats(min_value=0.0, max_value=max_delay, allow_nan=False, allow_infinity=False)),
        )
        for i in range(num_operations)
    ]

rate_burst_patterns

rate_burst_patterns(
    draw: DrawFn,
    max_burst_size: int = 1000,
    max_duration: float = 10.0,
) -> list[tuple[float, int]]

Generate traffic burst patterns for rate limiting tests.

Returns a list of (timestamp, request_count) representing burst patterns.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_burst_size	int	Maximum number of requests in a burst	1000
max_duration	float	Maximum duration of burst pattern in seconds	10.0

Returns:

Type	Description
list[tuple[float, int]]	List of (time_offset, request_count) tuples

Example

@given(bursts=rate_burst_patterns())
async def test_rate_limiter(bursts):
    for time_offset, count in bursts:
        await asyncio.sleep(time_offset)
        for _ in range(count):
            await rate_limited_operation()

Source code in provide/testkit/chaos/time_strategies.py

@composite
def rate_burst_patterns(
    draw: DrawFn,
    max_burst_size: int = 1000,
    max_duration: float = 10.0,
) -> list[tuple[float, int]]:
    """Generate traffic burst patterns for rate limiting tests.

    Returns a list of (timestamp, request_count) representing burst patterns.

    Args:
        draw: Hypothesis draw function
        max_burst_size: Maximum number of requests in a burst
        max_duration: Maximum duration of burst pattern in seconds

    Returns:
        List of (time_offset, request_count) tuples

    Example:
        ```python
        @given(bursts=rate_burst_patterns())
        async def test_rate_limiter(bursts):
            for time_offset, count in bursts:
                await asyncio.sleep(time_offset)
                for _ in range(count):
                    await rate_limited_operation()
        ```
    """
    num_bursts = draw(st.integers(min_value=1, max_value=20))
    bursts = []

    current_time = 0.0
    for _ in range(num_bursts):
        # Time between bursts
        time_offset = draw(st.floats(min_value=0.0, max_value=max_duration / num_bursts))
        current_time += time_offset

        # Burst size
        burst_size = draw(st.integers(min_value=1, max_value=max_burst_size))

        bursts.append((current_time, burst_size))

    return bursts

resource_limits

resource_limits(
    draw: DrawFn,
    min_memory: int = 1024,
    max_memory: int = 1024 * 1024 * 100,
    min_timeout: float = 0.01,
    max_timeout: float = 60.0,
) -> dict[str, Any]

Generate resource constraint scenarios.

Creates various resource limit configurations for testing behavior under constraints.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_memory	int	Minimum memory limit in bytes	1024
max_memory	int	Maximum memory limit in bytes	1024 * 1024 * 100
min_timeout	float	Minimum timeout in seconds	0.01
max_timeout	float	Maximum timeout in seconds	60.0

Returns:

Type	Description
dict[str, Any]	Dictionary with resource limit configuration

Example

@given(limits=resource_limits())
def test_with_limits(limits):
    with resource_limiter(limits['memory'], limits['timeout']):
        perform_operation()

Source code in provide/testkit/chaos/strategies.py

@composite
def resource_limits(
    draw: DrawFn,
    min_memory: int = 1024,
    max_memory: int = 1024 * 1024 * 100,  # 100MB
    min_timeout: float = 0.01,
    max_timeout: float = 60.0,
) -> dict[str, Any]:
    """Generate resource constraint scenarios.

    Creates various resource limit configurations for testing behavior under constraints.

    Args:
        draw: Hypothesis draw function
        min_memory: Minimum memory limit in bytes
        max_memory: Maximum memory limit in bytes
        min_timeout: Minimum timeout in seconds
        max_timeout: Maximum timeout in seconds

    Returns:
        Dictionary with resource limit configuration

    Example:
        ```python
        @given(limits=resource_limits())
        def test_with_limits(limits):
            with resource_limiter(limits['memory'], limits['timeout']):
                perform_operation()
        ```
    """
    return {
        "memory": draw(st.integers(min_value=min_memory, max_value=max_memory)),
        "timeout": draw(st.floats(min_value=min_timeout, max_value=max_timeout)),
        "cpu_count": draw(st.integers(min_value=1, max_value=64)),
        "max_threads": draw(st.integers(min_value=1, max_value=1000)),
        "max_open_files": draw(st.integers(min_value=10, max_value=10000)),
    }

retry_backoff_patterns

retry_backoff_patterns(
    draw: DrawFn, max_retries: int = 10
) -> dict[str, Any]

Generate retry and backoff patterns for resilience testing.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
max_retries	int	Maximum number of retries	10

Returns:

Type	Description
dict[str, Any]	Dictionary containing retry configuration

Example

@given(pattern=retry_backoff_patterns())
def test_retry_logic(pattern):
    for attempt in range(pattern['max_attempts']):
        delay = pattern['backoff_strategy'](attempt)
        time.sleep(delay)

Source code in provide/testkit/chaos/time_strategies.py

@composite
def retry_backoff_patterns(
    draw: DrawFn,
    max_retries: int = 10,
) -> dict[str, Any]:
    """Generate retry and backoff patterns for resilience testing.

    Args:
        draw: Hypothesis draw function
        max_retries: Maximum number of retries

    Returns:
        Dictionary containing retry configuration

    Example:
        ```python
        @given(pattern=retry_backoff_patterns())
        def test_retry_logic(pattern):
            for attempt in range(pattern['max_attempts']):
                delay = pattern['backoff_strategy'](attempt)
                time.sleep(delay)
        ```
    """
    num_retries = draw(st.integers(min_value=1, max_value=max_retries))

    backoff_type = draw(st.sampled_from(["constant", "linear", "exponential", "jittered"]))

    config: dict[str, Any] = {
        "max_attempts": num_retries,
        "backoff_type": backoff_type,
    }

    if backoff_type == "constant":
        config["base_delay"] = draw(st.floats(min_value=0.01, max_value=5.0))
    elif backoff_type == "linear":
        config["base_delay"] = draw(st.floats(min_value=0.01, max_value=1.0))
        config["increment"] = draw(st.floats(min_value=0.1, max_value=2.0))
    elif backoff_type == "exponential":
        config["base_delay"] = draw(st.floats(min_value=0.01, max_value=1.0))
        config["multiplier"] = draw(st.floats(min_value=1.5, max_value=3.0))
        config["max_delay"] = draw(st.floats(min_value=10.0, max_value=60.0))
    elif backoff_type == "jittered":
        config["base_delay"] = draw(st.floats(min_value=0.01, max_value=1.0))
        config["multiplier"] = draw(st.floats(min_value=1.5, max_value=3.0))
        config["jitter_percent"] = draw(st.floats(min_value=0.0, max_value=50.0))

    return config

task_cancellation_patterns

task_cancellation_patterns(
    draw: DrawFn, num_tasks: int = 20
) -> list[dict[str, Any]]

Generate task cancellation scenarios for async testing.

Creates patterns of task creation and cancellation to test cleanup.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
num_tasks	int	Number of tasks in the pattern	20

Returns:

Type	Description
list[dict[str, Any]]	List of task configuration dictionaries

Example

@given(tasks=task_cancellation_patterns())
async def test_task_cancellation(tasks):
    for task_config in tasks:
        task = asyncio.create_task(long_running())
        if task_config['should_cancel']:
            await asyncio.sleep(task_config['cancel_after'])
            task.cancel()

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def task_cancellation_patterns(
    draw: DrawFn,
    num_tasks: int = 20,
) -> list[dict[str, Any]]:
    """Generate task cancellation scenarios for async testing.

    Creates patterns of task creation and cancellation to test cleanup.

    Args:
        draw: Hypothesis draw function
        num_tasks: Number of tasks in the pattern

    Returns:
        List of task configuration dictionaries

    Example:
        ```python
        @given(tasks=task_cancellation_patterns())
        async def test_task_cancellation(tasks):
            for task_config in tasks:
                task = asyncio.create_task(long_running())
                if task_config['should_cancel']:
                    await asyncio.sleep(task_config['cancel_after'])
                    task.cancel()
        ```
    """
    tasks = []
    for i in range(num_tasks):
        should_cancel = draw(st.booleans())

        config: dict[str, Any] = {
            "task_id": i,
            "should_cancel": should_cancel,
        }

        if should_cancel:
            config["cancel_after"] = draw(st.floats(min_value=0.0, max_value=1.0))
            config["expect_cancellation_error"] = draw(st.booleans())
        else:
            config["expected_duration"] = draw(st.floats(min_value=0.1, max_value=2.0))

        tasks.append(config)

    return tasks

thread_counts

thread_counts(
    draw: DrawFn,
    min_threads: int = 1,
    max_threads: int = 100,
    include_extremes: bool = True,
) -> int

Generate thread count scenarios for concurrency testing.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_threads	int	Minimum number of threads	1
max_threads	int	Maximum number of threads	100
include_extremes	bool	Include edge cases (1 thread, max threads)	True

Returns:

Type	Description
int	Number of threads to spawn

Example

@given(num_threads=thread_counts())
def test_concurrent_access(num_threads):
    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        # Test concurrent operations
        pass

Source code in provide/testkit/chaos/concurrency_strategies.py

@composite
def thread_counts(
    draw: DrawFn,
    min_threads: int = 1,
    max_threads: int = 100,
    include_extremes: bool = True,
) -> int:
    """Generate thread count scenarios for concurrency testing.

    Args:
        draw: Hypothesis draw function
        min_threads: Minimum number of threads
        max_threads: Maximum number of threads
        include_extremes: Include edge cases (1 thread, max threads)

    Returns:
        Number of threads to spawn

    Example:
        ```python
        @given(num_threads=thread_counts())
        def test_concurrent_access(num_threads):
            with ThreadPoolExecutor(max_workers=num_threads) as executor:
                # Test concurrent operations
                pass
        ```
    """
    if include_extremes:
        return draw(
            st.one_of(
                st.just(1), st.just(max_threads), st.integers(min_value=min_threads, max_value=max_threads)
            )
        )
    return draw(st.integers(min_value=min_threads, max_value=max_threads))

time_advances

time_advances(
    draw: DrawFn,
    min_advance: float = 0.0,
    max_advance: float = 3600.0,
    allow_backwards: bool = False,
) -> float

Generate time progression patterns for time manipulation testing.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_advance	float	Minimum time advance in seconds	0.0
max_advance	float	Maximum time advance in seconds	3600.0
allow_backwards	bool	Allow negative time advances (backwards jumps)	False

Returns:

Type	Description
float	Time advance value in seconds (can be negative if allowed)

Example

@given(advance=time_advances(allow_backwards=True))
def test_with_time_jump(advance, chaos_time_source):
    chaos_time_source.advance(advance)
    # Test behavior after time jump

Source code in provide/testkit/chaos/time_strategies.py

@composite
def time_advances(
    draw: DrawFn,
    min_advance: float = 0.0,
    max_advance: float = 3600.0,
    allow_backwards: bool = False,
) -> float:
    """Generate time progression patterns for time manipulation testing.

    Args:
        draw: Hypothesis draw function
        min_advance: Minimum time advance in seconds
        max_advance: Maximum time advance in seconds
        allow_backwards: Allow negative time advances (backwards jumps)

    Returns:
        Time advance value in seconds (can be negative if allowed)

    Example:
        ```python
        @given(advance=time_advances(allow_backwards=True))
        def test_with_time_jump(advance, chaos_time_source):
            chaos_time_source.advance(advance)
            # Test behavior after time jump
        ```
    """
    if allow_backwards:
        return draw(
            st.floats(min_value=-max_advance, max_value=max_advance, allow_nan=False, allow_infinity=False)
        )
    return draw(st.floats(min_value=min_advance, max_value=max_advance, allow_nan=False, allow_infinity=False))

timeout_patterns

timeout_patterns(
    draw: DrawFn,
    min_timeout: float = 0.01,
    max_timeout: float = 60.0,
    include_none: bool = True,
) -> float | None

Generate realistic timeout scenarios.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
min_timeout	float	Minimum timeout value in seconds	0.01
max_timeout	float	Maximum timeout value in seconds	60.0
include_none	bool	Whether to include None (no timeout) as a possibility	True

Returns:

Type	Description
float | None	Timeout value in seconds, or None

Example

@given(timeout=timeout_patterns())
async def test_operation_timeout(timeout):
    await asyncio.wait_for(operation(), timeout=timeout)

Source code in provide/testkit/chaos/time_strategies.py

@composite
def timeout_patterns(  # type: ignore[misc]
    draw: DrawFn,
    min_timeout: float = 0.01,
    max_timeout: float = 60.0,
    include_none: bool = True,
) -> float | None:
    """Generate realistic timeout scenarios.

    Args:
        draw: Hypothesis draw function
        min_timeout: Minimum timeout value in seconds
        max_timeout: Maximum timeout value in seconds
        include_none: Whether to include None (no timeout) as a possibility

    Returns:
        Timeout value in seconds, or None

    Example:
        ```python
        @given(timeout=timeout_patterns())
        async def test_operation_timeout(timeout):
            await asyncio.wait_for(operation(), timeout=timeout)
        ```
    """
    strategies = [
        st.floats(min_value=min_timeout, max_value=max_timeout, allow_nan=False, allow_infinity=False),
        # Edge case: very short timeouts
        st.floats(min_value=0.001, max_value=0.01),
    ]

    if include_none:
        strategies.append(st.just(None))

    return draw(st.one_of(*strategies))

unicode_chaos

unicode_chaos(
    draw: DrawFn,
    include_emoji: bool = True,
    include_rtl: bool = True,
    include_zero_width: bool = True,
    include_surrogates: bool = False,
) -> str

Generate problematic Unicode strings for testing.

Creates strings with emoji, RTL text, zero-width characters, and other Unicode edge cases that often cause issues.

Parameters:

Name	Type	Description	Default
draw	DrawFn	Hypothesis draw function	required
include_emoji	bool	Include emoji characters	True
include_rtl	bool	Include right-to-left text	True
include_zero_width	bool	Include zero-width characters	True
include_surrogates	bool	Include surrogate pairs (can be problematic)	False

Returns:

Type	Description
str	A Unicode string with chaotic properties

Example

@given(text=unicode_chaos())
def test_text_handling(text):
    result = process_text(text)

Source code in provide/testkit/chaos/strategies.py

@composite
def unicode_chaos(  # type: ignore[misc]
    draw: DrawFn,
    include_emoji: bool = True,
    include_rtl: bool = True,
    include_zero_width: bool = True,
    include_surrogates: bool = False,
) -> str:
    """Generate problematic Unicode strings for testing.

    Creates strings with emoji, RTL text, zero-width characters, and other
    Unicode edge cases that often cause issues.

    Args:
        draw: Hypothesis draw function
        include_emoji: Include emoji characters
        include_rtl: Include right-to-left text
        include_zero_width: Include zero-width characters
        include_surrogates: Include surrogate pairs (can be problematic)

    Returns:
        A Unicode string with chaotic properties

    Example:
        ```python
        @given(text=unicode_chaos())
        def test_text_handling(text):
            result = process_text(text)
        ```
    """
    strategies = []

    # Basic Unicode text
    strategies.append(st.text(min_size=0, max_size=100))

    # Emoji ranges
    if include_emoji:
        # Emoticons and pictographs
        strategies.append(
            st.text(
                alphabet=st.characters(min_codepoint=0x1F300, max_codepoint=0x1F9FF),
                min_size=1,
                max_size=20,
            )
        )

    # RTL characters (Arabic, Hebrew)
    if include_rtl:
        strategies.append(
            st.text(
                alphabet=st.characters(min_codepoint=0x0590, max_codepoint=0x08FF),
                min_size=1,
                max_size=50,
            )
        )

    # Zero-width characters
    if include_zero_width:
        zero_width_chars = "\u200b\u200c\u200d\ufeff"  # ZWSP, ZWNJ, ZWJ, ZWNBSP
        strategies.append(st.text(alphabet=zero_width_chars, min_size=1, max_size=10))

    # Control characters
    strategies.append(st.text(alphabet=st.characters(max_codepoint=0x001F), min_size=1, max_size=10))

    # Combining characters
    strategies.append(
        st.text(alphabet=st.characters(min_codepoint=0x0300, max_codepoint=0x036F), min_size=1, max_size=20)
    )

    return draw(st.one_of(*strategies))