224 lines
7.9 KiB
Python
224 lines
7.9 KiB
Python
|
|
""" reporter tests.
|
|
|
|
XXX there are a few disabled reporting tests because
|
|
they test for exact formatting as far as i can see.
|
|
I think it's rather better to directly invoke a
|
|
reporter and pass it some hand-prepared events to see
|
|
that running the reporter doesn't break shallowly.
|
|
|
|
Otherwise, i suppose that some "visual" testing can usually be driven
|
|
manually by user-input. And when passing particular events
|
|
to a reporter it's also easier to check for one line
|
|
instead of having to know the order in which things are printed
|
|
etc.
|
|
|
|
|
|
"""
|
|
|
|
|
|
import py, os
|
|
#py.test.skip("in progress")
|
|
from py.__.test.rsession.rsession import LocalReporter, AbstractSession,\
|
|
RemoteReporter
|
|
from py.__.test.rsession import repevent
|
|
from py.__.test.rsession.outcome import ReprOutcome, Outcome
|
|
from py.__.test.rsession.hostmanage import HostInfo
|
|
from py.__.test.rsession.box import Box
|
|
from py.__.test.rsession.testing.basetest import BasicRsessionTest
|
|
import sys
|
|
from StringIO import StringIO
|
|
|
|
class DummyGateway(object):
|
|
def __init__(self, host):
|
|
self.host = host
|
|
|
|
class DummyChannel(object):
|
|
def __init__(self, host):
|
|
self.gateway = DummyGateway(host)
|
|
|
|
class AbstractTestReporter(BasicRsessionTest):
|
|
def prepare_outcomes(self):
|
|
# possible outcomes
|
|
try:
|
|
1/0
|
|
except:
|
|
exc = py.code.ExceptionInfo()
|
|
|
|
outcomes = [Outcome(()),
|
|
Outcome(skipped=True),
|
|
Outcome(excinfo=exc),
|
|
Outcome()]
|
|
|
|
outcomes = [ReprOutcome(outcome.make_repr()) for outcome in outcomes]
|
|
outcomes[3].signal = 11
|
|
outcomes[0].passed = False
|
|
|
|
return outcomes
|
|
|
|
def report_received_item_outcome(self):
|
|
item = self.getexample("pass")
|
|
outcomes = self.prepare_outcomes()
|
|
|
|
def boxfun(config, item, outcomes):
|
|
hosts = [HostInfo("localhost")]
|
|
r = self.reporter(config, hosts)
|
|
ch = DummyChannel(hosts[0])
|
|
for outcome in outcomes:
|
|
r.report(repevent.ReceivedItemOutcome(ch, item, outcome))
|
|
|
|
cap = py.io.StdCaptureFD()
|
|
boxfun(self.config, item, outcomes)
|
|
out, err = cap.reset()
|
|
assert not err
|
|
return out
|
|
|
|
def _test_module(self):
|
|
funcitem = self.getexample("pass")
|
|
moditem = self.getmod()
|
|
outcomes = self.prepare_outcomes()
|
|
|
|
def boxfun(config, item, funcitem, outcomes):
|
|
hosts = [HostInfo('localhost')]
|
|
r = self.reporter(config, hosts)
|
|
r.report(repevent.ItemStart(item))
|
|
ch = DummyChannel(hosts[0])
|
|
for outcome in outcomes:
|
|
r.report(repevent.ReceivedItemOutcome(ch, funcitem, outcome))
|
|
|
|
cap = py.io.StdCaptureFD()
|
|
boxfun(self.config, moditem, funcitem, outcomes)
|
|
out, err = cap.reset()
|
|
assert not err
|
|
return out
|
|
|
|
def _test_full_module(self):
|
|
tmpdir = py.test.ensuretemp("repmod")
|
|
tmpdir.ensure("__init__.py")
|
|
tmpdir.ensure("test_one.py").write(py.code.Source("""
|
|
def test_x():
|
|
pass
|
|
"""))
|
|
tmpdir.ensure("test_two.py").write(py.code.Source("""
|
|
import py
|
|
py.test.skip("reason")
|
|
"""))
|
|
tmpdir.ensure("test_three.py").write(py.code.Source("""
|
|
sadsadsa
|
|
"""))
|
|
|
|
def boxfun():
|
|
config = py.test.config._reparse([str(tmpdir)])
|
|
rootcol = py.test.collect.Directory(tmpdir)
|
|
hosts = [HostInfo('localhost')]
|
|
r = self.reporter(config, hosts)
|
|
list(rootcol._tryiter(reporterror=lambda x : AbstractSession.reporterror(r.report, x)))
|
|
|
|
cap = py.io.StdCaptureFD()
|
|
boxfun()
|
|
out, err = cap.reset()
|
|
assert not err
|
|
return out
|
|
|
|
def test_failed_to_load(self):
|
|
tmpdir = py.test.ensuretemp("failedtoload")
|
|
tmpdir.ensure("__init__.py")
|
|
tmpdir.ensure("test_three.py").write(py.code.Source("""
|
|
sadsadsa
|
|
"""))
|
|
def boxfun():
|
|
config = py.test.config._reparse([str(tmpdir)])
|
|
rootcol = py.test.collect.Directory(tmpdir)
|
|
host = HostInfo('localhost')
|
|
r = self.reporter(config, [host])
|
|
r.report(repevent.TestStarted([host], config.topdir, ["a"]))
|
|
r.report(repevent.RsyncFinished())
|
|
list(rootcol._tryiter(reporterror=lambda x : AbstractSession.reporterror(r.report, x)))
|
|
r.report(repevent.TestFinished())
|
|
|
|
cap = py.io.StdCaptureFD()
|
|
boxfun()
|
|
out, err = cap.reset()
|
|
assert not err
|
|
assert out.find("NameError: name 'sadsadsa' is not defined") != -1
|
|
|
|
def _test_still_to_go(self):
|
|
tmpdir = py.test.ensuretemp("stilltogo")
|
|
tmpdir.ensure("__init__.py")
|
|
cap = py.io.StdCaptureFD()
|
|
config = py.test.config._reparse([str(tmpdir)])
|
|
hosts = [HostInfo(i) for i in ["host1", "host2", "host3"]]
|
|
r = self.reporter(config, hosts)
|
|
r.report(repevent.TestStarted(hosts, config.topdir, ["a", "b", "c"]))
|
|
for host in hosts:
|
|
r.report(repevent.HostGatewayReady(host, ["a", "b", "c"]))
|
|
for host in hosts:
|
|
for root in ["a", "b", "c"]:
|
|
r.report(repevent.HostRSyncRootReady(host, root))
|
|
out, err = cap.reset()
|
|
assert not err
|
|
expected1 = "Test started, hosts: host1[0], host2[0], host3[0]"
|
|
assert out.find(expected1) != -1
|
|
for expected in py.code.Source("""
|
|
host1[0]: READY (still 2 to go)
|
|
host2[0]: READY (still 1 to go)
|
|
host3[0]: READY
|
|
""").lines:
|
|
expected = expected.strip()
|
|
assert out.find(expected) != -1
|
|
|
|
class TestLocalReporter(AbstractTestReporter):
|
|
reporter = LocalReporter
|
|
|
|
def test_report_received_item_outcome(self):
|
|
#py.test.skip("XXX rewrite test to not rely on exact formatting")
|
|
assert self.report_received_item_outcome() == 'FsF.'
|
|
|
|
def test_module(self):
|
|
#py.test.skip("XXX rewrite test to not rely on exact formatting")
|
|
output = self._test_module()
|
|
assert output.find("test_one") != -1
|
|
assert output.endswith("FsF."), output
|
|
|
|
def test_full_module(self):
|
|
#py.test.skip("XXX rewrite test to not rely on exact formatting")
|
|
received = self._test_full_module()
|
|
expected = """
|
|
repmod/test_one.py[1]
|
|
repmod/test_three.py[0] - FAILED TO LOAD MODULE
|
|
repmod/test_two.py[0] - skipped (reason)"""
|
|
assert received.find(expected) != -1
|
|
|
|
class TestRemoteReporter(AbstractTestReporter):
|
|
reporter = RemoteReporter
|
|
|
|
def test_still_to_go(self):
|
|
self._test_still_to_go()
|
|
|
|
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
|
|
"""
|
|
assert val.find(expected) != -1
|
|
|
|
def test_module(self):
|
|
py.test.skip("XXX rewrite test to not rely on exact formatting")
|
|
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
|
|
"""
|
|
assert val.find(expected) != -1
|
|
|
|
def test_full_module(self):
|
|
#py.test.skip("XXX rewrite test to not rely on exact formatting")
|
|
val = self._test_full_module()
|
|
assert val.find('FAILED TO LOAD MODULE: repmod/test_three.py\n'\
|
|
'\nSkipped (reason) repmod/test_two.py') != -1
|