-import fudge
import pytest
from mock import patch, Mock
-from teuthology.orchestra import cluster, remote
+from teuthology.orchestra import cluster, remote, run
class TestCluster(object):
- @fudge.with_fakes
def test_init_empty(self):
- fudge.clear_expectations()
c = cluster.Cluster()
assert c.remotes == {}
- @fudge.with_fakes
def test_init(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('Remote')
- r2 = fudge.Fake('Remote')
+ r1 = Mock()
+ r2 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
(r2, ['baz']),
],
)
- r3 = fudge.Fake('Remote')
+ r3 = Mock()
c.add(r3, ['xyzzy', 'thud', 'foo'])
assert c.remotes == {
r1: ['foo', 'bar'],
r3: ['xyzzy', 'thud', 'foo'],
}
- @fudge.with_fakes
def test_repr(self):
- fudge.clear_expectations()
- r1 = remote.Remote('r1', ssh=fudge.Fake('SSH'))
- r2 = remote.Remote('r2', ssh=fudge.Fake('SSH'))
+ r1 = remote.Remote('r1', ssh=Mock())
+ r2 = remote.Remote('r2', ssh=Mock())
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
(r2, ['baz']),
],
)
- assert repr(c) == "Cluster(remotes=[[Remote(name='r1'), ['foo', 'bar']], [Remote(name='r2'), ['baz']]])" # noqa
+ assert repr(c) == \
+ "Cluster(remotes=[[Remote(name='r1'), ['foo', 'bar']], " \
+ "[Remote(name='r2'), ['baz']]])"
- @fudge.with_fakes
def test_str(self):
- fudge.clear_expectations()
- r1 = remote.Remote('r1', ssh=fudge.Fake('SSH'))
- r2 = remote.Remote('r2', ssh=fudge.Fake('SSH'))
+ r1 = remote.Remote('r1', ssh=Mock())
+ r2 = remote.Remote('r2', ssh=Mock())
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
)
assert str(c) == "r1[foo,bar] r2[baz]"
- @fudge.with_fakes
def test_run_all(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('Remote').has_attr(name='r1')
- ret1 = fudge.Fake('RemoteProcess')
- r1.expects('run').with_args(args=['test']).returns(ret1)
- r2 = fudge.Fake('Remote').has_attr(name='r2')
- ret2 = fudge.Fake('RemoteProcess')
- r2.expects('run').with_args(args=['test']).returns(ret2)
+ r1 = Mock(spec=remote.Remote)
+ r1.configure_mock(name='r1')
+ ret1 = Mock(spec=run.RemoteProcess)
+ r1.run.return_value = ret1
+ r2 = Mock(spec=remote.Remote)
+ r2.configure_mock(name='r2')
+ ret2 = Mock(spec=run.RemoteProcess)
+ r2.run.return_value = ret2
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
(r2, ['baz']),
],
)
+ assert r1.run.called_once_with(args=['test'])
+ assert r2.run.called_once_with(args=['test'])
got = c.run(args=['test'])
assert len(got) == 2
assert got, [ret1 == ret2]
assert got[0] is ret1
assert got[1] is ret2
- @fudge.with_fakes
def test_only_one(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_foo = c.only('foo')
assert c_foo.remotes == {r1: ['foo', 'bar'], r3: ['foo']}
- @fudge.with_fakes
def test_only_two(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_both = c.only('foo', 'bar')
assert c_both.remotes, {r1: ['foo' == 'bar']}
- @fudge.with_fakes
def test_only_none(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_none = c.only('impossible')
assert c_none.remotes == {}
- @fudge.with_fakes
def test_only_match(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_foo = c.only('foo', lambda role: role.startswith('b'))
assert c_foo.remotes, {r1: ['foo' == 'bar']}
- @fudge.with_fakes
def test_exclude_one(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_foo = c.exclude('foo')
assert c_foo.remotes == {r2: ['bar']}
- @fudge.with_fakes
def test_exclude_two(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_both = c.exclude('foo', 'bar')
assert c_both.remotes == {r2: ['bar'], r3: ['foo']}
- @fudge.with_fakes
def test_exclude_none(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
c_none = c.exclude('impossible')
assert c_none.remotes == {r1: ['foo', 'bar'], r2: ['bar'], r3: ['foo']}
- @fudge.with_fakes
def test_exclude_match(self):
- fudge.clear_expectations()
- r1 = fudge.Fake('r1')
- r2 = fudge.Fake('r2')
- r3 = fudge.Fake('r3')
+ r1 = Mock()
+ r2 = Mock()
+ r3 = Mock()
c = cluster.Cluster(
remotes=[
(r1, ['foo', 'bar']),
-import fudge
+from mock import patch, Mock
from teuthology import config
from teuthology.orchestra import connection
class TestConnection(object):
def setup(self):
- import time
- time.sleep = lambda s: True
+ self.start_patchers()
+
+ def teardown(self):
+ self.stop_patchers()
+
+ def start_patchers(self):
+ self.patcher_sleep = patch(
+ 'time.sleep',
+ )
+ self.patcher_sleep.start()
+ self.m_ssh = Mock()
+ self.patcher_ssh = patch(
+ 'teuthology.orchestra.connection.paramiko.SSHClient',
+ self.m_ssh,
+ )
+ self.patcher_ssh.start()
+
+ def stop_patchers(self):
+ self.patcher_ssh.stop()
+ self.patcher_sleep.stop()
def clear_config(self):
config.config.teuthology_yaml = ''
e = assert_raises(AssertionError, connection.split_user, s)
assert str(e) == 'Bad input to split_user: {s!r}'.format(s=s)
- @fudge.with_fakes
def test_connect(self):
self.clear_config()
config.config.verify_host_keys = True
- fudge.clear_expectations()
- ssh = fudge.Fake('SSHClient')
- ssh.expects_call().with_args().returns(ssh)
- ssh.expects('set_missing_host_key_policy')
- ssh.expects('load_system_host_keys').with_args()
- ssh.expects('connect').with_args(
+ m_ssh_instance = self.m_ssh.return_value = Mock();
+ m_transport = Mock()
+ m_ssh_instance.get_transport.return_value = m_transport
+ got = connection.connect(
+ 'jdoe@orchestra.test.newdream.net.invalid',
+ _SSHClient=self.m_ssh,
+ )
+ self.m_ssh.assert_called_once()
+ m_ssh_instance.set_missing_host_key_policy.assert_called_once()
+ m_ssh_instance.load_system_host_keys.assert_called_once_with()
+ m_ssh_instance.connect.assert_called_once_with(
hostname='orchestra.test.newdream.net.invalid',
username='jdoe',
timeout=60,
)
- transport = ssh.expects('get_transport').with_args().returns_fake()
- transport.remember_order()
- transport.expects('set_keepalive').with_args(False)
- got = connection.connect(
- 'jdoe@orchestra.test.newdream.net.invalid',
- _SSHClient=ssh,
- )
- assert got is ssh
+ m_transport.set_keepalive.assert_called_once_with(False)
+ assert got is m_ssh_instance
- @fudge.with_fakes
def test_connect_no_verify_host_keys(self):
self.clear_config()
config.config.verify_host_keys = False
- fudge.clear_expectations()
- ssh = fudge.Fake('SSHClient')
- ssh.expects_call().with_args().returns(ssh)
- ssh.expects('set_missing_host_key_policy')
- ssh.expects('connect').with_args(
+ m_ssh_instance = self.m_ssh.return_value = Mock();
+ m_transport = Mock()
+ m_ssh_instance.get_transport.return_value = m_transport
+ got = connection.connect(
+ 'jdoe@orchestra.test.newdream.net.invalid',
+ _SSHClient=self.m_ssh,
+ )
+ self.m_ssh.assert_called_once()
+ m_ssh_instance.set_missing_host_key_policy.assert_called_once()
+ assert not m_ssh_instance.load_system_host_keys.called
+ m_ssh_instance.connect.assert_called_once_with(
hostname='orchestra.test.newdream.net.invalid',
username='jdoe',
timeout=60,
)
- transport = ssh.expects('get_transport').with_args().returns_fake()
- transport.remember_order()
- transport.expects('set_keepalive').with_args(False)
- got = connection.connect(
- 'jdoe@orchestra.test.newdream.net.invalid',
- _SSHClient=ssh,
- )
- assert got is ssh
+ m_transport.set_keepalive.assert_called_once_with(False)
+ assert got is m_ssh_instance
- @fudge.with_fakes
def test_connect_override_hostkeys(self):
self.clear_config()
- fudge.clear_expectations()
- sshclient = fudge.Fake('SSHClient')
- ssh = sshclient.expects_call().with_args().returns_fake()
- ssh.remember_order()
- host_keys = fudge.Fake('HostKeys')
- host_keys.expects('add').with_args(
+ m_ssh_instance = self.m_ssh.return_value = Mock();
+ m_transport = Mock()
+ m_ssh_instance.get_transport.return_value = m_transport
+ m_host_keys = Mock()
+ m_ssh_instance.get_host_keys.return_value = m_host_keys
+ m_create_key = Mock()
+ m_create_key.return_value = "frobnitz"
+ got = connection.connect(
+ 'jdoe@orchestra.test.newdream.net.invalid',
+ host_key='ssh-rsa testkey',
+ _SSHClient=self.m_ssh,
+ _create_key=m_create_key,
+ )
+ self.m_ssh.assert_called_once()
+ m_ssh_instance.get_host_keys.assert_called_once()
+ m_host_keys.add.assert_called_once_with(
hostname='orchestra.test.newdream.net.invalid',
keytype='ssh-rsa',
key='frobnitz',
- )
- ssh.expects('get_host_keys').with_args().returns(host_keys)
- ssh.expects('connect').with_args(
+ )
+ m_create_key.assert_called_once_with('ssh-rsa', 'testkey')
+ m_ssh_instance.connect.assert_called_once_with(
hostname='orchestra.test.newdream.net.invalid',
username='jdoe',
timeout=60,
- )
- transport = ssh.expects('get_transport').with_args().returns_fake()
- transport.remember_order()
- transport.expects('set_keepalive').with_args(False)
- create_key = fudge.Fake('create_key')
- create_key.expects_call().with_args('ssh-rsa',
- 'testkey').returns('frobnitz')
- got = connection.connect(
- 'jdoe@orchestra.test.newdream.net.invalid',
- host_key='ssh-rsa testkey',
- _SSHClient=sshclient,
- _create_key=create_key,
- )
- assert got is ssh
+ )
+ m_transport.set_keepalive.assert_called_once_with(False)
+ assert got is m_ssh_instance