1# Copyright (c) 2016 by Kaminario Technologies, Ltd. 2# All Rights Reserved. 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); you may 5# not use this file except in compliance with the License. You may obtain 6# a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 13# License for the specific language governing permissions and limitations 14# under the License. 15"""Unit tests for kaminario driver.""" 16import re 17 18import ddt 19import mock 20from oslo_utils import units 21import time 22 23from cinder import context 24from cinder import exception 25from cinder import objects 26from cinder.objects import fields 27from cinder import test 28from cinder.tests.unit import fake_snapshot 29from cinder.tests.unit import fake_volume 30from cinder import utils 31from cinder.volume import configuration 32from cinder.volume.drivers.kaminario import kaminario_common 33from cinder.volume.drivers.kaminario import kaminario_fc 34from cinder.volume.drivers.kaminario import kaminario_iscsi 35from cinder.volume import utils as vol_utils 36 37CONNECTOR = {'initiator': 'iqn.1993-08.org.debian:01:12aa12aa12aa', 38 'ip': '192.168.2.5', 'platform': 'x86_64', 'host': 'test-k2', 39 'wwpns': ['12341a2a00001234', '12341a2a00001235'], 40 'wwnns': ['12351a2a00001234', '12361a2a00001234'], 41 'os_type': 'linux2', 'multipath': False} 42 43 44class FakeK2Obj(object): 45 id = 548 46 lun = 548 47 48 49class FakeSaveObject(FakeK2Obj): 50 def __init__(self, *args, **kwargs): 51 item = kwargs.pop('item', 1) 52 self.ntype = kwargs.get('ntype') 53 self.ip_address = '10.0.0.%s' % item 54 self.iscsi_qualified_target_name = "xyztlnxyz" 55 self.snapshot = FakeK2Obj() 56 self.name = 'test' 57 self.pwwn = '50024f405330030%s' % item 58 self.volume_group = self 59 self.is_dedup = True 60 self.size = units.Mi 61 self.replication_status = None 62 self.state = 'in_sync' 63 self.generation_number = 548 64 self.current_role = 'target' 65 self.current_snapshot_progress = 100 66 self.current_snapshot_id = None 67 self.wan_port = None 68 69 def refresh(self): 70 return 71 72 def save(self): 73 return FakeSaveObject() 74 75 def delete(self): 76 return None 77 78 79class FakeSaveObjectExp(FakeSaveObject): 80 def save(self): 81 raise exception.KaminarioCinderDriverException("test") 82 83 def delete(self): 84 raise exception.KaminarioCinderDriverException("test") 85 86 87class FakeSearchObject(object): 88 hits = [FakeSaveObject(item=1), FakeSaveObject(item=2)] 89 total = 2 90 91 def __init__(self, *args): 92 if args and "mappings" in args[0]: 93 self.total = 0 94 95 96class FakeSearchObjectExp(object): 97 hits = [FakeSaveObjectExp()] 98 total = 1 99 100 101class FakeKrest(object): 102 def search(self, *args, **argv): 103 return FakeSearchObject(*args) 104 105 def new(self, *args, **argv): 106 return FakeSaveObject() 107 108 109class FakeKrestException(object): 110 def search(self, *args, **argv): 111 return FakeSearchObjectExp() 112 113 def new(self, *args, **argv): 114 return FakeSaveObjectExp() 115 116 117class Replication(object): 118 backend_id = '10.0.0.1' 119 login = 'login' 120 password = 'password' 121 rpo = 500 122 123 124class TestKaminarioCommon(test.TestCase): 125 driver = None 126 conf = None 127 128 def setUp(self): 129 self._setup_config() 130 self._setup_driver() 131 super(TestKaminarioCommon, self).setUp() 132 self.context = context.get_admin_context() 133 self.vol = fake_volume.fake_volume_obj(self.context) 134 self.vol.volume_type = fake_volume.fake_volume_type_obj(self.context) 135 self.vol.volume_type.extra_specs = {'foo': None} 136 self.snap = fake_snapshot.fake_snapshot_obj(self.context) 137 self.snap.volume = self.vol 138 self.patch('eventlet.sleep') 139 140 def _setup_config(self): 141 self.conf = mock.Mock(spec=configuration.Configuration) 142 self.conf.kaminario_dedup_type_name = "dedup" 143 self.conf.volume_dd_blocksize = 2 144 self.conf.unique_fqdn_network = True 145 self.conf.disable_discovery = False 146 147 def _setup_driver(self): 148 self.driver = (kaminario_iscsi. 149 KaminarioISCSIDriver(configuration=self.conf)) 150 device = mock.Mock(return_value={'device': {'path': '/dev'}}) 151 self.driver._connect_device = device 152 self.driver.client = FakeKrest() 153 154 def test_create_volume(self): 155 """Test create_volume.""" 156 result = self.driver.create_volume(self.vol) 157 self.assertIsNone(result) 158 159 def test_create_volume_with_exception(self): 160 """Test create_volume_with_exception.""" 161 self.driver.client = FakeKrestException() 162 self.assertRaises(exception.KaminarioCinderDriverException, 163 self.driver.create_volume, self.vol) 164 165 def test_delete_volume(self): 166 """Test delete_volume.""" 167 result = self.driver.delete_volume(self.vol) 168 self.assertIsNone(result) 169 170 def test_delete_volume_with_exception(self): 171 """Test delete_volume_with_exception.""" 172 self.driver.client = FakeKrestException() 173 self.assertRaises(exception.KaminarioCinderDriverException, 174 self.driver.delete_volume, self.vol) 175 176 def test_create_snapshot(self): 177 """Test create_snapshot.""" 178 self.snap.id = "253b2878-ec60-4793-ad19-e65496ec7aab" 179 self.driver.client.new = mock.Mock() 180 result = self.driver.create_snapshot(self.snap) 181 self.assertIsNone(result) 182 fake_object = self.driver.client.search().hits[0] 183 self.driver.client.new.assert_called_once_with( 184 "snapshots", 185 short_name='cs-253b2878-ec60-4793-ad19-e65496ec7aab', 186 source=fake_object, retention_policy=fake_object, 187 is_auto_deleteable=False) 188 189 def test_create_snapshot_with_exception(self): 190 """Test create_snapshot_with_exception.""" 191 self.driver.client = FakeKrestException() 192 self.assertRaises(exception.KaminarioCinderDriverException, 193 self.driver.create_snapshot, self.snap) 194 195 def test_delete_snapshot(self): 196 """Test delete_snapshot.""" 197 result = self.driver.delete_snapshot(self.snap) 198 self.assertIsNone(result) 199 200 def test_delete_snapshot_with_exception(self): 201 """Test delete_snapshot_with_exception.""" 202 self.driver.client = FakeKrestException() 203 self.assertRaises(exception.KaminarioCinderDriverException, 204 self.driver.delete_snapshot, self.snap) 205 206 @mock.patch.object(utils, 'brick_get_connector_properties') 207 @mock.patch.object(vol_utils, 'copy_volume') 208 def test_create_volume_from_snapshot(self, mock_copy_volume, 209 mock_brick_get): 210 """Test create_volume_from_snapshot.""" 211 mock_brick_get.return_value = CONNECTOR 212 mock_copy_volume.return_value = None 213 self.driver._kaminario_disconnect_volume = mock.Mock() 214 result = self.driver.create_volume_from_snapshot(self.vol, self.snap) 215 self.assertIsNone(result) 216 217 @mock.patch.object(utils, 'brick_get_connector_properties') 218 @mock.patch.object(vol_utils, 'copy_volume') 219 def test_create_volume_from_snapshot_with_exception(self, mock_copy_volume, 220 mock_brick_get): 221 """Test create_volume_from_snapshot_with_exception.""" 222 mock_brick_get.return_value = CONNECTOR 223 mock_copy_volume.return_value = None 224 self.driver.client = FakeKrestException() 225 self.assertRaises(exception.KaminarioCinderDriverException, 226 self.driver.create_volume_from_snapshot, self.vol, 227 self.snap) 228 229 @mock.patch.object(utils, 'brick_get_connector_properties') 230 @mock.patch.object(vol_utils, 'copy_volume') 231 def test_create_cloned_volume(self, mock_copy_volume, mock_brick_get): 232 """Test create_cloned_volume.""" 233 mock_brick_get.return_value = CONNECTOR 234 mock_copy_volume.return_value = None 235 self.driver._kaminario_disconnect_volume = mock.Mock() 236 result = self.driver.create_cloned_volume(self.vol, self.vol) 237 self.assertIsNone(result) 238 239 @mock.patch.object(utils, 'brick_get_connector_properties') 240 @mock.patch.object(vol_utils, 'copy_volume') 241 def test_create_cloned_volume_with_exception(self, mock_copy_volume, 242 mock_brick_get): 243 """Test create_cloned_volume_with_exception.""" 244 mock_brick_get.return_value = CONNECTOR 245 mock_copy_volume.return_value = None 246 self.driver.terminate_connection = mock.Mock() 247 self.driver.client = FakeKrestException() 248 self.assertRaises(exception.KaminarioCinderDriverException, 249 self.driver.create_cloned_volume, self.vol, self.vol) 250 251 def test_extend_volume(self): 252 """Test extend_volume.""" 253 new_size = 256 254 result = self.driver.extend_volume(self.vol, new_size) 255 self.assertIsNone(result) 256 257 def test_extend_volume_with_exception(self): 258 """Test extend_volume_with_exception.""" 259 self.driver.client = FakeKrestException() 260 new_size = 256 261 self.assertRaises(exception.KaminarioCinderDriverException, 262 self.driver.extend_volume, self.vol, new_size) 263 264 def test_initialize_connection_with_exception(self): 265 """Test initialize_connection_with_exception.""" 266 self.driver.client = FakeKrestException() 267 self.assertRaises(exception.KaminarioCinderDriverException, 268 self.driver.initialize_connection, self.vol, 269 CONNECTOR) 270 271 def test_get_lun_number(self): 272 """Test _get_lun_number.""" 273 host, host_rs, host_name = self.driver._get_host_object(CONNECTOR) 274 result = self.driver._get_lun_number(self.vol, host) 275 self.assertEqual(548, result) 276 277 def test_get_volume_object(self): 278 """Test _get_volume_object.""" 279 result = self.driver._get_volume_object(self.vol) 280 self.assertEqual(548, result.id) 281 282 def test_get_host_object(self): 283 """Test _get_host_object.""" 284 host, host_rs, host_name = self.driver._get_host_object(CONNECTOR) 285 self.assertEqual(548, host.id) 286 self.assertEqual(2, host_rs.total) 287 self.assertEqual('test-k2', host_name) 288 289 def test_k2_initialize_connection(self): 290 """Test k2_initialize_connection.""" 291 result = self.driver.k2_initialize_connection(self.vol, CONNECTOR) 292 self.assertEqual(548, result) 293 294 @mock.patch.object(FakeSearchObject, 'total', 1) 295 def test_manage_existing(self): 296 """Test manage_existing.""" 297 self.driver._get_replica_status = mock.Mock(return_value=False) 298 result = self.driver.manage_existing(self.vol, {'source-name': 'test'}) 299 self.assertIsNone(result) 300 301 def test_manage_existing_exp(self): 302 self.driver._get_replica_status = mock.Mock(return_value=True) 303 self.assertRaises(exception.ManageExistingInvalidReference, 304 self.driver.manage_existing, self.vol, 305 {'source-name': 'test'}) 306 307 def test_manage_vg_volumes(self): 308 self.driver.nvol = 2 309 self.assertRaises(exception.ManageExistingInvalidReference, 310 self.driver.manage_existing, self.vol, 311 {'source-name': 'test'}) 312 313 def test_manage_existing_get_size(self): 314 """Test manage_existing_get_size.""" 315 self.driver.client.search().hits[0].size = units.Mi 316 result = self.driver.manage_existing_get_size(self.vol, 317 {'source-name': 'test'}) 318 self.assertEqual(1, result) 319 320 def test_get_is_dedup(self): 321 """Test _get_is_dedup.""" 322 result = self.driver._get_is_dedup(self.vol.volume_type) 323 self.assertTrue(result) 324 325 def test_get_is_dedup_false(self): 326 """Test _get_is_dedup_false.""" 327 specs = {'kaminario:thin_prov_type': 'nodedup'} 328 self.vol.volume_type.extra_specs = specs 329 result = self.driver._get_is_dedup(self.vol.volume_type) 330 self.assertFalse(result) 331 332 def test_get_replica_status(self): 333 """Test _get_replica_status.""" 334 result = self.driver._get_replica_status(self.vol) 335 self.assertTrue(result) 336 337 def test_create_volume_replica(self): 338 """Test _create_volume_replica.""" 339 vg = FakeSaveObject() 340 rep = Replication() 341 self.driver.replica = rep 342 session_name = self.driver.get_session_name('1234567890987654321') 343 self.assertEqual('ssn-1234567890987654321', session_name) 344 rsession_name = self.driver.get_rep_name(session_name) 345 self.assertEqual('rssn-1234567890987654321', rsession_name) 346 src_ssn = self.driver.client.new("replication/sessions").save() 347 self.assertEqual('in_sync', src_ssn.state) 348 result = self.driver._create_volume_replica(self.vol, vg, vg, rep.rpo) 349 self.assertIsNone(result) 350 351 def test_create_volume_replica_exp(self): 352 """Test _create_volume_replica_exp.""" 353 vg = FakeSaveObject() 354 rep = Replication() 355 self.driver.replica = rep 356 self.driver.client = FakeKrestException() 357 self.assertRaises(exception.KaminarioCinderDriverException, 358 self.driver._create_volume_replica, self.vol, 359 vg, vg, rep.rpo) 360 361 def test_delete_by_ref(self): 362 """Test _delete_by_ref.""" 363 result = self.driver._delete_by_ref(self.driver.client, 'volume', 364 'name', 'message') 365 self.assertIsNone(result) 366 367 def test_failover_volume(self): 368 """Test _failover_volume.""" 369 self.driver.target = FakeKrest() 370 session_name = self.driver.get_session_name('1234567890987654321') 371 self.assertEqual('ssn-1234567890987654321', session_name) 372 rsession_name = self.driver.get_rep_name(session_name) 373 self.assertEqual('rssn-1234567890987654321', rsession_name) 374 result = self.driver._failover_volume(self.vol) 375 self.assertIsNone(result) 376 377 @mock.patch.object(kaminario_common.KaminarioCinderDriver, 378 '_check_for_status') 379 @mock.patch.object(objects.service.Service, 'get_by_args') 380 def test_failover_host(self, get_by_args, check_stauts): 381 """Test failover_host.""" 382 mock_args = mock.Mock() 383 mock_args.active_backend_id = '10.0.0.1' 384 self.vol.replication_status = 'failed-over' 385 self.driver.configuration.san_ip = '10.0.0.1' 386 get_by_args.side_effect = [mock_args, mock_args] 387 self.driver.host = 'host' 388 volumes = [self.vol, self.vol] 389 self.driver.replica = Replication() 390 self.driver.target = FakeKrest() 391 self.driver.target.search().total = 1 392 self.driver.client.search().total = 1 393 backend_ip, res_volumes, __ = self.driver.failover_host( 394 None, volumes, []) 395 self.assertEqual('10.0.0.1', backend_ip) 396 status = res_volumes[0]['updates']['replication_status'] 397 self.assertEqual(fields.ReplicationStatus.FAILED_OVER, status) 398 # different backend ip 399 self.driver.configuration.san_ip = '10.0.0.2' 400 self.driver.client.search().hits[0].state = 'in_sync' 401 backend_ip, res_volumes, __ = self.driver.failover_host( 402 None, volumes, []) 403 self.assertEqual('10.0.0.2', backend_ip) 404 status = res_volumes[0]['updates']['replication_status'] 405 self.assertEqual(fields.ReplicationStatus.DISABLED, status) 406 407 def test_delete_volume_replica(self): 408 """Test _delete_volume_replica.""" 409 self.driver.replica = Replication() 410 self.driver.target = FakeKrest() 411 session_name = self.driver.get_session_name('1234567890987654321') 412 self.assertEqual('ssn-1234567890987654321', session_name) 413 rsession_name = self.driver.get_rep_name(session_name) 414 self.assertEqual('rssn-1234567890987654321', rsession_name) 415 res = self.driver._delete_by_ref(self.driver.client, 'volumes', 416 'test', 'test') 417 self.assertIsNone(res) 418 result = self.driver._delete_volume_replica(self.vol, 'test', 'test') 419 self.assertIsNone(result) 420 src_ssn = self.driver.client.search("replication/sessions").hits[0] 421 self.assertEqual('idle', src_ssn.state) 422 423 def test_delete_volume_replica_exp(self): 424 """Test _delete_volume_replica_exp.""" 425 self.driver.replica = Replication() 426 self.driver.target = FakeKrestException() 427 self.driver._check_for_status = mock.Mock() 428 self.assertRaises(exception.KaminarioCinderDriverException, 429 self.driver._delete_volume_replica, self.vol, 430 'test', 'test') 431 432 def test_get_is_replica(self): 433 """Test get_is_replica.""" 434 result = self.driver._get_is_replica(self.vol.volume_type) 435 self.assertFalse(result) 436 437 def test_get_is_replica_true(self): 438 """Test get_is_replica_true.""" 439 self.driver.replica = Replication() 440 self.vol.volume_type.extra_specs = {'kaminario:replication': 'enabled'} 441 result = self.driver._get_is_replica(self.vol.volume_type) 442 self.assertTrue(result) 443 444 def test_after_volume_copy(self): 445 """Test after_volume_copy.""" 446 result = self.driver.after_volume_copy(None, self.vol, 447 self.vol.volume_type) 448 self.assertIsNone(result) 449 450 def test_retype(self): 451 """Test retype.""" 452 replica_status = self.driver._get_replica_status('test') 453 self.assertTrue(replica_status) 454 replica = self.driver._get_is_replica(self.vol.volume_type) 455 self.assertFalse(replica) 456 self.driver.replica = Replication() 457 result = self.driver._add_replication(self.vol) 458 self.assertIsNone(result) 459 self.driver.target = FakeKrest() 460 self.driver._check_for_status = mock.Mock() 461 result = self.driver._delete_replication(self.vol) 462 self.assertIsNone(result) 463 self.driver._delete_volume_replica = mock.Mock() 464 result = self.driver.retype(None, self.vol, 465 self.vol.volume_type, None, None) 466 self.assertTrue(result) 467 new_vol_type = fake_volume.fake_volume_type_obj(self.context) 468 new_vol_type.extra_specs = {'kaminario:thin_prov_type': 'nodedup'} 469 result2 = self.driver.retype(None, self.vol, 470 new_vol_type, None, None) 471 self.assertFalse(result2) 472 473 def test_add_replication(self): 474 """"Test _add_replication.""" 475 self.driver.replica = Replication() 476 result = self.driver._add_replication(self.vol) 477 self.assertIsNone(result) 478 479 def test_delete_replication(self): 480 """Test _delete_replication.""" 481 self.driver.replica = Replication() 482 self.driver.target = FakeKrest() 483 self.driver._check_for_status = mock.Mock() 484 result = self.driver._delete_replication(self.vol) 485 self.assertIsNone(result) 486 487 def test_create_failover_volume_replica(self): 488 """Test _create_failover_volume_replica.""" 489 self.driver.replica = Replication() 490 self.driver.target = FakeKrest() 491 self.driver.configuration.san_ip = '10.0.0.1' 492 result = self.driver._create_failover_volume_replica(self.vol, 493 'test', 'test') 494 self.assertIsNone(result) 495 496 def test_create_volume_replica_user_snap(self): 497 """Test create_volume_replica_user_snap.""" 498 result = self.driver._create_volume_replica_user_snap(FakeKrest(), 499 'sess') 500 self.assertEqual(548, result) 501 502 def test_is_user_snap_sync_finished(self): 503 """Test _is_user_snap_sync_finished.""" 504 sess_mock = mock.Mock() 505 sess_mock.refresh = mock.Mock() 506 sess_mock.generation_number = 548 507 sess_mock.current_snapshot_id = None 508 sess_mock.current_snapshot_progress = 100 509 sess_mock.current_snapshot_id = None 510 self.driver.snap_updates = [{'tgt_ssn': sess_mock, 'gno': 548, 511 'stime': time.time()}] 512 result = self.driver._is_user_snap_sync_finished() 513 self.assertIsNone(result) 514 515 def test_delete_failover_volume_replica(self): 516 """Test _delete_failover_volume_replica.""" 517 self.driver.target = FakeKrest() 518 result = self.driver._delete_failover_volume_replica(self.vol, 'test', 519 'test') 520 self.assertIsNone(result) 521 522 def test_get_initiator_host_name(self): 523 result = self.driver.get_initiator_host_name(CONNECTOR) 524 self.assertEqual(CONNECTOR['host'], result) 525 526 def test_get_initiator_host_name_unique(self): 527 self.driver.configuration.unique_fqdn_network = False 528 result = self.driver.get_initiator_host_name(CONNECTOR) 529 expected = re.sub('[:.]', '_', CONNECTOR['initiator'][::-1][:32]) 530 self.assertEqual(expected, result) 531 532 533@ddt.ddt 534class TestKaminarioISCSI(TestKaminarioCommon): 535 def test_get_target_info(self): 536 """Test get_target_info.""" 537 iscsi_portals, target_iqns = self.driver.get_target_info(self.vol) 538 self.assertEqual(['10.0.0.1:3260', '10.0.0.2:3260'], 539 iscsi_portals) 540 self.assertEqual(['xyztlnxyz', 'xyztlnxyz'], 541 target_iqns) 542 543 @ddt.data(True, False) 544 def test_initialize_connection(self, multipath): 545 """Test initialize_connection.""" 546 connector = CONNECTOR.copy() 547 connector['multipath'] = multipath 548 self.driver.configuration.disable_discovery = False 549 550 conn_info = self.driver.initialize_connection(self.vol, CONNECTOR) 551 expected = { 552 'data': { 553 'target_discovered': True, 554 'target_iqn': 'xyztlnxyz', 555 'target_lun': 548, 556 'target_portal': '10.0.0.1:3260', 557 }, 558 'driver_volume_type': 'iscsi', 559 } 560 self.assertEqual(expected, conn_info) 561 562 def test_initialize_connection_multipath(self): 563 """Test initialize_connection with multipath.""" 564 connector = CONNECTOR.copy() 565 connector['multipath'] = True 566 self.driver.configuration.disable_discovery = True 567 568 conn_info = self.driver.initialize_connection(self.vol, connector) 569 570 expected = { 571 'data': { 572 'target_discovered': True, 573 'target_iqn': 'xyztlnxyz', 574 'target_iqns': ['xyztlnxyz', 'xyztlnxyz'], 575 'target_lun': 548, 576 'target_luns': [548, 548], 577 'target_portal': '10.0.0.1:3260', 578 'target_portals': ['10.0.0.1:3260', '10.0.0.2:3260'], 579 }, 580 'driver_volume_type': 'iscsi', 581 } 582 self.assertEqual(expected, conn_info) 583 584 def test_terminate_connection(self): 585 """Test terminate_connection.""" 586 result = self.driver.terminate_connection(self.vol, CONNECTOR) 587 self.assertIsNone(result) 588 589 def test_terminate_connection_without_connector(self): 590 """Test terminate_connection_without_connector.""" 591 result = self.driver.terminate_connection(self.vol, None) 592 self.assertIsNone(result) 593 594 595class TestKaminarioFC(TestKaminarioCommon): 596 597 def _setup_driver(self): 598 self.driver = (kaminario_fc. 599 KaminarioFCDriver(configuration=self.conf)) 600 device = mock.Mock(return_value={'device': {'path': '/dev'}}) 601 self.driver._connect_device = device 602 self.driver.client = FakeKrest() 603 self.driver._lookup_service = mock.Mock() 604 605 def test_initialize_connection(self): 606 """Test initialize_connection.""" 607 conn_info = self.driver.initialize_connection(self.vol, CONNECTOR) 608 self.assertIn('data', conn_info) 609 self.assertIn('target_wwn', conn_info['data']) 610 611 def test_get_target_info(self): 612 """Test get_target_info.""" 613 target_wwpn = self.driver.get_target_info(self.vol) 614 self.assertEqual(['50024f4053300301', '50024f4053300302'], 615 target_wwpn) 616 617 def test_terminate_connection(self): 618 """Test terminate_connection.""" 619 result = self.driver.terminate_connection(self.vol, CONNECTOR) 620 self.assertIn('data', result) 621 622 def test_terminate_connection_without_connector(self): 623 """Test terminate_connection_without_connector.""" 624 result = self.driver.terminate_connection(self.vol, None) 625 self.assertIn('data', result) 626 627 def test_get_initiator_host_name_unique(self): 628 connector = CONNECTOR.copy() 629 del connector['initiator'] 630 self.driver.configuration.unique_fqdn_network = False 631 result = self.driver.get_initiator_host_name(connector) 632 expected = re.sub('[:.]', '_', connector['wwnns'][0][::-1][:32]) 633 self.assertEqual(expected, result) 634