4
from curtin.block import iscsi
5
from curtin import util
6
from .helpers import CiTestCase
9
class TestBlockIscsiPortalParsing(CiTestCase):
11
def test_iscsi_portal_parsing_string(self):
12
with self.assertRaisesRegexp(ValueError, 'not a string'):
13
iscsi.assert_valid_iscsi_portal(1234)
15
def test_iscsi_portal_parsing_no_port(self):
16
# port must be specified
17
with self.assertRaisesRegexp(ValueError, 'not in the format'):
18
iscsi.assert_valid_iscsi_portal('192.168.1.12')
19
with self.assertRaisesRegexp(ValueError, 'not in the format'):
20
iscsi.assert_valid_iscsi_portal('fe80::a634:d9ff:fe40:768a')
21
with self.assertRaisesRegexp(ValueError, 'not in the format'):
22
iscsi.assert_valid_iscsi_portal('192.168.1.12:')
23
with self.assertRaisesRegexp(ValueError, 'not in the format'):
24
iscsi.assert_valid_iscsi_portal('test.example.com:')
26
def test_iscsi_portal_parsing_valid_ip(self):
27
# IP must be in [] for IPv6, if not we misparse
28
host, port = iscsi.assert_valid_iscsi_portal(
29
'fe80::a634:d9ff:fe40:768a:9999')
30
self.assertEquals(host, 'fe80::a634:d9ff:fe40:768a')
31
self.assertEquals(port, 9999)
32
# IP must not be in [] if port is specified for IPv4
33
with self.assertRaisesRegexp(ValueError, 'Invalid IPv6 address'):
34
iscsi.assert_valid_iscsi_portal('[192.168.1.12]:9000')
35
with self.assertRaisesRegexp(ValueError, 'Invalid IPv6 address'):
36
iscsi.assert_valid_iscsi_portal('[test.example.com]:8000')
38
def test_iscsi_portal_parsing_ip(self):
39
with self.assertRaisesRegexp(ValueError, 'Invalid IPv6 address'):
40
iscsi.assert_valid_iscsi_portal(
41
'[1200::AB00:1234::2552:7777:1313]:9999')
42
# cannot distinguish between bad IP and bad hostname
43
host, port = iscsi.assert_valid_iscsi_portal('192.168:9000')
44
self.assertEquals(host, '192.168')
45
self.assertEquals(port, 9000)
47
def test_iscsi_portal_parsing_port(self):
48
with self.assertRaisesRegexp(ValueError, 'not in the format'):
49
iscsi.assert_valid_iscsi_portal('192.168.1.12:ABCD')
50
with self.assertRaisesRegexp(ValueError, 'not in the format'):
51
iscsi.assert_valid_iscsi_portal('[fe80::a634:d9ff:fe40:768a]:ABCD')
52
with self.assertRaisesRegexp(ValueError, 'not in the format'):
53
iscsi.assert_valid_iscsi_portal('test.example.com:ABCD')
55
def test_iscsi_portal_parsing_good_portals(self):
56
host, port = iscsi.assert_valid_iscsi_portal('192.168.1.12:9000')
57
self.assertEquals(host, '192.168.1.12')
58
self.assertEquals(port, 9000)
60
host, port = iscsi.assert_valid_iscsi_portal(
61
'[fe80::a634:d9ff:fe40:768a]:9999')
62
self.assertEquals(host, 'fe80::a634:d9ff:fe40:768a')
63
self.assertEquals(port, 9999)
65
host, port = iscsi.assert_valid_iscsi_portal('test.example.com:8000')
66
self.assertEquals(host, 'test.example.com')
67
self.assertEquals(port, 8000)
70
# TARGETSPEC=host:proto:port:lun:targetname
71
# root=iscsi:$TARGETSPEC
72
# root=iscsi:user:password@$TARGETSPEC
73
# root=iscsi:user:password:initiatoruser:initiatorpassword@$TARGETSPEC
74
def test_iscsi_disk_basic(self):
75
with self.assertRaisesRegexp(ValueError, 'must be specified'):
79
with self.assertRaisesRegexp(ValueError, 'must be specified'):
80
iscsi.IscsiDisk('iscs:')
83
with self.assertRaisesRegexp(ValueError, 'must be specified'):
84
iscsi.IscsiDisk('iscsi:')
85
with self.assertRaisesRegexp(ValueError, 'Both host and targetname'):
86
iscsi.IscsiDisk('iscsi:::::')
88
def test_iscsi_disk_ip_valid(self):
89
# these are all misparses we cannot catch trivially
90
i = iscsi.IscsiDisk('iscsi:192.168::::target')
91
self.assertEquals(i.user, None)
92
self.assertEquals(i.password, None)
93
self.assertEquals(i.iuser, None)
94
self.assertEquals(i.ipassword, None)
95
self.assertEquals(i.host, '192.168')
96
self.assertEquals(i.proto, '6')
97
self.assertEquals(i.port, 3260)
98
self.assertEquals(i.lun, 0)
99
self.assertEquals(i.target, 'target')
101
i = iscsi.IscsiDisk('iscsi:[fe80::]::::target')
102
self.assertEquals(i.user, None)
103
self.assertEquals(i.password, None)
104
self.assertEquals(i.iuser, None)
105
self.assertEquals(i.ipassword, None)
106
self.assertEquals(i.host, 'fe80::')
107
self.assertEquals(i.proto, '6')
108
self.assertEquals(i.port, 3260)
109
self.assertEquals(i.lun, 0)
110
self.assertEquals(i.target, 'target')
112
i = iscsi.IscsiDisk('iscsi:test.example::::target')
113
self.assertEquals(i.user, None)
114
self.assertEquals(i.password, None)
115
self.assertEquals(i.iuser, None)
116
self.assertEquals(i.ipassword, None)
117
self.assertEquals(i.host, 'test.example')
118
self.assertEquals(i.proto, '6')
119
self.assertEquals(i.port, 3260)
120
self.assertEquals(i.lun, 0)
121
self.assertEquals(i.target, 'target')
123
def test_iscsi_disk_port(self):
124
with self.assertRaisesRegexp(ValueError, 'Specified iSCSI port'):
125
iscsi.IscsiDisk('iscsi:192.168.1.12::ABCD::target')
126
with self.assertRaisesRegexp(ValueError, 'Specified iSCSI port'):
127
iscsi.IscsiDisk('iscsi:[fe80::a634:d9ff:fe40:768a:6]::ABCD::'
129
with self.assertRaisesRegexp(ValueError, 'Specified iSCSI port'):
130
iscsi.IscsiDisk('iscsi:test.example.com::ABCD::target')
132
def test_iscsi_disk_target(self):
133
with self.assertRaisesRegexp(ValueError, 'Both host and targetname'):
134
iscsi.IscsiDisk('iscsi:192.168.1.12::::')
135
with self.assertRaisesRegexp(ValueError, 'Both host and targetname'):
136
iscsi.IscsiDisk('iscsi:[fe80::a634:d9ff:fe40:768a:6]::::')
137
with self.assertRaisesRegexp(ValueError, 'Both host and targetname'):
138
iscsi.IscsiDisk('iscsi:test.example.com::::')
140
def test_iscsi_disk_ip(self):
141
with self.assertRaisesRegexp(ValueError, 'Both host and targetname'):
142
iscsi.IscsiDisk('iscsi:::::target')
144
def test_iscsi_disk_auth(self):
145
# user without password
146
with self.assertRaises(ValueError):
147
iscsi.IscsiDisk('iscsi:user@192.168.1.12::::target')
148
with self.assertRaises(ValueError):
149
iscsi.IscsiDisk('iscsi:user@[fe80::a634:d9ff:fe40:768a:6]::::'
151
with self.assertRaises(ValueError):
152
iscsi.IscsiDisk('iscsi:user@test.example.com::::target')
154
# iuser without password
155
with self.assertRaises(ValueError):
156
iscsi.IscsiDisk('iscsi:user:password:iuser@192.168.1.12::::target')
157
with self.assertRaises(ValueError):
158
iscsi.IscsiDisk('iscsi:user:password:iuser@'
159
'[fe80::a634:d9ff:fe40:768a:6]::::target')
160
with self.assertRaises(ValueError):
162
'iscsi:user:password:iuser@test.example.com::::target')
164
def test_iscsi_disk_good_ipv4(self):
165
i = iscsi.IscsiDisk('iscsi:192.168.1.12:6:3260:1:target')
166
self.assertEquals(i.user, None)
167
self.assertEquals(i.password, None)
168
self.assertEquals(i.iuser, None)
169
self.assertEquals(i.ipassword, None)
170
self.assertEquals(i.host, '192.168.1.12')
171
self.assertEquals(i.proto, '6')
172
self.assertEquals(i.port, 3260)
173
self.assertEquals(i.lun, 1)
174
self.assertEquals(i.target, 'target')
176
i = iscsi.IscsiDisk('iscsi:192.168.1.12::3260:1:target')
177
self.assertEquals(i.user, None)
178
self.assertEquals(i.password, None)
179
self.assertEquals(i.iuser, None)
180
self.assertEquals(i.ipassword, None)
181
self.assertEquals(i.host, '192.168.1.12')
182
self.assertEquals(i.proto, '6')
183
self.assertEquals(i.port, 3260)
184
self.assertEquals(i.lun, 1)
185
self.assertEquals(i.target, 'target')
187
i = iscsi.IscsiDisk('iscsi:192.168.1.12:::1:target')
188
self.assertEquals(i.user, None)
189
self.assertEquals(i.password, None)
190
self.assertEquals(i.iuser, None)
191
self.assertEquals(i.ipassword, None)
192
self.assertEquals(i.host, '192.168.1.12')
193
self.assertEquals(i.proto, '6')
194
self.assertEquals(i.port, 3260)
195
self.assertEquals(i.lun, 1)
196
self.assertEquals(i.target, 'target')
198
i = iscsi.IscsiDisk('iscsi:user:password@192.168.1.12:::1:target')
199
self.assertEquals(i.user, 'user')
200
self.assertEquals(i.password, 'password')
201
self.assertEquals(i.iuser, None)
202
self.assertEquals(i.ipassword, None)
203
self.assertEquals(i.host, '192.168.1.12')
204
self.assertEquals(i.proto, '6')
205
self.assertEquals(i.port, 3260)
206
self.assertEquals(i.lun, 1)
207
self.assertEquals(i.target, 'target')
209
i = iscsi.IscsiDisk('iscsi:user:@192.168.1.12:::1:target')
210
self.assertEquals(i.user, 'user')
211
self.assertEquals(i.password, '')
212
self.assertEquals(i.iuser, None)
213
self.assertEquals(i.ipassword, None)
214
self.assertEquals(i.host, '192.168.1.12')
215
self.assertEquals(i.proto, '6')
216
self.assertEquals(i.port, 3260)
217
self.assertEquals(i.lun, 1)
218
self.assertEquals(i.target, 'target')
220
i = iscsi.IscsiDisk('iscsi:user:password:iuser:ipassword@'
221
'192.168.1.12:::1:target')
222
self.assertEquals(i.user, 'user')
223
self.assertEquals(i.password, 'password')
224
self.assertEquals(i.iuser, 'iuser')
225
self.assertEquals(i.ipassword, 'ipassword')
226
self.assertEquals(i.host, '192.168.1.12')
227
self.assertEquals(i.proto, '6')
228
self.assertEquals(i.port, 3260)
229
self.assertEquals(i.lun, 1)
230
self.assertEquals(i.target, 'target')
232
i = iscsi.IscsiDisk('iscsi:user:password:iuser:@'
233
'192.168.1.12:::1:target')
234
self.assertEquals(i.user, 'user')
235
self.assertEquals(i.password, 'password')
236
self.assertEquals(i.iuser, 'iuser')
237
self.assertEquals(i.ipassword, '')
238
self.assertEquals(i.host, '192.168.1.12')
239
self.assertEquals(i.proto, '6')
240
self.assertEquals(i.port, 3260)
241
self.assertEquals(i.lun, 1)
242
self.assertEquals(i.target, 'target')
244
i = iscsi.IscsiDisk('iscsi:user::iuser:@192.168.1.12:::1:target')
245
self.assertEquals(i.user, 'user')
246
self.assertEquals(i.password, '')
247
self.assertEquals(i.iuser, 'iuser')
248
self.assertEquals(i.ipassword, '')
249
self.assertEquals(i.host, '192.168.1.12')
250
self.assertEquals(i.proto, '6')
251
self.assertEquals(i.port, 3260)
252
self.assertEquals(i.lun, 1)
253
self.assertEquals(i.target, 'target')
255
def test_iscsi_disk_good_ipv6(self):
257
'iscsi:[fe80::a634:d9ff:fe40:768a:6]:5:3260:1:target')
258
self.assertEquals(i.user, None)
259
self.assertEquals(i.password, None)
260
self.assertEquals(i.iuser, None)
261
self.assertEquals(i.ipassword, None)
262
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
263
self.assertEquals(i.proto, '6')
264
self.assertEquals(i.port, 3260)
265
self.assertEquals(i.lun, 1)
266
self.assertEquals(i.target, 'target')
269
'iscsi:[fe80::a634:d9ff:fe40:768a:6]::3260:1:target')
270
self.assertEquals(i.user, None)
271
self.assertEquals(i.password, None)
272
self.assertEquals(i.iuser, None)
273
self.assertEquals(i.ipassword, None)
274
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
275
self.assertEquals(i.proto, '6')
276
self.assertEquals(i.port, 3260)
277
self.assertEquals(i.lun, 1)
278
self.assertEquals(i.target, 'target')
280
i = iscsi.IscsiDisk('iscsi:[fe80::a634:d9ff:fe40:768a:6]:::1:target')
281
self.assertEquals(i.user, None)
282
self.assertEquals(i.password, None)
283
self.assertEquals(i.iuser, None)
284
self.assertEquals(i.ipassword, None)
285
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
286
self.assertEquals(i.proto, '6')
287
self.assertEquals(i.port, 3260)
288
self.assertEquals(i.lun, 1)
289
self.assertEquals(i.target, 'target')
291
i = iscsi.IscsiDisk('iscsi:user:password@'
292
'[fe80::a634:d9ff:fe40:768a:6]:::1:target')
293
self.assertEquals(i.user, 'user')
294
self.assertEquals(i.password, 'password')
295
self.assertEquals(i.iuser, None)
296
self.assertEquals(i.ipassword, None)
297
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
298
self.assertEquals(i.proto, '6')
299
self.assertEquals(i.port, 3260)
300
self.assertEquals(i.lun, 1)
301
self.assertEquals(i.target, 'target')
303
i = iscsi.IscsiDisk('iscsi:user:@'
304
'[fe80::a634:d9ff:fe40:768a:6]:::1:target')
305
self.assertEquals(i.user, 'user')
306
self.assertEquals(i.password, '')
307
self.assertEquals(i.iuser, None)
308
self.assertEquals(i.ipassword, None)
309
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
310
self.assertEquals(i.proto, '6')
311
self.assertEquals(i.port, 3260)
312
self.assertEquals(i.lun, 1)
313
self.assertEquals(i.target, 'target')
315
i = iscsi.IscsiDisk('iscsi:user:password:iuser:ipassword@'
316
'[fe80::a634:d9ff:fe40:768a:6]:::1:target')
317
self.assertEquals(i.user, 'user')
318
self.assertEquals(i.password, 'password')
319
self.assertEquals(i.iuser, 'iuser')
320
self.assertEquals(i.ipassword, 'ipassword')
321
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
322
self.assertEquals(i.proto, '6')
323
self.assertEquals(i.port, 3260)
324
self.assertEquals(i.lun, 1)
325
self.assertEquals(i.target, 'target')
327
i = iscsi.IscsiDisk('iscsi:user:password:iuser:@'
328
'[fe80::a634:d9ff:fe40:768a:6]:::1:target')
329
self.assertEquals(i.user, 'user')
330
self.assertEquals(i.password, 'password')
331
self.assertEquals(i.iuser, 'iuser')
332
self.assertEquals(i.ipassword, '')
333
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
334
self.assertEquals(i.proto, '6')
335
self.assertEquals(i.port, 3260)
336
self.assertEquals(i.lun, 1)
337
self.assertEquals(i.target, 'target')
339
i = iscsi.IscsiDisk('iscsi:user::iuser:@'
340
'[fe80::a634:d9ff:fe40:768a:6]:::1:target')
341
self.assertEquals(i.user, 'user')
342
self.assertEquals(i.password, '')
343
self.assertEquals(i.iuser, 'iuser')
344
self.assertEquals(i.ipassword, '')
345
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
346
self.assertEquals(i.proto, '6')
347
self.assertEquals(i.port, 3260)
348
self.assertEquals(i.lun, 1)
349
self.assertEquals(i.target, 'target')
351
def test_iscsi_disk_good_hostname(self):
352
i = iscsi.IscsiDisk('iscsi:test.example.com:6:3260:1:target')
353
self.assertEquals(i.user, None)
354
self.assertEquals(i.password, None)
355
self.assertEquals(i.iuser, None)
356
self.assertEquals(i.ipassword, None)
357
self.assertEquals(i.host, 'test.example.com')
358
self.assertEquals(i.proto, '6')
359
self.assertEquals(i.port, 3260)
360
self.assertEquals(i.lun, 1)
361
self.assertEquals(i.target, 'target')
363
i = iscsi.IscsiDisk('iscsi:test.example.com::3260:1:target')
364
self.assertEquals(i.user, None)
365
self.assertEquals(i.password, None)
366
self.assertEquals(i.iuser, None)
367
self.assertEquals(i.ipassword, None)
368
self.assertEquals(i.host, 'test.example.com')
369
self.assertEquals(i.proto, '6')
370
self.assertEquals(i.port, 3260)
371
self.assertEquals(i.lun, 1)
372
self.assertEquals(i.target, 'target')
374
i = iscsi.IscsiDisk('iscsi:test.example.com:::1:target')
375
self.assertEquals(i.user, None)
376
self.assertEquals(i.password, None)
377
self.assertEquals(i.iuser, None)
378
self.assertEquals(i.ipassword, None)
379
self.assertEquals(i.host, 'test.example.com')
380
self.assertEquals(i.proto, '6')
381
self.assertEquals(i.port, 3260)
382
self.assertEquals(i.lun, 1)
383
self.assertEquals(i.target, 'target')
385
i = iscsi.IscsiDisk('iscsi:user:password@test.example.com:::1:target')
386
self.assertEquals(i.user, 'user')
387
self.assertEquals(i.password, 'password')
388
self.assertEquals(i.iuser, None)
389
self.assertEquals(i.ipassword, None)
390
self.assertEquals(i.host, 'test.example.com')
391
self.assertEquals(i.proto, '6')
392
self.assertEquals(i.port, 3260)
393
self.assertEquals(i.lun, 1)
394
self.assertEquals(i.target, 'target')
396
i = iscsi.IscsiDisk('iscsi:user:@test.example.com:::1:target')
397
self.assertEquals(i.user, 'user')
398
self.assertEquals(i.password, '')
399
self.assertEquals(i.iuser, None)
400
self.assertEquals(i.ipassword, None)
401
self.assertEquals(i.host, 'test.example.com')
402
self.assertEquals(i.proto, '6')
403
self.assertEquals(i.port, 3260)
404
self.assertEquals(i.lun, 1)
405
self.assertEquals(i.target, 'target')
407
i = iscsi.IscsiDisk('iscsi:user:password:iuser:ipassword@'
408
'test.example.com:::1:target')
409
self.assertEquals(i.user, 'user')
410
self.assertEquals(i.password, 'password')
411
self.assertEquals(i.iuser, 'iuser')
412
self.assertEquals(i.ipassword, 'ipassword')
413
self.assertEquals(i.host, 'test.example.com')
414
self.assertEquals(i.proto, '6')
415
self.assertEquals(i.port, 3260)
416
self.assertEquals(i.lun, 1)
417
self.assertEquals(i.target, 'target')
419
i = iscsi.IscsiDisk('iscsi:user:password:iuser:@'
420
'test.example.com:::1:target')
421
self.assertEquals(i.user, 'user')
422
self.assertEquals(i.password, 'password')
423
self.assertEquals(i.iuser, 'iuser')
424
self.assertEquals(i.ipassword, '')
425
self.assertEquals(i.host, 'test.example.com')
426
self.assertEquals(i.proto, '6')
427
self.assertEquals(i.port, 3260)
428
self.assertEquals(i.lun, 1)
429
self.assertEquals(i.target, 'target')
431
i = iscsi.IscsiDisk('iscsi:user::iuser:@test.example.com:::1:target')
432
self.assertEquals(i.user, 'user')
433
self.assertEquals(i.password, '')
434
self.assertEquals(i.iuser, 'iuser')
435
self.assertEquals(i.ipassword, '')
436
self.assertEquals(i.host, 'test.example.com')
437
self.assertEquals(i.proto, '6')
438
self.assertEquals(i.port, 3260)
439
self.assertEquals(i.lun, 1)
440
self.assertEquals(i.target, 'target')
443
def test_iscsi_target_parsing(self):
445
'iscsi:192.168.1.12::::iqn.2017-04.com.example.test:target-name')
446
self.assertEquals(i.user, None)
447
self.assertEquals(i.password, None)
448
self.assertEquals(i.iuser, None)
449
self.assertEquals(i.ipassword, None)
450
self.assertEquals(i.host, '192.168.1.12')
451
self.assertEquals(i.proto, '6')
452
self.assertEquals(i.port, 3260)
453
self.assertEquals(i.lun, 0)
454
self.assertEquals(i.target, 'iqn.2017-04.com.example.test:target-name')
457
'iscsi:[fe80::a634:d9ff:fe40:768a:6]::::'
458
'iqn.2017-04.com.example.test:target-name')
459
self.assertEquals(i.user, None)
460
self.assertEquals(i.password, None)
461
self.assertEquals(i.iuser, None)
462
self.assertEquals(i.ipassword, None)
463
self.assertEquals(i.host, 'fe80::a634:d9ff:fe40:768a:6')
464
self.assertEquals(i.proto, '6')
465
self.assertEquals(i.port, 3260)
466
self.assertEquals(i.lun, 0)
467
self.assertEquals(i.target, 'iqn.2017-04.com.example.test:target-name')
470
'iscsi:test.example.com::::'
471
'iqn.2017-04.com.example.test:target-name')
472
self.assertEquals(i.user, None)
473
self.assertEquals(i.password, None)
474
self.assertEquals(i.iuser, None)
475
self.assertEquals(i.ipassword, None)
476
self.assertEquals(i.host, 'test.example.com')
477
self.assertEquals(i.proto, '6')
478
self.assertEquals(i.port, 3260)
479
self.assertEquals(i.lun, 0)
480
self.assertEquals(i.target, 'iqn.2017-04.com.example.test:target-name')
483
class TestBlockIscsiVolPath(CiTestCase):
484
# non-iscsi backed disk returns false
485
# regular iscsi-backed disk returns true
486
# layered setup without an iscsi member returns false
487
# layered setup with an iscsi member returns true
490
super(TestBlockIscsiVolPath, self).setUp()
491
self.add_patch('curtin.block.iscsi.get_device_slave_knames',
492
'mock_get_device_slave_knames')
493
self.add_patch('curtin.block.iscsi.path_to_kname',
494
'mock_path_to_kname')
495
self.add_patch('curtin.block.iscsi.kname_is_iscsi',
496
'mock_kname_is_iscsi')
498
def test_volpath_is_iscsi_false(self):
499
volume_path = '/dev/wark'
502
self.mock_get_device_slave_knames.return_value = slaves
503
self.mock_path_to_kname.return_value = kname
504
self.mock_kname_is_iscsi.return_value = 'iscsi' in kname
506
is_iscsi = iscsi.volpath_is_iscsi(volume_path)
508
self.assertFalse(is_iscsi)
509
self.mock_get_device_slave_knames.assert_called_with(volume_path)
510
self.mock_path_to_kname.assert_called_with(volume_path)
511
self.mock_kname_is_iscsi.assert_called_with(kname)
513
def test_volpath_is_iscsi_true(self):
514
volume_path = '/dev/wark'
515
kname = 'wark-iscsi-lun-2'
517
self.mock_get_device_slave_knames.return_value = slaves
518
self.mock_path_to_kname.return_value = kname
519
self.mock_kname_is_iscsi.return_value = 'iscsi' in kname
521
is_iscsi = iscsi.volpath_is_iscsi(volume_path)
523
self.assertTrue(is_iscsi)
524
self.mock_get_device_slave_knames.assert_called_with(volume_path)
525
self.mock_path_to_kname.assert_called_with(volume_path)
526
self.mock_kname_is_iscsi.assert_called_with(kname)
528
def test_volpath_is_iscsi_layered_true(self):
529
volume_path = '/dev/wark'
530
slaves = ['wark', 'bzr', 'super-iscsi-lun-27']
531
self.mock_get_device_slave_knames.return_value = slaves
532
self.mock_path_to_kname.side_effect = lambda x: x
533
self.mock_kname_is_iscsi.side_effect = lambda x: 'iscsi' in x
535
is_iscsi = iscsi.volpath_is_iscsi(volume_path)
537
self.assertTrue(is_iscsi)
538
self.mock_get_device_slave_knames.assert_called_with(volume_path)
539
self.mock_path_to_kname.assert_called_with(volume_path)
540
self.mock_kname_is_iscsi.assert_has_calls([
541
mock.call(x) for x in slaves])
543
def test_volpath_is_iscsi_layered_false(self):
544
volume_path = '/dev/wark'
545
slaves = ['wark', 'bzr', 'nvmen27p47']
546
self.mock_get_device_slave_knames.return_value = slaves
547
self.mock_path_to_kname.side_effect = lambda x: x
548
self.mock_kname_is_iscsi.side_effect = lambda x: 'iscsi' in x
550
is_iscsi = iscsi.volpath_is_iscsi(volume_path)
552
self.assertFalse(is_iscsi)
553
self.mock_get_device_slave_knames.assert_called_with(volume_path)
554
self.mock_path_to_kname.assert_called_with(volume_path)
555
self.mock_kname_is_iscsi.assert_has_calls([
556
mock.call(x) for x in slaves])
558
def test_volpath_is_iscsi_missing_param(self):
559
with self.assertRaises(ValueError):
560
iscsi.volpath_is_iscsi(None)
563
class TestBlockIscsiDiskFromConfig(CiTestCase):
564
# Test iscsi parsing of storage config for iscsi configure disks
567
super(TestBlockIscsiDiskFromConfig, self).setUp()
568
self.add_patch('curtin.block.iscsi.util.subp', 'mock_subp')
570
def test_parse_iscsi_disk_from_config(self):
571
"""Test parsing iscsi volume path creates the same iscsi disk"""
572
target = 'curtin-659d5f45-4f23-46cb-b826-f2937b896e09'
573
iscsi_path = 'iscsi:10.245.168.20::20112:1:' + target
576
'config': [{'type': 'disk',
579
'name': 'iscsi_disk1',
581
'wipe': 'superblock'}]
584
expected_iscsi_disk = iscsi.IscsiDisk(iscsi_path)
585
iscsi_disk = iscsi.get_iscsi_disks_from_config(cfg).pop()
586
# utilize IscsiDisk str method for equality check
587
self.assertEqual(str(expected_iscsi_disk), str(iscsi_disk))
589
def test_parse_iscsi_disk_from_config_no_iscsi(self):
590
"""Test parsing storage config with no iscsi disks included"""
593
'config': [{'type': 'disk',
595
'path': 'dev/slash/foo1',
596
'name': 'the-fast-one',
598
'wipe': 'superblock'}]
601
expected_iscsi_disks = []
602
iscsi_disks = iscsi.get_iscsi_disks_from_config(cfg)
603
self.assertEqual(expected_iscsi_disks, iscsi_disks)
605
def test_parse_iscsi_disk_from_config_invalid_iscsi(self):
606
"""Test parsing storage config with no iscsi disks included"""
609
'config': [{'type': 'disk',
611
'path': 'iscsi:garbage',
614
'wipe': 'superblock'}]
617
with self.assertRaises(ValueError):
618
iscsi.get_iscsi_disks_from_config(cfg)
620
def test_parse_iscsi_disk_from_config_empty(self):
621
"""Test parse_iscsi_disks handles empty/invalid config"""
622
expected_iscsi_disks = []
623
iscsi_disks = iscsi.get_iscsi_disks_from_config({})
624
self.assertEqual(expected_iscsi_disks, iscsi_disks)
626
cfg = {'storage': {'config': []}}
627
iscsi_disks = iscsi.get_iscsi_disks_from_config(cfg)
628
self.assertEqual(expected_iscsi_disks, iscsi_disks)
630
def test_parse_iscsi_disk_from_config_none(self):
631
"""Test parse_iscsi_disks handles no config"""
632
expected_iscsi_disks = []
633
iscsi_disks = iscsi.get_iscsi_disks_from_config({})
634
self.assertEqual(expected_iscsi_disks, iscsi_disks)
637
iscsi_disks = iscsi.get_iscsi_disks_from_config(cfg)
638
self.assertEqual(expected_iscsi_disks, iscsi_disks)
641
class TestBlockIscsiDisconnect(CiTestCase):
642
# test that when disconnecting iscsi targets we
643
# check that the target has an active session before
644
# issuing a disconnect command
647
super(TestBlockIscsiDisconnect, self).setUp()
648
self.add_patch('curtin.block.iscsi.util.subp', 'mock_subp')
649
self.add_patch('curtin.block.iscsi.iscsiadm_sessions',
650
'mock_iscsi_sessions')
651
# fake target_root + iscsi nodes dir
652
self.target_path = self.tmp_dir()
653
self.iscsi_nodes = os.path.join(self.target_path, 'etc/iscsi/nodes')
654
util.ensure_dir(self.iscsi_nodes)
656
def _fmt_disconnect(self, target, portal):
657
return ['iscsiadm', '--mode=node', '--targetname=%s' % target,
658
'--portal=%s' % portal, '--logout']
660
def _setup_nodes(self, sessions, connection):
661
# setup iscsi_nodes dir (<fakeroot>/etc/iscsi/nodes) with content
663
sdir = os.path.join(self.iscsi_nodes, s)
664
connpath = os.path.join(sdir, connection)
665
util.ensure_dir(sdir)
666
util.write_file(connpath, content="")
668
def test_disconnect_target_disk(self):
669
"""Test iscsi disconnecting multiple sessions, all present"""
672
'curtin-53ab23ff-a887-449a-80a8-288151208091',
673
'curtin-94b62de1-c579-42c0-879e-8a28178e64c5',
674
'curtin-556aeecd-a227-41b7-83d7-2bb471c574b4',
675
'curtin-fd0f644b-7858-420f-9997-3ea2aefe87b9'
677
connection = '10.245.168.20,16395,1'
678
self._setup_nodes(sessions, connection)
680
self.mock_iscsi_sessions.return_value = "\n".join(sessions)
682
iscsi.disconnect_target_disks(self.target_path)
685
for session in sessions:
686
(host, port, _) = connection.split(',')
687
disconnect = self._fmt_disconnect(session, "%s:%s" % (host, port))
690
mock.call(disconnect, capture=True, log_captured=True),
691
mock.call(['udevadm', 'settle']),
693
expected_calls.extend(calls)
695
self.mock_subp.assert_has_calls(expected_calls, any_order=True)
697
def test_disconnect_target_disk_skip_disconnected(self):
698
"""Test iscsi does not attempt to disconnect already closed sessions"""
700
'curtin-53ab23ff-a887-449a-80a8-288151208091',
701
'curtin-94b62de1-c579-42c0-879e-8a28178e64c5',
702
'curtin-556aeecd-a227-41b7-83d7-2bb471c574b4',
703
'curtin-fd0f644b-7858-420f-9997-3ea2aefe87b9'
705
connection = '10.245.168.20,16395,1'
706
self._setup_nodes(sessions, connection)
707
# Test with all sessions are already disconnected
708
self.mock_iscsi_sessions.return_value = ""
710
iscsi.disconnect_target_disks(self.target_path)
712
self.mock_subp.assert_has_calls([], any_order=True)
714
@mock.patch('curtin.block.iscsi.iscsiadm_logout')
715
def test_disconnect_target_disk_raises_runtime_error(self, mock_logout):
716
"""Test iscsi raises RuntimeError if we fail to logout"""
718
'curtin-53ab23ff-a887-449a-80a8-288151208091',
720
connection = '10.245.168.20,16395,1'
721
self._setup_nodes(sessions, connection)
722
self.mock_iscsi_sessions.return_value = "\n".join(sessions)
723
mock_logout.side_effect = util.ProcessExecutionError()
725
with self.assertRaises(RuntimeError):
726
iscsi.disconnect_target_disks(self.target_path)
729
for session in sessions:
730
(host, port, _) = connection.split(',')
731
disconnect = self._fmt_disconnect(session, "%s:%s" % (host, port))
734
mock.call(disconnect, capture=True, log_captured=True),
735
mock.call(['udevadm', 'settle']),
737
expected_calls.extend(calls)
739
self.mock_subp.assert_has_calls([], any_order=True)
741
# vi: ts=4 expandtab syntax=python
added
removed
tests/unittests/test_block_iscsi.py
