root / svc_extstorage / tests.py @ 2eba2338
History | View | Annotate | Download (35.1 kB)
1 |
#!/usr/bin/env python
|
---|---|
2 |
#
|
3 |
# Copyright (c) 2010-2013 Greek Research and Technology Network S.A.
|
4 |
#
|
5 |
# This program is free software; you can redistribute it and/or modify
|
6 |
# it under the terms of the GNU General Public License as published by
|
7 |
# the Free Software Foundation; either version 2 of the License, or
|
8 |
# (at your option) any later version.
|
9 |
#
|
10 |
# This program is distributed in the hope that it will be useful, but
|
11 |
# WITHOUT ANY WARRANTY; without even the implied warranty of
|
12 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
13 |
# General Public License for more details.
|
14 |
#
|
15 |
# You should have received a copy of the GNU General Public License
|
16 |
# along with this program; if not, write to the Free Software
|
17 |
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
|
18 |
# 02110-1301, USA.
|
19 |
|
20 |
|
21 |
try:
|
22 |
import unittest2 as unittest |
23 |
except ImportError: |
24 |
import unittest |
25 |
|
26 |
import os |
27 |
from mock import patch |
28 |
from storwize_svc import * |
29 |
import subprocess |
30 |
|
31 |
|
32 |
@patch('paramiko.SSHClient', autospec=True) |
33 |
class SVCConnectionTest(unittest.TestCase): |
34 |
def test_ssh_injection_1(self, ssh_conn): |
35 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
36 |
cmd_list = ['ls', '- l'] |
37 |
self.assertRaises(SVCError, self.conn.run_ssh_cmd, cmd_list) |
38 |
|
39 |
def test_ssh_injection_2(self, ssh_conn): |
40 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
41 |
cmd_list = ['ls', ';cat>/dev/sda</dev/zero'] |
42 |
self.assertRaises(SVCError, self.conn.run_ssh_cmd, cmd_list) |
43 |
|
44 |
@patch('paramiko.ChannelFile') |
45 |
def test_initial_connection_ok(self, chan, ssh_conn): |
46 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
47 |
cmd_list = ['ls', '-l'] |
48 |
chan.channel.recv_exit_status.return_value = 0
|
49 |
ssh_conn.return_value.exec_command.return_value = (chan, chan, chan) |
50 |
self.conn.run_ssh_cmd(cmd_list)
|
51 |
kwargs = {'port': 22, |
52 |
'username': 'user', |
53 |
'password': 'userpass', |
54 |
'timeout': 30, |
55 |
'look_for_keys': False, |
56 |
'allow_agent': False} |
57 |
|
58 |
ssh_conn.return_value.connect.assert_called_once_with('127.0.0.1',
|
59 |
**kwargs) |
60 |
|
61 |
@patch('paramiko.ChannelFile') |
62 |
def test_initial_connection_pooled(self, chan, ssh_conn): |
63 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
64 |
cmd_list = ['ls', '-l'] |
65 |
chan.channel.recv_exit_status.return_value = 0
|
66 |
ssh_conn.return_value.exec_command.return_value = (chan, chan, chan) |
67 |
self.conn.run_ssh_cmd(cmd_list)
|
68 |
self.conn.run_ssh_cmd(cmd_list)
|
69 |
kwargs = {'port': 22, |
70 |
'username': 'user', |
71 |
'password': 'userpass', |
72 |
'timeout': 30, |
73 |
'look_for_keys': False, |
74 |
'allow_agent': False} |
75 |
|
76 |
ssh_conn.return_value.connect.assert_called_once_with('127.0.0.1',
|
77 |
**kwargs) |
78 |
|
79 |
@patch('paramiko.ChannelFile') |
80 |
def test_initial_connection_fail(self, chan, ssh_conn): |
81 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
82 |
cmd_list = ['ls', '-l'] |
83 |
chan.channel.recv_exit_status.return_value = -1
|
84 |
ssh_conn.return_value.exec_command.return_value = (chan, chan, chan) |
85 |
self.assertRaises(SVCError, self.conn.run_ssh_cmd, cmd_list) |
86 |
|
87 |
@patch('paramiko.ChannelFile') |
88 |
def test_initial_connection_out(self, chan, ssh_conn): |
89 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
90 |
cmd_list = ['ls', '-l'] |
91 |
chan.channel.recv_exit_status.return_value = 0
|
92 |
chan.read.return_value = "test"
|
93 |
ssh_conn.return_value.exec_command.return_value = (chan, chan, chan) |
94 |
out, err = self.conn.run_ssh_cmd(cmd_list)
|
95 |
self.assertEqual(out, "test") |
96 |
self.assertEqual(err, "test") |
97 |
|
98 |
@patch('paramiko.ChannelFile') |
99 |
def test_initial_connection_pooled_retry(self, chan, ssh_conn): |
100 |
self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass') |
101 |
cmd_list = ['ls', '-l'] |
102 |
chan.channel.recv_exit_status.return_value = 0
|
103 |
ssh_conn.return_value.exec_command.return_value = (chan, chan, chan) |
104 |
self.conn.run_ssh_cmd(cmd_list)
|
105 |
transport = ssh_conn.return_value.get_transport.return_value |
106 |
transport.is_active.return_value = False
|
107 |
self.conn.run_ssh_cmd(cmd_list)
|
108 |
self.assertEqual(ssh_conn.return_value.connect.call_count, 2) |
109 |
|
110 |
|
111 |
@patch('storwize_svc.SVCConnection', autospec=True) |
112 |
class SVCInitTest(unittest.TestCase): |
113 |
def test_missing_config(self, svc_conn): |
114 |
self.assertRaises(TypeError, SVCProvider) |
115 |
|
116 |
def test_missing_env(self, svc_conn): |
117 |
self.assertRaises(SVCError, SVCProvider, './test_configs/some_config') |
118 |
|
119 |
@patch.dict(os.environ, {'VOL_SIZE': '1', 'EXTP_SVC_SAN': 'mysan'}) |
120 |
def test_missing_vol_name(self, svc_conn): |
121 |
self.assertRaises(SVCError, SVCProvider, './test_configs/some_config') |
122 |
|
123 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '1'}) |
124 |
def test_missing_san(self, svc_conn): |
125 |
self.assertRaises(SVCError, SVCProvider, './test_configs/some_config') |
126 |
|
127 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '1', |
128 |
'EXTP_SVC_SAN': 'mysan'}) |
129 |
def test_bad_config_path(self, svc_conn): |
130 |
self.assertRaises(SVCError, SVCProvider, './test_configs/some_config') |
131 |
|
132 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'EXTP_SVC_SAN': 'mysan'}) |
133 |
def test_bad_identifier(self, svc_conn): |
134 |
self.assertRaises(SVCError, SVCProvider, './test_configs/good_config') |
135 |
|
136 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', |
137 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
138 |
def test_missing_size(self, svc_conn): |
139 |
|
140 |
svc = SVCProvider('./test_configs/good_config')
|
141 |
|
142 |
self.assertEqual(svc._vol_name, 'myvol') |
143 |
self.assertIsNone(svc._vol_size)
|
144 |
self.assertEqual(svc._svc_pool, 'VM_POOL_SYNNEFO') |
145 |
self.assertEqual(svc._host_id, 2) |
146 |
|
147 |
svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser', |
148 |
'uberpass')
|
149 |
|
150 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
151 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
152 |
def test_good_config(self, svc_conn): |
153 |
svc = SVCProvider('./test_configs/good_config')
|
154 |
|
155 |
self.assertEqual(svc._vol_name, 'myvol') |
156 |
self.assertEqual(svc._vol_size, 100) |
157 |
self.assertEqual(svc._svc_pool, 'VM_POOL_SYNNEFO') |
158 |
self.assertEqual(svc._host_id, 2) |
159 |
|
160 |
svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser', |
161 |
'uberpass')
|
162 |
|
163 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
164 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
165 |
def test_good_config_extra_sections(self, svc_conn): |
166 |
svc = SVCProvider('./test_configs/good_config_extra_sections')
|
167 |
|
168 |
self.assertEqual(svc._vol_name, 'myvol') |
169 |
self.assertEqual(svc._vol_size, 100) |
170 |
self.assertEqual(svc._svc_pool, 'synnefo') |
171 |
self.assertEqual(svc._host_id, 2) |
172 |
|
173 |
svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser', |
174 |
'uberpass')
|
175 |
|
176 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
177 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
178 |
def test_good_config_extra_empty_sections(self, svc_conn): |
179 |
svc = SVCProvider('./test_configs/good_config_extra_empty_sections')
|
180 |
|
181 |
self.assertEqual(svc._vol_name, 'myvol') |
182 |
self.assertEqual(svc._vol_size, 100) |
183 |
self.assertEqual(svc._svc_pool, 'synnefo') |
184 |
self.assertEqual(svc._host_id, 2) |
185 |
|
186 |
svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser', |
187 |
'uberpass')
|
188 |
|
189 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
190 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
191 |
def test_bad_config_section(self, svc_conn): |
192 |
self.assertRaises(SVCError, SVCProvider,
|
193 |
'./test_configs/bad_config_section')
|
194 |
|
195 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
196 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
197 |
def test_bad_config_missing_section(self, svc_conn): |
198 |
self.assertRaises(SVCError, SVCProvider,
|
199 |
'./test_configs/bad_config_missing_section')
|
200 |
|
201 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
202 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
203 |
def test_bad_config_missing_fields(self, svc_conn): |
204 |
self.assertRaises(SVCError, SVCProvider,
|
205 |
'./test_configs/bad_config_missing_fields')
|
206 |
|
207 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
208 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
209 |
def test_bad_config_invalid_format(self, svc_conn): |
210 |
self.assertRaises(SVCError, SVCProvider,
|
211 |
'./test_configs/bad_config_invalid_format')
|
212 |
|
213 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
214 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
215 |
def test_bad_config_empty_host(self, svc_conn): |
216 |
self.assertRaises(SVCError, SVCProvider,
|
217 |
'./test_configs/bad_config_empty_host')
|
218 |
|
219 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
220 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
221 |
def test_bad_config_bad_port(self, svc_conn): |
222 |
self.assertRaises(SVCError, SVCProvider,
|
223 |
'./test_configs/bad_config_bad_port')
|
224 |
|
225 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
226 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
227 |
def test_bad_config_bad_hostid(self, svc_conn): |
228 |
self.assertRaises(SVCError, SVCProvider,
|
229 |
'./test_configs/bad_config_bad_hostid')
|
230 |
|
231 |
@patch.dict(os.environ, {'VOL_NAME': '', 'VOL_SIZE': '1', |
232 |
'EXTP_SVC_SAN': 'mysan'}) |
233 |
def test_empty_volume_name(self, svc_conn): |
234 |
self.assertRaises(SVCError, SVCProvider, './some_config') |
235 |
|
236 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '', |
237 |
'EXTP_SVC_SAN': 'mysan'}) |
238 |
def test_empty_volume_size(self, svc_conn): |
239 |
self.assertRaises(SVCError, SVCProvider, './some_config') |
240 |
|
241 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': 'test', |
242 |
'EXTP_SVC_SAN': 'mysan'}) |
243 |
def test_bad_volume_size(self, svc_conn): |
244 |
self.assertRaises(SVCError, SVCProvider, './some_config') |
245 |
|
246 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '-1', |
247 |
'EXTP_SVC_SAN': 'mysan'}) |
248 |
def test_negative_volume_size(self, svc_conn): |
249 |
self.assertRaises(SVCError, SVCProvider, './some_config') |
250 |
|
251 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '-1', |
252 |
'EXTP_SVC_SAN': ''}) |
253 |
def test_empty_svc_identifier(self, svc_conn): |
254 |
self.assertRaises(SVCError, SVCProvider, './some_config') |
255 |
|
256 |
|
257 |
class SVCEnvOpsTest(unittest.TestCase): |
258 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', |
259 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
260 |
def test_create_no_size(self): |
261 |
svc = SVCProvider('./test_configs/good_config')
|
262 |
self.assertRaises(SVCError, svc.create)
|
263 |
|
264 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', |
265 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
266 |
def test_grow_no_size(self): |
267 |
svc = SVCProvider('./test_configs/good_config')
|
268 |
self.assertRaises(SVCError, svc.grow)
|
269 |
|
270 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
271 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
272 |
def test_detach_size(self): |
273 |
svc = SVCProvider('./test_configs/good_config')
|
274 |
self.assertRaises(SVCError, svc.detach)
|
275 |
|
276 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
277 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
278 |
def test_attach_size(self): |
279 |
svc = SVCProvider('./test_configs/good_config')
|
280 |
self.assertRaises(SVCError, svc.attach)
|
281 |
|
282 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
283 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
284 |
def test_verify_size(self): |
285 |
svc = SVCProvider('./test_configs/good_config')
|
286 |
self.assertRaises(SVCError, svc.verify)
|
287 |
|
288 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
289 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
290 |
def test_remove_size(self): |
291 |
svc = SVCProvider('./test_configs/good_config')
|
292 |
self.assertRaises(SVCError, svc.remove)
|
293 |
|
294 |
|
295 |
@patch('subprocess.check_output', autospec=True) |
296 |
@patch('subprocess.check_call', autospec=True) |
297 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
298 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
299 |
class SVCInternalHostOpsTest(unittest.TestCase): |
300 |
def test_check_call_ok(self, call, out): |
301 |
svc = SVCProvider('./test_configs/good_config')
|
302 |
svc._check_call(['ls', '-l']) |
303 |
|
304 |
def test_check_call_exc(self, call, out): |
305 |
svc = SVCProvider('./test_configs/good_config')
|
306 |
call.side_effect = subprocess.CalledProcessError(1, "test", |
307 |
output="test")
|
308 |
self.assertRaises(SVCError, svc._check_call, ['ls', '-l']) |
309 |
|
310 |
def test_check_output_ok(self, call, out): |
311 |
svc = SVCProvider('./test_configs/good_config')
|
312 |
out.return_value = "test"
|
313 |
ret = svc._check_output(['ls', '-l']) |
314 |
self.assertEqual(ret, "test") |
315 |
|
316 |
def test_check_output_empty(self, call, out): |
317 |
svc = SVCProvider('./test_configs/good_config')
|
318 |
out.return_value = ""
|
319 |
ret = svc._check_output(['ls', '-l']) |
320 |
self.assertEqual(ret, "") |
321 |
|
322 |
def test_check_output_exc(self, call, out): |
323 |
svc = SVCProvider('./test_configs/good_config')
|
324 |
out.return_value = "test"
|
325 |
out.side_effect = subprocess.CalledProcessError(1, "test", |
326 |
output="test")
|
327 |
self.assertRaises(SVCError, svc._check_output, ['ls', '-l']) |
328 |
|
329 |
@patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc", |
330 |
"/dev/sdd"], autospec=True) |
331 |
def test_devices_add_ok(self, glob, call, out): |
332 |
svc = SVCProvider('./test_configs/good_config')
|
333 |
svc._devices_add(12)
|
334 |
|
335 |
@patch('glob.glob', return_value=[], autospec=True) |
336 |
def test_devices_add_not_discovered(self, glob, call, out): |
337 |
svc = SVCProvider('./test_configs/good_config')
|
338 |
self.assertRaises(SVCError, svc._devices_add, 12) |
339 |
|
340 |
@patch('glob.glob', return_value=["/dev/sda", "/dev/sdc"], autospec=True) |
341 |
def test_devices_add_partial_discovered(self, glob, call, out): |
342 |
svc = SVCProvider('./test_configs/good_config')
|
343 |
self.assertRaises(SVCError, svc._devices_add, 12) |
344 |
|
345 |
@patch('glob.glob', return_value=["/dev/sda", "/dev/sdc"], autospec=True) |
346 |
def test_devices_add_scan_exc(self, glob, call, out): |
347 |
svc = SVCProvider('./test_configs/good_config')
|
348 |
call.side_effect = SVCError("failed")
|
349 |
self.assertRaises(SVCError, svc._devices_add, 12) |
350 |
|
351 |
@patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc", |
352 |
"/dev/sdd"], autospec=True) |
353 |
@patch.object(SVCProvider, '_devices_discovered', return_value=False, |
354 |
autospec=True)
|
355 |
def test_devices_remove_ok(self, glob, disc, call, out): |
356 |
svc = SVCProvider('./test_configs/good_config')
|
357 |
with patch('__builtin__.open', autospec=True): |
358 |
svc._devices_remove(12)
|
359 |
|
360 |
@patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc", |
361 |
"/dev/sdd"], autospec=True) |
362 |
@patch.object(SVCProvider, '_devices_discovered', return_value=True, |
363 |
autospec=True)
|
364 |
def test_devices_remove_fail(self, glob, disc, call, out): |
365 |
svc = SVCProvider('./test_configs/good_config')
|
366 |
with patch('__builtin__.open', autospec=True): |
367 |
self.assertRaises(SVCError, svc._devices_remove, 12) |
368 |
|
369 |
@patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc", |
370 |
"/dev/sdd"], autospec=True) |
371 |
@patch.object(SVCProvider, '_devices_discovered', return_value=True, |
372 |
autospec=True)
|
373 |
def test_devices_remove_exc(self, glob, disc, call, out): |
374 |
svc = SVCProvider('./test_configs/good_config')
|
375 |
with patch('__builtin__.open', autospec=True) as op: |
376 |
op.side_effect = IOError("foo") |
377 |
self.assertRaises(SVCError, svc._devices_remove, 12) |
378 |
|
379 |
def test_multipath_ok_ok(self, call, out): |
380 |
svc = SVCProvider('./test_configs/good_config')
|
381 |
out.return_value = "multipath -ll command output not empty"
|
382 |
ret = svc._multipath_ok("/dev/mapper/3123")
|
383 |
self.assertEqual(ret, True) |
384 |
|
385 |
def test_multipath_ok_fail(self, call, out): |
386 |
svc = SVCProvider('./test_configs/good_config')
|
387 |
out.return_value = ""
|
388 |
ret = svc._multipath_ok("/dev/mapper/3123")
|
389 |
self.assertEqual(ret, False) |
390 |
|
391 |
def test_multipath_ok_exc(self, call, out): |
392 |
svc = SVCProvider('./test_configs/good_config')
|
393 |
call.side_effect = SVCError("command failed")
|
394 |
self.assertRaises(SVCError, svc._multipath_ok, "/dev/mapper/3123") |
395 |
|
396 |
def test_multipath_setup_ok(self, call, out): |
397 |
svc = SVCProvider('./test_configs/good_config')
|
398 |
out.return_value = "multipath -ll command output not empty"
|
399 |
svc._multipath_setup("/dev/mapper/3123")
|
400 |
|
401 |
def test_multipath_setup_fail(self, call, out): |
402 |
svc = SVCProvider('./test_configs/good_config')
|
403 |
out.return_value = ""
|
404 |
self.assertRaises(SVCError, svc._multipath_setup, "/dev/mapper/3123") |
405 |
|
406 |
def test_multipath_setup_exc(self, call, out): |
407 |
svc = SVCProvider('./test_configs/good_config')
|
408 |
call.side_effect = SVCError("command failed")
|
409 |
self.assertRaises(SVCError, svc._multipath_setup, "/dev/mapper/3123") |
410 |
|
411 |
def test_multipath_remove_ok(self, call, out): |
412 |
svc = SVCProvider('./test_configs/good_config')
|
413 |
out.return_value = ""
|
414 |
svc._multipath_remove("/dev/mapper/3123")
|
415 |
|
416 |
def test_multipath_remove_fail(self, call, out): |
417 |
svc = SVCProvider('./test_configs/good_config')
|
418 |
out.return_value = "multipath -ll cmd out"
|
419 |
self.assertRaises(SVCError, svc._multipath_remove, "/dev/mapper/3123") |
420 |
|
421 |
def test_multipath_remove_exc(self, call, out): |
422 |
svc = SVCProvider('./test_configs/good_config')
|
423 |
call.side_effect = SVCError("command failed")
|
424 |
self.assertRaises(SVCError, svc._multipath_remove, "/dev/mapper/3123") |
425 |
#TODO: host_device_setup/remove , fix testing for devices_* methods, rename
|
426 |
# devices to device
|
427 |
|
428 |
|
429 |
@patch.object(SVCConnection, 'run_ssh_cmd', autospec=True) |
430 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
431 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
432 |
class SVCInternalRemoteOpsTest(unittest.TestCase): |
433 |
def test_get_device_info_ok(self, run_ssh_cmd): |
434 |
svc = SVCProvider('./test_configs/good_config')
|
435 |
ret_str = "vdisk_UID!1\nSCSI_id!2\ntest3!3\ntest4!4\n"
|
436 |
run_ssh_cmd.return_value = (ret_str, "")
|
437 |
ret = svc._get_device_info('myvol')
|
438 |
ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol'] |
439 |
run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd) |
440 |
self.assertEqual(run_ssh_cmd.call_count, 2) |
441 |
exp_ret = ("/dev/mapper/31", 2) |
442 |
self.assertTupleEqual(ret, exp_ret)
|
443 |
|
444 |
def test_get_device_info_malformed(self, run_ssh_cmd): |
445 |
svc = SVCProvider('./test_configs/good_config')
|
446 |
ret_str = "vdisk_UID!ab\nSCSI_id!2\ntest3!3\ntest4!4\n"
|
447 |
run_ssh_cmd.return_value = (ret_str, "")
|
448 |
self.assertRaises(SVCError, svc._get_device_info, 'myvol') |
449 |
ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol'] |
450 |
run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd) |
451 |
self.assertEqual(run_ssh_cmd.call_count, 2) |
452 |
|
453 |
def test_get_device_info_missing(self, run_ssh_cmd): |
454 |
svc = SVCProvider('./test_configs/good_config')
|
455 |
ret_str = "vdisk_UID!ab\nsome_field!2\ntest3!3\ntest4!4\n"
|
456 |
run_ssh_cmd.return_value = (ret_str, "")
|
457 |
self.assertRaises(SVCError, svc._get_device_info, 'myvol') |
458 |
ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol'] |
459 |
run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd) |
460 |
self.assertEqual(run_ssh_cmd.call_count, 2) |
461 |
|
462 |
def test_get_device_info_empty(self, run_ssh_cmd): |
463 |
svc = SVCProvider('./test_configs/good_config')
|
464 |
run_ssh_cmd.return_value = ("", "") |
465 |
self.assertRaises(SVCError, svc._get_device_info, 'myvol') |
466 |
ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol'] |
467 |
run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd) |
468 |
self.assertEqual(run_ssh_cmd.call_count, 1) |
469 |
|
470 |
def test_vdisk_defined(self, run_ssh_cmd): |
471 |
svc = SVCProvider('./test_configs/good_config')
|
472 |
ret_str = "test1!1\ntest2!2\ntest3!3\ntest4!4\n"
|
473 |
run_ssh_cmd.return_value = (ret_str, "")
|
474 |
ret = svc._is_vdisk_defined('myvol')
|
475 |
ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol'] |
476 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
477 |
self.assertEqual(ret, True) |
478 |
|
479 |
#FIXME:
|
480 |
def test_vdisk_not_defined(self, run_ssh_cmd): |
481 |
svc = SVCProvider('./test_configs/good_config')
|
482 |
run_ssh_cmd.return_value = ("", "") |
483 |
ret = svc._is_vdisk_defined('myvol')
|
484 |
ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol'] |
485 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
486 |
self.assertEqual(ret, False) |
487 |
|
488 |
def test_driver_assert_ok(self, run_ssh_cmd): |
489 |
SVCProvider._driver_assert(1, 'ok') |
490 |
|
491 |
def test_driver_assert_fail(self, run_ssh_cmd): |
492 |
self.assertRaises(SVCError, SVCProvider._driver_assert, 0, 'fail') |
493 |
|
494 |
def test_assert_ssh_return_ok(self, run_ssh_cmd): |
495 |
svc = SVCProvider('./test_configs/good_config')
|
496 |
svc._assert_ssh_return(1, 'fun', ['ls', '-l'], 'test', 'test') |
497 |
|
498 |
def test_assert_ssh_return_fail(self, run_ssh_cmd): |
499 |
svc = SVCProvider('./test_configs/good_config')
|
500 |
self.assertRaises(SVCError, svc._assert_ssh_return, 0, |
501 |
'fun', ['ls', '-l'], 'test', 'test') |
502 |
|
503 |
def test_exec_cmd_parse_ok(self, run_ssh_cmd): |
504 |
svc = SVCProvider('./test_configs/good_config')
|
505 |
ret_str = "test1!1\ntest2!2\ntest3!3\ntest4!4\n"
|
506 |
run_ssh_cmd.return_value = (ret_str, "")
|
507 |
ret = svc._exec_cmd_and_parse_attrs('ssh_cmd placeholder')
|
508 |
ret_dict = {'test1': '1', 'test2': '2', 'test3': '3', 'test4': '4'} |
509 |
self.assertDictEqual(ret, ret_dict)
|
510 |
|
511 |
def test_exec_cmd_parse_empty(self, run_ssh_cmd): |
512 |
svc = SVCProvider('./test_configs/good_config')
|
513 |
run_ssh_cmd.return_value = ("", "") |
514 |
self.assertRaises(SVCError, svc._exec_cmd_and_parse_attrs,
|
515 |
'ssh_cmd placeholder')
|
516 |
|
517 |
def test_exec_cmd_parse_malformed(self, run_ssh_cmd): |
518 |
svc = SVCProvider('./test_configs/good_config')
|
519 |
ret_str = "test11 a\ntest2=2\ntest3\ntest4 test\n"
|
520 |
run_ssh_cmd.return_value = (ret_str, "")
|
521 |
ret = svc._exec_cmd_and_parse_attrs('ssh_cmd placeholder')
|
522 |
ret_dict = {'test11 a': '', 'test2=2': '', 'test3': '', |
523 |
'test4 test': ''} |
524 |
self.assertDictEqual(ret, ret_dict)
|
525 |
|
526 |
def test_get_vdisk_attributes_ok(self, run_ssh_cmd): |
527 |
svc = SVCProvider('./test_configs/good_config')
|
528 |
ret_str = "test1!1\ntest2!2\ntest3!3\ntest4!4\n"
|
529 |
run_ssh_cmd.return_value = (ret_str, "")
|
530 |
ret = svc._get_vdisk_attributes('ssh_cmd placeholder')
|
531 |
ret_dict = {'test1': '1', 'test2': '2', 'test3': '3', 'test4': '4'} |
532 |
self.assertDictEqual(ret, ret_dict)
|
533 |
|
534 |
def test_get_vdisk_attributes_empty(self, run_ssh_cmd): |
535 |
svc = SVCProvider('./test_configs/good_config')
|
536 |
run_ssh_cmd.return_value = ("", "") |
537 |
self.assertRaises(SVCError, svc._get_vdisk_attributes,
|
538 |
'ssh_cmd placeholder')
|
539 |
|
540 |
def test_get_vdisk_attributes_malformed(self, run_ssh_cmd): |
541 |
svc = SVCProvider('./test_configs/good_config')
|
542 |
ret_str = "test11 a\ntest2=2\ntest3\ntest4 test\n"
|
543 |
run_ssh_cmd.return_value = (ret_str, "")
|
544 |
ret = svc._get_vdisk_attributes('ssh_cmd placeholder')
|
545 |
ret_dict = {'test11 a': '', 'test2=2': '', 'test3': '', |
546 |
'test4 test': ''} |
547 |
self.assertDictEqual(ret, ret_dict)
|
548 |
self.assertDictEqual(ret, ret_dict)
|
549 |
|
550 |
def test_get_hdr_dic_ok(self, run_ssh_cmd): |
551 |
svc = SVCProvider('./test_configs/good_config')
|
552 |
hdr_str = "field1!field2!field3!field4"
|
553 |
row_str = "test1!test2!test3!test4"
|
554 |
ret = svc._get_hdr_dic(hdr_str, row_str, '!')
|
555 |
ret_dict = dict(zip(hdr_str.split('!'), row_str.split('!'))) |
556 |
self.assertDictEqual(ret, ret_dict)
|
557 |
|
558 |
def test_get_hdr_dic_malformed(self, run_ssh_cmd): |
559 |
svc = SVCProvider('./test_configs/good_config')
|
560 |
hdr_str = "field1!field2!field3!field4"
|
561 |
row_str = "test1!test2!test3"
|
562 |
self.assertRaises(SVCError, svc._get_hdr_dic, hdr_str, row_str, '!') |
563 |
|
564 |
def test_get_hdr_dic_empty(self, run_ssh_cmd): |
565 |
svc = SVCProvider('./test_configs/good_config')
|
566 |
ret = svc._get_hdr_dic("", "", '!') |
567 |
self.assertDictEqual(ret, {"": ""}) |
568 |
|
569 |
def test_get_hostvdisk_maps_ok(self, run_ssh_cmd): |
570 |
svc = SVCProvider('./test_configs/good_config')
|
571 |
hdr_str = "field1!vdisk_name!field3!field4\n"
|
572 |
row1_str = "test1!test2!test3!test4\n"
|
573 |
row2_str = "test11!test12!test13!test14\n"
|
574 |
ret_str = hdr_str + row1_str + row2_str |
575 |
run_ssh_cmd.return_value = (ret_str, "")
|
576 |
ret = svc._get_hostvdisk_mappings(5)
|
577 |
ssh_cmd = ['svcinfo', 'lshostvdiskmap', '-delim', '!', '5'] |
578 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
579 |
row1_dict = {'field1': 'test1', |
580 |
'vdisk_name': 'test2', |
581 |
'field3': 'test3', |
582 |
'field4': 'test4'} |
583 |
row2_dict = {'field1': 'test11', |
584 |
'vdisk_name': 'test12', |
585 |
'field3': 'test13', |
586 |
'field4': 'test14'} |
587 |
ret_dict = {'test2': row1_dict, 'test12': row2_dict} |
588 |
self.assertDictEqual(ret, ret_dict)
|
589 |
|
590 |
def test_get_hostvdisk_maps_malformed(self, run_ssh_cmd): |
591 |
svc = SVCProvider('./test_configs/good_config')
|
592 |
hdr_str = "field1!field2!field3!field4\n"
|
593 |
row1_str = "test1!test2!test3!test4\n"
|
594 |
row2_str = "test11!test12!test13!test14\n"
|
595 |
ret_str = hdr_str + row1_str + row2_str |
596 |
run_ssh_cmd.return_value = (ret_str, "")
|
597 |
self.assertRaises(Exception, svc._get_hostvdisk_mappings, 5) |
598 |
|
599 |
def test_get_hostvdisk_maps_empty(self, run_ssh_cmd): |
600 |
svc = SVCProvider('./test_configs/good_config')
|
601 |
run_ssh_cmd.return_value = ("", "") |
602 |
ret = svc._get_hostvdisk_mappings(5)
|
603 |
self.assertDictEqual(ret, {})
|
604 |
|
605 |
def test_get_hostvdisk_maps_only_header(self, run_ssh_cmd): |
606 |
svc = SVCProvider('./test_configs/good_config')
|
607 |
hdr_str = "field1!vdisk_name!field3!field4\n"
|
608 |
run_ssh_cmd.return_value = (hdr_str, "")
|
609 |
ret = svc._get_hostvdisk_mappings(5)
|
610 |
self.assertDictEqual(ret, {})
|
611 |
|
612 |
def test_get_hostvdisk_maps_exc(self, run_ssh_cmd): |
613 |
svc = SVCProvider('./test_configs/good_config')
|
614 |
run_ssh_cmd.side_effect = SVCError("Cannot run ssh cmd")
|
615 |
self.assertRaises(SVCError, svc._get_hostvdisk_mappings, 5) |
616 |
|
617 |
def test_map_vol_to_host(self, run_ssh_cmd): |
618 |
svc = SVCProvider('./test_configs/good_config')
|
619 |
run_ssh_cmd.return_value = ("successfully created vol", "") |
620 |
svc._map_vol_to_host('myvol', 7) |
621 |
ssh_cmd = ['svctask', 'mkvdiskhostmap', '-host', '7', 'myvol'] |
622 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
623 |
|
624 |
def test_map_vol_to_host_malformed(self, run_ssh_cmd): |
625 |
svc = SVCProvider('./test_configs/good_config')
|
626 |
run_ssh_cmd.return_value = ("malformed output", "") |
627 |
self.assertRaises(SVCError, svc._map_vol_to_host, 'myvol', 7) |
628 |
|
629 |
def test_map_vol_to_host_empty(self, run_ssh_cmd): |
630 |
svc = SVCProvider('./test_configs/good_config')
|
631 |
run_ssh_cmd.return_value = ("", "") |
632 |
self.assertRaises(SVCError, svc._map_vol_to_host, 'myvol', 7) |
633 |
|
634 |
def test_map_vol_to_host_assert(self, run_ssh_cmd): |
635 |
svc = SVCProvider('./test_configs/good_config')
|
636 |
self.assertRaises(AssertionError, svc._map_vol_to_host, 'myvol', 5) |
637 |
|
638 |
def test_unmap_vol_from_host(self, run_ssh_cmd): |
639 |
svc = SVCProvider('./test_configs/good_config')
|
640 |
run_ssh_cmd.return_value = ("", "") |
641 |
svc._unmap_vol_from_host('myvol', 7) |
642 |
ssh_cmd = ['svctask', 'rmvdiskhostmap', '-host', '7', 'myvol'] |
643 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
644 |
|
645 |
def test_unmap_vol_from_host_assert(self, run_ssh_cmd): |
646 |
svc = SVCProvider('./test_configs/good_config')
|
647 |
self.assertRaises(AssertionError, svc._unmap_vol_from_host, 'myvol', 5) |
648 |
|
649 |
def test_unmap_vol_from_host_malformed(self, run_ssh_cmd): |
650 |
svc = SVCProvider('./test_configs/good_config')
|
651 |
run_ssh_cmd.return_value = ("malformed output", "") |
652 |
self.assertRaises(SVCError, svc._unmap_vol_from_host, 'myvol', 7) |
653 |
|
654 |
|
655 |
@patch.object(SVCConnection, 'run_ssh_cmd', autospec=True) |
656 |
@patch.object(SVCProvider, '_is_vdisk_defined', autospec=True) |
657 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100', |
658 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
659 |
class SVCOpsTest(unittest.TestCase): |
660 |
def test_create_ok(self, is_vdisk_defined, run_ssh_cmd): |
661 |
svc = SVCProvider('./test_configs/good_config')
|
662 |
is_vdisk_defined.return_value = False
|
663 |
ret_str = "Virtual Disk, id [1], successfully created"
|
664 |
run_ssh_cmd.return_value = (ret_str, "")
|
665 |
svc.create() |
666 |
is_vdisk_defined.assert_called_once_with(svc, 'myvol')
|
667 |
ssh_cmd = ['svctask', 'mkvdisk', '-name', 'myvol', '-mdiskgrp', |
668 |
'VM_POOL_SYNNEFO', '-size', '100'] |
669 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
670 |
|
671 |
def test_create_exists(self, is_vdisk_defined, run_ssh_cmd): |
672 |
svc = SVCProvider('./test_configs/good_config')
|
673 |
is_vdisk_defined.return_value = True
|
674 |
self.assertRaises(SVCError, svc.create)
|
675 |
is_vdisk_defined.assert_called_once_with(svc, 'myvol')
|
676 |
|
677 |
@patch.object(SVCProvider, '_get_vdisk_attributes', autospec=True, |
678 |
return_value={'capacity': '83886080', |
679 |
'mdisk_grp_name': 'VM_POOL_SYNNEFO'}) |
680 |
def test_grow_expand_ok(self, attrs, is_vdisk_defined, run_ssh_cmd): |
681 |
svc = SVCProvider('./test_configs/good_config')
|
682 |
run_ssh_cmd.return_value = ("", "") |
683 |
svc.grow() |
684 |
attrs.assert_called_once_with(svc, 'myvol')
|
685 |
diff = (100 << 20) - (80 << 20) |
686 |
ssh_cmd = ['svctask', 'expandvdisksize', '-size', str(diff), '-unit', |
687 |
'bytes', 'myvol'] |
688 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
689 |
|
690 |
@patch.object(SVCProvider, '_get_vdisk_attributes', autospec=True, |
691 |
return_value={'capacity': str(120 << 20), |
692 |
'mdisk_grp_name': 'VM_POOL_SYNNEFO'}) |
693 |
def test_grow_shrink_ok(self, attrs, is_vdisk_defined, run_ssh_cmd): |
694 |
svc = SVCProvider('./test_configs/good_config')
|
695 |
run_ssh_cmd.return_value = ("", "") |
696 |
svc.grow() |
697 |
attrs.assert_called_once_with(svc, 'myvol')
|
698 |
diff = (120 << 20) - (100 << 20) |
699 |
ssh_cmd = ['svctask', 'shrinkvdisksize', '-size', str(diff), '-unit', |
700 |
'bytes', 'myvol'] |
701 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
702 |
#FIXME: failure tests
|
703 |
|
704 |
@patch.object(SVCProvider, '_verify_checklist', autospec=True, |
705 |
return_value=[True] * 4) |
706 |
def test_verify_ok(self, check, is_vdisk_defined, run_ssh_cmd): |
707 |
del os.environ['VOL_SIZE'] |
708 |
svc = SVCProvider('./test_configs/good_config')
|
709 |
svc.verify() |
710 |
|
711 |
@patch.dict(os.environ, {'VOL_NAME': 'myvol', |
712 |
'EXTP_SVC_SAN': 'svc_san_identifier'}) |
713 |
@patch.object(SVCProvider, '_verify_checklist', autospec=True, |
714 |
return_value=[False] * 4) |
715 |
def test_verify_not_ok(self, check, is_vdisk_defined, run_ssh_cmd): |
716 |
del os.environ['VOL_SIZE'] |
717 |
svc = SVCProvider('./test_configs/good_config')
|
718 |
self.assertRaises(SVCError, svc.verify)
|
719 |
|
720 |
@patch.object(SVCProvider, '_verify_checklist', autospec=True, |
721 |
return_value=[False] * 2 + [True] * 2) |
722 |
def test_inconsistent_unmap(self, check, is_vdisk_defined, run_ssh_cmd): |
723 |
svc = SVCProvider('./test_configs/good_config')
|
724 |
ret = svc._consistent_unmap(svc._verify_checklist()) |
725 |
self.assertEqual(ret, False) |
726 |
|
727 |
@patch.object(SVCProvider, '_is_vdisk_mapped', autospec=True, |
728 |
return_value=False)
|
729 |
@patch.object(SVCProvider, '_get_vdisk_attributes', autospec=True, |
730 |
return_value={'mdisk_grp_name': 'VM_POOL_SYNNEFO'}) |
731 |
def test_remove_ok(self, attrs, check, is_vdisk_defined, run_ssh_cmd): |
732 |
is_vdisk_defined.return_value = True
|
733 |
del os.environ['VOL_SIZE'] |
734 |
run_ssh_cmd.return_value = ("", "") |
735 |
svc = SVCProvider('./test_configs/good_config')
|
736 |
svc.remove() |
737 |
ssh_cmd = ['svctask', 'rmvdisk', '-force', 'myvol'] |
738 |
run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd) |
739 |
|
740 |
@patch.object(SVCProvider, '_is_vdisk_mapped', autospec=True, |
741 |
return_value=True)
|
742 |
@patch.object(SVCProvider, '_get_vdisk_attributes', autospec=True, |
743 |
return_value={'mdisk_grp_name': 'VM_POOL_SYNNEFO'}) |
744 |
def test_remove_mapped(self, attrs, check, is_vdisk_defined, run_ssh_cmd): |
745 |
is_vdisk_defined.return_value = True
|
746 |
del os.environ['VOL_SIZE'] |
747 |
run_ssh_cmd.return_value = ("", "") |
748 |
svc = SVCProvider('./test_configs/good_config')
|
749 |
self.assertRaises(SVCError, svc.remove)
|
750 |
|
751 |
@patch.object(SVCProvider, '_is_vdisk_mapped', autospec=True, |
752 |
return_value=False)
|
753 |
@patch.object(SVCProvider, '_get_vdisk_attributes', autospec=True, |
754 |
return_value={'mdisk_grp_name': 'VM_POOL_SYNNEFO'}) |
755 |
def test_remove_undef(self, attrs, check, is_vdisk_defined, run_ssh_cmd): |
756 |
is_vdisk_defined.return_value = False
|
757 |
del os.environ['VOL_SIZE'] |
758 |
run_ssh_cmd.return_value = ("", "") |
759 |
svc = SVCProvider('./test_configs/good_config')
|
760 |
self.assertRaises(SVCError, svc.remove)
|
761 |
|
762 |
@patch.object(SVCProvider, '_is_vdisk_mapped', autospec=True, |
763 |
return_value=False)
|
764 |
@patch.object(SVCProvider, '_get_vdisk_attributes', autospec=True, |
765 |
return_value={'mdisk_grp_name': 'VM_POOL_SYNNEFO1'}) |
766 |
def test_remove_assert(self, attrs, check, is_vdisk_defined, run_ssh_cmd): |
767 |
is_vdisk_defined.return_value = True
|
768 |
del os.environ['VOL_SIZE'] |
769 |
run_ssh_cmd.return_value = ("", "") |
770 |
svc = SVCProvider('./test_configs/good_config')
|
771 |
self.assertRaises(AssertionError, svc.remove) |
772 |
#TODO: Detach, attach tests
|
773 |
|
774 |
if __name__ == '__main__': |
775 |
unittest.main() |