pytest-xdist-hacks / xdist /

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'")
           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=[],
           help=("add a test execution environment. some examples: "
                 "--tx popen//python=python2.5 --tx socket= "
           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

# -------------------------------------------------------------------------
# distributed testing initialization
# -------------------------------------------------------------------------

def pytest_cmdline_main(config):
    if config.getvalue("looponfail"):
        from xdist.looponfail import looponfail_main
        return 2 # looponfail only can get stop with ctrl-C anyway

def pytest_configure(config, __multicall__):
    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:
        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
    import marshal
    from xdist.remote import serialize_report
    from xdist.slavemanage import unserialize_report
    def runforked():
            item.nextitem = None
            reports = runtestprotocol(item, log=False)
        except KeyboardInterrupt:
        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]
        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
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.