Fixtures

provide.testkit.process.fixtures

Process Test Fixtures.

Core process testing fixtures with re-exports from specialized modules. Utilities for testing async code, managing event loops, handling async subprocess mocking, and bash script testing across the provide-io ecosystem.

Classes

ScriptExecutionContext

Context for executing bash scripts in isolation.

Provides a temporary workspace with environment variable control and automatic cleanup.

Attributes:

Name	Type	Description
workspace	Path	Path to temporary workspace directory.
env	dict[str, str]	Environment variables for script execution.
timeout	int	Timeout in seconds for script execution.

Functions

create_directory

create_directory(name: str) -> Path

Create a directory in the workspace.

Parameters:

Name	Type	Description	Default
name	str	Directory name (can be nested path).	required

Returns:

Type	Description
Path	Path to the created directory.

Source code in provide/testkit/process/script_fixtures.py

def create_directory(self, name: str) -> Path:
    """Create a directory in the workspace.

    Args:
        name: Directory name (can be nested path).

    Returns:
        Path to the created directory.
    """
    dir_path = self.workspace / name
    dir_path.mkdir(parents=True, exist_ok=True)
    return dir_path

create_file

create_file(name: str, content: str = '') -> Path

Create a file in the workspace.

Parameters:

Name	Type	Description	Default
name	str	File name (can include subdirectories).	required
content	str	File content.	''

Returns:

Type	Description
Path	Path to the created file.

Source code in provide/testkit/process/script_fixtures.py

def create_file(self, name: str, content: str = "") -> Path:
    """Create a file in the workspace.

    Args:
        name: File name (can include subdirectories).
        content: File content.

    Returns:
        Path to the created file.
    """
    file_path = self.workspace / name
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content)
    return file_path

file_exists

file_exists(name: str) -> bool

Check if a file or directory exists in workspace.

Parameters:

Name	Type	Description	Default
name	str	File or directory name.	required

Returns:

Type	Description
bool	True if path exists, False otherwise.

Source code in provide/testkit/process/script_fixtures.py

def file_exists(self, name: str) -> bool:
    """Check if a file or directory exists in workspace.

    Args:
        name: File or directory name.

    Returns:
        True if path exists, False otherwise.
    """
    return (self.workspace / name).exists()

init_git_repo

init_git_repo() -> None

Initialize a git repository in the workspace.

Raises:

Type	Description
RuntimeError	If git is not available.

Source code in provide/testkit/process/script_fixtures.py

def init_git_repo(self) -> None:
    """Initialize a git repository in the workspace.

    Raises:
        RuntimeError: If git is not available.
    """
    if not shutil.which("git"):
        raise RuntimeError("git command not found")

    run(["git", "init"], cwd=self.workspace, check=True)
    run(["git", "config", "user.email", "[email protected]"], cwd=self.workspace, check=True)
    run(["git", "config", "user.name", "Test User"], cwd=self.workspace, check=True)

read_file

read_file(name: str) -> str

Read a file from the workspace.

Parameters:

Name	Type	Description	Default
name	str	File name.	required

Returns:

Type	Description
str	File contents as string.

Raises:

Type	Description
FileNotFoundError	If file does not exist.

Source code in provide/testkit/process/script_fixtures.py

def read_file(self, name: str) -> str:
    """Read a file from the workspace.

    Args:
        name: File name.

    Returns:
        File contents as string.

    Raises:
        FileNotFoundError: If file does not exist.
    """
    return (self.workspace / name).read_text()

run_command

run_command(
    command: str | list[str], check: bool = False
) -> ScriptResult

Execute a shell command in the isolated workspace.

Parameters:

Name	Type	Description	Default
command	str | list[str]	Shell command to execute (string or list).	required
check	bool	If True, raise exception on non-zero exit code.	False

Returns:

Type	Description
ScriptResult	ScriptResult with execution details.

Raises:

Type	Description
CalledProcessError	If check=True and command fails.
TimeoutExpired	If execution exceeds timeout.

Source code in provide/testkit/process/script_fixtures.py

def run_command(
    self,
    command: str | list[str],
    check: bool = False,
) -> ScriptResult:
    """Execute a shell command in the isolated workspace.

    Args:
        command: Shell command to execute (string or list).
        check: If True, raise exception on non-zero exit code.

    Returns:
        ScriptResult with execution details.

    Raises:
        subprocess.CalledProcessError: If check=True and command fails.
        subprocess.TimeoutExpired: If execution exceeds timeout.
    """
    import time

    start_time = time.time()

    try:
        result = run(
            command,
            cwd=self.workspace,
            env=self.env if self.env else None,
            timeout=self.timeout,
            check=check,
            shell=isinstance(command, str),
        )
        duration = time.time() - start_time

        return ScriptResult(
            returncode=result.returncode,
            stdout=result.stdout,
            stderr=result.stderr,
            command=command,
            cwd=self.workspace,
            duration=duration,
        )
    except subprocess.CalledProcessError as e:
        duration = time.time() - start_time
        return ScriptResult(
            returncode=e.returncode,
            stdout=e.stdout or "",
            stderr=e.stderr or "",
            command=command,
            cwd=self.workspace,
            duration=duration,
        )

