floraachy/pytest2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[svn r37278] move files from branch to trunk (and thus complete

Browse Source 
the merge of the config branch into the trunk)

--HG--
branch : trunk
This commit is contained in:
hpk
2007-01-24 17:46:46 +01:00
parent 638e4318e4
commit 7cf9824680
74 changed files with 4201 additions and 6210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/test/rsession/testing/test_boxing.py
+10 -13
View File
@@ -9,17 +9,16 @@ if sys.platform == 'win32':
from py.__.test.rsession.box import Box
from py.__.test.rsession.testing import example2
from py.__.test.rsession.conftest import option
def setup_module(mod):
from py.__.test.rsession.rsession import remote_options
remote_options['nice_level'] = 0
mod.rootdir = py.path.local(py.__file__).dirpath().dirpath()
mod.config = py.test.config._reparse([mod.rootdir])
def test_basic_boxing():
# XXX: because we do not have option transfer
## if not hasattr(option, 'nocapture') or not option.nocapture:
## py.test.skip("Interacts with pylib i/o skipping which is bad actually")
b = Box(example2.boxf1)
b = Box(example2.boxf1, config=config)
b.run()
assert b.stdoutrepr == "some out\n"
assert b.stderrrepr == "some err\n"
@@ -28,7 +27,7 @@ def test_basic_boxing():
assert b.retval == 1
def test_boxing_on_fds():
b = Box(example2.boxf2)
b = Box(example2.boxf2, config=config)
b.run()
assert b.stdoutrepr == "someout"
assert b.stderrrepr == "someerr"
@@ -37,13 +36,13 @@ def test_boxing_on_fds():
assert b.retval == 2
def test_boxing_signal():
b = Box(example2.boxseg)
b = Box(example2.boxseg, config=config)
b.run()
assert b.signal == 11
assert b.retval is None
def test_boxing_huge_data():
b = Box(example2.boxhuge)
b = Box(example2.boxhuge, config=config)
b.run()
assert b.stdoutrepr
assert b.exitstat == 0
@@ -53,7 +52,7 @@ def test_boxing_huge_data():
def test_box_seq():
# we run many boxes with huge data, just one after another
for i in xrange(100):
b = Box(example2.boxhuge)
b = Box(example2.boxhuge, config=config)
b.run()
assert b.stdoutrepr
assert b.exitstat == 0
@@ -62,13 +61,13 @@ def test_box_seq():
def test_box_in_a_box():
def boxfun():
b = Box(example2.boxf2)
b = Box(example2.boxf2, config=config)
b.run()
print b.stdoutrepr
print >>sys.stderr, b.stderrrepr
return b.retval
b = Box(boxfun)
b = Box(boxfun, config=config)
b.run()
assert b.stdoutrepr == "someout\n"
assert b.stderrrepr == "someerr\n"
@@ -77,8 +76,6 @@ def test_box_in_a_box():
assert b.retval == 2
def test_box_killer():
if option.skip_kill_test:
py.test.skip('skipping kill test')
class A:
pass
info = A()
@@ -87,7 +84,7 @@ def test_box_killer():
def box_fun():
time.sleep(10) # we don't want to last forever here
b = Box(box_fun)
b = Box(box_fun, config=config)
par, pid = b.run(continuation=True)
os.kill(pid, 15)
par(pid)
py/test/rsession/testing/test_config.py
-24
View File
@@ -1,24 +0,0 @@
""" test session config options
"""
import py
from py.__.test.rsession.rsession import session_options, SessionOptions,\
remote_options
def test_session_opts():
tmpdir = py.test.ensuretemp("sessionopts")
tmpdir.ensure("conftest.py").write("""class SessionOptions:
max_tasks_per_node = 5
nice_level = 10
""")
tmp2 = py.test.ensuretemp("xxx")
args = [str(tmpdir)]
config = py.test.config._reparse(args)
session_options.bind_config(config)
assert session_options.max_tasks_per_node == 5
assert remote_options.nice_level == 10
config = py.test.config._reparse([str(tmp2)])
session_options.bind_config(config)
assert session_options.max_tasks_per_node == \
SessionOptions.defaults['max_tasks_per_node']
py/test/rsession/testing/test_executor.py
+10 -11
View File
@@ -10,8 +10,7 @@ from py.__.test.rsession.testing.test_slave import funcprint_spec, \
def setup_module(mod):
mod.rootdir = py.path.local(py.__file__).dirpath().dirpath()
from py.__.test.rsession.rsession import remote_options
remote_options['nice_level'] = 0
mod.config = py.test.config._reparse([mod.rootdir])
def XXXtest_executor_passing_function():
ex = Executor(example1.f1)
@@ -61,32 +60,32 @@ class ItemTestSkipping(py.test.Item):
py.test.skip("hello")
def test_run_executor():
ex = RunExecutor(ItemTestPassing("pass"))
ex = RunExecutor(ItemTestPassing("pass"), config=config)
outcome = ex.execute()
assert outcome.passed
ex = RunExecutor(ItemTestFailing("fail"))
ex = RunExecutor(ItemTestFailing("fail"), config=config)
outcome = ex.execute()
assert not outcome.passed
ex = RunExecutor(ItemTestSkipping("skip"))
ex = RunExecutor(ItemTestSkipping("skip"), config=config)
outcome = ex.execute()
assert outcome.skipped
assert not outcome.passed
assert not outcome.excinfo
def test_box_executor():
ex = BoxExecutor(ItemTestPassing("pass"))
ex = BoxExecutor(ItemTestPassing("pass"), config=config)
outcome_repr = ex.execute()
outcome = ReprOutcome(outcome_repr)
assert outcome.passed
ex = BoxExecutor(ItemTestFailing("fail"))
ex = BoxExecutor(ItemTestFailing("fail"), config=config)
outcome_repr = ex.execute()
outcome = ReprOutcome(outcome_repr)
assert not outcome.passed
ex = BoxExecutor(ItemTestSkipping("skip"))
ex = BoxExecutor(ItemTestSkipping("skip"), config=config)
outcome_repr = ex.execute()
outcome = ReprOutcome(outcome_repr)
assert outcome.skipped
@@ -96,7 +95,7 @@ def test_box_executor():
def test_box_executor_stdout():
rootcol = py.test.collect.Directory(rootdir)
item = rootcol.getitembynames(funcprint_spec)
ex = BoxExecutor(item)
ex = BoxExecutor(item, config=config)
outcome_repr = ex.execute()
outcome = ReprOutcome(outcome_repr)
assert outcome.passed
@@ -105,7 +104,7 @@ def test_box_executor_stdout():
def test_box_executor_stdout_error():
rootcol = py.test.collect.Directory(rootdir)
item = rootcol.getitembynames(funcprintfail_spec)
ex = BoxExecutor(item)
ex = BoxExecutor(item, config=config)
outcome_repr = ex.execute()
outcome = ReprOutcome(outcome_repr)
assert not outcome.passed
@@ -114,7 +113,7 @@ def test_box_executor_stdout_error():
def test_cont_executor():
rootcol = py.test.collect.Directory(rootdir)
item = rootcol.getitembynames(funcprintfail_spec)
ex = AsyncExecutor(item)
ex = AsyncExecutor(item, config=config)
cont, pid = ex.execute()
assert pid
outcome_repr = cont()
py/test/rsession/testing/test_lsession.py
+48 -16
View File
@@ -5,7 +5,7 @@
import py
from py.__.test.rsession.rsession import LSession
from py.__.test.rsession import report
from py.__.test.rsession.local import box_runner, plain_runner
from py.__.test.rsession.local import box_runner, plain_runner, apigen_runner
def setup_module(mod):
mod.tmp = py.test.ensuretemp("lsession_module")
@@ -82,22 +82,26 @@ class TestLSession(object):
l = []
def some_fun(*args):
l.append(args)
pdb.post_mortem = some_fun
args = [str(tmpdir.join(subdir)), '--pdb']
config = py.test.config._reparse(args)
lsession = LSession(config)
allevents = []
try:
lsession.main(reporter=allevents.append, runner=plain_runner)
except SystemExit:
pass
else:
py.test.fail("Didn't raise system exit")
failure_events = [event for event in allevents if isinstance(event,
report.ImmediateFailure)]
assert len(failure_events) == 1
assert len(l) == 1
post_mortem = pdb.post_mortem
pdb.post_mortem = some_fun
args = [str(tmpdir.join(subdir)), '--pdb']
config = py.test.config._reparse(args)
lsession = LSession(config)
allevents = []
try:
lsession.main(reporter=allevents.append, runner=plain_runner)
except SystemExit:
pass
else:
py.test.fail("Didn't raise system exit")
failure_events = [event for event in allevents if isinstance(event,
report.ImmediateFailure)]
assert len(failure_events) == 1
assert len(l) == 1
finally:
pdb.post_mortem = post_mortem
def test_minus_x(self):
if not hasattr(py.std.os, 'fork'):
@@ -262,3 +266,31 @@ class TestLSession(object):
assert testevents[0].outcome.passed
assert testevents[0].outcome.stderr == ""
assert testevents[0].outcome.stdout == "1\n2\n3\n"
def test_runner_selection(self):
tmpdir = py.test.ensuretemp("lsession_runner_selection")
tmpdir.ensure("apigen.py").write(py.code.Source("""
def get_documentable_items(*args):
return {}
"""))
opt_mapping = {
'': plain_runner,
'--box': box_runner,
'--apigen=%s/apigen.py' % str(tmpdir): apigen_runner,
}
pkgdir = tmpdir.dirpath()
for opt in opt_mapping.keys():
if opt:
all = opt + " " + str(tmpdir)
else:
all = str(tmpdir)
config = py.test.config._reparse(all.split(" "))
lsession = LSession(config)
assert lsession.init_runner(pkgdir) is opt_mapping[opt]
#tmpdir.dirpath().ensure("conftest.py").write(py.code.Source("""
#dist_boxing=True
#"""))
#config = py.test.config._reparse([str(tmpdir)])
#lsession = LSession(config)
#assert lsession.init_runner(pkgdir) is box_runner
# XXX check why it fails
py/test/rsession/testing/test_master.py
+35 -8
View File
@@ -11,17 +11,14 @@ if sys.platform == 'win32':
from py.__.test.rsession.master import dispatch_loop, setup_slave, MasterNode, randomgen
from py.__.test.rsession.outcome import ReprOutcome, Outcome
from py.__.test.rsession.testing.test_slave import funcpass_spec, funcfail_spec
from py.__.test.rsession.testing.test_slave import funcpass_spec, funcfail_spec, funchang_spec
from py.__.test.rsession import report
from py.__.test.rsession.rsession import session_options, remote_options
from py.__.test.rsession.hostmanage import HostInfo
def setup_module(mod):
# bind an empty config
config = py.test.config._reparse([])
config.option.max_tasks_per_node = 10
session_options.bind_config(config)
#assert not remote_options.exitfirst
config._overwrite('dist_taskspernode', 10)
mod.pkgdir = py.path.local(py.__file__).dirpath()
class DummyGateway(object):
@@ -92,11 +89,12 @@ def test_dispatch_loop():
def test_slave_setup():
gw = py.execnet.PopenGateway()
channel = setup_slave(gw, pkgdir, remote_options.d)
config = py.test.config._reparse([])
channel = setup_slave(gw, pkgdir, config)
channel.send(funcpass_spec)
output = ReprOutcome(channel.receive())
assert output.passed
channel.send(None)
channel.send(42)
channel.waitclose(10)
gw.exit()
@@ -114,7 +112,8 @@ def test_slave_running():
gw = py.execnet.PopenGateway()
gw.host = HostInfo("localhost")
gw.host.gw = gw
channel = setup_slave(gw, pkgdir, remote_options.d)
config = py.test.config._reparse([])
channel = setup_slave(gw, pkgdir, config)
mn = MasterNode(channel, simple_report, {})
return mn
@@ -127,6 +126,34 @@ def test_slave_running():
shouldstop = lambda : False
dispatch_loop(master_nodes, itemgenerator, shouldstop)
def test_slave_running_interrupted():
#def simple_report(event):
# if not isinstance(event, report.ReceivedItemOutcome):
# return
# item = event.item
# if item.code.name == 'funcpass':
# assert event.outcome.passed
# else:
# assert not event.outcome.passed
reports = []
def open_gw():
gw = py.execnet.PopenGateway()
gw.host = HostInfo("localhost")
gw.host.gw = gw
config = py.test.config._reparse([])
channel = setup_slave(gw, pkgdir, config)
mn = MasterNode(channel, reports.append, {})
return mn, gw, channel
mn, gw, channel = open_gw()
rootcol = py.test.collect.Directory(pkgdir.dirpath())
funchang_item = rootcol.getitembynames(funchang_spec)
mn.send(funchang_item)
mn.send(StopIteration)
# XXX: We have to wait here a bit to make sure that it really did happen
channel.waitclose(2)
def test_randomgen():
d = {}
gen = randomgen({1:True, 2:True, 3:True}, d)
py/test/rsession/testing/test_outcome.py
+1 -1
View File
@@ -28,7 +28,7 @@ def test_exception_info_repr():
except:
outcome = Outcome(excinfo=py.code.ExceptionInfo())
repr = outcome.make_excinfo_repr()
repr = outcome.make_excinfo_repr("long")
assert marshal.dumps(repr)
excinfo = ExcInfoRepr(repr)
py/test/rsession/testing/test_reporter.py
+15 -13
View File
@@ -166,11 +166,12 @@ class AbstractTestReporter(object):
r.report(report.HostReady(hosts[2]))
out, err = cap.reset()
assert not err
expected = """================= Test started, hosts: host1, host2, host3 ==================
host1: READY (still 2 to go)
expected1 = "Test started, hosts: host1, host2, host3"
expected2 = """host1: READY (still 2 to go)
host2: READY (still 1 to go)
host3: READY"""
assert out.find(expected) != -1
assert out.find(expected1) != -1
assert out.find(expected2) != -1
class TestLocalReporter(AbstractTestReporter):
reporter = LocalReporter
@@ -181,8 +182,9 @@ class TestLocalReporter(AbstractTestReporter):
def test_module(self):
#py.test.skip("XXX rewrite test to not rely on exact formatting")
assert self._test_module().endswith("test_slave.py[9] FsF."),\
self._test_module()
output = self._test_module()
assert output.find("test_slave") != -1
assert output.endswith("FsF."), output
def test_full_module(self):
#py.test.skip("XXX rewrite test to not rely on exact formatting")
@@ -202,20 +204,20 @@ class TestRemoteReporter(AbstractTestReporter):
def test_report_received_item_outcome(self):
#py.test.skip("XXX rewrite test to not rely on exact formatting")
val = self.report_received_item_outcome()
expected = """ localhost: FAILED py test rsession testing test_slave.py funcpass
localhost: SKIPPED py test rsession testing test_slave.py funcpass
localhost: FAILED py test rsession testing test_slave.py funcpass
localhost: PASSED py test rsession testing test_slave.py funcpass
expected = """ localhost: FAILED py.test.rsession.testing.test_slave.py funcpass
localhost: SKIPPED py.test.rsession.testing.test_slave.py funcpass
localhost: FAILED py.test.rsession.testing.test_slave.py funcpass
localhost: PASSED py.test.rsession.testing.test_slave.py funcpass
"""
assert val.find(expected) != -1
def test_module(self):
val = self._test_module()
print val
expected = """ localhost: FAILED py test rsession testing test_slave.py funcpass
localhost: SKIPPED py test rsession testing test_slave.py funcpass
localhost: FAILED py test rsession testing test_slave.py funcpass
localhost: PASSED py test rsession testing test_slave.py funcpass
expected = """ localhost: FAILED py.test.rsession.testing.test_slave.py funcpass
localhost: SKIPPED py.test.rsession.testing.test_slave.py funcpass
localhost: FAILED py.test.rsession.testing.test_slave.py funcpass
localhost: PASSED py.test.rsession.testing.test_slave.py funcpass
"""
assert val.find(expected) != -1
py/test/rsession/testing/test_rest.py
+2
View File
@@ -10,6 +10,8 @@ from py.__.test.rsession.rest import RestReporter, NoLinkWriter
from py.__.rest.rst import *
from py.__.test.rsession.hostmanage import HostInfo
py.test.skip("This tests are not really testing, needs rewrite")
class RestTestReporter(RestReporter):
def __init__(self, *args, **kwargs):
if args:
py/test/rsession/testing/test_rsession.py
+59 -47
View File
@@ -5,12 +5,12 @@
import py
from py.__.test.rsession import report
from py.__.test.rsession.rsession import RSession, parse_directories,\
session_options, remote_options, parse_directories
from py.__.test.rsession.hostmanage import init_hosts, teardown_hosts,\
parse_directories
from py.__.test.rsession.hostmanage import HostOptions, HostManager,\
HostInfo
from py.__.test.rsession.testing.test_slave import funcfail_spec,\
funcpass_spec, funcskip_spec, funcprint_spec, funcprintfail_spec, \
funcoptioncustom_spec, funcoption_spec
funcoptioncustom_spec
def setup_module(mod):
mod.pkgdir = py.path.local(py.__file__).dirpath()
@@ -18,7 +18,8 @@ def setup_module(mod):
def test_setup_non_existing_hosts():
setup_events = []
hosts = [HostInfo("alskdjalsdkjasldkajlsd")]
cmd = "init_hosts(setup_events.append, hosts, pkgdir)"
hm = HostManager(hosts, None, pkgdir)
cmd = "hm.init_hosts(setup_events.append)"
py.test.raises((py.process.cmdexec.Error, IOError, EOFError), cmd)
#assert setup_events
@@ -105,7 +106,7 @@ class TestRSessionRemote:
tmpdir = py.test.ensuretemp("example_distribution")
tmpdir.ensure(subdir, "conftest.py").write(py.code.Source("""
disthosts = [%r]
distrsync_roots = ["%s"]
distrsync_roots = ["%s", "py"]
""" % ('localhost', subdir)))
tmpdir.ensure(subdir, "__init__.py")
tmpdir.ensure(subdir, "test_one.py").write(py.code.Source("""
@@ -119,7 +120,11 @@ class TestRSessionRemote:
pass
def test_5():
assert __file__ != '%s'
""" % str(tmpdir.join(subdir))))
def test_6():
import py
assert py.__file__ != '%s'
""" % (str(tmpdir.join(subdir)), py.__file__)))
tmpdir.join("py").mksymlinkto(py.path.local(py.__file__).dirpath())
args = [str(tmpdir.join(subdir))]
config = py.test.config._reparse(args)
rsession = RSession(config, optimise_localhost=False)
@@ -131,21 +136,21 @@ class TestRSessionRemote:
passevents = [i for i in testevents if i.outcome.passed]
failevents = [i for i in testevents if i.outcome.excinfo]
skippedevents = [i for i in testevents if i.outcome.skipped]
assert len(testevents) == 5
assert len(passevents) == 2
assert len(testevents) == 6
assert len(passevents) == 3
assert len(failevents) == 3
tb = failevents[0].outcome.excinfo.traceback
assert tb[0].path.find("test_one") != -1
assert str(tb[0].path).find("test_one") != -1
assert tb[0].source.find("test_2") != -1
assert failevents[0].outcome.excinfo.typename == 'AssertionError'
tb = failevents[1].outcome.excinfo.traceback
assert tb[0].path.find("test_one") != -1
assert str(tb[0].path).find("test_one") != -1
assert tb[0].source.find("test_3") != -1
assert failevents[1].outcome.excinfo.typename == 'ValueError'
assert failevents[1].outcome.excinfo.value == '23'
tb = failevents[2].outcome.excinfo.traceback
assert failevents[2].outcome.excinfo.typename == 'TypeError'
assert tb[0].path.find("executor") != -1
assert str(tb[0].path).find("executor") != -1
assert tb[0].source.find("execute") != -1
def test_setup_teardown_ssh(self):
@@ -155,10 +160,10 @@ class TestRSessionRemote:
teardown_events = []
config = py.test.config._reparse([])
session_options.bind_config(config)
nodes = init_hosts(setup_events.append, hosts, pkgdir,
rsync_roots=["py"], optimise_localhost=False, remote_options=remote_options.d)
teardown_hosts(teardown_events.append,
opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(setup_events.append)
hm.teardown_hosts(teardown_events.append,
[node.channel for node in nodes], nodes)
count_rsyn_calls = [i for i in setup_events
@@ -182,9 +187,9 @@ class TestRSessionRemote:
allevents = []
config = py.test.config._reparse([])
session_options.bind_config(config)
nodes = init_hosts(allevents.append, hosts, pkgdir,
rsync_roots=["py"], optimise_localhost=False, remote_options=remote_options.d)
opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(allevents.append)
from py.__.test.rsession.testing.test_executor \
import ItemTestPassing, ItemTestFailing, ItemTestSkipping
@@ -202,7 +207,7 @@ class TestRSessionRemote:
node.send(itemskip)
node.send(itemprint)
teardown_hosts(allevents.append, [node.channel for node in nodes], nodes)
hm.teardown_hosts(allevents.append, [node.channel for node in nodes], nodes)
events = [i for i in allevents
if isinstance(i, report.ReceivedItemOutcome)]
@@ -224,31 +229,33 @@ class TestRSessionRemote:
hosts = [HostInfo('localhost')]
parse_directories(hosts)
config = py.test.config._reparse([])
session_options.bind_config(config)
d = remote_options.d.copy()
d['custom'] = 'custom'
nodes = init_hosts(allevents.append, hosts, pkgdir,
rsync_roots=["py"], remote_options=d,
optimise_localhost=False)
rootcol = py.test.collect.Directory(pkgdir.dirpath())
itempass = rootcol.getitembynames(funcoption_spec)
itempassaswell = rootcol.getitembynames(funcoptioncustom_spec)
for node in nodes:
node.send(itempass)
node.send(itempassaswell)
teardown_hosts(allevents.append, [node.channel for node in nodes], 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]
assert len(passed) == 2 * len(nodes)
assert len(skipped) == 0
assert len(events) == len(passed)
config._overwrite('custom', 'custom')
# we need to overwrite default list to serialize
from py.__.test.rsession.master import defaultconftestnames
defaultconftestnames.append("custom")
try:
opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(allevents.append)
rootcol = py.test.collect.Directory(pkgdir.dirpath())
itempass = rootcol.getitembynames(funcoptioncustom_spec)
for node in nodes:
node.send(itempass)
hm.teardown_hosts(allevents.append, [node.channel for node in nodes], 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]
assert len(passed) == 1 * len(nodes)
assert len(skipped) == 0
assert len(events) == len(passed)
finally:
defaultconftestnames.remove("custom")
def test_nice_level(self):
""" Tests if nice level behaviour is ok
@@ -258,14 +265,16 @@ class TestRSessionRemote:
parse_directories(hosts)
tmpdir = py.test.ensuretemp("nice")
tmpdir.ensure("__init__.py")
tmpdir.ensure("conftest.py").write("""disthosts = ['localhost']""")
tmpdir.ensure("conftest.py").write(py.code.Source("""
disthosts = ['localhost']
dist_nicelevel = 10
"""))
tmpdir.ensure("test_one.py").write("""def test_nice():
import os
assert os.nice(0) == 10
""")
config = py.test.config._reparse([tmpdir])
config.option.nice_level = 10
rsession = RSession(config)
allevents = []
rsession.main(reporter=allevents.append)
@@ -298,7 +307,10 @@ class TestInithosts(object):
hostnames = ['h1:/tmp', 'h1:/tmp', 'h1:/other', 'h2', 'h2:home']
hosts = [HostInfo(i) for i in hostnames]
parse_directories(hosts)
init_hosts(testevents.append, hosts, pkgdir, do_sync=False)
config = py.test.config._reparse([])
opts = HostOptions(do_sync=False, create_gateways=False)
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(testevents.append)
events = [i for i in testevents if isinstance(i, report.HostRSyncing)]
assert len(events) == 4
assert events[0].host.hostname == 'h1'
py/test/rsession/testing/test_rsession.py.merge.tmp
-324
View File
@@ -1,324 +0,0 @@
""" Tests various aspects of rsession, like ssh hosts setup/teardown
"""
import py
from py.__.test.rsession import report
from py.__.test.rsession.rsession import RSession, parse_directories,\
parse_directories
from py.__.test.rsession.hostmanage import HostOptions, HostManager,\
HostInfo
from py.__.test.rsession.testing.test_slave import funcfail_spec,\
funcpass_spec, funcskip_spec, funcprint_spec, funcprintfail_spec, \
funcoptioncustom_spec
def setup_module(mod):
mod.pkgdir = py.path.local(py.__file__).dirpath()
def test_setup_non_existing_hosts():
setup_events = []
hosts = [HostInfo("alskdjalsdkjasldkajlsd")]
hm = HostManager(hosts, None, pkgdir)
cmd = "hm.init_hosts(setup_events.append)"
py.test.raises((py.process.cmdexec.Error, IOError, EOFError), cmd)
#assert setup_events
def test_getpkdir():
one = pkgdir.join("initpkg.py")
two = pkgdir.join("path", "__init__.py")
p1 = RSession.getpkgdir(one)
p2 = RSession.getpkgdir(two)
assert p1 == p2
assert p1 == pkgdir
def test_getpkdir_no_inits():
tmp = py.test.ensuretemp("getpkdir1")
fn = tmp.ensure("hello.py")
assert RSession.getpkgdir(fn) == fn
def test_make_colitems():
one = pkgdir.join("initpkg.py")
two = pkgdir.join("path", "__init__.py")
cols = RSession.make_colitems([one, two], baseon=pkgdir)
assert len(cols) == 2
col_one, col_two = cols
assert col_one.listnames() == ["py", "initpkg.py"]
assert col_two.listnames() == ["py", "path", "__init__.py"]
cols = RSession.make_colitems([one, two], baseon=pkgdir.dirpath())
assert len(cols) == 2
col_one, col_two = cols
assert col_one.listnames() == [pkgdir.dirpath().basename,
"py", "initpkg.py"]
assert col_two.listnames() == [pkgdir.dirpath().basename,
"py", "path", "__init__.py"]
def test_example_tryiter():
events = []
tmpdir = py.test.ensuretemp("tryitertest")
tmpdir.ensure("a", "__init__.py")
tmpdir.ensure("conftest.py").write(py.code.Source("""
import py
py.test.skip("Reason")
"""))
tmpdir.ensure("a", "test_empty.py").write(py.code.Source("""
def test_empty():
pass
"""))
rootcol = py.test.collect.Directory(tmpdir)
data = list(rootcol.tryiter(reporterror=events.append))
assert len(events) == 2
assert str(events[1][0].value) == "Reason"
class TestRSessionRemote:
def test_example_distribution_minus_x(self):
tmpdir = py.test.ensuretemp("example_distribution_minus_x")
tmpdir.ensure("sub", "conftest.py").write(py.code.Source("""
disthosts = [%r]
""" % ('localhost',)))
tmpdir.ensure("sub", "__init__.py")
tmpdir.ensure("sub", "test_one.py").write(py.code.Source("""
def test_1():
pass
def test_x():
import py
py.test.skip("aaa")
def test_2():
assert 0
def test_3():
raise ValueError(23)
def test_4(someargs):
pass
"""))
args = [str(tmpdir.join("sub")), "-x"]
config = py.test.config._reparse(args)
rsession = RSession(config)
allevents = []
rsession.main(reporter=allevents.append)
testevents = [x for x in allevents
if isinstance(x, report.ReceivedItemOutcome)]
assert len(testevents) == 3
assert rsession.checkfun()
def test_example_distribution(self):
subdir = "sub_example_dist"
tmpdir = py.test.ensuretemp("example_distribution")
tmpdir.ensure(subdir, "conftest.py").write(py.code.Source("""
disthosts = [%r]
distrsync_roots = ["%s", "py"]
""" % ('localhost', subdir)))
tmpdir.ensure(subdir, "__init__.py")
tmpdir.ensure(subdir, "test_one.py").write(py.code.Source("""
def test_1():
pass
def test_2():
assert 0
def test_3():
raise ValueError(23)
def test_4(someargs):
pass
def test_5():
assert __file__ != '%s'
def test_6():
import py
assert py.__file__ != '%s'
""" % (str(tmpdir.join(subdir)), py.__file__)))
tmpdir.join("py").mksymlinkto(py.path.local(py.__file__).dirpath())
args = [str(tmpdir.join(subdir))]
config = py.test.config._reparse(args)
rsession = RSession(config, optimise_localhost=False)
allevents = []
rsession.main(reporter=allevents.append)
testevents = [x for x in allevents
if isinstance(x, report.ReceivedItemOutcome)]
assert len(testevents)
passevents = [i for i in testevents if i.outcome.passed]
failevents = [i for i in testevents if i.outcome.excinfo]
skippedevents = [i for i in testevents if i.outcome.skipped]
assert len(testevents) == 6
assert len(passevents) == 3
assert len(failevents) == 3
tb = failevents[0].outcome.excinfo.traceback
assert str(tb[0].path).find("test_one") != -1
assert tb[0].source.find("test_2") != -1
assert failevents[0].outcome.excinfo.typename == 'AssertionError'
tb = failevents[1].outcome.excinfo.traceback
assert str(tb[0].path).find("test_one") != -1
assert tb[0].source.find("test_3") != -1
assert failevents[1].outcome.excinfo.typename == 'ValueError'
assert failevents[1].outcome.excinfo.value == '23'
tb = failevents[2].outcome.excinfo.traceback
assert failevents[2].outcome.excinfo.typename == 'TypeError'
assert str(tb[0].path).find("executor") != -1
assert tb[0].source.find("execute") != -1
def test_setup_teardown_ssh(self):
hosts = [HostInfo('localhost')]
parse_directories(hosts)
setup_events = []
teardown_events = []
config = py.test.config._reparse([])
opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(setup_events.append)
hm.teardown_hosts(teardown_events.append,
[node.channel for node in nodes], nodes)
count_rsyn_calls = [i for i in setup_events
if isinstance(i, report.HostRSyncing)]
assert len(count_rsyn_calls) == len([i for i in hosts])
count_ready_calls = [i for i in setup_events
if isinstance(i, report.HostReady)]
assert len(count_ready_calls) == len([i for i in hosts])
# same for teardown events
teardown_wait_starts = [i for i in teardown_events
if isinstance(i, report.CallStart)]
teardown_wait_ends = [i for i in teardown_events
if isinstance(i, report.CallFinish)]
assert len(teardown_wait_starts) == len(hosts)
assert len(teardown_wait_ends) == len(hosts)
def test_setup_teardown_run_ssh(self):
hosts = [HostInfo('localhost')]
parse_directories(hosts)
allevents = []
config = py.test.config._reparse([])
opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(allevents.append)
from py.__.test.rsession.testing.test_executor \
import ItemTestPassing, ItemTestFailing, ItemTestSkipping
rootcol = py.test.collect.Directory(pkgdir.dirpath())
itempass = rootcol.getitembynames(funcpass_spec)
itemfail = rootcol.getitembynames(funcfail_spec)
itemskip = rootcol.getitembynames(funcskip_spec)
itemprint = rootcol.getitembynames(funcprint_spec)
# actually run some tests
for node in nodes:
node.send(itempass)
node.send(itemfail)
node.send(itemskip)
node.send(itemprint)
hm.teardown_hosts(allevents.append, [node.channel for node in nodes], 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]
assert len(passed) == 2 * len(nodes)
assert len(skipped) == len(nodes)
assert len(events) == 4 * len(nodes)
# one of passed for each node has non-empty stdout
passed_stdout = [i for i in passed if i.outcome.stdout.find('samfing') != -1]
assert len(passed_stdout) == len(nodes), passed
def test_config_pass(self):
""" Tests options object passing master -> server
"""
allevents = []
hosts = [HostInfo('localhost')]
parse_directories(hosts)
config = py.test.config._reparse([])
config._overwrite('custom', 'custom')
# we need to overwrite default list to serialize
from py.__.test.rsession.master import defaultconftestnames
defaultconftestnames.append("custom")
try:
opts = HostOptions(optimise_localhost=False, rsync_roots=['py'])
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(allevents.append)
rootcol = py.test.collect.Directory(pkgdir.dirpath())
itempass = rootcol.getitembynames(funcoptioncustom_spec)
for node in nodes:
node.send(itempass)
hm.teardown_hosts(allevents.append, [node.channel for node in nodes], 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]
assert len(passed) == 1 * len(nodes)
assert len(skipped) == 0
assert len(events) == len(passed)
finally:
defaultconftestnames.remove("custom")
def test_nice_level(self):
""" Tests if nice level behaviour is ok
"""
allevents = []
hosts = [HostInfo('localhost')]
parse_directories(hosts)
tmpdir = py.test.ensuretemp("nice")
tmpdir.ensure("__init__.py")
tmpdir.ensure("conftest.py").write(py.code.Source("""
disthosts = ['localhost']
dist_nicelevel = 10
"""))
tmpdir.ensure("test_one.py").write("""def test_nice():
import os
assert os.nice(0) == 10
""")
config = py.test.config._reparse([tmpdir])
rsession = RSession(config)
allevents = []
rsession.main(reporter=allevents.append)
testevents = [x for x in allevents
if isinstance(x, report.ReceivedItemOutcome)]
passevents = [x for x in testevents if x.outcome.passed]
assert len(passevents) == 1
class XxxTestDirectories(object):
# need complete rewrite, and unsure if it makes sense at all
def test_simple_parse(self):
sshhosts = [HostInfo(i) for i in ['h1', 'h2', 'h3']]
parse_directories(sshhosts)
def test_sophisticated_parse(self):
sshhosts = ['a@h1:/tmp', 'h2:tmp', 'h3']
dirs = parse_directories(sshhosts)
assert py.builtin.sorted(
dirs.values()) == ['/tmp', 'pytestcache', 'tmp']
def test_parse_multiple_hosts(self):
hosts = ['h1', 'h1', 'h1:/tmp']
dirs = parse_directories(hosts)
assert dirs == {(0, 'h1'): 'pytestcache', (1, 'h1'): 'pytestcache',
(2, 'h1'):'/tmp'}
class TestInithosts(object):
def test_inithosts(self):
testevents = []
hostnames = ['h1:/tmp', 'h1:/tmp', 'h1:/other', 'h2', 'h2:home']
hosts = [HostInfo(i) for i in hostnames]
parse_directories(hosts)
config = py.test.config._reparse([])
opts = HostOptions(do_sync=False, create_gateways=False)
hm = HostManager(hosts, config, pkgdir, opts)
nodes = hm.init_hosts(testevents.append)
events = [i for i in testevents if isinstance(i, report.HostRSyncing)]
assert len(events) == 4
assert events[0].host.hostname == 'h1'
assert events[0].host.relpath == '/tmp-h1'
assert events[1].host.hostname == 'h1'
assert events[1].host.relpath == '/other-h1'
assert events[2].host.hostname == 'h2'
assert events[2].host.relpath == 'pytestcache-h2'
assert events[3].host.hostname == 'h2'
assert events[3].host.relpath == 'home-h2'
py/test/rsession/testing/test_rsync.py
-80
View File
@@ -1,80 +0,0 @@
import py
from py.__.test.rsession.rsync import RSync
def setup_module(mod):
mod.gw = py.execnet.PopenGateway()
mod.gw2 = py.execnet.PopenGateway()
def teardown_module(mod):
mod.gw.exit()
mod.gw2.exit()
def test_dirsync():
base = py.test.ensuretemp('dirsync')
dest = base.join('dest')
dest2 = base.join('dest2')
source = base.mkdir('source')
for s in ('content1', 'content2-a-bit-longer'):
source.ensure('subdir', 'file1').write(s)
rsync = RSync()
rsync.add_target(gw, dest)
rsync.add_target(gw2, dest2)
rsync.send(source)
assert dest.join('subdir').check(dir=1)
assert dest.join('subdir', 'file1').check(file=1)
assert dest.join('subdir', 'file1').read() == s
assert dest2.join('subdir').check(dir=1)
assert dest2.join('subdir', 'file1').check(file=1)
assert dest2.join('subdir', 'file1').read() == s
source.join('subdir').remove('file1')
rsync = RSync()
rsync.add_target(gw2, dest2)
rsync.add_target(gw, dest)
rsync.send(source)
assert dest.join('subdir', 'file1').check(file=1)
assert dest2.join('subdir', 'file1').check(file=1)
rsync = RSync(delete=True)
rsync.add_target(gw2, dest2)
rsync.add_target(gw, dest)
rsync.send(source)
assert not dest.join('subdir', 'file1').check()
assert not dest2.join('subdir', 'file1').check()
def test_symlink_rsync():
if py.std.sys.platform == 'win32':
py.test.skip("symlinks are unsupported on Windows.")
base = py.test.ensuretemp('symlinkrsync')
dest = base.join('dest')
source = base.join('source')
source.ensure("existant")
source.join("rellink").mksymlinkto(source.join("existant"), absolute=0)
source.join('abslink').mksymlinkto(source.join("existant"))
rsync = RSync()
rsync.add_target(gw, dest)
rsync.send(source)
assert dest.join('rellink').readlink() == dest.join("existant")
assert dest.join('abslink').readlink() == dest.join("existant")
def test_callback():
base = py.test.ensuretemp('callback')
dest = base.join("dest")
source = base.join("source")
source.ensure("existant").write("a" * 100)
source.ensure("existant2").write("a" * 10)
total = {}
def callback(cmd, lgt, channel):
total[(cmd, lgt)] = True
rsync = RSync(callback=callback)
#rsync = RSync()
rsync.add_target(gw, dest)
rsync.send(source)
assert total == {("list", 110):True, ("ack", 100):True, ("ack", 10):True}
py/test/rsession/testing/test_slave.py
+38 -24
View File
@@ -1,6 +1,6 @@
""" Testing the slave side node code (in a local way). """
from py.__.test.rsession.slave import SlaveNode, slave_main
from py.__.test.rsession.slave import SlaveNode, slave_main, setup, PidInfo
from py.__.test.rsession.outcome import ReprOutcome
import py, sys
@@ -11,10 +11,7 @@ if sys.platform == 'win32':
py.test.skip("rsession is unsupported on Windows.")
def setup_module(module):
from py.__.test.rsession.rsession import session_options
module.rootdir = py.path.local(py.__file__).dirpath().dirpath()
config = py.test.config._reparse([])
session_options.bind_config(config)
# ----------------------------------------------------------------------
# inlined testing functions used below
@@ -34,13 +31,8 @@ def funcprintfail():
print "samfing elz"
asddsa
def funcoption():
from py.__.test.rsession.rsession import remote_options
assert remote_options.we_are_remote
def funcoptioncustom():
from py.__.test.rsession.rsession import remote_options
assert remote_options.custom == "custom"
assert py.test.config.getvalue("custom")
def funchang():
import time
@@ -52,7 +44,6 @@ funcfail_spec = (BASE + "funcfail").split("/")
funcskip_spec = (BASE + "funcskip").split("/")
funcprint_spec = (BASE + "funcprint").split("/")
funcprintfail_spec = (BASE + "funcprintfail").split("/")
funcoption_spec = (BASE + "funcoption").split("/")
funcoptioncustom_spec = (BASE + "funcoptioncustom").split("/")
funchang_spec = (BASE + "funchang").split("/")
mod_spec = BASE[:-1].split("/")
@@ -63,7 +54,9 @@ from py.__.test.rsession.executor import RunExecutor
def gettestnode():
rootcol = py.test.collect.Directory(rootdir)
node = SlaveNode(rootcol, executor=RunExecutor)
config = py.test.config._reparse([rootdir])
pidinfo = PidInfo()
node = SlaveNode(rootcol, config, pidinfo, executor=RunExecutor)
return node
def test_slave_run_passing():
@@ -116,7 +109,9 @@ def test_slave_main_simple():
funcpass_spec,
funcfail_spec
]
slave_main(q.pop, res.append, str(rootdir))
config = py.test.config._reparse([])
pidinfo = PidInfo()
slave_main(q.pop, res.append, str(rootdir), config, pidinfo)
assert len(res) == 2
res_repr = [ReprOutcome(r) for r in res]
assert not res_repr[0].passed and res_repr[1].passed
@@ -126,8 +121,8 @@ def test_slave_run_different_stuff():
node.run("py doc log.txt".split())
def test_slave_setup_fails_on_import_error():
from py.__.test.rsession.slave import setup
tmp = py.test.ensuretemp("slavesetup")
config = py.test.config._reparse([tmp])
class C:
def __init__(self):
self.count = 0
@@ -136,12 +131,15 @@ def test_slave_setup_fails_on_import_error():
if self.count == 0:
retval = str(tmp)
elif self.count == 1:
from py.__.test.rsession.rsession import remote_options
retval = remote_options.d
retval = config.make_repr(conftestnames=['dist_nicelevel'])
else:
raise NotImplementedError("more data")
self.count += 1
return retval
def close(self):
pass
try:
exec py.code.Source(setup, "setup()").compile() in {
'channel': C()}
@@ -153,16 +151,14 @@ def test_slave_setup_fails_on_import_error():
def test_slave_setup_exit():
tmp = py.test.ensuretemp("slaveexit")
tmp.ensure("__init__.py")
from py.__.test.rsession.slave import setup
from Queue import Queue
q = Queue()
q = py.std.Queue.Queue()
config = py.test.config._reparse([tmp])
class C:
res = []
def __init__(self):
from py.__.test.rsession.rsession import remote_options
self.q = [str(tmp),
remote_options.d,
config.make_repr(conftestnames=['dist_nicelevel']),
funchang_spec,
42,
funcpass_spec]
@@ -178,6 +174,9 @@ def test_slave_setup_exit():
callback(self.q.pop())
f()
#thread.start_new_thread(f, ())
def close(self):
pass
send = res.append
try:
@@ -188,8 +187,8 @@ def test_slave_setup_exit():
py.test.fail("Did not exit")
def test_slave_setup_fails_on_missing_pkg():
from py.__.test.rsession.slave import setup
tmp = py.test.ensuretemp("slavesetup2")
config = py.test.config._reparse([tmp])
x = tmp.ensure("sometestpackage", "__init__.py")
class C:
def __init__(self):
@@ -199,8 +198,7 @@ def test_slave_setup_fails_on_missing_pkg():
if self.count == 0:
retval = str(x.dirpath())
elif self.count == 1:
from py.__.test.rsession.rsession import remote_options
retval = remote_options.d
retval = config.make_repr(conftestnames=['dist_nicelevel'])
else:
raise NotImplementedError("more data")
self.count += 1
@@ -223,3 +221,19 @@ def test_slave_setup_fails_on_missing_pkg():
else:
py.test.fail("missing exception")
def test_pidinfo():
if not hasattr(os, 'fork') or not hasattr(os, 'waitpid'):
py.test.skip("Platform does not support fork")
pidinfo = PidInfo()
pid = os.fork()
if pid:
pidinfo.set_pid(pid)
pidinfo.waitandclear(pid, 0)
else:
import time, sys
time.sleep(.3)
os._exit(0)
# check if this really exits
py.test.raises(OSError, "os.waitpid(pid, 0)")
py/test/rsession/testing/test_web.py
+1 -3
View File
@@ -14,9 +14,7 @@ except ImportError:
def setup_module(mod):
config = py.test.config._reparse([])
from py.__.test.rsession.rsession import session_options
session_options.bind_config(config)
session_options.import_pypy = True
config._overwrite('_dist_import_pypy', True)
from py.__.test.rsession.web import TestHandler as _TestHandler
from py.__.test.rsession.web import MultiQueue
mod._TestHandler = _TestHandler
py/test/rsession/testing/test_webjs.py
+1 -3
View File
@@ -23,9 +23,7 @@ def setup_module(mod):
dom.window = dom.Window(html)
dom.document = dom.window.document
config = py.test.config._reparse([])
from py.__.test.rsession.rsession import session_options
session_options.bind_config(config)
session_options.import_pypy = True
config._overwrite('_dist_import_pypy', True)
from py.__.test.rsession import webjs
from py.__.test.rsession.web import exported_methods
mod.webjs = webjs
py/test/rsession/testing/test_webjs.py.merge.tmp
-138
View File
@@ -1,138 +0,0 @@
import py
try:
import pypy
from pypy.translator.js.modules import dom
from pypy.translator.js.tester import schedule_callbacks
from py.__.test.rsession.rsession import session_options
dom.Window # check whether dom was properly imported or is just a
# leftover in sys.modules
except (ImportError, AttributeError):
py.test.skip('PyPy not found')
from py.__.test.rsession import webjs
from py.__.test.rsession.web import exported_methods
here = py.magic.autopath().dirpath()
def setup_module(mod):
# load HTML into window object
html = here.join('../webdata/index.html').read()
mod.html = html
from pypy.translator.js.modules import dom
mod.dom = dom
dom.window = dom.Window(html)
dom.document = dom.window.document
config = py.test.config._reparse([])
config._overwrite('_dist_import_pypy', True)
from py.__.test.rsession import webjs
from py.__.test.rsession.web import exported_methods
mod.webjs = webjs
mod.exported_methods = exported_methods
def setup_function(f):
dom.window = dom.Window(html)
dom.document = dom.window.document
def test_html_loaded():
body = dom.window.document.getElementsByTagName('body')[0]
assert len(body.childNodes) > 0
assert str(body.childNodes[1].nodeName) == 'A'
def test_set_msgbox():
msgbox = dom.window.document.getElementById('messagebox')
assert len(msgbox.childNodes) == 0
webjs.set_msgbox('foo', 'bar')
assert len(msgbox.childNodes) == 1
assert msgbox.childNodes[0].nodeName == 'PRE'
assert msgbox.childNodes[0].childNodes[0].nodeValue == 'foo\nbar'
def test_show_info():
info = dom.window.document.getElementById('info')
info.style.visibility = 'hidden'
info.innerHTML = ''
webjs.show_info('foobar')
content = info.innerHTML
assert content == 'foobar'
bgcolor = info.style.backgroundColor
assert bgcolor == 'beige'
def test_hide_info():
info = dom.window.document.getElementById('info')
info.style.visibility = 'visible'
webjs.hide_info()
assert info.style.visibility == 'hidden'
def test_process():
main_t = dom.window.document.getElementById('main_table')
assert len(main_t.getElementsByTagName('tr')) == 0
assert not webjs.process({})
msg = {'type': 'ItemStart',
'itemtype': 'Module',
'itemname': 'foo.py',
'fullitemname': 'modules/foo.py',
'length': 10,
}
assert webjs.process(msg)
trs = main_t.getElementsByTagName('tr')
assert len(trs) == 1
tr = trs[0]
assert len(tr.childNodes) == 2
assert tr.childNodes[0].nodeName == 'TD'
assert tr.childNodes[0].innerHTML == 'foo.py[0/10]'
assert tr.childNodes[1].nodeName == 'TD'
assert tr.childNodes[1].childNodes[0].nodeName == 'TABLE'
assert len(tr.childNodes[1].getElementsByTagName('tr')) == 0
def test_process_two():
main_t = dom.window.document.getElementById('main_table')
msg = {'type': 'ItemStart',
'itemtype': 'Module',
'itemname': 'foo.py',
'fullitemname': 'modules/foo.py',
'length': 10,
}
webjs.process(msg)
msg = {'type': 'ReceivedItemOutcome',
'fullmodulename': 'modules/foo.py',
'passed' : 'True',
'fullitemname' : 'modules/foo.py/test_item',
'hostkey': None,
}
webjs.process(msg)
trs = main_t.getElementsByTagName('tr')
tds = trs[0].getElementsByTagName('td')
# two cells in the row, one in the table inside one of the cells
assert len(tds) == 3
html = tds[0].innerHTML
assert html == 'foo.py[1/10]'
assert tds[2].innerHTML == '.'
def test_signal():
main_t = dom.window.document.getElementById('main_table')
msg = {'type': 'ItemStart',
'itemtype': 'Module',
'itemname': 'foo.py',
'fullitemname': 'modules/foo.py',
'length': 10,
}
webjs.process(msg)
msg = {'type': 'ReceivedItemOutcome',
'fullmodulename': 'modules/foo.py',
'passed' : 'False',
'fullitemname' : 'modules/foo.py/test_item',
'hostkey': None,
'signal': '10',
'skipped': 'False',
}
exported_methods.fail_reasons['modules/foo.py/test_item'] = 'Received signal 10'
exported_methods.stdout['modules/foo.py/test_item'] = ''
exported_methods.stderr['modules/foo.py/test_item'] = ''
webjs.process(msg)
schedule_callbacks(exported_methods)
# ouch
assert dom.document.getElementById('modules/foo.py').childNodes[0].\
childNodes[0].childNodes[0].childNodes[0].nodeValue == 'F'
# XXX: Write down test for full run