Skip to content

Subprocess fixtures

provide.testkit.process.subprocess_fixtures

Subprocess-specific test fixtures for process testing.

Provides fixtures for mocking and testing subprocess operations, stream handling, and process communication.

Classes

AsyncMockServer 

AsyncMockServer()

Simple async HTTP-like server mock.

Source code in provide/testkit/process/subprocess_fixtures.py 
24
25
26
27
28
29
def __init__(self) -> None:
    self.started = False
    self.host = "localhost"
    self.port = 8080
    self.connections: list[dict[str, Any]] = []
    self.requests: list[bytes] = []
Functions
get_url 
get_url() -> str

Return the server URL.

Source code in provide/testkit/process/subprocess_fixtures.py 
59
60
61
def get_url(self) -> str:
    """Return the server URL."""
    return f"http://{self.host}:{self.port}"
handle_connection async 
handle_connection(reader: Any, writer: Any) -> None

Mock connection handler.

Source code in provide/testkit/process/subprocess_fixtures.py 
46
47
48
49
50
51
52
53
54
55
56
57
async def handle_connection(self, reader: Any, writer: Any) -> None:
    """Mock connection handler."""
    connection = {"reader": reader, "writer": writer}
    self.connections.append(connection)

    data = await reader.read(1024)
    self.requests.append(data)

    writer.write(b"HTTP/1.1 200 OK\r\n\r\nOK")
    await writer.drain()
    writer.close()
    await writer.wait_closed()
start async 
start(host: str = 'localhost', port: int = 8080) -> None

Start the mock server.

Source code in provide/testkit/process/subprocess_fixtures.py 
31
32
33
34
35
async def start(self, host: str = "localhost", port: int = 8080) -> None:
    """Start the mock server."""
    self.started = True
    self.host = host
    self.port = port
stop async 
stop() -> None

Stop the mock server and close connections.

Source code in provide/testkit/process/subprocess_fixtures.py 
37
38
39
40
41
42
43
44
async def stop(self) -> None:
    """Stop the mock server and close connections."""
    self.started = False
    for conn in self.connections:
        writer = conn.get("writer")
        if writer is not None:
            writer.close()
            await writer.wait_closed()

AsyncTestClient 

AsyncTestClient()

Lightweight async HTTP client mock.

Source code in provide/testkit/process/subprocess_fixtures.py 
67
68
69
def __init__(self) -> None:
    self.responses: dict[str, dict[str, Any]] = {}
    self.requests: list[dict[str, Any]] = []
Functions
close async 
close() -> None

Close the client.

Source code in provide/testkit/process/subprocess_fixtures.py 
85
86
87
async def close(self) -> None:
    """Close the client."""
    return None
get async 
get(url: str, **kwargs: Any) -> dict[str, Any]

Mock GET request.

Source code in provide/testkit/process/subprocess_fixtures.py 
75
76
77
78
async def get(self, url: str, **kwargs: Any) -> dict[str, Any]:
    """Mock GET request."""
    self.requests.append({"method": "GET", "url": url, "kwargs": kwargs})
    return self.responses.get(url, {"status": 404, "body": "Not Found"})
post async 
post(
    url: str, data: Any | None = None, **kwargs: Any
) -> dict[str, Any]

Mock POST request.

Source code in provide/testkit/process/subprocess_fixtures.py 
80
81
82
83
async def post(self, url: str, data: Any | None = None, **kwargs: Any) -> dict[str, Any]:
    """Mock POST request."""
    self.requests.append({"method": "POST", "url": url, "data": data, "kwargs": kwargs})
    return self.responses.get(url, {"status": 200, "body": "OK"})
set_response 
set_response(url: str, response: dict[str, Any]) -> None

Register a mock response for a URL.

Source code in provide/testkit/process/subprocess_fixtures.py 
71
72
73
def set_response(self, url: str, response: dict[str, Any]) -> None:
    """Register a mock response for a URL."""
    self.responses[url] = response

Functions

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 
199
200
201
202
203
204
205
206
207
208
@pytest.fixture
def async_mock_server() -> AsyncMockServer:
    """
    Create a mock async server for testing.

    Returns:
        Mock server with async methods.
    """

    return AsyncMockServer()

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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@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_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 
211
212
213
214
215
216
217
218
219
220
@pytest.fixture
def async_test_client() -> AsyncTestClient:
    """
    Create an async HTTP test client.

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

    return AsyncTestClient()

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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@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