run_script

run_script(
    script_path: Path | str,
    args: list[str] | None = None,
    check: bool = False,
) -> ScriptResult

Execute a bash script in the isolated workspace.

Parameters:

Name	Type	Description	Default
script_path	Path | str	Path to the bash script to execute.	required
args	list[str] | None	Optional arguments to pass to the script.	None
check	bool	If True, raise exception on non-zero exit code.	False

Returns:

Type	Description
ScriptResult	ScriptResult with execution details.

Raises:

Type	Description
CalledProcessError	If check=True and script fails.
TimeoutExpired	If execution exceeds timeout.

Source code in provide/testkit/process/script_fixtures.py

def run_script(
    self,
    script_path: Path | str,
    args: list[str] | None = None,
    check: bool = False,
) -> ScriptResult:
    """Execute a bash script in the isolated workspace.

    Args:
        script_path: Path to the bash script to execute.
        args: Optional arguments to pass to the script.
        check: If True, raise exception on non-zero exit code.

    Returns:
        ScriptResult with execution details.

    Raises:
        subprocess.CalledProcessError: If check=True and script fails.
        subprocess.TimeoutExpired: If execution exceeds timeout.
    """
    script_path = Path(script_path)
    if not script_path.exists():
        raise FileNotFoundError(f"Script not found: {script_path}")

    command = ["bash", str(script_path)]
    if args:
        command.extend(args)

    import time

    start_time = time.time()

    try:
        result = run(
            command,
            cwd=self.workspace,
            env=self.env if self.env else None,
            timeout=self.timeout,
            check=check,
        )
        duration = time.time() - start_time

        return ScriptResult(
            returncode=result.returncode,
            stdout=result.stdout,
            stderr=result.stderr,
            command=command,
            cwd=self.workspace,
            duration=duration,
        )
    except subprocess.CalledProcessError as e:
        duration = time.time() - start_time
        return ScriptResult(
            returncode=e.returncode,
            stdout=e.stdout or "",
            stderr=e.stderr or "",
            command=command,
            cwd=self.workspace,
            duration=duration,
        )

ScriptResult

Result of a bash script execution.

Attributes:

Name	Type	Description
returncode	int	Exit code from the script.
stdout	str	Standard output as string.
stderr	str	Standard error as string.
command	str | list[str]	Command that was executed.
cwd	Path	Working directory where script was executed.
duration	float	Execution duration in seconds.

Attributes

failed property

failed: bool

Check if script failed (non-zero exit code).

Returns:

Type	Description
bool	True if returncode is non-zero, False otherwise.

success property

success: bool

Check if script succeeded (exit code 0).

Returns:

Type	Description
bool	True if returncode is 0, False otherwise.

Functions

assert_directory_exists

assert_directory_exists(
    dir_path: Path, message: str | None = None
) -> None

Assert that a directory exists.

Parameters:

Name	Type	Description	Default
dir_path	Path	Path to the directory.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If directory does not exist.

Source code in provide/testkit/process/script_assertions.py

def assert_directory_exists(dir_path: Path, message: str | None = None) -> None:
    """Assert that a directory exists.

    Args:
        dir_path: Path to the directory.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If directory does not exist.
    """
    if message is None:
        message = f"Directory does not exist: {dir_path}"
    assert dir_path.exists() and dir_path.is_dir(), message

assert_file_contains

assert_file_contains(
    file_path: Path,
    content: str,
    message: str | None = None,
) -> None

Assert that a file contains specific content.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to the file.	required
content	str	Content to search for.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If file does not contain the content.

Source code in provide/testkit/process/script_assertions.py

def assert_file_contains(
    file_path: Path,
    content: str,
    message: str | None = None,
) -> None:
    """Assert that a file contains specific content.

    Args:
        file_path: Path to the file.
        content: Content to search for.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If file does not contain the content.
    """
    if not file_path.exists():
        raise AssertionError(f"File does not exist: {file_path}")

    file_content = file_path.read_text()
    if message is None:
        message = (
            f"File {file_path} does not contain expected content\n"
            f"Expected: {content!r}\n"
            f"File contents: {file_content!r}"
        )
    assert content in file_content, message

assert_file_created

assert_file_created(
    file_path: Path, message: str | None = None
) -> None

Assert that a file was created.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to the file.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If file does not exist.

Source code in provide/testkit/process/script_assertions.py

def assert_file_created(file_path: Path, message: str | None = None) -> None:
    """Assert that a file was created.

    Args:
        file_path: Path to the file.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If file does not exist.
    """
    if message is None:
        message = f"File does not exist: {file_path}"
    assert file_path.exists() and file_path.is_file(), message

assert_file_executable

assert_file_executable(
    file_path: Path, message: str | None = None
) -> None

