[svn r63209] polish command line options for distributed testing.
--HG-- branch : trunk
This commit is contained in:
parent
c81ffeb027
commit
ad6afe21ff
|
@ -253,21 +253,17 @@ class Config(object):
|
|||
raise self.Error("unknown io capturing: " + iocapture)
|
||||
|
||||
def getxspecs(self):
|
||||
config = self
|
||||
if config.option.numprocesses:
|
||||
xspeclist = ['popen'] * config.option.numprocesses
|
||||
else:
|
||||
xspeclist = []
|
||||
for xspec in config.getvalue("tx"):
|
||||
i = xspec.find("*")
|
||||
try:
|
||||
num = int(xspec[:i])
|
||||
except ValueError:
|
||||
xspeclist.append(xspec)
|
||||
else:
|
||||
xspeclist.extend([xspec[i+1:]] * num)
|
||||
xspeclist = []
|
||||
for xspec in self.getvalue("tx"):
|
||||
i = xspec.find("*")
|
||||
try:
|
||||
num = int(xspec[:i])
|
||||
except ValueError:
|
||||
xspeclist.append(xspec)
|
||||
else:
|
||||
xspeclist.extend([xspec[i+1:]] * num)
|
||||
if not xspeclist:
|
||||
raise config.Error("MISSING test execution (tx) nodes: please specify --tx")
|
||||
raise self.Error("MISSING test execution (tx) nodes: please specify --tx")
|
||||
return [py.execnet.XSpec(x) for x in xspeclist]
|
||||
|
||||
def getrsyncdirs(self):
|
||||
|
|
|
@ -1,21 +1,14 @@
|
|||
import py
|
||||
|
||||
Module = py.test.collect.Module
|
||||
#DoctestFile = py.test.collect.DoctestFile
|
||||
Directory = py.test.collect.Directory
|
||||
File = py.test.collect.File
|
||||
|
||||
# python collectors
|
||||
Class = py.test.collect.Class
|
||||
Generator = py.test.collect.Generator
|
||||
Function = py.test.collect.Function
|
||||
Instance = py.test.collect.Instance
|
||||
|
||||
|
||||
pytest_plugins = "default terminal xfail tmpdir execnetcleanup monkeypatch".split()
|
||||
|
||||
# ===================================================
|
||||
# settings in conftest only (for now) - for distribution
|
||||
|
||||
if hasattr(py.std.os, 'nice'):
|
||||
dist_nicelevel = py.std.os.nice(0) # nice py.test works
|
||||
else:
|
||||
dist_nicelevel = 0
|
||||
|
||||
|
|
|
@ -62,19 +62,14 @@ class DSession(Session):
|
|||
self.item2nodes = {}
|
||||
super(DSession, self).__init__(config=config)
|
||||
|
||||
def pytest_configure(self, config):
|
||||
if self.config.getvalue("usepdb"):
|
||||
raise self.config.Error("--pdb does not work for distributed tests (yet).")
|
||||
def pytest_configure(self, __call__, config):
|
||||
__call__.execute()
|
||||
try:
|
||||
self.config.getxspecs()
|
||||
except self.config.Error:
|
||||
print "Please specify test environments for distribution of tests:"
|
||||
print "py.test --tx ssh=user@somehost --tx popen//python=python2.5"
|
||||
print "conftest.py: pytest_option_tx=['ssh=user@somehost','popen']"
|
||||
print "environment: PYTEST_OPTION_TX=ssh=@somehost,popen"
|
||||
config.getxspecs()
|
||||
except config.Error:
|
||||
print
|
||||
#print "see also: http://codespeak.net/py/current/doc/test.html#automated-distributed-testing"
|
||||
raise SystemExit
|
||||
raise config.Error("dist mode %r needs test execution environments, "
|
||||
"none found." %(config.option.dist))
|
||||
|
||||
def main(self, colitems=None):
|
||||
colitems = self.getinitialitems(colitems)
|
||||
|
@ -126,6 +121,7 @@ class DSession(Session):
|
|||
loopstate.exitstatus = outcome.EXIT_TESTSFAILED
|
||||
else:
|
||||
loopstate.exitstatus = outcome.EXIT_OK
|
||||
#self.config.bus.unregister(loopstate)
|
||||
|
||||
def _initloopstate(self, colitems):
|
||||
loopstate = LoopState(self, colitems)
|
||||
|
|
|
@ -65,26 +65,12 @@ class NodeManager(object):
|
|||
|
||||
def setup_nodes(self, putevent):
|
||||
self.rsync_roots()
|
||||
nice = self.config.getvalue("dist_nicelevel")
|
||||
if nice != 0:
|
||||
self.gwmanager.multi_exec("""
|
||||
import os
|
||||
if hasattr(os, 'nice'):
|
||||
os.nice(%r)
|
||||
""" % nice).waitclose()
|
||||
|
||||
self.trace_nodestatus()
|
||||
multigw = self.gwmanager.getgateways(inplacelocal=False, remote=True)
|
||||
multigw.remote_exec("""
|
||||
import os, sys
|
||||
sys.path.insert(0, os.getcwd())
|
||||
""").waitclose()
|
||||
|
||||
for gateway in self.gwmanager.gateways:
|
||||
node = MasterNode(gateway, self.config, putevent)
|
||||
self.nodes.append(node)
|
||||
|
||||
def teardown_nodes(self):
|
||||
# XXX teardown nodes?
|
||||
# XXX do teardown nodes?
|
||||
self.gwmanager.exit()
|
||||
|
||||
|
|
|
@ -356,3 +356,30 @@ class TestDSession:
|
|||
session.loop_once(loopstate)
|
||||
assert loopstate.colitems == colreport.result
|
||||
assert loopstate.exitstatus is None, "loop did not care for colitems"
|
||||
|
||||
def test_dist_some_tests(self, testdir):
|
||||
from py.__.test.dist.testing.test_txnode import EventQueue
|
||||
p1 = testdir.makepyfile(test_one="""
|
||||
def test_1():
|
||||
pass
|
||||
def test_x():
|
||||
import py
|
||||
py.test.skip("aaa")
|
||||
def test_fail():
|
||||
assert 0
|
||||
""")
|
||||
config = testdir.parseconfig('-d', p1, '--tx=popen')
|
||||
dsession = DSession(config)
|
||||
eq = EventQueue(config.bus)
|
||||
dsession.main([config.getfsnode(p1)])
|
||||
ev, = eq.geteventargs("itemtestreport")
|
||||
assert ev.passed
|
||||
ev, = eq.geteventargs("itemtestreport")
|
||||
assert ev.skipped
|
||||
ev, = eq.geteventargs("itemtestreport")
|
||||
assert ev.failed
|
||||
# see that the node is really down
|
||||
node, error = eq.geteventargs("testnodedown")
|
||||
assert node.gateway.spec.popen
|
||||
eq.geteventargs("testrunfinish")
|
||||
|
||||
|
|
|
@ -1,94 +0,0 @@
|
|||
import py
|
||||
from py.__.test.dist.dsession import DSession
|
||||
from test_txnode import EventQueue
|
||||
|
||||
class TestAsyncFunctional:
|
||||
def test_conftest_options(self, testdir):
|
||||
p1 = testdir.tmpdir.ensure("dir", 'p1.py')
|
||||
p1.dirpath("__init__.py").write("")
|
||||
p1.dirpath("conftest.py").write(py.code.Source("""
|
||||
print "importing conftest", __file__
|
||||
import py
|
||||
Option = py.test.config.Option
|
||||
option = py.test.config.addoptions("someopt",
|
||||
Option('--someopt', action="store_true", dest="someopt", default=False))
|
||||
dist_rsync_roots = ['../dir']
|
||||
print "added options", option
|
||||
print "config file seen from conftest", py.test.config
|
||||
"""))
|
||||
p1.write(py.code.Source("""
|
||||
import py, conftest
|
||||
def test_1():
|
||||
print "config from test_1", py.test.config
|
||||
print "conftest from test_1", conftest.__file__
|
||||
print "test_1: py.test.config.option.someopt", py.test.config.option.someopt
|
||||
print "test_1: conftest", conftest
|
||||
print "test_1: conftest.option.someopt", conftest.option.someopt
|
||||
assert conftest.option.someopt
|
||||
"""))
|
||||
result = testdir.runpytest('-d', '--tx=popen', p1, '--someopt')
|
||||
assert result.ret == 0
|
||||
extra = result.stdout.fnmatch_lines([
|
||||
"*1 passed*",
|
||||
])
|
||||
|
||||
def test_dist_some_tests(self, testdir):
|
||||
p1 = testdir.makepyfile(test_one="""
|
||||
def test_1():
|
||||
pass
|
||||
def test_x():
|
||||
import py
|
||||
py.test.skip("aaa")
|
||||
def test_fail():
|
||||
assert 0
|
||||
""")
|
||||
config = testdir.parseconfig('-d', p1, '--tx=popen')
|
||||
dsession = DSession(config)
|
||||
eq = EventQueue(config.bus)
|
||||
dsession.main([config.getfsnode(p1)])
|
||||
ev, = eq.geteventargs("itemtestreport")
|
||||
assert ev.passed
|
||||
ev, = eq.geteventargs("itemtestreport")
|
||||
assert ev.skipped
|
||||
ev, = eq.geteventargs("itemtestreport")
|
||||
assert ev.failed
|
||||
# see that the node is really down
|
||||
node, error = eq.geteventargs("testnodedown")
|
||||
assert node.gateway.spec.popen
|
||||
eq.geteventargs("testrunfinish")
|
||||
|
||||
def test_distribution_rsyncdirs_example(self, testdir):
|
||||
source = testdir.mkdir("source")
|
||||
dest = testdir.mkdir("dest")
|
||||
subdir = source.mkdir("example_pkg")
|
||||
subdir.ensure("__init__.py")
|
||||
p = subdir.join("test_one.py")
|
||||
p.write("def test_5(): assert not __file__.startswith(%r)" % str(p))
|
||||
result = testdir.runpytest("-d", "--rsyncdir=%(subdir)s" % locals(),
|
||||
"--tx=popen//chdir=%(dest)s" % locals(), p)
|
||||
assert result.ret == 0
|
||||
result.stdout.fnmatch_lines([
|
||||
"*1* *popen*platform*",
|
||||
#"RSyncStart: [G1]",
|
||||
#"RSyncFinished: [G1]",
|
||||
"*1 passed*"
|
||||
])
|
||||
assert dest.join(subdir.basename).check(dir=1)
|
||||
|
||||
def test_nice_level(self, testdir):
|
||||
""" Tests if nice level behaviour is ok """
|
||||
import os
|
||||
if not hasattr(os, 'nice'):
|
||||
py.test.skip("no os.nice() available")
|
||||
testdir.makepyfile(conftest="""
|
||||
dist_nicelevel = 10
|
||||
""")
|
||||
p1 = testdir.makepyfile("""
|
||||
def test_nice():
|
||||
import os
|
||||
assert os.nice(0) == 10
|
||||
""")
|
||||
evrec = testdir.inline_run('-d', p1, '--tx=popen')
|
||||
ev = evrec.getfirstnamed('itemtestreport')
|
||||
assert ev.passed
|
||||
|
|
@ -70,6 +70,8 @@ class MasterNode(object):
|
|||
# setting up slave code
|
||||
def install_slave(gateway, config):
|
||||
channel = gateway.remote_exec(source="""
|
||||
import os, sys
|
||||
sys.path.insert(0, os.getcwd())
|
||||
from py.__.test.dist.mypickle import PickleChannel
|
||||
from py.__.test.dist.txnode import SlaveNode
|
||||
channel = PickleChannel(channel)
|
||||
|
|
|
@ -70,7 +70,7 @@ class DefaultPlugin:
|
|||
"and instantiate 'HelloPlugin' from the module."))
|
||||
group._addoption('-f', '--looponfail',
|
||||
action="store_true", dest="looponfail", default=False,
|
||||
help="run tests, loop on failing test set, until all pass. repeat forever.")
|
||||
help="run tests, re-run failing test set until all pass.")
|
||||
|
||||
group = parser.addgroup("test process debugging")
|
||||
group.addoption('--collectonly',
|
||||
|
@ -94,33 +94,43 @@ class DefaultPlugin:
|
|||
action="store_true", dest="debug", default=False,
|
||||
help="generate and show debugging information.")
|
||||
|
||||
group = parser.addgroup("xplatform", "distributed/cross platform testing")
|
||||
group._addoption('-d', '--dist',
|
||||
action="store_true", dest="dist", default=False,
|
||||
help="ad-hoc distribute tests across machines (requires conftest settings)")
|
||||
group._addoption('-n', dest="numprocesses", default=0, metavar="numprocesses",
|
||||
group = parser.addgroup("dist", "distributed testing") # see http://pytest.org/help/dist")
|
||||
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('-n', dest="numprocesses", metavar="numprocesses",
|
||||
action="store", type="int",
|
||||
help="number of local test processes. conflicts with --dist.")
|
||||
help="shortcut for '--dist=load --tx=NUM*popen'")
|
||||
group.addoption('--rsyncdir', action="append", default=[], metavar="dir1",
|
||||
help="add local directory for rsync to remote test nodes.")
|
||||
group._addoption('--tx', dest="tx", action="append", default=[],
|
||||
help=("add a test environment, specified in XSpec syntax. examples: "
|
||||
"--tx popen//python=python2.5 --tx socket=192.168.1.102"))
|
||||
#group._addoption('--rest',
|
||||
# action='store_true', dest="restreport", default=False,
|
||||
# help="restructured text output reporting."),
|
||||
help="add directory for rsyncing to remote tx nodes.")
|
||||
|
||||
def pytest_configure(self, config):
|
||||
self.fixoptions(config)
|
||||
self.setsession(config)
|
||||
self.loadplugins(config)
|
||||
self.fixoptions(config)
|
||||
|
||||
def fixoptions(self, 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"
|
||||
if config.getvalue("usepdb"):
|
||||
if config.getvalue("looponfail"):
|
||||
raise config.Error("--pdb incompatible with --looponfail.")
|
||||
if config.getvalue("dist"):
|
||||
raise config.Error("--pdb incomptaible with distributed testing.")
|
||||
if config.option.dist != "no":
|
||||
raise config.Error("--pdb incomptaible with distributing tests.")
|
||||
|
||||
def loadplugins(self, config):
|
||||
for name in config.getvalue("plugin"):
|
||||
|
@ -136,7 +146,7 @@ class DefaultPlugin:
|
|||
if val("looponfail"):
|
||||
from py.__.test.looponfail.remote import LooponfailingSession
|
||||
config.setsessionclass(LooponfailingSession)
|
||||
elif val("numprocesses") or val("dist"):
|
||||
elif val("dist") != "no":
|
||||
from py.__.test.dist.dsession import DSession
|
||||
config.setsessionclass(DSession)
|
||||
|
||||
|
@ -153,10 +163,10 @@ def test_implied_different_sessions(tmpdir):
|
|||
return Exception
|
||||
return getattr(config._sessionclass, '__name__', None)
|
||||
assert x() == None
|
||||
assert x('--dist') == 'DSession'
|
||||
assert x('-d') == 'DSession'
|
||||
assert x('--dist=each') == 'DSession'
|
||||
assert x('-n3') == 'DSession'
|
||||
assert x('-f') == 'LooponfailingSession'
|
||||
assert x('--dist', '--collectonly') == 'Session'
|
||||
|
||||
def test_generic(plugintester):
|
||||
plugintester.apicheck(DefaultPlugin)
|
||||
|
@ -173,16 +183,48 @@ def test_plugin_already_exists(testdir):
|
|||
assert config.option.plugin == ['default']
|
||||
config.pytestplugins.do_configure(config)
|
||||
|
||||
def test_conflict_options():
|
||||
def check_conflict_option(opts):
|
||||
print "testing if options conflict:", " ".join(opts)
|
||||
config = py.test.config._reparse(opts)
|
||||
py.test.raises(config.Error,
|
||||
"config.pytestplugins.do_configure(config)")
|
||||
conflict_options = (
|
||||
'--looponfail --pdb',
|
||||
'--dist --pdb',
|
||||
)
|
||||
for spec in conflict_options:
|
||||
opts = spec.split()
|
||||
yield check_conflict_option, opts
|
||||
|
||||
class TestDistOptions:
|
||||
def test_getxspecs(self, testdir):
|
||||
config = testdir.parseconfigure("--tx=popen", "--tx", "ssh=xyz")
|
||||
xspecs = config.getxspecs()
|
||||
assert len(xspecs) == 2
|
||||
print xspecs
|
||||
assert xspecs[0].popen
|
||||
assert xspecs[1].ssh == "xyz"
|
||||
|
||||
def test_xspecs_multiplied(self, testdir):
|
||||
xspecs = testdir.parseconfigure("--tx=3*popen",).getxspecs()
|
||||
assert len(xspecs) == 3
|
||||
assert xspecs[1].popen
|
||||
|
||||
def test_getrsyncdirs(self, testdir):
|
||||
config = testdir.parseconfigure('--rsyncdir=' + str(testdir.tmpdir))
|
||||
roots = config.getrsyncdirs()
|
||||
assert len(roots) == 1 + 1
|
||||
assert testdir.tmpdir in roots
|
||||
|
||||
def test_getrsyncdirs_with_conftest(self, testdir):
|
||||
p = py.path.local()
|
||||
for bn in 'x y z'.split():
|
||||
p.mkdir(bn)
|
||||
testdir.makeconftest("""
|
||||
rsyncdirs= 'x',
|
||||
""")
|
||||
config = testdir.parseconfigure(testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
|
||||
roots = config.getrsyncdirs()
|
||||
assert len(roots) == 3 + 1
|
||||
assert py.path.local('y') in roots
|
||||
assert py.path.local('z') in roots
|
||||
assert testdir.tmpdir.join('x') in roots
|
||||
|
||||
def test_dist_options(testdir):
|
||||
py.test.raises(Exception, "testdir.parseconfigure('--pdb', '--looponfail')")
|
||||
py.test.raises(Exception, "testdir.parseconfigure('--pdb', '-n 3')")
|
||||
py.test.raises(Exception, "testdir.parseconfigure('--pdb', '-d')")
|
||||
config = testdir.parseconfigure("-n 2")
|
||||
assert config.option.dist == "load"
|
||||
assert config.option.tx == ['popen'] * 2
|
||||
|
||||
config = testdir.parseconfigure("-d")
|
||||
assert config.option.dist == "load"
|
||||
|
|
|
@ -55,6 +55,7 @@ class TmpTestdir:
|
|||
self.tmpdir = tmpdir.mkdir(name)
|
||||
self.plugins = []
|
||||
self._syspathremove = []
|
||||
self.chdir() # always chdir
|
||||
|
||||
def Config(self, pyplugins=None, topdir=None):
|
||||
if topdir is None:
|
||||
|
@ -163,6 +164,11 @@ class TmpTestdir:
|
|||
config.parse(args)
|
||||
return config
|
||||
|
||||
def parseconfigure(self, *args):
|
||||
config = self.parseconfig(*args)
|
||||
config.pytestplugins.do_configure(config)
|
||||
return config
|
||||
|
||||
def getitem(self, source, funcname="test_func"):
|
||||
modcol = self.getmodulecol(source)
|
||||
item = modcol.join(funcname)
|
||||
|
|
|
@ -6,12 +6,12 @@ class RestdocPlugin:
|
|||
group.addoption('-R', '--urlcheck',
|
||||
action="store_true", dest="urlcheck", default=False,
|
||||
help="urlopen() remote links found in ReST text files.")
|
||||
group.addoption('--urlcheck-timeout', action="store",
|
||||
group.addoption('--urltimeout', action="store", metavar="secs",
|
||||
type="int", dest="urlcheck_timeout", default=5,
|
||||
help="timeout in seconds for urlcheck")
|
||||
help="timeout in seconds for remote urlchecks")
|
||||
group.addoption('--forcegen',
|
||||
action="store_true", dest="forcegen", default=False,
|
||||
help="force generation of html files even if they appear up-to-date")
|
||||
help="force generation of html files.")
|
||||
|
||||
def pytest_collect_file(self, path, parent):
|
||||
if path.ext == ".txt":
|
||||
|
|
|
@ -4,7 +4,7 @@ pydir = py.path.local(py.__file__).dirpath()
|
|||
pytestpath = pydir.join("bin", "py.test")
|
||||
EXPECTTIMEOUT=10.0
|
||||
|
||||
class TestPyTest:
|
||||
class TestGeneralUsage:
|
||||
def test_config_error(self, testdir):
|
||||
testdir.makeconftest("""
|
||||
class ConftestPlugin:
|
||||
|
@ -251,8 +251,61 @@ class TestPyTest:
|
|||
"y* = 'xxxxxx*"
|
||||
])
|
||||
|
||||
def test_verbose_reporting(self, testdir):
|
||||
p1 = testdir.makepyfile("""
|
||||
import py
|
||||
def test_fail():
|
||||
raise ValueError()
|
||||
def test_pass():
|
||||
pass
|
||||
class TestClass:
|
||||
def test_skip(self):
|
||||
py.test.skip("hello")
|
||||
def test_gen():
|
||||
def check(x):
|
||||
assert x == 1
|
||||
yield check, 0
|
||||
""")
|
||||
result = testdir.runpytest(p1, '-v')
|
||||
result.stdout.fnmatch_lines([
|
||||
"*FAIL*test_verbose_reporting.py:2: test_fail*",
|
||||
"*PASS*test_verbose_reporting.py:4: test_pass*",
|
||||
"*SKIP*test_verbose_reporting.py:7: TestClass.test_skip*",
|
||||
"*FAIL*test_verbose_reporting.py:10: test_gen*",
|
||||
])
|
||||
assert result.ret == 1
|
||||
|
||||
def test_dist_testing(self, testdir):
|
||||
class TestDistribution:
|
||||
def test_dist_conftest_options(self, testdir):
|
||||
p1 = testdir.tmpdir.ensure("dir", 'p1.py')
|
||||
p1.dirpath("__init__.py").write("")
|
||||
p1.dirpath("conftest.py").write(py.code.Source("""
|
||||
print "importing conftest", __file__
|
||||
import py
|
||||
Option = py.test.config.Option
|
||||
option = py.test.config.addoptions("someopt",
|
||||
Option('--someopt', action="store_true", dest="someopt", default=False))
|
||||
dist_rsync_roots = ['../dir']
|
||||
print "added options", option
|
||||
print "config file seen from conftest", py.test.config
|
||||
"""))
|
||||
p1.write(py.code.Source("""
|
||||
import py, conftest
|
||||
def test_1():
|
||||
print "config from test_1", py.test.config
|
||||
print "conftest from test_1", conftest.__file__
|
||||
print "test_1: py.test.config.option.someopt", py.test.config.option.someopt
|
||||
print "test_1: conftest", conftest
|
||||
print "test_1: conftest.option.someopt", conftest.option.someopt
|
||||
assert conftest.option.someopt
|
||||
"""))
|
||||
result = testdir.runpytest('-d', '--tx=popen', p1, '--someopt')
|
||||
assert result.ret == 0
|
||||
extra = result.stdout.fnmatch_lines([
|
||||
"*1 passed*",
|
||||
])
|
||||
|
||||
def test_manytests_to_one_popen(self, testdir):
|
||||
p1 = testdir.makepyfile("""
|
||||
import py
|
||||
def test_fail0():
|
||||
|
@ -273,7 +326,7 @@ class TestPyTest:
|
|||
])
|
||||
assert result.ret == 1
|
||||
|
||||
def test_dist_testing_conftest_specified(self, testdir):
|
||||
def test_dist_conftest_specified(self, testdir):
|
||||
p1 = testdir.makepyfile("""
|
||||
import py
|
||||
def test_fail0():
|
||||
|
@ -329,44 +382,24 @@ class TestPyTest:
|
|||
])
|
||||
assert result.ret == 1
|
||||
|
||||
def test_keyboard_interrupt(self, testdir):
|
||||
p1 = testdir.makepyfile("""
|
||||
import py
|
||||
def test_fail():
|
||||
raise ValueError()
|
||||
def test_inter():
|
||||
raise KeyboardInterrupt()
|
||||
""")
|
||||
result = testdir.runpytest(p1)
|
||||
def test_distribution_rsyncdirs_example(self, testdir):
|
||||
source = testdir.mkdir("source")
|
||||
dest = testdir.mkdir("dest")
|
||||
subdir = source.mkdir("example_pkg")
|
||||
subdir.ensure("__init__.py")
|
||||
p = subdir.join("test_one.py")
|
||||
p.write("def test_5(): assert not __file__.startswith(%r)" % str(p))
|
||||
result = testdir.runpytest("-d", "--rsyncdir=%(subdir)s" % locals(),
|
||||
"--tx=popen//chdir=%(dest)s" % locals(), p)
|
||||
assert result.ret == 0
|
||||
result.stdout.fnmatch_lines([
|
||||
#"*test_inter() INTERRUPTED",
|
||||
"*KEYBOARD INTERRUPT*",
|
||||
"*1 failed*",
|
||||
"*1* *popen*platform*",
|
||||
#"RSyncStart: [G1]",
|
||||
#"RSyncFinished: [G1]",
|
||||
"*1 passed*"
|
||||
])
|
||||
assert dest.join(subdir.basename).check(dir=1)
|
||||
|
||||
def test_verbose_reporting(self, testdir):
|
||||
p1 = testdir.makepyfile("""
|
||||
import py
|
||||
def test_fail():
|
||||
raise ValueError()
|
||||
def test_pass():
|
||||
pass
|
||||
class TestClass:
|
||||
def test_skip(self):
|
||||
py.test.skip("hello")
|
||||
def test_gen():
|
||||
def check(x):
|
||||
assert x == 1
|
||||
yield check, 0
|
||||
""")
|
||||
result = testdir.runpytest(p1, '-v')
|
||||
result.stdout.fnmatch_lines([
|
||||
"*FAIL*test_verbose_reporting.py:2: test_fail*",
|
||||
"*PASS*test_verbose_reporting.py:4: test_pass*",
|
||||
"*SKIP*test_verbose_reporting.py:7: TestClass.test_skip*",
|
||||
"*FAIL*test_verbose_reporting.py:10: test_gen*",
|
||||
])
|
||||
assert result.ret == 1
|
||||
|
||||
class TestInteractive:
|
||||
def getspawn(self, tmpdir):
|
||||
|
@ -444,3 +477,19 @@ class TestInteractive:
|
|||
"*2 failed*"
|
||||
])
|
||||
|
||||
class TestKeyboardInterrupt:
|
||||
def test_raised_in_testfunction(self, testdir):
|
||||
p1 = testdir.makepyfile("""
|
||||
import py
|
||||
def test_fail():
|
||||
raise ValueError()
|
||||
def test_inter():
|
||||
raise KeyboardInterrupt()
|
||||
""")
|
||||
result = testdir.runpytest(p1)
|
||||
result.stdout.fnmatch_lines([
|
||||
#"*test_inter() INTERRUPTED",
|
||||
"*KEYBOARD INTERRUPT*",
|
||||
"*1 failed*",
|
||||
])
|
||||
|
||||
|
|
|
@ -218,57 +218,13 @@ class TestConfigApi_getcolitems:
|
|||
assert col.config is config
|
||||
|
||||
|
||||
class TestOptionsAndConfiguration:
|
||||
def test_getxspecs_numprocesses(self, testdir):
|
||||
config = testdir.parseconfig("-n3")
|
||||
xspecs = config.getxspecs()
|
||||
assert len(xspecs) == 3
|
||||
|
||||
def test_getxspecs(self, testdir):
|
||||
testdir.chdir()
|
||||
config = testdir.parseconfig("--tx=popen", "--tx", "ssh=xyz")
|
||||
xspecs = config.getxspecs()
|
||||
assert len(xspecs) == 2
|
||||
print xspecs
|
||||
assert xspecs[0].popen
|
||||
assert xspecs[1].ssh == "xyz"
|
||||
|
||||
def test_xspecs_multiplied(self, testdir):
|
||||
testdir.chdir()
|
||||
xspecs = testdir.parseconfig("--tx=3*popen",).getxspecs()
|
||||
assert len(xspecs) == 3
|
||||
assert xspecs[1].popen
|
||||
|
||||
def test_getrsyncdirs(self, testdir):
|
||||
config = testdir.parseconfig('--rsyncdir=' + str(testdir.tmpdir))
|
||||
roots = config.getrsyncdirs()
|
||||
assert len(roots) == 1 + 1
|
||||
assert testdir.tmpdir in roots
|
||||
|
||||
def test_getrsyncdirs_with_conftest(self, testdir):
|
||||
testdir.chdir()
|
||||
p = py.path.local()
|
||||
for bn in 'x y z'.split():
|
||||
p.mkdir(bn)
|
||||
testdir.makeconftest("""
|
||||
rsyncdirs= 'x',
|
||||
""")
|
||||
config = testdir.parseconfig(testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
|
||||
roots = config.getrsyncdirs()
|
||||
assert len(roots) == 3 + 1
|
||||
assert py.path.local('y') in roots
|
||||
assert py.path.local('z') in roots
|
||||
assert testdir.tmpdir.join('x') in roots
|
||||
|
||||
|
||||
|
||||
class TestOptionEffects:
|
||||
def test_boxed_option_default(self, testdir):
|
||||
tmpdir = testdir.tmpdir.ensure("subdir", dir=1)
|
||||
config = py.test.config._reparse([tmpdir])
|
||||
config.initsession()
|
||||
assert not config.option.boxed
|
||||
config = py.test.config._reparse(['--dist', tmpdir])
|
||||
config = py.test.config._reparse(['-d', tmpdir])
|
||||
config.initsession()
|
||||
assert not config.option.boxed
|
||||
|
||||
|
|
|
@ -87,9 +87,7 @@ class TestConfigPickling:
|
|||
assert getattr(config2.option, name) == value
|
||||
assert config2.getvalue("x") == 1
|
||||
|
||||
def test_config_rconfig(self, testdir):
|
||||
tmp = testdir.tmpdir
|
||||
tmp.ensure("__init__.py")
|
||||
def test_config_pickling_customoption(self, testdir):
|
||||
testdir.makeconftest("""
|
||||
class ConftestPlugin:
|
||||
def pytest_addoption(self, parser):
|
||||
|
@ -97,6 +95,27 @@ class TestConfigPickling:
|
|||
group.addoption('-G', '--glong', action="store", default=42,
|
||||
type="int", dest="gdest", help="g value.")
|
||||
""")
|
||||
config = testdir.parseconfig("-G", "11")
|
||||
assert config.option.gdest == 11
|
||||
repr = config.__getstate__()
|
||||
|
||||
config = testdir.Config()
|
||||
py.test.raises(AttributeError, "config.option.gdest")
|
||||
|
||||
config2 = testdir.Config()
|
||||
config2.__setstate__(repr)
|
||||
assert config2.option.gdest == 11
|
||||
|
||||
def test_config_pickling_and_conftest_deprecated(self, testdir):
|
||||
tmp = testdir.tmpdir.ensure("w1", "w2", dir=1)
|
||||
tmp.ensure("__init__.py")
|
||||
tmp.join("conftest.py").write(py.code.Source("""
|
||||
class ConftestPlugin:
|
||||
def pytest_addoption(self, parser):
|
||||
group = parser.addgroup("testing group")
|
||||
group.addoption('-G', '--glong', action="store", default=42,
|
||||
type="int", dest="gdest", help="g value.")
|
||||
"""))
|
||||
config = testdir.parseconfig(tmp, "-G", "11")
|
||||
assert config.option.gdest == 11
|
||||
repr = config.__getstate__()
|
||||
|
@ -106,10 +125,11 @@ class TestConfigPickling:
|
|||
|
||||
config2 = testdir.Config()
|
||||
config2.__setstate__(repr)
|
||||
assert config2.option.gdest == 11
|
||||
|
||||
option = config2.addoptions("testing group",
|
||||
config2.Option('-G', '--glong', action="store", default=42,
|
||||
type="int", dest="gdest", help="g value."))
|
||||
assert config2.option.gdest == 11
|
||||
assert option.gdest == 11
|
||||
|
||||
def test_config_picklability(self, testdir):
|
||||
|
|
Loading…
Reference in New Issue