# Tests that spawn processes

> Tests that run CLIs with subprocess or start multiprocessing pools and Docker containers pay for a new process each time. Call code in-process, fake subprocess.run, share containers.

Starting a process is expensive. A test that runs `subprocess.run([sys.executable, "-m", "myapp", ...])` starts a new Python interpreter, which imports your whole application again before it does anything, and then the test waits for it to exit. For a Django or data science project that's often a second or more per call, and none of it is the code you meant to test.

The same goes for `multiprocessing` and `ProcessPoolExecutor` in the code under test, and for tests that start a Docker container (directly or through testcontainers) for every test or module.

## Why it's slow

- **Interpreter startup and imports.** A child Python process starts from scratch: the interpreter, `site`, and every import your entry point pulls in. That's the same cost as [pytest's own startup](https://flakytest.dev/guides/slow-pytest-startup/), paid once per call.
- **`spawn` and `forkserver` start methods.** These start fresh interpreters for multiprocessing workers, which re-import the modules they need. `spawn` is the default on macOS and Windows.
- **Containers.** Pulling, creating and starting a container, then waiting for the service inside to accept connections, takes seconds.
- **Waiting.** The test blocks until the child exits, including any timeouts or slow shutdown in the child.

## Call the code in-process

Most CLI tests check what the command does with its arguments, not that the operating system can start a process. Call the entry point as a function.

### Click and Typer: CliRunner

```python
from click.testing import CliRunner

from myapp.cli import cli


def test_export_writes_rows(tmp_path):
    runner = CliRunner()
    result = runner.invoke(cli, ["export", "--out", str(tmp_path / "rows.csv")])

    assert result.exit_code == 0, result.output
    assert "Exported 3 rows" in result.output
```

Typer has the same API in `typer.testing.CliRunner`.

### argparse: call main(argv)

Write `main` so it takes the arguments and returns the exit code, then call it directly and read the output with `capsys`:

```python
# myapp/__main__.py
import argparse
import sys


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("name")
    args = parser.parse_args(argv)
    print(f"Hello, {args.name}")
    return 0


if __name__ == "__main__":
    sys.exit(main())
```

```python
from myapp.__main__ import main


def test_main(capsys):
    assert main(["Ada"]) == 0
    assert capsys.readouterr().out == "Hello, Ada\n"
```

### Django: call_command

```python
from io import StringIO

from django.core.management import call_command
from django.test import TestCase
from model_bakery import baker

from polls.models import Poll


class ClosePollsTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.poll = baker.make(Poll, is_open=True)

    def test_closes_open_polls(self):
        out = StringIO()
        call_command("closepolls", stdout=out)

        self.poll.refresh_from_db()
        self.assertFalse(self.poll.is_open)
        self.assertIn("Closed 1 poll", out.getvalue())
```

`call_command` runs in the test's transaction, so it also sees data from `setUpTestData`, which a `manage.py` subprocess can't.

## Fake subprocess.run for wrapper code

When your code shells out to another program (`git`, `ffmpeg`, `pg_dump`) and the test only checks the command line or how the output is parsed, fake the process. pytest-subprocess hooks `subprocess.Popen`, so `run`, `check_output` and friends work too:

```python
def test_current_branch(fp):
    fp.register(["git", "branch", "--show-current"], stdout="main\n")

    assert current_branch() == "main"
```

An unregistered command raises an error, so the test also fails if the command line changes. Plain `unittest.mock.patch("myapp.git.subprocess.run")` works as well, returning a `subprocess.CompletedProcess(args=[...], returncode=0, stdout="main\n")`.

## Process pools: inject the executor

If the code under test uses `ProcessPoolExecutor` or `multiprocessing.Pool`, make the executor a parameter (or a setting) and use a thread pool or a synchronous executor in tests. The work runs in the test process, exceptions have normal tracebacks, and coverage and the profiler see it. A synchronous executor is shown in [tests blocked on threads, queues and the event loop](https://flakytest.dev/guides/waiting-in-tests/#run-background-work-synchronously).

## Containers: start them once per session

Starting a container per test, or per module, multiplies the startup cost. Start it once with a session-scoped fixture and reset its state per test:

```python
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer


@pytest.fixture(scope="session")
def postgres_url():
    with PostgresContainer("postgres:17") as postgres:
        yield postgres.get_connection_url()


@pytest.fixture(scope="session")
def redis_container():
    with RedisContainer("redis:7") as redis:
        yield redis


@pytest.fixture
def redis_client(redis_container):
    client = redis_container.get_client()
    yield client
    client.flushdb()
```

For the database, combine the session-scoped server with a transaction that's rolled back per test (Django's `TestCase`, or [a SAVEPOINT per test in SQLAlchemy](https://flakytest.dev/guides/sqlalchemy-rollback-per-test/)). With pytest-xdist each worker starts its own containers; in CI, a service container defined in the CI job is started once for all of them.

## Keep the tests correct

- Keep one end-to-end test that runs the real command (the installed entry point or `python -m myapp`), so packaging, the `__main__` guard and exit codes stay covered.
- In-process calls share state with the test process: module-level caches, environment variables, `sys.argv`, the working directory. Use `monkeypatch.setenv` and `monkeypatch.chdir` instead of changing them for good.
- A fake for `subprocess.run` checks your command line, not that the program accepts it. Keep a test against the real program where its flags matter.
- Shared containers keep data between tests: reset it (truncate, `flushdb`, roll back) in a function-scoped fixture.

## How flakytest finds it

flakytest puts time spent in `subprocess`, `asyncio.subprocess`, `multiprocessing`, testcontainers and the Docker SDK in the *subprocess* category. When it's more than a second and 2% of test time, it suggests **Tests spawn processes**, with the tests and lines that start the most expensive processes. If part of that time is in testcontainers or Docker, the suggestion also says to start containers once per session. The child processes themselves aren't profiled, so their time shows up as the parent waiting.

## Further reading

- [Click: testing click applications](https://click.palletsprojects.com/en/stable/testing/)
- [Django: running management commands from your code](https://docs.djangoproject.com/en/stable/ref/django-admin/#running-management-commands-from-your-code)
- [pytest-subprocess](https://pytest-subprocess.readthedocs.io/)
- [testcontainers-python](https://testcontainers-python.readthedocs.io/)