Assert that a file has executable permissions.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to the file.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If file is not executable.

Source code in provide/testkit/process/script_assertions.py

def assert_file_executable(file_path: Path, message: str | None = None) -> None:
    """Assert that a file has executable permissions.

    Args:
        file_path: Path to the file.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If file is not executable.
    """
    if not file_path.exists():
        raise AssertionError(f"File does not exist: {file_path}")

    import os

    is_executable = os.access(file_path, os.X_OK)

    if message is None:
        message = f"File is not executable: {file_path}"

    assert is_executable, message

assert_file_not_contains

assert_file_not_contains(
    file_path: Path,
    content: str,
    message: str | None = None,
) -> None

Assert that a file does not contain specific content.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to the file.	required
content	str	Content that should not be present.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If file contains the content.

Source code in provide/testkit/process/script_assertions.py

def assert_file_not_contains(
    file_path: Path,
    content: str,
    message: str | None = None,
) -> None:
    """Assert that a file does not contain specific content.

    Args:
        file_path: Path to the file.
        content: Content that should not be present.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If file contains the content.
    """
    if not file_path.exists():
        raise AssertionError(f"File does not exist: {file_path}")

    file_content = file_path.read_text()
    if message is None:
        message = (
            f"File {file_path} unexpectedly contains content\n"
            f"Unexpected: {content!r}\n"
            f"File contents: {file_content!r}"
        )
    assert content not in file_content, message

assert_git_repo_cloned

assert_git_repo_cloned(
    repo_path: Path, message: str | None = None
) -> None

Assert that a git repository was cloned.

Checks that the directory exists and contains a .git directory.

Parameters:

Name	Type	Description	Default
repo_path	Path	Path to the repository.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If repository was not cloned.

Source code in provide/testkit/process/script_assertions.py

def assert_git_repo_cloned(repo_path: Path, message: str | None = None) -> None:
    """Assert that a git repository was cloned.

    Checks that the directory exists and contains a .git directory.

    Args:
        repo_path: Path to the repository.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If repository was not cloned.
    """
    if message is None:
        message = f"Git repository not found at: {repo_path}"

    assert repo_path.exists() and repo_path.is_dir(), f"{message} (directory missing)"

    git_dir = repo_path / ".git"
    assert git_dir.exists() and git_dir.is_dir(), f"{message} (.git directory missing)"

assert_script_exit_code

assert_script_exit_code(
    result: ScriptResult,
    expected_code: int,
    message: str | None = None,
) -> None

Assert that a script exited with a specific code.

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
expected_code	int	Expected exit code.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If exit code does not match.

Source code in provide/testkit/process/script_assertions.py

def assert_script_exit_code(
    result: ScriptResult,
    expected_code: int,
    message: str | None = None,
) -> None:
    """Assert that a script exited with a specific code.

    Args:
        result: Script execution result.
        expected_code: Expected exit code.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If exit code does not match.
    """
    if message is None:
        message = (
            f"Script exit code {result.returncode} does not match expected {expected_code}\n"
            f"Command: {result.command}\n"
            f"Stdout: {result.stdout}\n"
            f"Stderr: {result.stderr}"
        )
    assert result.returncode == expected_code, message

assert_script_failure

assert_script_failure(
    result: ScriptResult, message: str | None = None
) -> None

Assert that a script failed (non-zero exit code).

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If script succeeded.

Source code in provide/testkit/process/script_assertions.py

def assert_script_failure(result: ScriptResult, message: str | None = None) -> None:
    """Assert that a script failed (non-zero exit code).

    Args:
        result: Script execution result.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If script succeeded.
    """
    if message is None:
        message = (
            f"Script unexpectedly succeeded with exit code 0\n"
            f"Command: {result.command}\n"
            f"Stdout: {result.stdout}"
        )
    assert result.returncode != 0, message

assert_script_success

assert_script_success(
    result: ScriptResult, message: str | None = None
) -> None

Assert that a script executed successfully (exit code 0).

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If script did not succeed.

Source code in provide/testkit/process/script_assertions.py

def assert_script_success(result: ScriptResult, message: str | None = None) -> None:
    """Assert that a script executed successfully (exit code 0).

    Args:
        result: Script execution result.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If script did not succeed.
    """
    if message is None:
        message = (
            f"Script failed with exit code {result.returncode}\n"
            f"Command: {result.command}\n"
            f"Stdout: {result.stdout}\n"
            f"Stderr: {result.stderr}"
        )
    assert result.returncode == 0, message

assert_stderr_contains

assert_stderr_contains(
    result: ScriptResult,
    content: str,
    message: str | None = None,
) -> None

Assert that script stderr contains specific content.

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
content	str	Content to search for in stderr.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If stderr does not contain the content.

Source code in provide/testkit/process/script_assertions.py

