def test_setup_teardown_run_ssh(self): |
hosts = [HostInfo('localhost:%s' % self.dest)] |
allevents = [] |
|
hm = HostManager(self.config, hosts=hosts) |
nodes = hm.setup_hosts(allevents.append) |
|
from py.__.test.rsession.testing.test_executor \ |
import ItemTestPassing, ItemTestFailing, ItemTestSkipping |
|
itempass = self.getexample("pass") |
itemfail = self.getexample("fail") |
itemskip = self.getexample("skip") |
itemprint = self.getexample("print") |
|
|
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, repevent.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) |
|
passed_stdout = [i for i in passed if i.outcome.stdout.find('samfing') != -1] |
assert len(passed_stdout) == len(nodes), passed |