[py-svn] r31563 - in py/branch/distributed/py/test/rsession: . testing
hpk at codespeak.net
hpk at codespeak.net
Wed Aug 23 18:16:15 CEST 2006
Author: hpk
Date: Wed Aug 23 18:16:11 2006
New Revision: 31563
Added:
py/branch/distributed/py/test/rsession/testing/test_outcome.py (contents, props changed)
Modified:
py/branch/distributed/py/test/rsession/rsession.py
py/branch/distributed/py/test/rsession/testing/test_master.py
py/branch/distributed/py/test/rsession/testing/test_rsession.py
Log:
(fijal, hpk) use pkgdir and pkdir.dirpath() more consistently.
Modified: py/branch/distributed/py/test/rsession/rsession.py
==============================================================================
--- py/branch/distributed/py/test/rsession/rsession.py (original)
+++ py/branch/distributed/py/test/rsession/rsession.py Wed Aug 23 18:16:11 2006
@@ -12,12 +12,16 @@
from py.__.test.rsession.master import (
setup_slave, bin_rsync, MasterNode, dispatch_loop)
-def init_hosts(reporter, sshhosts, relpath, rootdir):
+def init_hosts(reporter, sshhosts, relpath, pkgdir):
+ assert pkgdir.join("__init__.py").check(), (
+ "%s probably wrong" %(pkgdir,))
+ assert relpath, relpath
nodes = []
nodes_lock = threading.Condition()
def host_init(host):
- report.wrapcall(reporter, bin_rsync, str(rootdir)+'/', host, relpath)
+ report.wrapcall(reporter, bin_rsync,
+ str(pkgdir.dirpath())+"/", host, relpath)
gw = py.execnet.SshGateway(host)
ch = setup_slave(gw, relpath)
nodes_lock.acquire()
@@ -63,8 +67,8 @@
colitems = self._map2colitems(args)
sshhosts = self.config.getinitialvalue("disthosts")
relpath = "pytestcache"
- rootdir = getrootdir(args)
- nodes = init_hosts(self.report, sshhosts, relpath, rootdir)
+ pkgdir = getrootdir(args)
+ nodes = init_hosts(self.report, sshhosts, relpath, pkgdir)
# XXX: We're just ignoring errors from the tryiter, so skipped tests
# does not appear in output
Modified: py/branch/distributed/py/test/rsession/testing/test_master.py
==============================================================================
--- py/branch/distributed/py/test/rsession/testing/test_master.py (original)
+++ py/branch/distributed/py/test/rsession/testing/test_master.py Wed Aug 23 18:16:11 2006
@@ -11,7 +11,7 @@
from py.__.test.rsession import report
def setup_module(mod):
- mod.rootdir = py.path.local(py.__file__).dirpath().dirpath()
+ mod.pkgdir = py.path.local(py.__file__).dirpath()
class DummyChannel(object):
def __init__(self):
@@ -61,7 +61,7 @@
def test_slave_setup():
gw = py.execnet.PopenGateway()
- channel = setup_slave(gw, rootdir)
+ channel = setup_slave(gw, pkgdir.dirpath())
channel.send(funcpass_spec)
output = ReprOutcome(channel.receive())
assert output.passed
@@ -81,12 +81,12 @@
def open_gw():
gw = py.execnet.PopenGateway()
- channel = setup_slave(gw, rootdir)
+ channel = setup_slave(gw, pkgdir.dirpath())
mn = MasterNode(channel, simple_report)
return mn
master_nodes = [open_gw(), open_gw(), open_gw()]
- rootcol = py.test.collect.Directory(rootdir)
+ rootcol = py.test.collect.Directory(pkgdir.dirpath())
funcpass_item = rootcol.getitembynames(funcpass_spec)
funcfail_item = rootcol.getitembynames(funcfail_spec)
itemgenerator = iter([funcfail_item] +
Added: py/branch/distributed/py/test/rsession/testing/test_outcome.py
==============================================================================
--- (empty file)
+++ py/branch/distributed/py/test/rsession/testing/test_outcome.py Wed Aug 23 18:16:11 2006
@@ -0,0 +1,9 @@
+
+from py.__.test.rsession.outcome import Outcome, ReprOutcome
+
+def test_critical_debugging_flag():
+ outcome = Outcome(is_critical=True)
+ r = ReprOutcome(outcome.make_repr())
+ assert r.is_critical
+
+
Modified: py/branch/distributed/py/test/rsession/testing/test_rsession.py
==============================================================================
--- py/branch/distributed/py/test/rsession/testing/test_rsession.py (original)
+++ py/branch/distributed/py/test/rsession/testing/test_rsession.py Wed Aug 23 18:16:11 2006
@@ -9,23 +9,25 @@
funcpass_spec, funcskip_spec)
def setup_module(mod):
- mod.rootdir = py.path.local(py.__file__).dirpath().dirpath()
+ mod.pkgdir = py.path.local(py.__file__).dirpath()
from py.__.test.rsession.conftest import option
if not option.disthosts:
py.test.skip("no test distribution ssh hosts specified")
mod.hosts = option.disthosts.split(",")
def test_setup_teardown_ssh():
-
setup_events = []
teardown_events = []
- nodes = init_hosts(setup_events.append, hosts, 'pytestest', rootdir)
- teardown_hosts(teardown_events.append, [node.channel for node in nodes])
+ nodes = init_hosts(setup_events.append, hosts, 'pytestest', pkgdir)
+ teardown_hosts(teardown_events.append,
+ [node.channel for node in nodes])
- count_rsyn_calls = [i for i in setup_events if isinstance(i, report.CallStart)]
+ count_rsyn_calls = [i for i in setup_events
+ if isinstance(i, report.CallStart)]
assert len(count_rsyn_calls) == len(hosts)
- count_rsyn_ends = [i for i in setup_events if isinstance(i, report.CallFinish)]
+ count_rsyn_ends = [i for i in setup_events
+ if isinstance(i, report.CallFinish)]
assert len(count_rsyn_ends) == len(hosts)
# same for teardown events
@@ -38,14 +40,13 @@
def test_setup_teardown_run_ssh():
allevents = []
- hosts = ['codespeak.net']
- nodes = init_hosts(allevents.append, hosts, 'pytestest', rootdir)
+ nodes = init_hosts(allevents.append, hosts, 'pytestest', pkgdir)
from py.__.test.rsession.testing.test_executor \
import ItemTestPassing, ItemTestFailing, ItemTestSkipping
- rootcol = py.test.collect.Directory(rootdir)
+ rootcol = py.test.collect.Directory(pkgdir.dirpath())
itempass = rootcol.getitembynames(funcpass_spec)
itemfail = rootcol.getitembynames(funcfail_spec)
itemskip = rootcol.getitembynames(funcskip_spec)
@@ -58,9 +59,12 @@
teardown_hosts(allevents.append, [node.channel for node in nodes])
- events = [i for i in allevents if isinstance(i, report.ReceivedItemOutcome)]
- passed = [i for i in events if i.outcome.passed]
- skipped = [i for i in events if i.outcome.skipped]
+ events = [i for i in allevents
+ if isinstance(i, report.ReceivedItemOutcome)]
+ passed = [i for i in events
+ if i.outcome.passed]
+ skipped = [i for i in events
+ if i.outcome.skipped]
assert len(passed) == len(nodes)
assert len(skipped) == len(nodes)
assert len(events) == 3 * len(nodes)
More information about the pytest-commit
mailing list