def assert_stderr_contains(
    result: ScriptResult,
    content: str,
    message: str | None = None,
) -> None:
    """Assert that script stderr contains specific content.

    Args:
        result: Script execution result.
        content: Content to search for in stderr.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If stderr does not contain the content.
    """
    if message is None:
        message = f"Stderr does not contain expected content\nExpected: {content!r}\nStderr: {result.stderr!r}"
    assert content in result.stderr, message

assert_stderr_empty

assert_stderr_empty(
    result: ScriptResult, message: str | None = None
) -> None

Assert that script stderr is empty.

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If stderr is not empty.

Source code in provide/testkit/process/script_assertions.py

def assert_stderr_empty(result: ScriptResult, message: str | None = None) -> None:
    """Assert that script stderr is empty.

    Args:
        result: Script execution result.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If stderr is not empty.
    """
    if message is None:
        message = f"Stderr is not empty: {result.stderr!r}"
    assert not result.stderr or result.stderr.strip() == "", message

assert_stdout_contains

assert_stdout_contains(
    result: ScriptResult,
    content: str,
    message: str | None = None,
) -> None

Assert that script stdout contains specific content.

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
content	str	Content to search for in stdout.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If stdout does not contain the content.

Source code in provide/testkit/process/script_assertions.py

def assert_stdout_contains(
    result: ScriptResult,
    content: str,
    message: str | None = None,
) -> None:
    """Assert that script stdout contains specific content.

    Args:
        result: Script execution result.
        content: Content to search for in stdout.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If stdout does not contain the content.
    """
    if message is None:
        message = f"Stdout does not contain expected content\nExpected: {content!r}\nStdout: {result.stdout!r}"
    assert content in result.stdout, message

assert_stdout_empty

assert_stdout_empty(
    result: ScriptResult, message: str | None = None
) -> None

Assert that script stdout is empty.

Parameters:

Name	Type	Description	Default
result	ScriptResult	Script execution result.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If stdout is not empty.

Source code in provide/testkit/process/script_assertions.py

def assert_stdout_empty(result: ScriptResult, message: str | None = None) -> None:
    """Assert that script stdout is empty.

    Args:
        result: Script execution result.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If stdout is not empty.
    """
    if message is None:
        message = f"Stdout is not empty: {result.stdout!r}"
    assert not result.stdout or result.stdout.strip() == "", message

assert_symlink_points_to

assert_symlink_points_to(
    symlink_path: Path,
    target_path: Path,
    message: str | None = None,
) -> None

Assert that a symlink points to a specific target.

Parameters:

Name	Type	Description	Default
symlink_path	Path	Path to the symlink.	required
target_path	Path	Expected target path.	required
message	str | None	Optional custom assertion message.	None

Raises:

Type	Description
AssertionError	If symlink does not point to target.

Source code in provide/testkit/process/script_assertions.py

def assert_symlink_points_to(
    symlink_path: Path,
    target_path: Path,
    message: str | None = None,
) -> None:
    """Assert that a symlink points to a specific target.

    Args:
        symlink_path: Path to the symlink.
        target_path: Expected target path.
        message: Optional custom assertion message.

    Raises:
        AssertionError: If symlink does not point to target.
    """
    if not symlink_path.exists():
        raise AssertionError(f"Symlink does not exist: {symlink_path}")

    if not symlink_path.is_symlink():
        raise AssertionError(f"Path is not a symlink: {symlink_path}")

    actual_target = symlink_path.resolve()
    expected_target = target_path.resolve()

    if message is None:
        message = (
            f"Symlink {symlink_path} does not point to expected target\n"
            f"Expected: {expected_target}\n"
            f"Actual: {actual_target}"
        )

    assert actual_target == expected_target, message

async_condition_waiter

async_condition_waiter() -> (
    Callable[
        [Callable[[], bool], float, float], Awaitable[bool]
    ]
)

Helper for waiting on async conditions in tests.

Returns:

Type	Description
Callable[[Callable[[], bool], float, float], Awaitable[bool]]	Function to wait for conditions with timeout.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_condition_waiter() -> Callable[[Callable[[], bool], float, float], Awaitable[bool]]:
    """
    Helper for waiting on async conditions in tests.

    Returns:
        Function to wait for conditions with timeout.
    """

    async def _wait_for(condition: Callable[[], bool], timeout: float = 5.0, interval: float = 0.1) -> bool:
        """
        Wait for a condition to become true.

        Args:
            condition: Function that returns True when condition is met
            timeout: Maximum wait time
            interval: Check interval

        Returns:
            True if condition met, False if timeout
        """
        start = asyncio.get_event_loop().time()

        while asyncio.get_event_loop().time() - start < timeout:
            if condition():
                return True
            await asyncio.sleep(interval)

        return False

    return _wait_for

async_context_manager async

async_context_manager() -> (
    Callable[[Any | None, Any | None], AsyncMock]
)

Factory for creating mock async context managers.

Returns:

