Skip to content

Instantly share code, notes, and snippets.

@sublee
Last active December 24, 2015 07:48
Show Gist options
  • Save sublee/6765822 to your computer and use it in GitHub Desktop.
Save sublee/6765822 to your computer and use it in GitHub Desktop.
A mark decorator "pytest.mark.boxed" to run a test in forked process.
# -*- coding: utf-8 -*-
import multiprocessing
from _pytest import runner
def pytest_runtest_protocol(item):
if 'boxed' in item.keywords:
reports = boxed_run_report(item)
for rep in reports:
item.ihook.pytest_runtest_logreport(report=rep)
return True
def boxed_run_report(item):
"""Seems to be :func:`xdist.plugin.forked_run_report` but it uses
:mod:`multiprocessing`.
"""
EXITSTATUS_TESTEXIT = 4
def run_test(pipe):
pipe.send(runner.runtestprotocol(item, log=False))
pipe_in, pipe_out = multiprocessing.Pipe()
proc = multiprocessing.Process(target=run_test, args=(pipe_in,))
proc.start()
proc.join()
if pipe_out.poll():
return pipe_out.recv()
else:
path, lineno = item._getfslineno()
call = runner.CallInfo(lambda: 0/0, '???')
call.excinfo = ('{0}:{1}: running the test CRASHED with '
'exit code {2}'.format(path, lineno, proc.exitcode))
return [runner.pytest_runtest_makereport(item, call)]
# -*- coding: utf-8 -*-
import os
import pytest
PID = os.getpid()
def test_non_boxing():
assert PID == os.getpid()
@pytest.mark.boxed
def test_boxing():
assert PID != os.getpid()
assert PID == os.getppid()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment