openstack/rally
Code Issues Proposed changes

Replace ' with " in tests/unit/[common,deploy,doc,fixtures]

Browse Source 
Partial bug: 1405884

Change-Id: I0100649c32bb911dab345c40492aa148ed1832c6
This commit is contained in:
jacobliberman 2015-02-03 10:48:15 -06:00
parent 11cc77b406
commit 3498071401
6 changed files with 197 additions and 197 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
tests/unit/common/test_fileutils.py
View File

@ -23,25 +23,25 @@ from tests.unit import test
class FileUtilsTestCase(test.TestCase): class FileUtilsTestCase(test.TestCase):
@mock.patch('os.path.exists', return_value=True) @mock.patch("os.path.exists", return_value=True)
@mock.patch.dict('os.environ', values={}, clear=True) @mock.patch.dict("os.environ", values={}, clear=True)
def test_load_env_vile(self, mock_path): def test_load_env_vile(self, mock_path):
file_data = ["FAKE_ENV=fake_env\n"] file_data = ["FAKE_ENV=fake_env\n"]
with mock.patch('rally.common.fileutils.open', mock.mock_open( with mock.patch("rally.common.fileutils.open", mock.mock_open(
read_data=file_data), create=True) as mock_file: read_data=file_data), create=True) as mock_file:
mock_file.return_value.readlines.return_value = file_data mock_file.return_value.readlines.return_value = file_data
fileutils.load_env_file('path_to_file') fileutils.load_env_file("path_to_file")
self.assertIn('FAKE_ENV', os.environ) self.assertIn("FAKE_ENV", os.environ)
mock_file.return_value.readlines.assert_called_once_with() mock_file.return_value.readlines.assert_called_once_with()
@mock.patch('os.path.exists', return_value=True) @mock.patch("os.path.exists", return_value=True)
def test_update_env_file(self, mock_path): def test_update_env_file(self, mock_path):
file_data = ["FAKE_ENV=old_value\n", "FAKE_ENV2=any\n"] file_data = ["FAKE_ENV=old_value\n", "FAKE_ENV2=any\n"]
with mock.patch('rally.common.fileutils.open', mock.mock_open( with mock.patch("rally.common.fileutils.open", mock.mock_open(
read_data=file_data), create=True) as mock_file: read_data=file_data), create=True) as mock_file:
mock_file.return_value.readlines.return_value = file_data mock_file.return_value.readlines.return_value = file_data
fileutils.update_env_file('path_to_file', 'FAKE_ENV', 'new_value') fileutils.update_env_file("path_to_file", "FAKE_ENV", "new_value")
calls = [mock.call('FAKE_ENV2=any\n'), mock.call( calls = [mock.call("FAKE_ENV2=any\n"), mock.call(
'FAKE_ENV=new_value')] "FAKE_ENV=new_value")]
mock_file.return_value.readlines.assert_called_once_with() mock_file.return_value.readlines.assert_called_once_with()
mock_file.return_value.write.assert_has_calls(calls) mock_file.return_value.write.assert_has_calls(calls)

168
tests/unit/common/test_sshutils.py
View File