Type	Description
Callable[[Any | None, Any | None], AsyncMock]	Function that creates configured async context manager mocks.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
async def async_context_manager() -> Callable[[Any | None, Any | None], AsyncMock]:
    """
    Factory for creating mock async context managers.

    Returns:
        Function that creates configured async context manager mocks.
    """

    def _create_async_cm(enter_value: Any | None = None, exit_value: Any | None = None) -> AsyncMock:
        """
        Create a mock async context manager.

        Args:
            enter_value: Value to return from __aenter__
            exit_value: Value to return from __aexit__

        Returns:
            AsyncMock configured as context manager
        """
        mock_cm = AsyncMock()
        mock_cm.__aenter__ = AsyncMock(return_value=enter_value)
        mock_cm.__aexit__ = AsyncMock(return_value=exit_value)
        return mock_cm

    return _create_async_cm

async_gather_helper

async_gather_helper() -> (
    Callable[..., Awaitable[list[Any]]]
)

Helper for testing asyncio.gather operations.

Returns:

Type	Description
Callable[..., Awaitable[list[Any]]]	Function to gather async results with error handling.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_gather_helper() -> Callable[..., Awaitable[list[Any]]]:
    """
    Helper for testing asyncio.gather operations.

    Returns:
        Function to gather async results with error handling.
    """

    async def _gather(*coroutines: Awaitable[Any], return_exceptions: bool = False) -> list[Any]:
        """
        Gather results from multiple coroutines.

        Args:
            *coroutines: Coroutines to gather
            return_exceptions: Whether to return exceptions as results

        Returns:
            List of results from coroutines
        """
        return await asyncio.gather(*coroutines, return_exceptions=return_exceptions)

    return _gather

async_iterator async

async_iterator() -> (
    Callable[[Sequence[T]], AsyncIterable[T]]
)

Factory for creating mock async iterators.

Returns:

Type	Description
Callable[[Sequence[T]], AsyncIterable[T]]	Function that creates async iterator mocks with specified values.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
async def async_iterator() -> Callable[[Sequence[T]], AsyncIterable[T]]:
    """
    Factory for creating mock async iterators.

    Returns:
        Function that creates async iterator mocks with specified values.
    """

    def _create_async_iter(values: Sequence[T]) -> AsyncIterable[T]:
        """
        Create a mock async iterator.

        Args:
            values: List of values to yield

        Returns:
            Async iterator that yields the specified values
        """
        return _AsyncIterator(values)

    return _create_async_iter

async_lock async

async_lock() -> asyncio.Lock

Create an async lock for testing synchronization.

Returns:

Type	Description
Lock	asyncio.Lock instance for testing.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
async def async_lock() -> asyncio.Lock:
    """
    Create an async lock for testing synchronization.

    Returns:
        asyncio.Lock instance for testing.
    """
    return asyncio.Lock()

async_mock_server

async_mock_server() -> AsyncMockServer

Create a mock async server for testing.

Returns:

Type	Description
AsyncMockServer	Mock server with async methods.

Source code in provide/testkit/process/subprocess_fixtures.py

@pytest.fixture
def async_mock_server() -> AsyncMockServer:
    """
    Create a mock async server for testing.

    Returns:
        Mock server with async methods.
    """

    return AsyncMockServer()

async_pipeline

async_pipeline() -> AsyncPipeline

Create an async pipeline for testing data flow.

Returns:

Type	Description
AsyncPipeline	AsyncPipeline instance for chaining async operations.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_pipeline() -> AsyncPipeline:
    """
    Create an async pipeline for testing data flow.

    Returns:
        AsyncPipeline instance for chaining async operations.
    """

    return AsyncPipeline()

async_queue

async_queue() -> asyncio.Queue[Any]

Create an async queue for testing producer/consumer patterns.

Returns:

Type	Description
Queue[Any]	asyncio.Queue instance for testing.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_queue() -> asyncio.Queue[Any]:
    """
    Create an async queue for testing producer/consumer patterns.

    Returns:
        asyncio.Queue instance for testing.
    """
    return asyncio.Queue()

async_rate_limiter

async_rate_limiter() -> AsyncRateLimiter

Create an async rate limiter for testing.

Returns:

Type	Description
AsyncRateLimiter	AsyncRateLimiter instance for controlling request rates.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_rate_limiter() -> AsyncRateLimiter:
    """
    Create an async rate limiter for testing.

    Returns:
        AsyncRateLimiter instance for controlling request rates.
    """

    return AsyncRateLimiter()

async_stream_reader async

async_stream_reader() -> AsyncMock

Mock async stream reader for subprocess stdout/stderr.

Returns:

Type	Description
AsyncMock	AsyncMock configured as a stream reader.

Source code in provide/testkit/process/subprocess_fixtures.py

@pytest.fixture
async def async_stream_reader() -> AsyncMock:
    """
    Mock async stream reader for subprocess stdout/stderr.

    Returns:
        AsyncMock configured as a stream reader.
    """
    reader = AsyncMock()

    # Simulate reading lines
    async def readline_side_effect() -> AsyncGenerator[bytes, None]:
        for line in [b"line1\n", b"line2\n", b""]:
            yield line

    reader.readline = AsyncMock(side_effect=readline_side_effect().__anext__)
    reader.read = AsyncMock(return_value=b"full content")
    reader.at_eof = Mock(side_effect=[False, False, True])

    return reader

