def test_setup_teardown_ssh(self): |
hosts = [HostInfo('localhost:%s' % self.dest)] |
setup_events = [] |
teardown_events = [] |
tmpdir = self.source |
config = py.test.config._reparse([tmpdir]) |
hm = HostManager(config, hosts) |
nodes = hm.setup_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, repevent.HostRSyncing)] |
assert len(count_rsyn_calls) == len([i for i in hosts]) |
count_ready_calls = [i for i in setup_events |
if isinstance(i, repevent.HostRSyncRootReady)] |
assert len(count_ready_calls) == len([i for i in hosts]) |
|
|
teardown_wait_starts = [i for i in teardown_events |
if isinstance(i, repevent.CallStart)] |
teardown_wait_ends = [i for i in teardown_events |
if isinstance(i, repevent.CallFinish)] |
assert len(teardown_wait_starts) == len(hosts) |
assert len(teardown_wait_ends) == len(hosts) |