3 # Copyright (c) 2018 Orange and others.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # pylint: disable=missing-docstring
15 from paramiko import ssh_exception
20 from functest.opnfv_tests.openstack.vping import vping_ssh
21 from functest.utils import config
24 class VpingSSHTesting(unittest.TestCase):
27 with mock.patch('functest.core.singlevm.SingleVm2.__init__'):
28 self.vping = vping_ssh.VPingSSH()
29 self.vping.cloud = mock.Mock()
30 self.vping.case_name = 'vping'
33 @mock.patch('functest.core.singlevm.SingleVm2.prepare',
34 side_effect=Exception)
35 def test_prepare_exc1(self, *args):
36 with self.assertRaises(Exception):
38 args[0].assert_called_once_with()
40 @mock.patch('functest.opnfv_tests.openstack.vping.vping_ssh.VPingSSH.'
42 side_effect=Exception)
43 @mock.patch('functest.core.singlevm.SingleVm2.prepare')
44 def test_prepare_exc2(self, *args):
45 self.vping.sec = munch.Munch(id='foo')
46 with self.assertRaises(Exception):
48 args[0].assert_called_once_with()
49 args[1].assert_called_once_with(
50 '{}-vm2_{}'.format(self.vping.case_name, self.vping.guid),
51 security_groups=[self.vping.sec.id])
53 @mock.patch('functest.opnfv_tests.openstack.vping.vping_ssh.VPingSSH.'
55 @mock.patch('functest.core.singlevm.SingleVm2.prepare')
56 def test_prepare(self, *args):
57 self.vping.sec = munch.Munch(id='foo')
59 args[0].assert_called_once_with()
60 args[1].assert_called_once_with(
61 '{}-vm2_{}'.format(self.vping.case_name, self.vping.guid),
62 security_groups=[self.vping.sec.id])
64 def test_execute_exc(self):
65 self.vping.vm2 = munch.Munch(private_v4='127.0.0.1')
66 self.vping.ssh = mock.Mock()
67 self.vping.ssh.exec_command.side_effect = ssh_exception.SSHException
68 with self.assertRaises(ssh_exception.SSHException):
70 self.vping.ssh.exec_command.assert_called_once_with(
71 'ping -c 1 {}'.format(self.vping.vm2.private_v4))
73 def _test_execute(self, ret=0):
74 self.vping.vm2 = munch.Munch(private_v4='127.0.0.1')
75 self.vping.ssh = mock.Mock()
77 stdout.channel.recv_exit_status.return_value = ret
78 self.vping.ssh.exec_command.return_value = (None, stdout, None)
79 self.assertEqual(self.vping.execute(), ret)
80 self.vping.ssh.exec_command.assert_called_once_with(
81 'ping -c 1 {}'.format(self.vping.vm2.private_v4))
83 def test_execute1(self):
86 def test_execute2(self):
89 def test_clean_exc1(self):
90 self.vping.cloud = None
91 with self.assertRaises(AssertionError):
94 def test_clean_exc2(self):
95 self.vping.vm2 = munch.Munch(id='vm2')
96 mdelete_server = self.vping.cloud.delete_server
97 mdelete_server.side_effect = shade.OpenStackCloudException(None)
98 with self.assertRaises(shade.OpenStackCloudException):
101 @mock.patch('functest.core.singlevm.SingleVm2.clean',
102 side_effect=Exception)
103 def test_clean_exc3(self, *args):
104 self.vping.vm2 = munch.Munch(id='vm2')
105 with self.assertRaises(Exception):
107 self.vping.cloud.delete_server.assert_called_once_with(
108 self.vping.vm2, wait=True,
109 timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
110 args[0].assert_called_once_with()
112 @mock.patch('functest.core.singlevm.SingleVm2.clean')
113 def test_clean1(self, *args):
114 self.vping.vm2 = None
116 self.vping.cloud.delete_server.assert_not_called()
117 args[0].assert_called_once_with()
119 @mock.patch('functest.core.singlevm.SingleVm2.clean')
120 def test_clean2(self, *args):
121 self.vping.vm2 = munch.Munch(id='vm2')
123 self.vping.cloud.delete_server.assert_called_once_with(
124 self.vping.vm2, wait=True,
125 timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
126 args[0].assert_called_once_with()
129 if __name__ == '__main__':
130 logging.disable(logging.CRITICAL)
131 unittest.main(verbosity=2)