async_subprocess

async_subprocess() -> (
    Callable[[int, bytes, bytes, int], AsyncMock]
)

Create mock async subprocess for testing.

Returns:

Type	Description
Callable[[int, bytes, bytes, int], AsyncMock]	Function that creates mock subprocess with configurable behavior.

Source code in provide/testkit/process/subprocess_fixtures.py

@pytest.fixture
def async_subprocess() -> Callable[[int, bytes, bytes, int], AsyncMock]:
    """
    Create mock async subprocess for testing.

    Returns:
        Function that creates mock subprocess with configurable behavior.
    """

    def _create_subprocess(
        returncode: int = 0,
        stdout: bytes = b"",
        stderr: bytes = b"",
        pid: int = 12345,
    ) -> AsyncMock:
        """
        Create a mock async subprocess.

        Args:
            returncode: Process return code
            stdout: Process stdout output
            stderr: Process stderr output
            pid: Process ID

        Returns:
            AsyncMock configured as subprocess
        """
        process = AsyncMock()
        process.returncode = returncode
        process.pid = pid
        process.communicate = AsyncMock(return_value=(stdout, stderr))
        process.wait = AsyncMock(return_value=returncode)
        process.kill = Mock()
        process.terminate = Mock()
        process.send_signal = Mock()

        # Add stdout/stderr as async stream readers
        process.stdout = AsyncMock()
        process.stdout.read = AsyncMock(return_value=stdout)
        process.stdout.readline = AsyncMock(side_effect=[stdout, b""])
        process.stdout.at_eof = Mock(side_effect=[False, True])

        process.stderr = AsyncMock()
        process.stderr.read = AsyncMock(return_value=stderr)
        process.stderr.readline = AsyncMock(side_effect=[stderr, b""])
        process.stderr.at_eof = Mock(side_effect=[False, True])

        process.stdin = AsyncMock()
        process.stdin.write = AsyncMock()
        process.stdin.drain = AsyncMock()
        process.stdin.close = Mock()

        return process

    return _create_subprocess

async_task_group

async_task_group() -> AsyncTaskGroup

Manage a group of async tasks with cleanup.

Returns:

Type	Description
AsyncTaskGroup	AsyncTaskGroup instance for managing tasks.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_task_group() -> AsyncTaskGroup:
    """
    Manage a group of async tasks with cleanup.

    Returns:
        AsyncTaskGroup instance for managing tasks.
    """

    return AsyncTaskGroup()

async_test_client

async_test_client() -> AsyncTestClient

Create an async HTTP test client.

Returns:

Type	Description
AsyncTestClient	Mock async HTTP client for testing.

Source code in provide/testkit/process/subprocess_fixtures.py

@pytest.fixture
def async_test_client() -> AsyncTestClient:
    """
    Create an async HTTP test client.

    Returns:
        Mock async HTTP client for testing.
    """

    return AsyncTestClient()

async_timeout

async_timeout() -> (
    Callable[[Awaitable[T], float], Awaitable[T]]
)

Provide configurable timeout wrapper for async operations.

Returns:

Type	Description
Callable[[Awaitable[T], float], Awaitable[T]]	A function that wraps async operations with a timeout.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def async_timeout() -> Callable[[Awaitable[T], float], Awaitable[T]]:
    """
    Provide configurable timeout wrapper for async operations.

    Returns:
        A function that wraps async operations with a timeout.
    """

    def _timeout_wrapper(coro: Awaitable[T], seconds: float = 5.0) -> Awaitable[T]:
        """
        Wrap a coroutine with a timeout.

        Args:
            coro: Coroutine to wrap
            seconds: Timeout in seconds

        Returns:
            Result of the coroutine or raises asyncio.TimeoutError
        """
        return asyncio.wait_for(coro, timeout=seconds)

    return _timeout_wrapper

bash_script_runner

bash_script_runner(
    isolated_workspace: Path,
) -> Callable[
    [Path | str, list[str] | None, dict[str, str] | None],
    ScriptResult,
]

Create a bash script runner fixture.

Returns:

Type	Description
Callable[[Path | str, list[str] | None, dict[str, str] | None], ScriptResult]	Function that executes bash scripts and returns results.

Example

def test_my_script(bash_script_runner): result = bash_script_runner("path/to/script.sh", args=["--flag"]) assert result.success assert "expected output" in result.stdout

Source code in provide/testkit/process/script_fixtures.py