@ -28,132 +28,132 @@ class SSHTestCase(test.TestCase):
def setUp(self): def setUp(self):
super(SSHTestCase, self).setUp() super(SSHTestCase, self).setUp()
self.ssh = sshutils.SSH('root', 'example.net') self.ssh = sshutils.SSH("root", "example.net")
@mock.patch('rally.common.sshutils.SSH._get_pkey') @mock.patch("rally.common.sshutils.SSH._get_pkey")
def test_construct(self, m_pkey): def test_construct(self, m_pkey):
m_pkey.return_value = 'pkey' m_pkey.return_value = "pkey"
ssh = sshutils.SSH('root', 'example.net', port=33, pkey='key', ssh = sshutils.SSH("root", "example.net", port=33, pkey="key",
key_filename='kf', password='secret') key_filename="kf", password="secret")
m_pkey.assert_called_once_with('key') m_pkey.assert_called_once_with("key")
self.assertEqual('root', ssh.user) self.assertEqual("root", ssh.user)
self.assertEqual('example.net', ssh.host) self.assertEqual("example.net", ssh.host)
self.assertEqual(33, ssh.port) self.assertEqual(33, ssh.port)
self.assertEqual('pkey', ssh.pkey) self.assertEqual("pkey", ssh.pkey)
self.assertEqual('kf', ssh.key_filename) self.assertEqual("kf", ssh.key_filename)
self.assertEqual('secret', ssh.password) self.assertEqual("secret", ssh.password)
def test_construct_default(self): def test_construct_default(self):
self.assertEqual('root', self.ssh.user) self.assertEqual("root", self.ssh.user)
self.assertEqual('example.net', self.ssh.host) self.assertEqual("example.net", self.ssh.host)
self.assertEqual(22, self.ssh.port) self.assertEqual(22, self.ssh.port)
self.assertIsNone(self.ssh.pkey) self.assertIsNone(self.ssh.pkey)
self.assertIsNone(self.ssh.key_filename) self.assertIsNone(self.ssh.key_filename)
self.assertIsNone(self.ssh.password) self.assertIsNone(self.ssh.password)
@mock.patch('rally.common.sshutils.paramiko') @mock.patch("rally.common.sshutils.paramiko")
def test__get_pkey_invalid(self, m_paramiko): def test__get_pkey_invalid(self, m_paramiko):
m_paramiko.SSHException = FakeParamikoException m_paramiko.SSHException = FakeParamikoException
rsa = m_paramiko.rsakey.RSAKey rsa = m_paramiko.rsakey.RSAKey
dss = m_paramiko.dsskey.DSSKey dss = m_paramiko.dsskey.DSSKey
rsa.from_private_key.side_effect = m_paramiko.SSHException rsa.from_private_key.side_effect = m_paramiko.SSHException
dss.from_private_key.side_effect = m_paramiko.SSHException dss.from_private_key.side_effect = m_paramiko.SSHException
self.assertRaises(sshutils.SSHError, self.ssh._get_pkey, 'key') self.assertRaises(sshutils.SSHError, self.ssh._get_pkey, "key")
@mock.patch('rally.common.sshutils.six.moves') @mock.patch("rally.common.sshutils.six.moves")
@mock.patch('rally.common.sshutils.paramiko') @mock.patch("rally.common.sshutils.paramiko")
def test__get_pkey_dss(self, m_paramiko, m_stringio): def test__get_pkey_dss(self, m_paramiko, m_stringio):
m_paramiko.SSHException = FakeParamikoException m_paramiko.SSHException = FakeParamikoException
m_stringio.StringIO.return_value = 'string_key' m_stringio.StringIO.return_value = "string_key"
m_paramiko.dsskey.DSSKey.from_private_key.return_value = 'dss_key' m_paramiko.dsskey.DSSKey.from_private_key.return_value = "dss_key"
rsa = m_paramiko.rsakey.RSAKey rsa = m_paramiko.rsakey.RSAKey
rsa.from_private_key.side_effect = m_paramiko.SSHException rsa.from_private_key.side_effect = m_paramiko.SSHException
key = self.ssh._get_pkey('key') key = self.ssh._get_pkey("key")
dss_calls = m_paramiko.dsskey.DSSKey.from_private_key.mock_calls dss_calls = m_paramiko.dsskey.DSSKey.from_private_key.mock_calls
self.assertEqual([mock.call('string_key')], dss_calls) self.assertEqual([mock.call("string_key")], dss_calls)
self.assertEqual(key, 'dss_key') self.assertEqual(key, "dss_key")
m_stringio.StringIO.assert_called_once_with('key') m_stringio.StringIO.assert_called_once_with("key")
@mock.patch('rally.common.sshutils.six.moves') @mock.patch("rally.common.sshutils.six.moves")
@mock.patch('rally.common.sshutils.paramiko') @mock.patch("rally.common.sshutils.paramiko")
def test__get_pkey_rsa(self, m_paramiko, m_stringio): def test__get_pkey_rsa(self, m_paramiko, m_stringio):
m_paramiko.SSHException = FakeParamikoException m_paramiko.SSHException = FakeParamikoException
m_stringio.StringIO.return_value = 'string_key' m_stringio.StringIO.return_value = "string_key"
m_paramiko.rsakey.RSAKey.from_private_key.return_value = 'rsa_key' m_paramiko.rsakey.RSAKey.from_private_key.return_value = "rsa_key"
dss = m_paramiko.dsskey.DSSKey dss = m_paramiko.dsskey.DSSKey
dss.from_private_key.side_effect = m_paramiko.SSHException dss.from_private_key.side_effect = m_paramiko.SSHException
key = self.ssh._get_pkey('key') key = self.ssh._get_pkey("key")
rsa_calls = m_paramiko.rsakey.RSAKey.from_private_key.mock_calls rsa_calls = m_paramiko.rsakey.RSAKey.from_private_key.mock_calls
self.assertEqual([mock.call('string_key')], rsa_calls) self.assertEqual([mock.call("string_key")], rsa_calls)
self.assertEqual(key, 'rsa_key') self.assertEqual(key, "rsa_key")
m_stringio.StringIO.assert_called_once_with('key') m_stringio.StringIO.assert_called_once_with("key")
@mock.patch('rally.common.sshutils.SSH._get_pkey') @mock.patch("rally.common.sshutils.SSH._get_pkey")
@mock.patch('rally.common.sshutils.paramiko') @mock.patch("rally.common.sshutils.paramiko")
def test__get_client(self, m_paramiko, m_pkey): def test__get_client(self, m_paramiko, m_pkey):
m_pkey.return_value = 'key' m_pkey.return_value = "key"
fake_client = mock.Mock() fake_client = mock.Mock()
m_paramiko.SSHClient.return_value = fake_client m_paramiko.SSHClient.return_value = fake_client
m_paramiko.AutoAddPolicy.return_value = 'autoadd' m_paramiko.AutoAddPolicy.return_value = "autoadd"
ssh = sshutils.SSH('admin', 'example.net', pkey='key') ssh = sshutils.SSH("admin", "example.net", pkey="key")
client = ssh._get_client() client = ssh._get_client()
self.assertEqual(fake_client, client) self.assertEqual(fake_client, client)
client_calls = [ client_calls = [
mock.call.set_missing_host_key_policy('autoadd'), mock.call.set_missing_host_key_policy("autoadd"),
mock.call.connect('example.net', username='admin', mock.call.connect("example.net", username="admin",
port=22, pkey='key', key_filename=None, port=22, pkey="key", key_filename=None,
password=None, timeout=1), password=None, timeout=1),
] ]
self.assertEqual(client_calls, client.mock_calls) self.assertEqual(client_calls, client.mock_calls)
def test_close(self): def test_close(self):
with mock.patch.object(self.ssh, '_client') as m_client: with mock.patch.object(self.ssh, "_client") as m_client:
self.ssh.close() self.ssh.close()
m_client.close.assert_called_once_with() m_client.close.assert_called_once_with()
self.assertFalse(self.ssh._client) self.assertFalse(self.ssh._client)
@mock.patch('rally.common.sshutils.six.moves') @mock.patch("rally.common.sshutils.six.moves")
def test_execute(self, m_stringio): def test_execute(self, m_stringio):
m_stringio.StringIO.side_effect = stdio = [mock.Mock(), mock.Mock()] m_stringio.StringIO.side_effect = stdio = [mock.Mock(), mock.Mock()]
stdio[0].read.return_value = 'stdout fake data' stdio[0].read.return_value = "stdout fake data"
stdio[1].read.return_value = 'stderr fake data' stdio[1].read.return_value = "stderr fake data"
with mock.patch.object(self.ssh, 'run', return_value=0) as m_run: with mock.patch.object(self.ssh, "run", return_value=0) as m_run:
status, stdout, stderr = self.ssh.execute('cmd', status, stdout, stderr = self.ssh.execute("cmd",
stdin='fake_stdin', stdin="fake_stdin",
timeout=43) timeout=43)
m_run.assert_called_once_with('cmd', stdin='fake_stdin', m_run.assert_called_once_with("cmd", stdin="fake_stdin",
stdout=stdio[0], stdout=stdio[0],
stderr=stdio[1], timeout=43, stderr=stdio[1], timeout=43,
raise_on_error=False) raise_on_error=False)
self.assertEqual(0, status) self.assertEqual(0, status)
self.assertEqual('stdout fake data', stdout) self.assertEqual("stdout fake data", stdout)
self.assertEqual('stderr fake data', stderr) self.assertEqual("stderr fake data", stderr)
@mock.patch('rally.common.sshutils.time') @mock.patch("rally.common.sshutils.time")
def test_wait_timeout(self, m_time): def test_wait_timeout(self, m_time):
m_time.time.side_effect = [1, 50, 150] m_time.time.side_effect = [1, 50, 150]
self.ssh.execute = mock.Mock(side_effect=[sshutils.SSHError, self.ssh.execute = mock.Mock(side_effect=[sshutils.SSHError,
sshutils.SSHError, sshutils.SSHError,
0]) 0])
self.assertRaises(sshutils.SSHTimeout, self.ssh.wait) self.assertRaises(sshutils.SSHTimeout, self.ssh.wait)
self.assertEqual([mock.call('uname')] * 2, self.ssh.execute.mock_calls) self.assertEqual([mock.call("uname")] * 2, self.ssh.execute.mock_calls)
@mock.patch('rally.common.sshutils.time') @mock.patch("rally.common.sshutils.time")
def test_wait(self, m_time): def test_wait(self, m_time):
m_time.time.side_effect = [1, 50, 100] m_time.time.side_effect = [1, 50, 100]
self.ssh.execute = mock.Mock(side_effect=[sshutils.SSHError, self.ssh.execute = mock.Mock(side_effect=[sshutils.SSHError,
sshutils.SSHError, sshutils.SSHError,
0]) 0])
self.ssh.wait() self.ssh.wait()
self.assertEqual([mock.call('uname')] * 3, self.ssh.execute.mock_calls) self.assertEqual([mock.call("uname")] * 3, self.ssh.execute.mock_calls)
class SSHRunTestCase(test.TestCase): class SSHRunTestCase(test.TestCase):
"""Test SSH.run method in different aspects. """Test SSH.run method in different aspects.
Also tested method 'execute'. Also tested method "execute".
""" """
def setUp(self): def setUp(self):
@ -172,85 +172,85 @@ class SSHRunTestCase(test.TestCase):
self.fake_session.exit_status_ready.return_value = True self.fake_session.exit_status_ready.return_value = True
self.fake_session.recv_exit_status.return_value = 0 self.fake_session.recv_exit_status.return_value = 0
self.ssh = sshutils.SSH('admin', 'example.net') self.ssh = sshutils.SSH("admin", "example.net")
self.ssh._get_client = mock.Mock(return_value=self.fake_client) self.ssh._get_client = mock.Mock(return_value=self.fake_client)
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_execute(self, m_select): def test_execute(self, m_select):
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.fake_session.recv_ready.side_effect = [1, 0, 0] self.fake_session.recv_ready.side_effect = [1, 0, 0]
self.fake_session.recv_stderr_ready.side_effect = [1, 0] self.fake_session.recv_stderr_ready.side_effect = [1, 0]
self.fake_session.recv.return_value = 'ok' self.fake_session.recv.return_value = "ok"
self.fake_session.recv_stderr.return_value = 'error' self.fake_session.recv_stderr.return_value = "error"
self.fake_session.exit_status_ready.return_value = 1 self.fake_session.exit_status_ready.return_value = 1
self.fake_session.recv_exit_status.return_value = 127 self.fake_session.recv_exit_status.return_value = 127
self.assertEqual((127, 'ok', 'error'), self.ssh.execute('cmd')) self.assertEqual((127, "ok", "error"), self.ssh.execute("cmd"))
self.fake_session.exec_command.assert_called_once_with('cmd') self.fake_session.exec_command.assert_called_once_with("cmd")
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run(self, m_select): def test_run(self, m_select):
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.assertEqual(0, self.ssh.run('cmd')) self.assertEqual(0, self.ssh.run("cmd"))
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run_nonzero_status(self, m_select): def test_run_nonzero_status(self, m_select):
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.fake_session.recv_exit_status.return_value = 1 self.fake_session.recv_exit_status.return_value = 1
self.assertRaises(sshutils.SSHError, self.ssh.run, 'cmd') self.assertRaises(sshutils.SSHError, self.ssh.run, "cmd")
self.assertEqual(1, self.ssh.run('cmd', raise_on_error=False)) self.assertEqual(1, self.ssh.run("cmd", raise_on_error=False))
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run_stdout(self, m_select): def test_run_stdout(self, m_select):
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.fake_session.recv_ready.side_effect = [True, True, False] self.fake_session.recv_ready.side_effect = [True, True, False]
self.fake_session.recv.side_effect = ['ok1', 'ok2'] self.fake_session.recv.side_effect = ["ok1", "ok2"]
stdout = mock.Mock() stdout = mock.Mock()
self.ssh.run('cmd', stdout=stdout) self.ssh.run("cmd", stdout=stdout)
self.assertEqual([mock.call('ok1'), mock.call('ok2')], self.assertEqual([mock.call("ok1"), mock.call("ok2")],
stdout.write.mock_calls) stdout.write.mock_calls)
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run_stderr(self, m_select): def test_run_stderr(self, m_select):
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.fake_session.recv_stderr_ready.side_effect = [True, False] self.fake_session.recv_stderr_ready.side_effect = [True, False]
self.fake_session.recv_stderr.return_value = 'error' self.fake_session.recv_stderr.return_value = "error"
stderr = mock.Mock() stderr = mock.Mock()
self.ssh.run('cmd', stderr=stderr) self.ssh.run("cmd", stderr=stderr)
stderr.write.assert_called_once_with('error') stderr.write.assert_called_once_with("error")
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run_stdin(self, m_select): def test_run_stdin(self, m_select):
"""Test run method with stdin. """Test run method with stdin.
Third send call was called with 'e2' because only 3 bytes was sent Third send call was called with "e2" because only 3 bytes was sent
by second call. So remainig 2 bytes of 'line2' was sent by third call. by second call. So remainig 2 bytes of "line2" was sent by third call.
""" """
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.fake_session.exit_status_ready.side_effect = [0, 0, 0, True] self.fake_session.exit_status_ready.side_effect = [0, 0, 0, True]
self.fake_session.send_ready.return_value = True self.fake_session.send_ready.return_value = True
self.fake_session.send.side_effect = [5, 3, 2] self.fake_session.send.side_effect = [5, 3, 2]
fake_stdin = mock.Mock() fake_stdin = mock.Mock()
fake_stdin.read.side_effect = ['line1', 'line2', ''] fake_stdin.read.side_effect = ["line1", "line2", ""]
fake_stdin.closed = False fake_stdin.closed = False
def close(): def close():
fake_stdin.closed = True fake_stdin.closed = True
fake_stdin.close = mock.Mock(side_effect=close) fake_stdin.close = mock.Mock(side_effect=close)
self.ssh.run('cmd', stdin=fake_stdin) self.ssh.run("cmd", stdin=fake_stdin)
call = mock.call call = mock.call
send_calls = [call('line1'), call('line2'), call('e2')] send_calls = [call("line1"), call("line2"), call("e2")]
self.assertEqual(send_calls, self.fake_session.send.mock_calls) self.assertEqual(send_calls, self.fake_session.send.mock_calls)
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run_select_error(self, m_select): def test_run_select_error(self, m_select):
self.fake_session.exit_status_ready.return_value = False self.fake_session.exit_status_ready.return_value = False
m_select.select.return_value = ([], [], [True]) m_select.select.return_value = ([], [], [True])
self.assertRaises(sshutils.SSHError, self.ssh.run, 'cmd') self.assertRaises(sshutils.SSHError, self.ssh.run, "cmd")
@mock.patch('rally.common.sshutils.time') @mock.patch("rally.common.sshutils.time")
@mock.patch('rally.common.sshutils.select') @mock.patch("rally.common.sshutils.select")
def test_run_timemout(self, m_select, m_time): def test_run_timemout(self, m_select, m_time):
m_time.time.side_effect = [1, 3700] m_time.time.side_effect = [1, 3700]
m_select.select.return_value = ([], [], []) m_select.select.return_value = ([], [], [])
self.fake_session.exit_status_ready.return_value = False self.fake_session.exit_status_ready.return_value = False
self.assertRaises(sshutils.SSHTimeout, self.ssh.run, 'cmd') self.assertRaises(sshutils.SSHTimeout, self.ssh.run, "cmd")

