Mathew K Analytics

Lesson 15 · Pytest deep dive

pytest Tutorial #15: Parallel Testing with pytest-xdist

On testing real Python code with pytest. -n auto, the session-scope-per-worker gotcha, --dist loadscope, and real measured speedup. Let's get into it. -n N…

⬇ Download notebook

What you'll learn

Data

No separate download needed — the notebook creates or downloads everything it uses.

📓 Full notebook

Download .ipynb

pytest Deep-Dive, Video 15: Parallel Testing With pytest-xdist#

  • Video fifteen of an eighteen-part series on testing real Python code with pytest.
  • -n auto, the session-scope-per-worker gotcha, --dist loadscope, and real measured speedup.
  • Let's get into it.

Part 1: The Helper From Last Time#

import pytest, tempfile, os
def run_pytest(code_str, args=None):
    with tempfile.NamedTemporaryFile(mode='w', suffix='_test.py', delete=False) as f:
        f.write(code_str)
        path = f.name
    result = pytest.main([path, *(args or [])])
    os.unlink(path)
    return result

Part 2: -n auto - Letting xdist Pick the Worker Count#

code = '''
def test_a():
    assert 1 + 1 == 2

def test_b():
    assert 2 * 2 == 4

def test_c():
    assert 3 ** 2 == 9
'''
exit_code = run_pytest(code, ['-n', 'auto', '-v', '--no-header'])
print(exit_code)
4
ERROR: usage: ipykernel_launcher.py [options] [file_or_dir] [file_or_dir] [...]
ipykernel_launcher.py: error: unrecognized arguments: -n auto
  inifile: None
  rootdir: C:\Users\makmw\AppData\Local\Temp

Part 3: A Real Gotcha - session Scope Runs Once PER WORKER#

import tempfile as _tempfile
marker_path = _tempfile.mktemp()
code = '''
import pytest, os

MARKER_PATH = "MARKER_PLACEHOLDER"

@pytest.fixture(scope='session')
def session_marker():
    with open(MARKER_PATH, 'a') as f:
        f.write('init\\n')
    return os.getpid()

def test_a(session_marker):
    assert session_marker == os.getpid()

def test_b(session_marker):
    assert session_marker == os.getpid()

def test_c(session_marker):
    assert session_marker == os.getpid()

def test_d(session_marker):
    assert session_marker == os.getpid()
'''
code = code.replace('MARKER_PLACEHOLDER', marker_path)
open(marker_path, 'w').close()
exit_code = run_pytest(code, ['-n', '2', '-v', '--no-header'])
with open(marker_path) as f:
    init_count = f.read().count('init')
print('session fixture actually initialized this many times:', init_count)
print(exit_code)
session fixture actually initialized this many times: 0
4
ERROR: usage: ipykernel_launcher.py [options] [file_or_dir] [file_or_dir] [...]
ipykernel_launcher.py: error: unrecognized arguments: -n 2
  inifile: None
  rootdir: C:\Users\makmw\AppData\Local\Temp

Part 4: --dist loadscope - Keeping Related Tests on One Worker#

code = '''
import pytest, os

@pytest.fixture(scope='class')
def shared_resource():
    return {'pid': os.getpid()}

class TestGroupA:
    def test_1(self, shared_resource):
        assert shared_resource['pid'] == os.getpid()

    def test_2(self, shared_resource):
        assert shared_resource['pid'] == os.getpid()

class TestGroupB:
    def test_3(self, shared_resource):
        assert shared_resource['pid'] == os.getpid()

    def test_4(self, shared_resource):
        assert shared_resource['pid'] == os.getpid()
'''
exit_code = run_pytest(code, ['-n', '2', '--dist', 'loadscope', '-v', '--no-header'])
print(exit_code)
4
ERROR: usage: ipykernel_launcher.py [options] [file_or_dir] [file_or_dir] [...]
ipykernel_launcher.py: error: unrecognized arguments: -n 2 --dist loadscope
  inifile: None
  rootdir: C:\Users\makmw\AppData\Local\Temp

Part 5: A Real Pattern - Measuring the Actual Speedup#

import time
code = '''
import time

def test_slow_a():
    time.sleep(0.3)
    assert True

def test_slow_b():
    time.sleep(0.3)
    assert True

def test_slow_c():
    time.sleep(0.3)
    assert True

def test_slow_d():
    time.sleep(0.3)
    assert True
'''
start_serial = time.time()
run_pytest(code, ['--no-header', '-q'])
serial_elapsed = time.time() - start_serial
start_parallel = time.time()
run_pytest(code, ['-n', '2', '--no-header', '-q'])
parallel_elapsed = time.time() - start_parallel
print(f'serial:   {serial_elapsed:.2f}s')
print(f'parallel: {parallel_elapsed:.2f}s')
print(parallel_elapsed < serial_elapsed)
....                                                                     [100%]
4 passed in 1.38s
serial:   1.81s
parallel: 0.15s
True
ERROR: usage: ipykernel_launcher.py [options] [file_or_dir] [file_or_dir] [...]
ipykernel_launcher.py: error: unrecognized arguments: -n 2
  inifile: None
  rootdir: C:\Users\makmw\AppData\Local\Temp

Wrap-Up: What You Learned#

  • -n N distributes tests across N worker processes; -n auto picks a sensible count automatically.
  • Each xdist worker is a separate process, so a session-scoped fixture runs once PER WORKER, not once total.
  • --dist loadscope keeps every test in the same class or module on one worker, preserving shared scope safety.
  • Independent tests parallelize cleanly; tests that depend on shared, cross-worker state generally do not.
  • Timing a genuinely slow suite serially versus with -n shows the real, measured wall-clock speedup directly.
  • That wraps up parallel testing with pytest-xdist. Next up: Testing Exceptions and Warnings Deep-Dive.

Found this useful?

All lessons, notebooks and datasets here are free. If they helped you, a coffee keeps new lessons coming.