@pytest.fixture
def bash_script_runner(
    isolated_workspace: Path,
) -> Callable[[Path | str, list[str] | None, dict[str, str] | None], ScriptResult]:
    """Create a bash script runner fixture.

    Returns:
        Function that executes bash scripts and returns results.

    Example:
        def test_my_script(bash_script_runner):
            result = bash_script_runner("path/to/script.sh", args=["--flag"])
            assert result.success
            assert "expected output" in result.stdout
    """

    def _run_script(
        script_path: Path | str,
        args: list[str] | None = None,
        env: dict[str, str] | None = None,
        timeout: int = 60,
    ) -> ScriptResult:
        """Execute a bash script.

        Args:
            script_path: Path to bash script.
            args: Optional script arguments.
            env: Optional environment variables.
            timeout: Timeout in seconds.

        Returns:
            ScriptResult with execution details.
        """
        context = ScriptExecutionContext(
            workspace=isolated_workspace,
            env=env or {},
            timeout=timeout,
        )
        return context.run_script(script_path, args=args)

    return _run_script

clean_event_loop async

clean_event_loop() -> AsyncGenerator[None, None]

Ensure clean event loop for async tests.

Cancels all pending tasks after the test to prevent event loop issues.

Yields:

Type	Description
AsyncGenerator[None, None]	None - fixture for test setup/teardown.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
async def clean_event_loop() -> AsyncGenerator[None, None]:
    """
    Ensure clean event loop for async tests.

    Cancels all pending tasks after the test to prevent event loop issues.

    Yields:
        None - fixture for test setup/teardown.
    """
    yield

    # Clean up any pending tasks
    loop = asyncio.get_event_loop()
    pending = asyncio.all_tasks(loop)

    for task in pending:
        if not task.done():
            task.cancel()

    # Wait for all tasks to complete cancellation
    if pending:
        await asyncio.gather(*pending, return_exceptions=True)

disable_setproctitle

disable_setproctitle() -> Generator[None, None, None]

Disables setproctitle during tests to prevent pytest-xdist performance issues.

The setproctitle module causes severe performance degradation when running tests with pytest-xdist parallelization: - Immediate slowdown on test start - Progressive performance degradation over time - High CPU usage due to frequent system calls

Mocks setproctitle as a no-op during regular test runs while preserving functionality for mutation testing tools like mutmut that use it for displaying progress information.

Autouse and session-scoped - applies automatically to all tests in a session.

Yields:

Name	Type	Description
None	None	Context manager for test execution with setproctitle disabled

Note

Only disables setproctitle when NOT running under mutmut. Mutmut detection uses sys.argv to check for "mutmut" in arguments.

Source code in provide/testkit/process/system_fixtures.py

@pytest.fixture(scope="session", autouse=True)
def disable_setproctitle() -> Generator[None, None, None]:
    """Disables setproctitle during tests to prevent pytest-xdist performance issues.

    The setproctitle module causes severe performance degradation when running
    tests with pytest-xdist parallelization:
    - Immediate slowdown on test start
    - Progressive performance degradation over time
    - High CPU usage due to frequent system calls

    Mocks setproctitle as a no-op during regular test runs while preserving
    functionality for mutation testing tools like mutmut that use it for
    displaying progress information.

    Autouse and session-scoped - applies automatically to all tests in a session.

    Yields:
        None: Context manager for test execution with setproctitle disabled

    Note:
        Only disables setproctitle when NOT running under mutmut.
        Mutmut detection uses sys.argv to check for "mutmut" in arguments.
    """
    # Only disable if not running under mutmut
    # mutmut needs setproctitle to show which mutations are being tested
    if not any("mutmut" in arg for arg in sys.argv):
        # Check if setproctitle is already imported
        original_module = sys.modules.get("setproctitle")

        # Create a mock module with common setproctitle functions
        mock_setproctitle = MagicMock()
        mock_setproctitle.setproctitle = MagicMock(return_value=None)
        mock_setproctitle.getproctitle = MagicMock(return_value="python")
        mock_setproctitle.setthreadtitle = MagicMock(return_value=None)
        mock_setproctitle.getthreadtitle = MagicMock(return_value="")

        # Inject the mock into sys.modules before any imports
        sys.modules["setproctitle"] = mock_setproctitle

        try:
            yield
        finally:
            # Restore original module if it existed
            if original_module is not None:
                sys.modules["setproctitle"] = original_module
            elif "setproctitle" in sys.modules:
                del sys.modules["setproctitle"]
    else:
        # When running under mutmut, don't disable setproctitle
        yield

event_loop_policy

event_loop_policy() -> (
    Generator[asyncio.AbstractEventLoopPolicy, None, None]
)

Set event loop policy for tests to avoid conflicts.

Returns:

Type	Description
None	New event loop policy for isolated testing.

Source code in provide/testkit/process/async_fixtures.py

@pytest.fixture
def event_loop_policy() -> Generator[asyncio.AbstractEventLoopPolicy, None, None]:
    """
    Set event loop policy for tests to avoid conflicts.

    Returns:
        New event loop policy for isolated testing.
    """
    policy = asyncio.get_event_loop_policy()
    new_policy = asyncio.DefaultEventLoopPolicy()
    asyncio.set_event_loop_policy(new_policy)

    yield new_policy

    # Restore original policy
    asyncio.set_event_loop_policy(policy)

git_workspace

git_workspace() -> Generator[Path, None, None]

Create a temporary workspace with git initialized.

Yields:

Type	Description
Path	Path to git-initialized workspace directory.

Source code in provide/testkit/process/script_fixtures.py

@pytest.fixture
def git_workspace() -> Generator[Path, None, None]:
    """Create a temporary workspace with git initialized.

    Yields:
        Path to git-initialized workspace directory.
    """
    with temp_dir(prefix="git_test_") as workspace:
        if not shutil.which("git"):
            pytest.skip("git command not available")

        run(["git", "init"], cwd=workspace, check=True)
        run(["git", "config", "user.email", "[email protected]"], cwd=workspace, check=True)
        run(["git", "config", "user.name", "Test User"], cwd=workspace, check=True)

        yield workspace

isolated_workspace

isolated_workspace() -> Generator[Path, None, None]

Create an isolated temporary workspace directory.

Yields:

Type	Description
Path	Path to temporary workspace directory.

Source code in provide/testkit/process/script_fixtures.py

@pytest.fixture
def isolated_workspace() -> Generator[Path, None, None]:
    """Create an isolated temporary workspace directory.

    Yields:
        Path to temporary workspace directory.
    """
    with temp_dir(prefix="workspace_test_") as workspace:
        yield workspace

mock_async_process

mock_async_process() -> AsyncMock

Mock async subprocess for testing.

Returns:

Type	Description
AsyncMock	AsyncMock configured as a subprocess with common attributes.

Source code in provide/testkit/process/subprocess_fixtures.py

@pytest.fixture
def mock_async_process() -> AsyncMock:
    """
    Mock async subprocess for testing.

    Returns:
        AsyncMock configured as a subprocess with common attributes.
    """
    mock_process = AsyncMock()
    mock_process.communicate = AsyncMock(return_value=(b"output", b""))
    mock_process.returncode = 0
    mock_process.pid = 12345
    mock_process.stdin = AsyncMock()
    mock_process.stdout = AsyncMock()
    mock_process.stderr = AsyncMock()
    mock_process.wait = AsyncMock(return_value=0)
    mock_process.kill = Mock()
    mock_process.terminate = Mock()

    return mock_process

mock_git_repo

mock_git_repo() -> Callable[[Path, str], Path]

Create a factory for mock git repositories.

Returns:

Type	Description
Callable[[Path, str], Path]	Function that creates a mock git repo with commits.

Example

def test_git_clone(mock_git_repo, isolated_workspace): repo = mock_git_repo(isolated_workspace, "test-repo") # repo is a path to a git repository with initial commit

Source code in provide/testkit/process/script_fixtures.py

@pytest.fixture
def mock_git_repo() -> Callable[[Path, str], Path]:
    """Create a factory for mock git repositories.

    Returns:
        Function that creates a mock git repo with commits.

    Example:
        def test_git_clone(mock_git_repo, isolated_workspace):
            repo = mock_git_repo(isolated_workspace, "test-repo")
            # repo is a path to a git repository with initial commit
    """

    def _create_repo(base_dir: Path, repo_name: str, add_commits: bool = True) -> Path:
        """Create a mock git repository.

        Args:
            base_dir: Directory to create repo in.
            repo_name: Name of the repository.
            add_commits: If True, add an initial commit.

        Returns:
            Path to the created git repository.
        """
        if not shutil.which("git"):
            pytest.skip("git command not available")

        repo_path = base_dir / repo_name
        repo_path.mkdir(parents=True, exist_ok=True)

        run(["git", "init"], cwd=repo_path, check=True)
        run(["git", "config", "user.email", "[email protected]"], cwd=repo_path, check=True)
        run(["git", "config", "user.name", "Test User"], cwd=repo_path, check=True)

        if add_commits:
            readme = repo_path / "README.md"
            readme.write_text(f"# {repo_name}\n\nTest repository.")
            run(["git", "add", "README.md"], cwd=repo_path, check=True)
            run(["git", "commit", "-m", "Initial commit"], cwd=repo_path, check=True)

        return repo_path

    return _create_repo

script_execution_context

script_execution_context() -> (
    Generator[ScriptExecutionContext, None, None]
)

Create an isolated script execution context.

Provides a temporary workspace directory with utilities for running bash scripts in isolation.

Yields:

Type	Description
ScriptExecutionContext	ScriptExecutionContext with temporary workspace.

Source code in provide/testkit/process/script_fixtures.py

@pytest.fixture
def script_execution_context() -> Generator[ScriptExecutionContext, None, None]:
    """Create an isolated script execution context.

    Provides a temporary workspace directory with utilities
    for running bash scripts in isolation.

    Yields:
        ScriptExecutionContext with temporary workspace.
    """
    with temp_dir(prefix="script_test_") as workspace:
        yield ScriptExecutionContext(workspace=workspace)