3 # Copyright (c) 2018 Enea AB 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
21 from functest.opnfv_tests.openstack.cinder import cinder_test
22 from functest.utils import config
23 from functest.utils import env
26 class CinderTesting(unittest.TestCase):
29 with mock.patch('functest.core.singlevm.SingleVm2.__init__'):
30 self.cinder = cinder_test.CinderCheck()
31 self.cinder.cloud = mock.Mock()
32 self.cinder.case_name = 'cinder'
33 self.cinder.guid = '1'
35 @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
36 'CinderCheck.connect')
37 @mock.patch('functest.core.singlevm.SingleVm2.prepare',
38 side_effect=Exception)
39 def test_prepare_exc1(self, *args):
40 self.cinder.cloud.boot_vm = mock.Mock()
41 with self.assertRaises(Exception):
43 args[0].assert_called_once_with()
44 args[1].assert_not_called()
45 self.cinder.cloud.boot_vm.assert_not_called()
46 self.cinder.cloud.create_volume.assert_not_called()
48 @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
49 'CinderCheck.connect')
50 @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
51 'CinderCheck.boot_vm',
52 side_effect=Exception)
53 @mock.patch('functest.core.singlevm.SingleVm2.prepare')
54 def test_prepare_exc2(self, *args):
55 self.cinder.sec = munch.Munch(id='foo')
56 self.cinder.keypair = munch.Munch(id='foo')
57 self.cinder.volume_timeout = munch.Munch(id='foo')
58 with self.assertRaises(Exception):
60 args[0].assert_called_with()
61 args[1].assert_called_once_with(
62 '{}-vm2_{}'.format(self.cinder.case_name, self.cinder.guid),
63 security_groups=[self.cinder.sec.id],
64 key_name=self.cinder.keypair.id)
65 self.cinder.cloud.create_volume.assert_not_called()
66 args[2].assert_not_called()
68 @mock.patch('functest.opnfv_tests.openstack.cinder.cinder_test.'
69 'CinderCheck.boot_vm', return_value=munch.Munch(id='vm2'))
70 @mock.patch('functest.core.singlevm.SingleVm2.prepare')
71 def test_prepare(self, *args):
72 self.cinder.sec = munch.Munch(id='foo')
73 self.cinder.keypair = munch.Munch(id='foo')
74 self.cinder.ext_net = mock.Mock(id='foo')
75 self.cinder.ssh2 = mock.Mock()
76 self.cinder.fip2 = munch.Munch(id='fip2')
77 self.cinder.connect = mock.Mock(
78 return_value=(self.cinder.fip2, self.cinder.ssh2))
79 self.cinder.cloud.create_volume = mock.Mock(
80 return_value=munch.Munch())
82 args[0].assert_called_once_with()
83 args[1].assert_called_once_with(
84 '{}-vm2_{}'.format(self.cinder.case_name, self.cinder.guid),
85 security_groups=[self.cinder.sec.id],
86 key_name=self.cinder.keypair.id)
87 self.cinder.connect.assert_called_once_with(args[1].return_value)
88 self.cinder.cloud.create_volume.assert_called_once_with(
89 name='{}-volume_{}'.format(
90 self.cinder.case_name, self.cinder.guid),
91 size='2', timeout=self.cinder.volume_timeout, wait=True)
93 @mock.patch('scp.SCPClient.put')
94 def test_write(self, *args):
95 # pylint: disable=protected-access
96 self.cinder.ssh = mock.Mock()
97 self.cinder.sshvm = mock.Mock(id='foo')
98 self.cinder.volume = mock.Mock(id='volume')
100 stdout.channel.recv_exit_status.return_value = 0
101 self.cinder.ssh.exec_command.return_value = (None, stdout, mock.Mock())
102 self.assertEqual(self.cinder._write_data(), 0)
103 self.cinder.ssh.exec_command.assert_called_once_with(
104 "sh ~/write_data.sh {}".format(env.get('VOLUME_DEVICE_NAME')))
105 self.cinder.cloud.attach_volume.assert_called_once_with(
106 self.cinder.sshvm, self.cinder.volume,
107 timeout=self.cinder.volume_timeout)
108 self.cinder.cloud.detach_volume.assert_called_once_with(
109 self.cinder.sshvm, self.cinder.volume,
110 timeout=self.cinder.volume_timeout)
111 args[0].assert_called_once_with(
112 pkg_resources.resource_filename(
113 'functest.opnfv_tests.openstack.cinder', 'write_data.sh'),
116 @mock.patch('scp.SCPClient.put', side_effect=Exception)
117 def test_write_exc1(self, *args):
118 # pylint: disable=protected-access
119 self.cinder.ssh = mock.Mock()
120 self.cinder.sshvm = mock.Mock(id='foo')
121 self.cinder.cloud.attach_volume = mock.Mock()
123 self.cinder._write_data(), self.cinder.EX_RUN_ERROR)
124 args[0].assert_called_once_with(
125 pkg_resources.resource_filename(
126 'functest.opnfv_tests.openstack.cinder', 'write_data.sh'),
129 @mock.patch('scp.SCPClient.put')
130 def test_read(self, *args):
131 # pylint: disable=protected-access
132 self.cinder.ssh2 = mock.Mock()
133 self.cinder.vm2 = mock.Mock(id='foo')
134 self.cinder.volume = mock.Mock(id='volume')
136 self.cinder.ssh2.exec_command.return_value = (
137 None, stdout, mock.Mock())
138 stdout.channel.recv_exit_status.return_value = 0
139 self.assertEqual(self.cinder._read_data(), 0)
140 self.cinder.ssh2.exec_command.assert_called_once_with(
141 "sh ~/read_data.sh {}".format(env.get('VOLUME_DEVICE_NAME')))
142 self.cinder.cloud.attach_volume.assert_called_once_with(
143 self.cinder.vm2, self.cinder.volume,
144 timeout=self.cinder.volume_timeout)
145 self.cinder.cloud.detach_volume.assert_called_once_with(
146 self.cinder.vm2, self.cinder.volume,
147 timeout=self.cinder.volume_timeout)
148 args[0].assert_called_once_with(
149 pkg_resources.resource_filename(
150 'functest.opnfv_tests.openstack.cinder', 'read_data.sh'),
153 @mock.patch('scp.SCPClient.put', side_effect=Exception)
154 def test_read_exc1(self, *args):
155 # pylint: disable=protected-access
156 self.cinder.ssh = mock.Mock()
157 self.cinder.ssh2 = mock.Mock()
158 self.cinder.sshvm = mock.Mock(id='foo')
159 self.cinder.cloud.attach_volume = mock.Mock()
161 self.cinder._read_data(), self.cinder.EX_RUN_ERROR)
162 args[0].assert_called_once_with(
163 pkg_resources.resource_filename(
164 'functest.opnfv_tests.openstack.cinder', 'read_data.sh'),
167 def test_execute_exc1(self):
168 # pylint: disable=protected-access
169 self.cinder._write_data = mock.Mock(side_effect=Exception)
170 self.cinder._read_data = mock.Mock()
171 with self.assertRaises(Exception):
172 self.cinder.execute()
173 self.cinder._write_data.assert_called_once_with()
174 self.cinder._read_data.assert_not_called()
176 def test_execute_exc2(self):
177 # pylint: disable=protected-access
178 self.cinder._write_data = mock.Mock(return_value=0)
179 self.cinder._read_data = mock.Mock(side_effect=Exception)
180 with self.assertRaises(Exception):
181 self.cinder.execute()
182 self.cinder._write_data.assert_called_once_with()
183 self.cinder._read_data.assert_called_once_with()
185 def test_execute_res1(self):
186 # pylint: disable=protected-access
187 self.cinder._write_data = mock.Mock(return_value=1)
188 self.cinder._read_data = mock.Mock()
189 self.assertEqual(self.cinder.execute(), 1)
190 self.cinder._write_data.assert_called_once_with()
191 self.cinder._read_data.assert_not_called()
193 def test_execute_res2(self):
194 # pylint: disable=protected-access
195 self.cinder._write_data = mock.Mock(return_value=0)
196 self.cinder._read_data = mock.Mock(return_value=1)
197 self.assertEqual(self.cinder.execute(), 1)
198 self.cinder._write_data.assert_called_once_with()
199 self.cinder._read_data.assert_called_once_with()
201 def test_execute_res3(self):
202 # pylint: disable=protected-access
203 self.cinder._write_data = mock.Mock(return_value=0)
204 self.cinder._read_data = mock.Mock(return_value=0)
205 self.assertEqual(self.cinder.execute(), 0)
206 self.cinder._write_data.assert_called_once_with()
207 self.cinder._read_data.assert_called_once_with()
209 def test_clean_exc1(self):
210 self.cinder.cloud = None
211 with self.assertRaises(AssertionError):
214 @mock.patch('functest.core.singlevm.SingleVm2.clean')
215 def test_clean_exc2(self, *args):
216 self.cinder.vm2 = munch.Munch(id='vm2')
217 self.cinder.cloud.delete_server = mock.Mock(
218 side_effect=shade.OpenStackCloudException("Foo"))
219 with self.assertRaises(shade.OpenStackCloudException):
221 self.cinder.cloud.delete_server.assert_called_once_with(
222 self.cinder.vm2, wait=True,
223 timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
224 self.cinder.cloud.delete_floating_ip.assert_not_called()
225 self.cinder.cloud.delete_volume.assert_not_called()
226 args[0].assert_not_called()
228 @mock.patch('functest.core.singlevm.SingleVm2.clean',
229 side_effect=Exception)
230 def test_clean_exc3(self, mock_clean):
231 self.cinder.vm2 = munch.Munch(id='vm2')
232 self.cinder.volume = munch.Munch(id='volume')
233 self.cinder.fip2 = munch.Munch(id='fip2')
234 with self.assertRaises(Exception):
236 self.cinder.cloud.delete_server.assert_called_once_with(
237 self.cinder.vm2, wait=True,
238 timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
239 self.cinder.cloud.delete_floating_ip.assert_called_once_with(
241 self.cinder.cloud.delete_volume.assert_called_once_with(
242 self.cinder.volume.id)
245 @mock.patch('functest.core.singlevm.SingleVm2.clean')
246 def test_clean(self, *args):
247 self.cinder.vm2 = munch.Munch(id='vm2')
248 self.cinder.volume = munch.Munch(id='volume')
249 self.cinder.fip2 = munch.Munch(id='fip2')
251 self.cinder.cloud.delete_server.assert_called_once_with(
252 self.cinder.vm2, wait=True,
253 timeout=getattr(config.CONF, 'vping_vm_delete_timeout'))
254 self.cinder.cloud.delete_floating_ip.assert_called_once_with(
256 self.cinder.cloud.delete_volume.assert_called_once_with(
257 self.cinder.volume.id)
258 args[0].assert_called_once_with()
260 @mock.patch('functest.core.singlevm.SingleVm2.clean')
261 def test_clean2(self, *args):
263 self.cinder.cloud.delete_server.assert_not_called()
264 self.cinder.cloud.delete_floating_ip.assert_not_called()
265 self.cinder.cloud.delete_volume.assert_not_called()
266 args[0].assert_called_once_with()
269 if __name__ == '__main__':
270 logging.disable(logging.CRITICAL)
271 unittest.main(verbosity=2)