/usr/share/pyshared/xdist/plugin.py is in python-pytest-xdist 1.8-0.1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import sys
import py, pytest
def pytest_addoption(parser):
group = parser.getgroup("xdist", "distributed and subprocess testing")
group._addoption('-f', '--looponfail',
action="store_true", dest="looponfail", default=False,
help="run tests in subprocess, wait for modified files "
"and re-run failing test set until all pass.")
group._addoption('-n', dest="numprocesses", metavar="numprocesses",
action="store", type="int",
help="shortcut for '--dist=load --tx=NUM*popen'")
group.addoption('--boxed',
action="store_true", dest="boxed", default=False,
help="box each test run in a separate process (unix)")
group._addoption('--dist', metavar="distmode",
action="store", choices=['load', 'each', 'no'],
type="choice", dest="dist", default="no",
help=("set mode for distributing tests to exec environments.\n\n"
"each: send each test to each available environment.\n\n"
"load: send each test to available environment.\n\n"
"(default) no: run tests inprocess, don't distribute."))
group._addoption('--tx', dest="tx", action="append", default=[],
metavar="xspec",
help=("add a test execution environment. some examples: "
"--tx popen//python=python2.5 --tx socket=192.168.1.102:8888 "
"--tx ssh=user@codespeak.net//chdir=testcache"))
group._addoption('-d',
action="store_true", dest="distload", default=False,
help="load-balance tests. shortcut for '--dist=load'")
group.addoption('--rsyncdir', action="append", default=[], metavar="dir1",
help="add directory for rsyncing to remote tx nodes.")
parser.addini('rsyncdirs', 'list of (relative) paths to be rsynced for'
' remote distributed testing.', type="pathlist")
parser.addini('rsyncignore', 'list of (relative) paths to be ignored '
'for rsyncing.', type="pathlist")
parser.addini("looponfailroots", type="pathlist",
help="directories to check for changes", default=[py.path.local()])
# -------------------------------------------------------------------------
# distributed testing hooks
# -------------------------------------------------------------------------
def pytest_addhooks(pluginmanager):
from xdist import newhooks
pluginmanager.addhooks(newhooks)
# -------------------------------------------------------------------------
# distributed testing initialization
# -------------------------------------------------------------------------
def pytest_cmdline_main(config):
check_options(config)
if config.getvalue("looponfail"):
from xdist.looponfail import looponfail_main
looponfail_main(config)
return 2 # looponfail only can get stop with ctrl-C anyway
def pytest_configure(config, __multicall__):
__multicall__.execute()
if config.getvalue("dist") != "no":
from xdist.dsession import DSession
session = DSession(config)
config.pluginmanager.register(session, "dsession")
def check_options(config):
if config.option.numprocesses:
config.option.dist = "load"
config.option.tx = ['popen'] * int(config.option.numprocesses)
if config.option.distload:
config.option.dist = "load"
val = config.getvalue
if not val("collectonly"):
usepdb = config.option.usepdb # a core option
if val("looponfail"):
if usepdb:
raise pytest.UsageError("--pdb incompatible with --looponfail.")
elif val("dist") != "no":
if usepdb:
raise pytest.UsageError("--pdb incompatible with distributing tests.")
def pytest_runtest_protocol(item):
if item.config.getvalue("boxed"):
reports = forked_run_report(item)
for rep in reports:
item.ihook.pytest_runtest_logreport(report=rep)
return True
def forked_run_report(item):
# for now, we run setup/teardown in the subprocess
# XXX optionally allow sharing of setup/teardown
from _pytest.runner import runtestprotocol
EXITSTATUS_TESTEXIT = 4
import marshal
from xdist.remote import serialize_report
from xdist.slavemanage import unserialize_report
def runforked():
try:
reports = runtestprotocol(item, log=False)
except KeyboardInterrupt:
py.std.os._exit(EXITSTATUS_TESTEXIT)
return marshal.dumps([serialize_report(x) for x in reports])
ff = py.process.ForkedFunc(runforked)
result = ff.waitfinish()
if result.retval is not None:
report_dumps = marshal.loads(result.retval)
return [unserialize_report("testreport", x) for x in report_dumps]
else:
if result.exitstatus == EXITSTATUS_TESTEXIT:
py.test.exit("forked test item %s raised Exit" %(item,))
return [report_process_crash(item, result)]
def report_process_crash(item, result):
path, lineno = item._getfslineno()
info = "%s:%s: running the test CRASHED with signal %d" %(
path, lineno, result.signal)
from _pytest import runner
call = runner.CallInfo(lambda: 0/0, "???")
call.excinfo = info
rep = runner.pytest_runtest_makereport(item, call)
return rep
|