本文整理汇总了Python中xdist.slavemanage.NodeManager类的典型用法代码示例。如果您正苦于以下问题:Python NodeManager类的具体用法?Python NodeManager怎么用?Python NodeManager使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了NodeManager类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_getrsyncdirs
def test_getrsyncdirs(self, testdir):
config = testdir.parseconfigure('--rsyncdir=' + str(testdir.tmpdir))
nm = NodeManager(config, specs=[execnet.XSpec("popen")])
assert not nm._getrsyncdirs()
nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")])
assert nm.roots
assert testdir.tmpdir in nm.roots
开发者ID:nicoddemus,项目名称:pytest-xdist,代码行数:7,代码来源:test_plugin.py
示例2: test_getxspecs
def test_getxspecs(self, testdir):
config = testdir.parseconfigure("--tx=popen", "--tx", "ssh=xyz")
nodemanager = NodeManager(config)
xspecs = nodemanager._getxspecs()
assert len(xspecs) == 2
print(xspecs)
assert xspecs[0].popen
assert xspecs[1].ssh == "xyz"
开发者ID:nicoddemus,项目名称:pytest-xdist,代码行数:8,代码来源:test_plugin.py
示例3: test_rsync_same_popen_twice
def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder):
source, dest = mysetup.source, mysetup.dest
hm = NodeManager(config, ["popen//chdir=%s" %dest] * 2)
hm.makegateways()
source.ensure("dir1", "dir2", "hello")
hm.rsync(source)
call = hookrecorder.popcall("pytest_xdist_rsyncstart")
assert call.source == source
assert len(call.gateways) == 1
assert call.gateways[0] in hm.group
call = hookrecorder.popcall("pytest_xdist_rsyncfinish")
开发者ID:curzona,项目名称:pytest-xdist,代码行数:11,代码来源:test_slavemanage.py
示例4: test_optimise_popen
def test_optimise_popen(self, testdir, mysetup, slavecontroller):
source = mysetup.source
specs = ["popen"] * 3
source.join("conftest.py").write("rsyncdirs = ['a']")
source.ensure('a', dir=1)
config = testdir.parseconfig(source)
nodemanager = NodeManager(config, specs)
nodemanager.setup_nodes(None) # calls .rysnc_roots()
for gwspec in nodemanager.specs:
assert gwspec._samefilesystem()
assert not gwspec.chdir
开发者ID:epeden,项目名称:pytest-xdist,代码行数:11,代码来源:test_slavemanage.py
示例5: test_optimise_popen
def test_optimise_popen(self, testdir, mysetup):
source, dest = mysetup.source, mysetup.dest
specs = ["popen"] * 3
source.join("conftest.py").write("rsyncdirs = ['a']")
source.ensure('a', dir=1)
config = testdir.parseconfig(source)
nodemanager = NodeManager(config, specs)
nodemanager.makegateways()
nodemanager.rsync_roots()
for gwspec in nodemanager.specs:
assert gwspec._samefilesystem()
assert not gwspec.chdir
开发者ID:curzona,项目名称:pytest-xdist,代码行数:12,代码来源:test_slavemanage.py
示例6: test_rsync_roots_no_roots
def test_rsync_roots_no_roots(self, testdir, mysetup):
mysetup.source.ensure("dir1", "file1").write("hello")
config = testdir.parseconfig(mysetup.source)
nodemanager = NodeManager(config, ["popen//chdir=%s" % mysetup.dest])
# assert nodemanager.config.topdir == source == config.topdir
nodemanager.makegateways()
nodemanager.rsync_roots()
p, = nodemanager.gwmanager.multi_exec(
"import os ; channel.send(os.getcwd())").receive_each()
p = py.path.local(p)
py.builtin.print_("remote curdir", p)
assert p == mysetup.dest.join(config.topdir.basename)
assert p.join("dir1").check()
assert p.join("dir1", "file1").check()
开发者ID:epeden,项目名称:pytest-xdist,代码行数:14,代码来源:test_slavemanage.py
示例7: test_popen_makegateway_events
def test_popen_makegateway_events(self, config, hookrecorder, _pytest):
hm = NodeManager(config, ["popen"] * 2)
hm.makegateways()
call = hookrecorder.popcall("pytest_xdist_setupnodes")
assert len(call.specs) == 2
call = hookrecorder.popcall("pytest_xdist_newgateway")
assert call.gateway.spec == execnet.XSpec("popen")
assert call.gateway.id == "gw0"
call = hookrecorder.popcall("pytest_xdist_newgateway")
assert call.gateway.id == "gw1"
assert len(hm.group) == 2
hm.teardown_nodes()
assert not len(hm.group)
开发者ID:curzona,项目名称:pytest-xdist,代码行数:14,代码来源:test_slavemanage.py
示例8: test_init_rsync_roots
def test_init_rsync_roots(self, testdir, mysetup, slavecontroller):
source, dest = mysetup.source, mysetup.dest
dir2 = source.ensure("dir1", "dir2", dir=1)
source.ensure("dir1", "somefile", dir=1)
dir2.ensure("hello")
source.ensure("bogusdir", "file")
source.join("tox.ini").write(py.std.textwrap.dedent("""
[pytest]
rsyncdirs=dir1/dir2
"""))
config = testdir.parseconfig(source)
nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
nodemanager.setup_nodes(None) # calls .rsync_roots()
assert dest.join("dir2").check()
assert not dest.join("dir1").check()
assert not dest.join("bogus").check()
开发者ID:epeden,项目名称:pytest-xdist,代码行数:16,代码来源:test_slavemanage.py
示例9: test_getrsyncdirs_with_conftest
def test_getrsyncdirs_with_conftest(self, testdir):
p = py.path.local()
for bn in 'x y z'.split():
p.mkdir(bn)
testdir.makeini("""
[pytest]
rsyncdirs= x
""")
config = testdir.parseconfigure(
testdir.tmpdir, '--rsyncdir=y', '--rsyncdir=z')
nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=xyz")])
roots = nm._getrsyncdirs()
# assert len(roots) == 3 + 1 # pylib
assert py.path.local('y') in roots
assert py.path.local('z') in roots
assert testdir.tmpdir.join('x') in roots
开发者ID:nicoddemus,项目名称:pytest-xdist,代码行数:16,代码来源:test_plugin.py
示例10: test_popen_rsync_subdir
def test_popen_rsync_subdir(self, testdir, mysetup, slavecontroller):
source, dest = mysetup.source, mysetup.dest
dir1 = mysetup.source.mkdir("dir1")
dir2 = dir1.mkdir("dir2")
dir2.ensure("hello")
for rsyncroot in (dir1, source):
dest.remove()
nodemanager = NodeManager(testdir.parseconfig(
"--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot,
source, ))
nodemanager.setup_nodes(None) # calls .rsync_roots()
if rsyncroot == source:
dest = dest.join("source")
assert dest.join("dir1").check()
assert dest.join("dir1", "dir2").check()
assert dest.join("dir1", "dir2", 'hello').check()
nodemanager.teardown_nodes()
开发者ID:epeden,项目名称:pytest-xdist,代码行数:17,代码来源:test_slavemanage.py
示例11: pytest_sessionstart
def pytest_sessionstart(self, session):
"""Creates and starts the nodes.
The nodes are setup to put their events onto self.queue. As
soon as nodes start they will emit the slave_slaveready event.
"""
self.nodemanager = NodeManager(self.config)
nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
self._active_nodes.update(nodes)
开发者ID:rdebliek,项目名称:oscar,代码行数:9,代码来源:dsession.py
示例12: test_rsyncignore
def test_rsyncignore(self, testdir, mysetup):
source, dest = mysetup.source, mysetup.dest
dir2 = source.ensure("dir1", "dir2", dir=1)
dir5 = source.ensure("dir5", "dir6", "bogus")
dirf = source.ensure("dir5", "file")
dir2.ensure("hello")
source.join("tox.ini").write(py.std.textwrap.dedent("""
[pytest]
rsyncdirs = dir1 dir5
rsyncignore = dir1/dir2 dir5/dir6
"""))
config = testdir.parseconfig(source)
nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
nodemanager.makegateways()
nodemanager.rsync_roots()
assert dest.join("dir1").check()
assert not dest.join("dir1", "dir2").check()
assert dest.join("dir5","file").check()
assert not dest.join("dir6").check()
开发者ID:curzona,项目名称:pytest-xdist,代码行数:19,代码来源:test_slavemanage.py
示例13: pytest_sessionstart
def pytest_sessionstart(self, session):
"""Creates and starts the nodes.
The nodes are setup to put their events onto self.queue. As
soon as nodes start they will emit the slave_slaveready event.
"""
self.nodemanager = NodeManager(self.config)
nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
for node in nodes:
node.channel.send(("RUNNING_MODE", self.dist))
self._active_nodes.update(nodes)
self._session = session
开发者ID:xsandr,项目名称:pytest-xdist,代码行数:13,代码来源:dsession.py
示例14: test_rsyncignore
def test_rsyncignore(self, testdir, mysetup, slavecontroller):
source, dest = mysetup.source, mysetup.dest
dir2 = source.ensure("dir1", "dir2", dir=1)
source.ensure("dir5", "dir6", "bogus")
source.ensure("dir5", "file")
dir2.ensure("hello")
source.ensure("foo", "bar")
source.ensure("bar", "foo")
source.join("tox.ini").write(py.std.textwrap.dedent("""
[pytest]
rsyncdirs = dir1 dir5
rsyncignore = dir1/dir2 dir5/dir6 foo*
"""))
config = testdir.parseconfig(source)
config.option.rsyncignore = ['bar']
nodemanager = NodeManager(config, ["popen//chdir=%s" % dest])
nodemanager.setup_nodes(None) # calls .rsync_roots()
assert dest.join("dir1").check()
assert not dest.join("dir1", "dir2").check()
assert dest.join("dir5", "file").check()
assert not dest.join("dir6").check()
assert not dest.join('foo').check()
assert not dest.join('bar').check()
开发者ID:epeden,项目名称:pytest-xdist,代码行数:23,代码来源:test_slavemanage.py
示例15: test_rsync_popen_with_path
def test_rsync_popen_with_path(self, config, mysetup):
source, dest = mysetup.source, mysetup.dest
hm = NodeManager(config, ["popen//chdir=%s" %dest] * 1)
hm.makegateways()
source.ensure("dir1", "dir2", "hello")
l = []
hm.rsync(source, notify=lambda *args: l.append(args))
assert len(l) == 1
assert l[0] == ("rsyncrootready", hm.group['gw0'].spec, source)
hm.teardown_nodes()
dest = dest.join(source.basename)
assert dest.join("dir1").check()
assert dest.join("dir1", "dir2").check()
assert dest.join("dir1", "dir2", 'hello').check()
开发者ID:curzona,项目名称:pytest-xdist,代码行数:14,代码来源:test_slavemanage.py
示例16: test_rsync_same_popen_twice
def test_rsync_same_popen_twice(self, config, mysetup, hookrecorder,
slavecontroller):
source, dest = mysetup.source, mysetup.dest
hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2)
hm.roots = []
hm.setup_nodes(None)
source.ensure("dir1", "dir2", "hello")
gw = hm.group[0]
hm.rsync(gw, source)
call = hookrecorder.popcall("pytest_xdist_rsyncstart")
assert call.source == source
assert len(call.gateways) == 1
assert call.gateways[0] in hm.group
call = hookrecorder.popcall("pytest_xdist_rsyncfinish")
开发者ID:epeden,项目名称:pytest-xdist,代码行数:14,代码来源:test_slavemanage.py
示例17: test_popens_rsync
def test_popens_rsync(self, config, mysetup):
source = mysetup.source
hm = NodeManager(config, ["popen"] * 2)
hm.makegateways()
assert len(hm.group) == 2
for gw in hm.group:
class pseudoexec:
args = []
def __init__(self, *args):
self.args.extend(args)
def waitclose(self):
pass
gw.remote_exec = pseudoexec
l = []
hm.rsync(source, notify=lambda *args: l.append(args))
assert not l
hm.teardown_nodes()
assert not len(hm.group)
assert "sys.path.insert" in gw.remote_exec.args[0]
开发者ID:curzona,项目名称:pytest-xdist,代码行数:19,代码来源:test_slavemanage.py
示例18: pytest_sessionstart
def pytest_sessionstart(self, session):
self.nodemanager = NodeManager(self.config)
self.nodemanager.setup_nodes(putevent=self.queue.put)
开发者ID:chetan51,项目名称:nupic-linux64,代码行数:3,代码来源:dsession.py
示例19: __init__
class DSession:
def __init__(self, config):
self.config = config
self.log = py.log.Producer("dsession")
if not config.option.debug:
py.log.setconsumer(self.log._keywords, None)
self.shuttingdown = False
self.countfailures = 0
self.maxfail = config.getvalue("maxfail")
self.queue = queue.Queue()
try:
self.terminal = config.pluginmanager.getplugin("terminalreporter")
except KeyError:
self.terminal = None
else:
self.trdist = TerminalDistReporter(config)
config.pluginmanager.register(self.trdist, "terminaldistreporter")
def report_line(self, line):
if self.terminal and self.config.option.verbose >= 0:
self.terminal.write_line(line)
@pytest.mark.trylast
def pytest_sessionstart(self, session):
self.nodemanager = NodeManager(self.config)
self.nodemanager.setup_nodes(putevent=self.queue.put)
def pytest_sessionfinish(self, session):
""" teardown any resources after a test run. """
nm = getattr(self, "nodemanager", None) # if not fully initialized
if nm is not None:
nm.teardown_nodes()
def pytest_collection(self):
# prohibit collection of test items in master process
return True
def pytest_runtestloop(self):
numnodes = len(self.nodemanager.specs)
dist = self.config.getvalue("dist")
if dist == "load":
self.sched = LoadScheduling(numnodes, log=self.log)
elif dist == "each":
self.sched = EachScheduling(numnodes, log=self.log)
else:
assert 0, dist
self.shouldstop = False
self.session_finished = False
while not self.session_finished:
self.loop_once()
if self.shouldstop:
raise Interrupted(str(self.shouldstop))
return True
def loop_once(self):
""" process one callback from one of the slaves. """
while 1:
try:
eventcall = self.queue.get(timeout=2.0)
break
except queue.Empty:
continue
callname, kwargs = eventcall
assert callname, kwargs
method = "slave_" + callname
call = getattr(self, method)
self.log("calling method: %s(**%s)" % (method, kwargs))
call(**kwargs)
if self.sched.tests_finished():
self.triggershutdown()
#
# callbacks for processing events from slaves
#
def slave_slaveready(self, node, slaveinfo):
node.slaveinfo = slaveinfo
node.slaveinfo["id"] = node.gateway.id
node.slaveinfo["spec"] = node.gateway.spec
self.config.hook.pytest_testnodeready(node=node)
self.sched.addnode(node)
if self.shuttingdown:
node.shutdown()
def slave_slavefinished(self, node):
self.config.hook.pytest_testnodedown(node=node, error=None)
if node.slaveoutput["exitstatus"] == 2: # keyboard-interrupt
self.shouldstop = "%s received keyboard-interrupt" % (node,)
self.slave_errordown(node, "keyboard-interrupt")
return
crashitem = self.sched.remove_node(node)
# assert not crashitem, (crashitem, node)
if self.shuttingdown and not self.sched.hasnodes():
self.session_finished = True
def slave_errordown(self, node, error):
self.config.hook.pytest_testnodedown(node=node, error=error)
try:
crashitem = self.sched.remove_node(node)
except KeyError:
#.........这里部分代码省略.........
开发者ID:chetan51,项目名称:nupic-linux64,代码行数:101,代码来源:dsession.py
示例20: __init__
class DSession:
"""A py.test plugin which runs a distributed test session
At the beginning of the test session this creates a NodeManager
instance which creates and starts all nodes. Nodes then emit
events processed in the pytest_runtestloop hook using the slave_*
methods.
Once a node is started it will automatically start running the
py.test mainloop with some custom hooks. This means a node
automatically starts collecting tests. Once tests are collected
it will wait for instructions.
"""
def __init__(self, config):
self.config = config
self.log = py.log.Producer("dsession")
if not config.option.debug:
py.log.setconsumer(self.log._keywords, None)
self.shuttingdown = False
self.countfailures = 0
self.maxfail = config.getvalue("maxfail")
self.queue = queue.Queue()
self._session = None
self._failed_collection_errors = {}
self._active_nodes = set()
self._failed_nodes_count = 0
self._max_slave_restart = self.config.getoption('max_slave_restart')
if self._max_slave_restart is not None:
self._max_slave_restart = int(self._max_slave_restart)
try:
self.terminal = config.pluginmanager.getplugin("terminalreporter")
except KeyError:
self.terminal = None
else:
self.trdist = TerminalDistReporter(config)
config.pluginmanager.register(self.trdist, "terminaldistreporter")
@property
def session_finished(self):
"""Return True if the distributed session has finished
This means all nodes have executed all test items. This is
used to by pytest_runtestloop to break out of it's loop.
"""
return bool(self.shuttingdown and not self._active_nodes)
def report_line(self, line):
if self.terminal and self.config.option.verbose >= 0:
self.terminal.write_line(line)
@pytest.mark.trylast
def pytest_sessionstart(self, session):
"""Creates and starts the nodes.
The nodes are setup to put their events onto self.queue. As
soon as nodes start they will emit the slave_slaveready event.
"""
self.nodemanager = NodeManager(self.config)
nodes = self.nodemanager.setup_nodes(putevent=self.queue.put)
self._active_nodes.update(nodes)
self._session = session
def pytest_sessionfinish(self, session):
"""Shutdown all nodes."""
nm = getattr(self, 'nodemanager', None) # if not fully initialized
if nm is not None:
nm.teardown_nodes()
self._session = None
def pytest_collection(self):
# prohibit collection of test items in master process
return True
def pytest_runtestloop(self):
numnodes = len(self.nodemanager.specs)
dist = self.config.getvalue("dist")
if dist == "load":
self.sched = LoadScheduling(numnodes, log=self.log,
config=self.config)
elif dist == "each":
self.sched = EachScheduling(numnodes, log=self.log)
else:
assert 0, dist
self.shouldstop = False
while not self.session_finished:
self.loop_once()
if self.shouldstop:
raise Interrupted(str(self.shouldstop))
return True
def loop_once(self):
"""Process one callback from one of the slaves."""
while 1:
try:
eventcall = self.queue.get(timeout=2.0)
break
except queue.Empty:
continue
callname, kwargs = eventcall
assert callname, kwargs
#.........这里部分代码省略.........
开发者ID:nchammas,项目名称:pytest-xdist,代码行数:101,代码来源:dsession.py
注:本文中的xdist.slavemanage.NodeManager类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论