44
tests/unit/common/test_utils.py
View File

@ -34,7 +34,7 @@ class ImmutableMixinTestCase(test.TestCase):
def test_without_base_values(self): def test_without_base_values(self):
im = utils.ImmutableMixin() im = utils.ImmutableMixin()
self.assertRaises(exceptions.ImmutableException, self.assertRaises(exceptions.ImmutableException,
im.__setattr__, 'test', 'test') im.__setattr__, "test", "test")
def test_with_base_values(self): def test_with_base_values(self):
@ -43,10 +43,10 @@ class ImmutableMixinTestCase(test.TestCase):
self.test = test self.test = test
super(A, self).__init__() super(A, self).__init__()
a = A('test') a = A("test")
self.assertRaises(exceptions.ImmutableException, self.assertRaises(exceptions.ImmutableException,
a.__setattr__, 'abc', 'test') a.__setattr__, "abc", "test")
self.assertEqual(a.test, 'test') self.assertEqual(a.test, "test")
class EnumMixinTestCase(test.TestCase): class EnumMixinTestCase(test.TestCase):
@ -65,22 +65,22 @@ class StdIOCaptureTestCase(test.TestCase):
def test_stdout_capture(self): def test_stdout_capture(self):
stdout = sys.stdout stdout = sys.stdout
messages = ['abcdef', 'defgaga'] messages = ["abcdef", "defgaga"]
with utils.StdOutCapture() as out: with utils.StdOutCapture() as out:
for msg in messages: for msg in messages:
print(msg) print(msg)
self.assertEqual(out.getvalue().rstrip('\n').split('\n'), messages) self.assertEqual(out.getvalue().rstrip("\n").split("\n"), messages)
self.assertEqual(stdout, sys.stdout) self.assertEqual(stdout, sys.stdout)
def test_stderr_capture(self): def test_stderr_capture(self):
stderr = sys.stderr stderr = sys.stderr
messages = ['abcdef', 'defgaga'] messages = ["abcdef", "defgaga"]
with utils.StdErrCapture() as err: with utils.StdErrCapture() as err:
for msg in messages: for msg in messages:
print(msg, file=sys.stderr) print(msg, file=sys.stderr)
self.assertEqual(err.getvalue().rstrip('\n').split('\n'), messages) self.assertEqual(err.getvalue().rstrip("\n").split("\n"), messages)
self.assertEqual(stderr, sys.stderr) self.assertEqual(stderr, sys.stderr)
@ -90,7 +90,7 @@ class TimerTestCase(test.TestCase):
start_time = time.time() start_time = time.time()
end_time = time.time() end_time = time.time()
with mock.patch('rally.common.utils.time') as mock_time: with mock.patch("rally.common.utils.time") as mock_time:
mock_time.time = mock.MagicMock(return_value=start_time) mock_time.time = mock.MagicMock(return_value=start_time)
with utils.Timer() as timer: with utils.Timer() as timer:
mock_time.time = mock.MagicMock(return_value=end_time) mock_time.time = mock.MagicMock(return_value=end_time)
@ -129,20 +129,20 @@ class IterSubclassesTestCase(test.TestCase):
class ImportModulesTestCase(test.TestCase): class ImportModulesTestCase(test.TestCase):
def test_try_append_module_into_sys_modules(self): def test_try_append_module_into_sys_modules(self):
modules = {} modules = {}
utils.try_append_module('rally.common.version', modules) utils.try_append_module("rally.common.version", modules)
self.assertIn('rally.common.version', modules) self.assertIn("rally.common.version", modules)
def test_try_append_broken_module(self): def test_try_append_broken_module(self):
modules = {} modules = {}
self.assertRaises(ImportError, self.assertRaises(ImportError,
utils.try_append_module, utils.try_append_module,
'tests.unit.fixtures.import.broken', "tests.unit.fixtures.import.broken",
modules) modules)
def test_import_modules_from_package(self): def test_import_modules_from_package(self):
utils.import_modules_from_package('tests.unit.fixtures.import.package') utils.import_modules_from_package("tests.unit.fixtures.import.package")
self.assertIn('tests.unit.fixtures.import.package.a', sys.modules) self.assertIn("tests.unit.fixtures.import.package.a", sys.modules)
self.assertIn('tests.unit.fixtures.import.package.b', sys.modules) self.assertIn("tests.unit.fixtures.import.package.b", sys.modules)
class LogTestCase(test.TestCase): class LogTestCase(test.TestCase):
@ -154,7 +154,7 @@ class LogTestCase(test.TestCase):
class TaskLog(object): class TaskLog(object):
def __init__(self): def __init__(self):
self.task = {'uuid': 'some_uuid'} self.task = {"uuid": "some_uuid"}
@utils.log_task_wrapper(mock_log, msg, a=10, b=20) @utils.log_task_wrapper(mock_log, msg, a=10, b=20)
def some_method(self, x, y): def some_method(self, x, y):
@ -163,7 +163,7 @@ class LogTestCase(test.TestCase):
t = TaskLog() t = TaskLog()
self.assertEqual(t.some_method.__name__, "some_method") self.assertEqual(t.some_method.__name__, "some_method")
self.assertEqual(t.some_method(2, 2), 4) self.assertEqual(t.some_method(2, 2), 4)
params = {'msg': msg % {'a': 10, 'b': 20}, 'uuid': t.task['uuid']} params = {"msg": msg % {"a": 10, "b": 20}, "uuid": t.task["uuid"]}
expected = [ expected = [
mock.call(_("Task %(uuid)s | Starting: %(msg)s") % params), mock.call(_("Task %(uuid)s | Starting: %(msg)s") % params),
mock.call(_("Task %(uuid)s | Completed: %(msg)s") % params) mock.call(_("Task %(uuid)s | Completed: %(msg)s") % params)
@ -177,10 +177,10 @@ class LoadExtraModulesTestCase(test.TestCase):
@mock.patch("rally.common.utils.imp.find_module", @mock.patch("rally.common.utils.imp.find_module",
return_value=(mock.MagicMock(), None, None)) return_value=(mock.MagicMock(), None, None))
@mock.patch("rally.common.utils.os.walk", return_value=[ @mock.patch("rally.common.utils.os.walk", return_value=[
('/somewhere', ('/subdir', ), ('plugin1.py', )), ("/somewhere", ("/subdir", ), ("plugin1.py", )),
('/somewhere/subdir', ('/subsubdir', ), ('plugin2.py', ("/somewhere/subdir", ("/subsubdir", ), ("plugin2.py",
'withoutextension')), "withoutextension")),
('/somewhere/subdir/subsubdir', [], ('plugin3.py', ))]) ("/somewhere/subdir/subsubdir", [], ("plugin3.py", ))])
@mock.patch("rally.common.utils.os.path.exists", return_value=True) @mock.patch("rally.common.utils.os.path.exists", return_value=True)
def test_load_plugins_successfull(self, mock_exists, def test_load_plugins_successfull(self, mock_exists,
mock_oswalk, mock_find_module, mock_oswalk, mock_find_module,
@ -209,7 +209,7 @@ class LoadExtraModulesTestCase(test.TestCase):
@mock.patch("rally.common.utils.imp.find_module") @mock.patch("rally.common.utils.imp.find_module")
@mock.patch("rally.common.utils.os.path", return_value=True) @mock.patch("rally.common.utils.os.path", return_value=True)
@mock.patch("rally.common.utils.os.walk", @mock.patch("rally.common.utils.os.walk",
return_value=[('/etc/.rally/plugins', [], ('load_it.py', ))]) return_value=[("/etc/.rally/plugins", [], ("load_it.py", ))])
def test_load_plugins_fails(self, mock_oswalk, mock_ospath, def test_load_plugins_fails(self, mock_oswalk, mock_ospath,
mock_load_module, mock_find_module): mock_load_module, mock_find_module):
# test no fails if module is broken # test no fails if module is broken

82
tests/unit/deploy/engines/test_devstack.py
View File

@ -21,17 +21,17 @@ from tests.unit import test
SAMPLE_CONFIG = { SAMPLE_CONFIG = {
'type': 'DevstackEngine', "type": "DevstackEngine",
'provider': { "provider": {
'name': 'ExistingServers', "name": "ExistingServers",
'credentials': [{'user': 'root', 'host': 'example.com'}], "credentials": [{"user": "root", "host": "example.com"}],
}, },
'localrc': { "localrc": {
'ADMIN_PASSWORD': 'secret', "ADMIN_PASSWORD": "secret",
}, },
} }
DEVSTACK_REPO = 'https://git.openstack.org/cgit/openstack-dev/devstack.git' DEVSTACK_REPO = "https://git.openstack.org/cgit/openstack-dev/devstack.git"
class DevstackEngineTestCase(test.TestCase): class DevstackEngineTestCase(test.TestCase):
@ -39,67 +39,67 @@ class DevstackEngineTestCase(test.TestCase):
def setUp(self): def setUp(self):
super(DevstackEngineTestCase, self).setUp() super(DevstackEngineTestCase, self).setUp()
self.deployment = { self.deployment = {
'uuid': 'de641026-dbe3-4abe-844a-ffef930a600a', "uuid": "de641026-dbe3-4abe-844a-ffef930a600a",
'config': SAMPLE_CONFIG, "config": SAMPLE_CONFIG,
} }
self.engine = devstack.DevstackEngine(self.deployment) self.engine = devstack.DevstackEngine(self.deployment)
def test_invalid_config(self): def test_invalid_config(self):
self.deployment = SAMPLE_CONFIG.copy() self.deployment = SAMPLE_CONFIG.copy()
self.deployment['config'] = {'type': 42} self.deployment["config"] = {"type": 42}
engine = devstack.DevstackEngine(self.deployment) engine = devstack.DevstackEngine(self.deployment)
self.assertRaises(jsonschema.ValidationError, self.assertRaises(jsonschema.ValidationError,
engine.validate) engine.validate)
def test_construct(self): def test_construct(self):
self.assertEqual(self.engine.localrc['ADMIN_PASSWORD'], 'secret') self.assertEqual(self.engine.localrc["ADMIN_PASSWORD"], "secret")
@mock.patch('rally.deploy.engines.devstack.open', create=True) @mock.patch("rally.deploy.engines.devstack.open", create=True)
def test_prepare_server(self, m_open): def test_prepare_server(self, m_open):
m_open.return_value = 'fake_file' m_open.return_value = "fake_file"
server = mock.Mock() server = mock.Mock()
server.password = 'secret' server.password = "secret"
self.engine.prepare_server(server) self.engine.prepare_server(server)
calls = [ calls = [
mock.call('/bin/sh -e', stdin='fake_file'), mock.call("/bin/sh -e", stdin="fake_file"),
mock.call('chpasswd', stdin='rally:secret'), mock.call("chpasswd", stdin="rally:secret"),
] ]
self.assertEqual(calls, server.ssh.run.mock_calls) self.assertEqual(calls, server.ssh.run.mock_calls)
filename = m_open.mock_calls[0][1][0] filename = m_open.mock_calls[0][1][0]
self.assertTrue(filename.endswith('rally/deploy/engines/' self.assertTrue(filename.endswith("rally/deploy/engines/"
'devstack/install.sh')) "devstack/install.sh"))
self.assertEqual([mock.call(filename, 'rb')], m_open.mock_calls) self.assertEqual([mock.call(filename, "rb")], m_open.mock_calls)
@mock.patch('rally.deploy.engine.EngineFactory.get_provider') @mock.patch("rally.deploy.engine.EngineFactory.get_provider")
@mock.patch('rally.deploy.engines.devstack.get_updated_server') @mock.patch("rally.deploy.engines.devstack.get_updated_server")
@mock.patch('rally.deploy.engines.devstack.get_script') @mock.patch("rally.deploy.engines.devstack.get_script")
@mock.patch('rally.deploy.serverprovider.provider.Server') @mock.patch("rally.deploy.serverprovider.provider.Server")
@mock.patch('rally.deploy.engines.devstack.objects.Endpoint') @mock.patch("rally.deploy.engines.devstack.objects.Endpoint")
def test_deploy(self, m_endpoint, m_server, m_gs, m_gus, m_gp): def test_deploy(self, m_endpoint, m_server, m_gs, m_gus, m_gp):
m_gp.return_value = fake_provider = mock.Mock() m_gp.return_value = fake_provider = mock.Mock()
server = mock.Mock() server = mock.Mock()
server.host = 'host' server.host = "host"
m_endpoint.return_value = 'fake_endpoint' m_endpoint.return_value = "fake_endpoint"
m_gus.return_value = ds_server = mock.Mock() m_gus.return_value = ds_server = mock.Mock()
m_gs.return_value = 'fake_script' m_gs.return_value = "fake_script"
server.get_credentials.return_value = 'fake_credentials' server.get_credentials.return_value = "fake_credentials"
fake_provider.create_servers.return_value = [server] fake_provider.create_servers.return_value = [server]
with mock.patch.object(self.engine, 'deployment') as m_d: with mock.patch.object(self.engine, "deployment") as m_d:
endpoints = self.engine.deploy() endpoints = self.engine.deploy()
self.assertEqual({"admin": "fake_endpoint"}, endpoints) self.assertEqual({"admin": "fake_endpoint"}, endpoints)
m_endpoint.assert_called_once_with('http://host:5000/v2.0/', 'admin', m_endpoint.assert_called_once_with("http://host:5000/v2.0/", "admin",
'secret', 'admin', 'admin') "secret", "admin", "admin")
m_d.add_resource.assert_called_once_with( m_d.add_resource.assert_called_once_with(
info='fake_credentials', info="fake_credentials",
provider_name='DevstackEngine', provider_name="DevstackEngine",
type='credentials') type="credentials")
repo = 'https://git.openstack.org/cgit/openstack-dev/devstack.git' repo = "https://git.openstack.org/cgit/openstack-dev/devstack.git"
cmd = '/bin/sh -e -s %s master' % repo cmd = "/bin/sh -e -s %s master" % repo
server.ssh.run.assert_called_once_with(cmd, stdin='fake_script') server.ssh.run.assert_called_once_with(cmd, stdin="fake_script")
ds_calls = [ ds_calls = [
mock.call.ssh.run('cat > ~/devstack/localrc', stdin=mock.ANY), mock.call.ssh.run("cat > ~/devstack/localrc", stdin=mock.ANY),
mock.call.ssh.run('~/devstack/stack.sh') mock.call.ssh.run("~/devstack/stack.sh")
] ]
self.assertEqual(ds_calls, ds_server.mock_calls) self.assertEqual(ds_calls, ds_server.mock_calls)
localrc = ds_server.mock_calls[0][2]['stdin'] localrc = ds_server.mock_calls[0][2]["stdin"]
self.assertIn('ADMIN_PASSWORD=secret', localrc) self.assertIn("ADMIN_PASSWORD=secret", localrc)

78
tests/unit/deploy/engines/test_fuel.py
View File

@ -23,17 +23,17 @@ import mock
SAMPLE_CONFIG = { SAMPLE_CONFIG = {
'type': 'FuelEngine', "type": "FuelEngine",
'deploy_name': 'TestDeploy01', "deploy_name": "TestDeploy01",
'net_provider': 'nova_network', "net_provider": "nova_network",
'release': 'Havana on Ubuntu 12.04', "release": "Havana on Ubuntu 12.04",
'api_url': 'http://example.net:8000/api/v1/', "api_url": "http://example.net:8000/api/v1/",
'mode': 'multinode', "mode": "multinode",
'networks': {'public': {'cidr': '10.1.1.0/24'}}, "networks": {"public": {"cidr": "10.1.1.0/24"}},
'nodes': { "nodes": {
'controller': {'amount': 1, 'filters': ['cpus==2']}, "controller": {"amount": 1, "filters": ["cpus==2"]},
'cinder+compute': {'amount': 4, 'filters': ['cpus==8', "cinder+compute": {"amount": 4, "filters": ["cpus==8",
'storage>=2T']}, "storage>=2T"]},
}, },
} }
@ -42,15 +42,15 @@ class FuelEngineTestCase(test.TestCase):
def setUp(self): def setUp(self):
super(FuelEngineTestCase, self).setUp() super(FuelEngineTestCase, self).setUp()
self.deployment = fakes.FakeDeployment({'config': SAMPLE_CONFIG}) self.deployment = fakes.FakeDeployment({"config": SAMPLE_CONFIG})
def test_construct(self): def test_construct(self):
fuel.FuelEngine(self.deployment) fuel.FuelEngine(self.deployment)
def test_validate_no_computes(self): def test_validate_no_computes(self):
config = SAMPLE_CONFIG.copy() config = SAMPLE_CONFIG.copy()
config['nodes'].pop('cinder+compute') config["nodes"].pop("cinder+compute")
deployment = {'config': config} deployment = {"config": config}
engine = fuel.FuelEngine(deployment) engine = fuel.FuelEngine(deployment)
self.assertRaises(exceptions.ValidationError, self.assertRaises(exceptions.ValidationError,
engine.validate) engine.validate)
@ -59,9 +59,9 @@ class FuelEngineTestCase(test.TestCase):
engine = fuel.FuelEngine(self.deployment) engine = fuel.FuelEngine(self.deployment)
engine.nodes = mock.MagicMock() engine.nodes = mock.MagicMock()
engine.nodes.pop.side_effect = [1, 2, 3, 4] engine.nodes.pop.side_effect = [1, 2, 3, 4]
nodes = engine._get_nodes('cinder+compute') nodes = engine._get_nodes("cinder+compute")
self.assertEqual([1, 2, 3, 4], nodes) self.assertEqual([1, 2, 3, 4], nodes)
expected_calls = [mock.call(['cpus==8', 'storage>=2T'])] * 4 expected_calls = [mock.call(["cpus==8", "storage>=2T"])] * 4
self.assertEqual(expected_calls, engine.nodes.pop.mock_calls) self.assertEqual(expected_calls, engine.nodes.pop.mock_calls)
def test__get_nodes_no_nodes(self): def test__get_nodes_no_nodes(self):
@ -69,30 +69,30 @@ class FuelEngineTestCase(test.TestCase):
engine.nodes = mock.MagicMock() engine.nodes = mock.MagicMock()
engine.nodes.pop.return_value = None engine.nodes.pop.return_value = None
self.assertRaises(exceptions.NoNodesFound, self.assertRaises(exceptions.NoNodesFound,
engine._get_nodes, 'controller') engine._get_nodes, "controller")
def test__get_nodes_empty(self): def test__get_nodes_empty(self):
engine = fuel.FuelEngine(self.deployment) engine = fuel.FuelEngine(self.deployment)
self.assertEqual([], engine._get_nodes('nonexistent')) self.assertEqual([], engine._get_nodes("nonexistent"))
def test__get_release_id(self): def test__get_release_id(self):
engine = fuel.FuelEngine(self.deployment) engine = fuel.FuelEngine(self.deployment)
engine.client = mock.Mock() engine.client = mock.Mock()
fake_releases = [{'name': 'fake', 'id': 1}, fake_releases = [{"name": "fake", "id": 1},
{'name': 'Havana on Ubuntu 12.04', 'id': 42}] {"name": "Havana on Ubuntu 12.04", "id": 42}]
engine.client.get_releases = mock.Mock(return_value=fake_releases) engine.client.get_releases = mock.Mock(return_value=fake_releases)
self.assertEqual(42, engine._get_release_id()) self.assertEqual(42, engine._get_release_id())
@mock.patch('rally.deploy.fuel.fuelclient.FuelClient') @mock.patch("rally.deploy.fuel.fuelclient.FuelClient")
@mock.patch('rally.deploy.fuel.fuelclient.FuelCluster') @mock.patch("rally.deploy.fuel.fuelclient.FuelCluster")
def test_deploy(self, m_cluster, m_client): def test_deploy(self, m_cluster, m_client):
attributes = {'editable': {'access': {'user': {'value': 'user'}, attributes = {"editable": {"access": {"user": {"value": "user"},
'password': {'value': 'pw'}, "password": {"value": "pw"},
'tenant': {'value': 'tn'}}}} "tenant": {"value": "tn"}}}}
client = mock.Mock() client = mock.Mock()
cluster = mock.Mock() cluster = mock.Mock()
cluster.cluster = {'id': 42} cluster.cluster = {"id": 42}
cluster.get_endpoint_ip = mock.Mock(return_value='2.3.4.5') cluster.get_endpoint_ip = mock.Mock(return_value="2.3.4.5")
cluster.get_attributes = mock.Mock(return_value=attributes) cluster.get_attributes = mock.Mock(return_value=attributes)
m_client.return_value = client m_client.return_value = client
m_cluster.return_value = cluster m_cluster.return_value = cluster
@ -107,18 +107,18 @@ class FuelEngineTestCase(test.TestCase):
self.assertEqual(["admin"], list(endpoint)) self.assertEqual(["admin"], list(endpoint))
endpoint = endpoint["admin"] endpoint = endpoint["admin"]
self.assertEqual('user', endpoint.username) self.assertEqual("user", endpoint.username)
self.assertEqual('pw', endpoint.password) self.assertEqual("pw", endpoint.password)
self.assertEqual('tn', endpoint.tenant_name) self.assertEqual("tn", endpoint.tenant_name)
self.assertEqual('http://2.3.4.5:5000/v2.0/', endpoint.auth_url) self.assertEqual("http://2.3.4.5:5000/v2.0/", endpoint.auth_url)
self.assertEqual(consts.EndpointPermission.ADMIN, endpoint.permission) self.assertEqual(consts.EndpointPermission.ADMIN, endpoint.permission)
expected_cluster_calls = [ expected_cluster_calls = [
mock.call.set_nodes(1, ['controller']), mock.call.set_nodes(1, ["controller"]),
mock.call.set_nodes(2, ['compute']), mock.call.set_nodes(2, ["compute"]),
mock.call.set_nodes(3, ['cinder']), mock.call.set_nodes(3, ["cinder"]),
mock.call.set_nodes(4, ['compute', 'cinder']), mock.call.set_nodes(4, ["compute", "cinder"]),
mock.call.configure_network({'public': {'cidr': '10.1.1.0/24'}}), mock.call.configure_network({"public": {"cidr": "10.1.1.0/24"}}),
mock.call.deploy(), mock.call.deploy(),
mock.call.get_endpoint_ip(), mock.call.get_endpoint_ip(),
mock.call.get_attributes() mock.call.get_attributes()
@ -126,10 +126,10 @@ class FuelEngineTestCase(test.TestCase):
self.assertEqual(expected_cluster_calls, cluster.mock_calls) self.assertEqual(expected_cluster_calls, cluster.mock_calls)
self.assertEqual([mock.call.get_nodes()], client.mock_calls) self.assertEqual([mock.call.get_nodes()], client.mock_calls)
@mock.patch('rally.deploy.fuel.fuelclient.FuelClient') @mock.patch("rally.deploy.fuel.fuelclient.FuelClient")
@mock.patch('rally.deploy.engines.fuel.objects.Deployment') @mock.patch("rally.deploy.engines.fuel.objects.Deployment")
def test_cleanup(self, m_deployment, m_client): def test_cleanup(self, m_deployment, m_client):
fake_resources = [{'id': 41, 'info': {'id': 42}}] fake_resources = [{"id": 41, "info": {"id": 42}}]
self.deployment.get_resources = mock.Mock(return_value=fake_resources) self.deployment.get_resources = mock.Mock(return_value=fake_resources)
engine = fuel.FuelEngine(self.deployment) engine = fuel.FuelEngine(self.deployment)

2
tests/unit/doc/test_task_samples.py
View File

@ -43,7 +43,7 @@ class TaskSampleTestCase(test.TestCase):
# NOTE(hughsaunders): Skip non config files # NOTE(hughsaunders): Skip non config files
# (bug https://bugs.launchpad.net/rally/+bug/1314369) # (bug https://bugs.launchpad.net/rally/+bug/1314369)
if not re.search('\.(ya?ml|json)$', filename, flags=re.I): if not re.search("\.(ya?ml|json)$", filename, flags=re.I):
continue continue
with open(full_path) as task_file: with open(full_path) as task_file: